"""Idempotent seed script. docker compose exec api python -m app.seed Always ensures the default departments/deviation categories and the notifications setting. When SEED_DEMO_DATA=true it also creates dev users (one per role — usable directly with AUTH_MODE=dev) and a spread of sample NCRs across every workflow stage for development and demos. """ import asyncio import logging import random from datetime import timedelta from decimal import Decimal from sqlalchemy import select from app.config import get_settings from app.database import get_session_factory from app.domain import Role, Stage from app.models import ( AppSetting, Department, DeviationCategory, Ncr, NcrLink, NcrSecondaryAssignee, RootCauseCategory, StageTransition, User, UserRole, ) from app.models.app_setting import NOTIFICATIONS_ENABLED_KEY from app.models.base import utcnow from app.services.numbering import allocate_ncr_number logging.basicConfig(level=logging.INFO, format="%(message)s") log = logging.getLogger("seed") DEPARTMENTS = [ "Machining", "Welding", "Fabrication", "Assembly", "Paint & Coating", "Shipping / Receiving", "Engineering", "Quality", ] CATEGORIES = [ "Dimensional", "Material Defect", "Weld Defect", "Documentation", "Process Deviation", "Supplier Nonconformance", "Damage / Handling", "Other", ] # Standard 6M root-cause categories (API Q1 §6.4.2 trend reporting). ROOT_CAUSE_CATEGORIES = [ "Man", "Machine", "Method", "Material", "Measurement", "Environment", ] ROOT_CAUSES = [ "Operator used a superseded revision of the work instruction.", "Fixture clamping force drifted out of spec; no PM interval defined.", "Incoming material certified to the wrong specification revision.", "Measurement performed with a gauge past its calibration due date.", "Setup sheet did not call out the datum change from the ECN.", ] DEV_USERS = [ ("admin@pescoinc.biz", "Dev Admin", list(r.value for r in Role)), ("dispo@pescoinc.biz", "Dana Disposition", [Role.REQUESTER.value, Role.DISPOSITION_AUTHORITY.value]), ("second@pescoinc.biz", "Sam Secondary", [Role.REQUESTER.value, Role.SECONDARY_DISPOSITION_AUTHORITY.value]), ("ops@pescoinc.biz", "Owen Operations", [Role.REQUESTER.value, Role.OPERATIONS.value]), ("qc@pescoinc.biz", "Quinn Inspector", [Role.REQUESTER.value, Role.QC_INSPECTOR.value]), ("cost@pescoinc.biz", "Casey Costing", [Role.REQUESTER.value, Role.COSTING.value]), ("req@pescoinc.biz", "Riley Requester", [Role.REQUESTER.value]), ] DETAILS = [ "Bore diameter measured 0.008\" over drawing tolerance on 3 of 12 pieces.", "Weld porosity found on the underside seam during visual inspection.", "Wrong material grade pulled from stock; heat number does not match the traveler.", "Paint runs and inadequate coverage on exterior panels after first coat.", "Fixture shifted during machining; datum surfaces out of parallel by 0.015\".", "Supplier-provided casting shows shrinkage cavity at the flange face.", "Part dropped during transfer between stations; visible dent on sealing surface.", "Traveler missing signed inspection step for operation 40.", ] async def seed() -> None: settings = get_settings() session_factory = get_session_factory() async with session_factory() as db: # ── lookups ────────────────────────────────────────────────────────── existing = { d.name for d in (await db.execute(select(Department))).scalars().all() } for name in DEPARTMENTS: if name not in existing: db.add(Department(name=name, is_active=True)) existing = { c.name for c in (await db.execute(select(DeviationCategory))).scalars().all() } for name in CATEGORIES: if name not in existing: db.add(DeviationCategory(name=name, is_active=True)) existing = { c.name for c in (await db.execute(select(RootCauseCategory))).scalars().all() } for name in ROOT_CAUSE_CATEGORIES: if name not in existing: db.add(RootCauseCategory(name=name, is_active=True)) if await db.get(AppSetting, NOTIFICATIONS_ENABLED_KEY) is None: db.add( AppSetting( key=NOTIFICATIONS_ENABLED_KEY, value="true" if settings.notifications_enabled_default else "false", ) ) await db.commit() log.info("Lookups + settings seeded.") if not settings.seed_demo_data: log.info("SEED_DEMO_DATA is false — skipping demo users/NCRs. Done.") return # ── dev users ──────────────────────────────────────────────────────── users: dict[str, User] = {} for email, name, roles in DEV_USERS: user = ( await db.execute(select(User).where(User.email == email)) ).scalar_one_or_none() if user is None: user = User(email=email, display_name=name, is_active=True) db.add(user) await db.flush() for role in roles: db.add(UserRole(user_id=user.id, role=role)) users[email] = user await db.commit() log.info("Dev users seeded: %s", ", ".join(u for u, _, _ in DEV_USERS)) # ── demo NCRs ──────────────────────────────────────────────────────── ncr_count = (await db.execute(select(Ncr.id).limit(1))).first() if ncr_count is not None: log.info("NCRs already exist — skipping demo NCR creation. Done.") return departments = (await db.execute(select(Department))).scalars().all() categories = (await db.execute(select(DeviationCategory))).scalars().all() root_causes = (await db.execute(select(RootCauseCategory))).scalars().all() rng = random.Random(42) seeded_ncrs: list[Ncr] = [] dispo = users["dispo@pescoinc.biz"] second = users["second@pescoinc.biz"] ops = users["ops@pescoinc.biz"] qc = users["qc@pescoinc.biz"] cost = users["cost@pescoinc.biz"] req = users["req@pescoinc.biz"] # (target_stage, count) plan = [ (Stage.NEW_REQUEST, 3), (Stage.SECONDARY_DISPOSITION, 2), (Stage.OPERATIONS, 3), (Stage.QC_INSPECTION, 2), (Stage.COSTING, 2), (Stage.CLOSED, 4), ] for target, count in plan: for i in range(count): days_ago = rng.randint(5, 120) created = utcnow() - timedelta(days=days_ago) number, year, seq = await allocate_ncr_number(db, now=created) use_secondary = rng.random() < 0.4 or target == Stage.SECONDARY_DISPOSITION ncr = Ncr( ncr_number=number, ncr_year=year, ncr_seq=seq, job_number=f"J{rng.randint(10000, 49999)}", department_id=rng.choice(departments).id, deviation_category_id=rng.choice(categories).id, disposition_authority_id=dispo.id, deviation_detail=rng.choice(DETAILS), requester_id=req.id, stage=Stage.NEW_REQUEST.value, created_at=created, stage_entered_at=created, updated_at=created, ) db.add(ncr) await db.flush() seeded_ncrs.append(ncr) t = created db.add(StageTransition( ncr_id=ncr.id, from_stage=None, to_stage=Stage.NEW_REQUEST.value, action="create", acted_by_id=req.id, acted_at=t, )) def hop(days_lo=1, days_hi=4): nonlocal t t = min(utcnow(), t + timedelta(days=rng.randint(days_lo, days_hi), hours=rng.randint(0, 8))) return t def advance(to_stage: Stage, action: str, actor: User, note=None): db.add(StageTransition( ncr_id=ncr.id, from_stage=ncr.stage, to_stage=to_stage.value, action=action, acted_by_id=actor.id, acted_at=hop(), note=note, )) ncr.stage = to_stage.value ncr.stage_entered_at = t if target == Stage.NEW_REQUEST: continue # initial disposition ncr.qc_authority = "AS9100 8.7" ncr.work_order = f"WO-{rng.randint(1000, 9999)}" ncr.disposition_notes = ( "

Disposition: Rework per attached instructions. " "Re-inspect all affected features.

" ) ncr.secondary_review_needed = use_secondary if use_secondary: db.add(NcrSecondaryAssignee(ncr_id=ncr.id, user_id=second.id)) advance(Stage.SECONDARY_DISPOSITION, "initial_disposition", dispo) if target == Stage.SECONDARY_DISPOSITION: continue advance(Stage.OPERATIONS, "secondary_release", second) else: advance(Stage.OPERATIONS, "initial_disposition", dispo) if target == Stage.OPERATIONS: continue ncr.operations_complete = True ncr.operations_completed_by_id = ops.id advance(Stage.QC_INSPECTION, "operations_complete", ops) ncr.operations_completed_at = t if target == Stage.QC_INSPECTION: continue ncr.qc_approval = "yes" ncr.inspection_notes = "Reworked features re-inspected; all within tolerance." ncr.qc_closed = True ncr.qc_closed_by_id = qc.id advance(Stage.COSTING, "qc_close", qc) ncr.qc_closed_at = t # CAPA — the API Q1 closure gate requires the CA question to be # answered (and any required action verified) before costing # can close the NCR, so seed it here. ncr.root_cause = rng.choice(ROOT_CAUSES) ncr.root_cause_category_id = rng.choice(root_causes).id # Alternate yes/no so every dashboard metric has data. if i % 2 == 0 or target == Stage.COSTING: ncr.corrective_action_required = True ncr.corrective_action_justification = ( "Recurrence risk without a process control update." ) ncr.corrective_action_plan = ( "Update the work instruction and add an in-process check " "at the affected operation." ) ncr.corrective_action_owner_id = ops.id ncr.corrective_action_opened_at = t ncr.corrective_action_due_date = (t + timedelta(days=14)).date() if target == Stage.CLOSED: ncr.effectiveness_result = "effective" ncr.effectiveness_notes = ( "No recurrence observed over three subsequent runs." ) ncr.effectiveness_verified_by_id = qc.id ncr.effectiveness_verified_at = hop() else: ncr.corrective_action_required = False ncr.corrective_action_justification = ( "Isolated incident fully contained by the disposition." ) if target == Stage.COSTING: continue ncr.labor_cost = Decimal(rng.randint(80, 2400)) ncr.material_cost = Decimal(rng.randint(0, 1800)) ncr.service_cost = Decimal(rng.choice([0, 0, 150, 450, 900])) ncr.other_cost = Decimal(rng.choice([0, 0, 0, 75, 200])) ncr.costing_completed_by_id = cost.id ncr.closed_by_id = cost.id advance(Stage.CLOSED, "complete_costing", cost) ncr.costing_completed_at = t ncr.closed_at = t # Recurring-issue demo: the newest open NCR references two closed ones. closed = [n for n in seeded_ncrs if n.stage == Stage.CLOSED.value] still_open = [n for n in seeded_ncrs if n.stage != Stage.CLOSED.value] if closed and still_open: repeat = still_open[0] repeat.is_recurring = True for prior in closed[:2]: db.add(NcrLink(ncr_id=repeat.id, related_ncr_id=prior.id)) await db.commit() log.info("Demo NCRs seeded. Done.") if __name__ == "__main__": asyncio.run(seed())