diff --git a/apps/core/management/__init__.py b/apps/core/management/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/apps/core/management/commands/__init__.py b/apps/core/management/commands/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/apps/core/management/commands/seed_walkthrough.py b/apps/core/management/commands/seed_walkthrough.py new file mode 100644 index 0000000..839b2fd --- /dev/null +++ b/apps/core/management/commands/seed_walkthrough.py @@ -0,0 +1,635 @@ +"""Seed one coherent, fully fictional batch chain for IXG ops training. + +Everything created here is synthetic: fictional companies, collector groups, +field officers, laboratories and vessels. No real buyer, supplier, collector or +producer data is used anywhere. + +The chain runs end to end so that the same batch reference appears on every +screen an ops trainer walks through: + + origin zone -> collector group -> harvest intakes -> lineage + -> processing batch -> QC result -> four-eyes verification + -> certificate of analysis (+ lab report, packing list, B/L, phytosanitary) + -> EUDR due-diligence case (evidence, risk assessment, DDS reference) + -> shipment -> ESG field report -> buyer subscription + +Three batches are prepared on purpose: + + IXG-BATCH-2026-014 completed happy path (delivered) - the buyer's view + IXG-BATCH-2026-005 mid-flight (QC passed, CoA published, in transit) + IXG-TEST-001 QC failure + CAPA - the exception branch + +Usage (inside the app container): + + python manage.py seed_walkthrough + python manage.py seed_walkthrough --reset # remove seeded rows first +""" + +from __future__ import annotations + +import hashlib +from datetime import date, datetime, timedelta +from decimal import Decimal +from pathlib import Path + +from django.conf import settings +from django.core.management.base import BaseCommand +from django.utils import timezone +from reportlab.lib import colors +from reportlab.lib.pagesizes import A4 +from reportlab.lib.units import mm +from reportlab.pdfgen import canvas + +from apps.batches.models import ProcessingBatch +from apps.capa.models import CAPARecord +from apps.compliance.models import ( + CollectorGroup, + DueDiligenceCase, + DueDiligenceStatementReference, + EUDREvidence, + Facility, + HarvestIntake, + IntakeBatchLineage, + MitigationAction, + OriginZone, + ProductSpecification, + RiskAssessment, +) +from apps.documents.models import CertificateOfAnalysis, Document, RegulatoryCert +from apps.esg.models import ESGFieldReport +from apps.qc.models import QCResult +from apps.shipments.models import Shipment +from apps.subscriptions.models import Subscription +from apps.users.models import User + +BUYER_EMAIL = "buyer@greenfieldfoods.example" +BUYER_NAME = "Greenfield Foods Ltd (dummy buyer)" +LAB_NAME = "Northgate Analytical Services Ltd (dummy laboratory)" +FORWARDER = "Meridian Freight Forwarding Ltd (dummy forwarder)" + +COMPLETED_BATCH = "IXG-BATCH-2026-014" +INFLIGHT_BATCH = "IXG-BATCH-2026-005" +FAILED_BATCH = "IXG-TEST-001" + +DOCS_ROOT = Path(settings.MEDIA_ROOT) / "ixg-documents" + + +def _pdf(rel_dir: str, filename: str, heading: str, subheading: str, + lines: list[str], table: list[list[str]] | None = None) -> dict: + """Render a small, realistic-enough document and return its file metadata.""" + folder = DOCS_ROOT / rel_dir + folder.mkdir(parents=True, exist_ok=True) + path = folder / filename + + c = canvas.Canvas(str(path), pagesize=A4) + width, height = A4 + c.setFillColor(colors.HexColor("#1A3D2B")) + c.rect(0, height - 28 * mm, width, 28 * mm, stroke=0, fill=1) + c.setFillColor(colors.white) + c.setFont("Helvetica-Bold", 16) + c.drawString(18 * mm, height - 15 * mm, "IXG LTD") + c.setFont("Helvetica", 9) + c.drawString(18 * mm, height - 21 * mm, "Unrefined shea butter - Nigeria to United Kingdom") + c.setFillColor(colors.HexColor("#1A1A1A")) + c.setFont("Helvetica-Bold", 13) + c.drawString(18 * mm, height - 40 * mm, heading) + c.setFont("Helvetica-Oblique", 9.5) + c.drawString(18 * mm, height - 46 * mm, subheading) + + y = height - 58 * mm + c.setFont("Helvetica", 10) + for line in lines: + c.drawString(18 * mm, y, line) + y -= 5.5 * mm + + if table: + y -= 3 * mm + c.setFont("Helvetica-Bold", 9.5) + c.drawString(18 * mm, y, table[0][0]) + c.drawString(90 * mm, y, table[0][1]) + c.drawString(140 * mm, y, table[0][2]) + y -= 2 * mm + c.setStrokeColor(colors.HexColor("#CCD3CF")) + c.line(18 * mm, y, 175 * mm, y) + c.setFont("Helvetica", 9.5) + for row in table[1:]: + y -= 5.5 * mm + c.drawString(18 * mm, y, row[0]) + c.drawString(90 * mm, y, row[1]) + c.drawString(140 * mm, y, row[2]) + + c.setFont("Helvetica-Oblique", 8) + c.setFillColor(colors.HexColor("#5B625F")) + c.drawString(18 * mm, 22 * mm, "SYNTHETIC DOCUMENT - generated for IXG platform training. Not a real certificate.") + c.showPage() + c.save() + + data = path.read_bytes() + return { + "file_url": f"{settings.MEDIA_URL}ixg-documents/{rel_dir}/{filename}", + "file_name": filename, + "sha256": hashlib.sha256(data).hexdigest(), + "size": len(data), + } + + +class Command(BaseCommand): + help = "Seed the fictional end-to-end batch chain used for ops training walkthroughs." + + def add_arguments(self, parser): + parser.add_argument( + "--reset", + action="store_true", + help="Delete the seeded records for the three demo batches before re-creating them.", + ) + + # ------------------------------------------------------------------ helpers + + def _user(self, email: str) -> User | None: + return User.objects.filter(email__iexact=email).first() + + def _ops_users(self) -> tuple[User | None, User | None]: + """Two different ops accounts, so the four-eyes step is genuinely two pairs of eyes.""" + enterer = self._user("test@ixg.local") or User.objects.filter(role="ops").order_by("email").first() + verifier = ( + User.objects.filter(role="ops") + .exclude(pk=getattr(enterer, "pk", None)) + .order_by("email") + .first() + ) or enterer + return enterer, verifier + + def _spec(self) -> ProductSpecification | None: + return ( + ProductSpecification.objects.filter(product__icontains="shea", is_current=True).first() + or ProductSpecification.current("Unrefined Shea Butter") + ) + + def _reset(self) -> None: + for ref in (COMPLETED_BATCH, INFLIGHT_BATCH): + ProcessingBatch.objects.filter(batch_reference=ref).delete() + self.stdout.write(" reset: removed seeded batches") + + # --------------------------------------------------------------------- steps + + def _origin_chain(self, verifier: User) -> tuple[OriginZone, list[HarvestIntake]]: + facility, _ = Facility.objects.update_or_create( + name="IXG Processing Facility - Katsina", + defaults={ + "country": "Nigeria", + "region": "Katsina State", + "address": "Industrial Layout, Katsina State, Nigeria (dummy address)", + "facility_type": Facility.FacilityType.PROCESSING, + "is_active": True, + }, + ) + + zone, _ = OriginZone.objects.update_or_create( + zone_code="NG-KT-Z14", + defaults={ + "facility": facility, + "country": "Nigeria", + "region": "Katsina State", + "geometry": { + "type": "Polygon", + "coordinates": [[[7.6012, 12.9905], [7.6180, 12.9905], [7.6180, 13.0042], [7.6012, 13.0042], [7.6012, 12.9905]]], + }, + "coordinate_source": "IFO handheld GPS capture (synthetic coordinates)", + "coordinates_collected_by": "IFO - A. Bello (dummy officer)", + "coordinates_collected_at": date(2026, 8, 18), + "evidence_file": "origin-zone-NG-KT-Z14-geolocation.pdf", + "verification_status": OriginZone.VerificationStatus.VERIFIED, + "verified_by": verifier, + "verified_at": timezone.now() - timedelta(days=20), + }, + ) + + group_a, _ = CollectorGroup.objects.update_or_create( + name="Kaduna Road Women's Shea Collective (dummy)", + defaults={ + "contact_reference": "CG-014", + "origin_zone": zone, + "notes": "Synthetic collector group used for training. 42 registered pickers.", + }, + ) + group_b, _ = CollectorGroup.objects.update_or_create( + name="Dutsin-Ma Shea Pickers Association (dummy)", + defaults={ + "contact_reference": "CG-015", + "origin_zone": zone, + "notes": "Synthetic collector group used for training. 28 registered pickers.", + }, + ) + + intakes = [] + intake_specs = [ + ("IXG-INTAKE-2026-0141", group_a, date(2026, 8, 25), Decimal("820.00"), "A"), + ("IXG-INTAKE-2026-0142", group_a, date(2026, 8, 27), Decimal("640.00"), "A"), + ("IXG-INTAKE-2026-0143", group_b, date(2026, 8, 29), Decimal("600.00"), "B"), + ] + for ref, group, when, weight, grade in intake_specs: + intake, _ = HarvestIntake.objects.update_or_create( + intake_reference=ref, + defaults={ + "collector_group": group, + "origin_zone": zone, + "intake_date": when, + "weight_kg": weight, + "grade": grade, + "receiving_officer": "IFO - A. Bello (dummy officer)", + "status": HarvestIntake.Status.VERIFIED, + "evidence_file": f"{ref}-weighbridge-ticket.pdf", + }, + ) + intakes.append(intake) + return zone, intakes + + def _batch(self, ref: str, intakes: list[HarvestIntake], enterer: User, **fields) -> ProcessingBatch: + batch, created = ProcessingBatch.objects.update_or_create( + batch_reference=ref, + defaults={"origin_facility": "ASI - Katsina State, Nigeria", "created_by": enterer, **fields}, + ) + allocations = [Decimal("800.00"), Decimal("640.00"), Decimal("560.00")] + for intake, weight in zip(intakes, allocations): + IntakeBatchLineage.objects.update_or_create( + batch=batch, + harvest_intake=intake, + defaults={"weight_allocated_kg": weight}, + ) + return batch + + def _qc(self, batch: ProcessingBatch, enterer: User, verifier: User, spec: ProductSpecification | None) -> QCResult: + qc = batch.qc_results.order_by("-tested_at").first() + if qc is None: + qc = QCResult(batch=batch) + qc.ffa_percentage = Decimal("0.42") + qc.moisture_percentage = Decimal("0.08") + qc.peroxide_value = Decimal("1.20") + qc.visual_grade = QCResult.VisualGrade.A + qc.overall_pass = True + qc.tested_by = LAB_NAME + qc.lab_reference = "NAS-2026-0914" + qc.spec_version = spec.version if spec else "1.0" + qc.entered_by = enterer + qc.verified_by = verifier + qc.verification_status = QCResult.VerificationStatus.APPROVED + qc.approved_at = timezone.now() - timedelta(days=10) + qc.method_ffa = "ISO 660" + qc.method_moisture = "ISO 662" + qc.method_peroxide = "ISO 3960" + qc.result_status = QCResult.ResultStatus.WITHIN_SPEC + qc.is_quarantine = False + qc.save() + return qc + + def _documents(self, batch: ProcessingBatch, qc: QCResult, enterer: User, + verifier: User, coa_issue: date, is_completed: bool, + coa_published: bool = True) -> CertificateOfAnalysis: + """Attach the lab report, and the CoA either released or awaiting release. + + Mirrors the real flow: the CoA record exists first (pending approval) and + only becomes a buyer-visible document when it is released. + """ + ref = batch.batch_reference + folder = ref.lower() + + lab = _pdf( + folder, f"lab-report-{ref}.pdf", + "Laboratory Report", f"Sample reference {qc.lab_reference}", + [ + f"Batch: {ref}", + "Product: Unrefined Shea Butter", + f"Laboratory: {LAB_NAME}", + f"Sample received: {coa_issue.strftime('%d %B %Y')}", + f"Specification: Unrefined Shea Butter v{qc.spec_version}", + ], + table=[ + ["Parameter", "Result", "Specification limit"], + ["Free Fatty Acid (%)", "0.42", "max 1.00"], + ["Moisture (%)", "0.08", "max 0.20"], + ["Peroxide value (meq/kg)", "1.20", "max 5.00"], + ["Visual grade", "A", "A or B"], + ], + ) + Document.objects.update_or_create( + batch=batch, doc_type=Document.DocType.LAB_REPORT, file_name=lab["file_name"], + defaults={ + "file_url": lab["file_url"], "file_size_bytes": lab["size"], + "visibility": Document.Visibility.OPS, "uploaded_by": enterer, + }, + ) + + coa_pdf = _pdf( + folder, f"certificate-of-analysis-{ref}.pdf", + "Certificate of Analysis", f"Certificate {qc.lab_reference} - issued {coa_issue.strftime('%d %B %Y')}", + [ + f"Batch: {ref}", + "Product: Unrefined Shea Butter (Grade A)", + "Origin: Katsina State, Nigeria", + f"Issued by: {LAB_NAME}", + f"Specification: Unrefined Shea Butter v{qc.spec_version}", + ], + table=[ + ["Parameter", "Result", "Specification limit"], + ["Free Fatty Acid (%)", "0.42", "max 1.00"], + ["Moisture (%)", "0.08", "max 0.20"], + ["Peroxide value (meq/kg)", "1.20", "max 5.00"], + ["Visual grade", "A", "A or B"], + ], + ) + source_doc = None + if coa_published: + source_doc = Document.objects.update_or_create( + batch=batch, doc_type=Document.DocType.COA, file_name=coa_pdf["file_name"], + defaults={ + "file_url": coa_pdf["file_url"], "file_size_bytes": coa_pdf["size"], + "visibility": Document.Visibility.BUYER, "uploaded_by": enterer, + }, + )[0] + else: + # not released yet, so nothing buyer-visible exists + Document.objects.filter( + batch=batch, doc_type=Document.DocType.COA, file_name=coa_pdf["file_name"] + ).delete() + + coa, _ = CertificateOfAnalysis.objects.update_or_create( + batch=batch, version=1, + defaults={ + "qc_result": qc, + "file_url": coa_pdf["file_url"], + "file_name": coa_pdf["file_name"], + "issuer": LAB_NAME, + "issue_date": coa_issue, + "expiry_date": coa_issue + timedelta(days=365), + "uploaded_by": enterer, + "status": ( + CertificateOfAnalysis.Status.PUBLISHED if coa_published + else CertificateOfAnalysis.Status.PENDING_APPROVAL + ), + "sha256_hash": coa_pdf["sha256"], + "spec_version": qc.spec_version, + "source_document": source_doc, + "quality_approved_by": verifier, + "quality_approved_at": timezone.now() - timedelta(days=9), + "released_by": enterer if coa_published else None, + "released_at": timezone.now() - timedelta(days=8) if coa_published else None, + "published_at": timezone.now() - timedelta(days=8) if coa_published else None, + }, + ) + + if is_completed: + packing = _pdf( + folder, f"packing-list-{ref}.pdf", "Packing List", + f"Consignment {ref} - 80 x 25 kg food-grade drums", + [ + f"Batch: {ref}", + "Product: Unrefined Shea Butter", + "Net weight: 2,000 kg", + "Packaging: 80 x 25 kg food-grade poly-lined drums", + "Container: IXGU2450163", + ], + ) + bl = _pdf( + folder, f"bill-of-lading-{ref}.pdf", "Bill of Lading", + "Original - port to port", + [ + f"Goods: Unrefined Shea Butter, batch {ref}", + "Port of loading: Apapa Port, Lagos, Nigeria", + "Port of discharge: Port of Felixstowe, United Kingdom", + "Vessel: MV Atlantic Shea Star (dummy vessel)", + "Container: IXGU2450163", + ], + ) + phyto = _pdf( + folder, f"phytosanitary-certificate-{ref}.pdf", "Phytosanitary Certificate", + "Nigeria Agricultural Quarantine Service (dummy)", + [ + f"Consignment: {ref}", + "Product: Unrefined Shea Butter (processed, non-viable)", + "Declared for export to the United Kingdom", + ], + ) + for meta, doc_type in ((packing, Document.DocType.PACKING_LIST), (bl, Document.DocType.BL)): + Document.objects.update_or_create( + batch=batch, doc_type=doc_type, file_name=meta["file_name"], + defaults={ + "file_url": meta["file_url"], "file_size_bytes": meta["size"], + "visibility": Document.Visibility.BUYER, "uploaded_by": enterer, + }, + ) + Document.objects.update_or_create( + batch=batch, doc_type=Document.DocType.REGULATORY, file_name=phyto["file_name"], + defaults={ + "file_url": phyto["file_url"], "file_size_bytes": phyto["size"], + "visibility": Document.Visibility.BUYER, "uploaded_by": enterer, + }, + ) + RegulatoryCert.objects.update_or_create( + batch=batch, cert_type=RegulatoryCert.CertType.PHYTOSANITARY, + defaults={ + "file_url": phyto["file_url"], "file_name": phyto["file_name"], + "cert_reference": f"NAQS-{ref}", "issue_date": coa_issue + timedelta(days=10), + "expiry_date": coa_issue + timedelta(days=100), + }, + ) + + return coa + + def _eudr(self, batch: ProcessingBatch, zone: OriginZone, intakes: list[HarvestIntake], + enterer: User, verifier: User, approved: bool) -> DueDiligenceCase: + case, _ = DueDiligenceCase.objects.update_or_create( + batch=batch, + defaults={ + "status": DueDiligenceCase.Status.APPROVED if approved + else DueDiligenceCase.Status.COMPLETE_FOR_REVIEW, + "risk_level": DueDiligenceCase.RiskLevel.LOW, + "assessed_by": verifier, + "assessed_at": timezone.now() - timedelta(days=7), + "approved_by": verifier if approved else None, + "approved_at": timezone.now() - timedelta(days=6) if approved else None, + }, + ) + + for etype, name in ( + (EUDREvidence.EvidenceType.GEOLOCATION, f"{zone.zone_code}-geolocation.json"), + (EUDREvidence.EvidenceType.HARVEST_DOCUMENT, f"{intakes[0].intake_reference}-weighbridge-ticket.pdf"), + (EUDREvidence.EvidenceType.SUPPLIER_IDENTITY, "collector-group-register.pdf"), + (EUDREvidence.EvidenceType.LEGALITY, "export-permit-and-land-use.pdf"), + ): + EUDREvidence.objects.update_or_create( + case=case, evidence_type=etype, file=name, + defaults={ + "source": "IFO field capture (dummy)", + "collected_at": date(2026, 8, 20), + "verified_by": verifier, + "verified_at": timezone.now() - timedelta(days=7), + "notes": "Synthetic evidence record for training.", + }, + ) + + RiskAssessment.objects.update_or_create( + case=case, risk_level=DueDiligenceCase.RiskLevel.LOW, + defaults={ + "basis": "Country risk benchmarking plus origin-zone geolocation screening (synthetic).", + "findings": "No deforestation, no protected-area overlap, no land-tenure dispute identified.", + "assessed_by": verifier, + }, + ) + MitigationAction.objects.update_or_create( + case=case, action="Annual re-verification of collector group membership register", + defaults={ + "status": MitigationAction.Status.COMPLETED, + "owner": "IXG Compliance (dummy owner)", + "due_date": date(2027, 8, 31), + "evidence_file": "collector-group-register-reverification.pdf", + "completed_at": timezone.now() - timedelta(days=5), + }, + ) + DueDiligenceStatementReference.objects.update_or_create( + case=case, dds_reference=f"DDS-2026-{batch.batch_reference[-3:]}", + defaults={ + "submission_date": date(2026, 9, 2), + "submitted_by": enterer, + "reconciliation_status": DueDiligenceStatementReference.ReconciliationStatus.RECONCILED, + "notes": "Synthetic DDS reference recorded for training.", + }, + ) + + # Run the REAL completeness check rather than inventing results, so the + # seeded case matches what the ops screen shows when the check is re-run. + from apps.compliance.services import run_completeness_check + + run_completeness_check(case) + case.refresh_from_db() + return case + + def _shipment(self, batch: ProcessingBatch, delivered: bool) -> Shipment: + defaults = { + "status": Shipment.Status.DELIVERED if delivered else Shipment.Status.IN_TRANSIT, + "origin_port": "Apapa Port, Lagos", + "destination_port": "Port of Felixstowe", + "vessel_name": "MV Atlantic Shea Star (dummy vessel)", + "container_ref": "IXGU2450163", + "etd_nigeria": date(2026, 9, 8) if delivered else date(2026, 9, 14), + "eta_uk": date(2026, 9, 28) if delivered else date(2026, 10, 5), + "bl_number": f"MAEU-260824-{batch.batch_reference[-3:]}", + "freight_forwarder": FORWARDER, + } + shipment = batch.shipments.first() + if shipment is None: + return Shipment.objects.create(batch=batch, **defaults) + for key, value in defaults.items(): + setattr(shipment, key, value) + shipment.save() + return shipment + + def _esg(self, batch: ProcessingBatch, reports: list[tuple[date, int, int, str]]) -> None: + for report_date, total, women, note in reports: + ESGFieldReport.objects.update_or_create( + batch=batch, report_date=report_date, + defaults={ + "women_employed": women, + "total_workers": total, + "fair_wage_paid": True, + "safety_equipment_used": True, + "community_notes": note, + "reported_by": "IFO - A. Bello (dummy officer)", + }, + ) + + def _buyer(self) -> User: + buyer, _ = User.objects.get_or_create( + email=BUYER_EMAIL, + defaults={"full_name": BUYER_NAME, "role": "buyer", "is_active": True, "is_staff": False}, + ) + buyer.role = "buyer" + buyer.full_name = BUYER_NAME + buyer.save(update_fields=["role", "full_name"]) + Subscription.objects.update_or_create( + user=buyer, plan=Subscription.Plan.PROFESSIONAL, + defaults={ + "status": Subscription.Status.ACTIVE, + "stripe_price_id": "price_dummy_professional", + "current_period_end": timezone.now() + timedelta(days=305), + }, + ) + return buyer + + # ---------------------------------------------------------------------- main + + def handle(self, *args, **options): + if options["reset"]: + self._reset() + + enterer, verifier = self._ops_users() + spec = self._spec() + self.stdout.write(f" ops enterer: {getattr(enterer, 'email', 'none')}") + self.stdout.write(f" ops verifier: {getattr(verifier, 'email', 'none')} (four-eyes)") + self.stdout.write(f" spec: {spec}") + + if enterer is None: + self.stderr.write("No ops user found - create one before seeding.") + return + + zone, intakes = self._origin_chain(verifier or enterer) + self.stdout.write(f" origin: {zone.zone_code}, {len(intakes)} harvest intakes") + + # 1. completed happy path + completed = self._batch( + COMPLETED_BATCH, intakes, enterer, + harvest_date=date(2026, 8, 30), weight_kg=Decimal("2000.00"), + grade=ProcessingBatch.Grade.A, status=ProcessingBatch.Status.DELIVERED, + notes="Synthetic batch used for IXG ops training walkthroughs.", + ) + qc = self._qc(completed, enterer, verifier or enterer, spec) + coa = self._documents(completed, qc, enterer, verifier or enterer, date(2026, 9, 1), is_completed=True) + self._eudr(completed, zone, intakes, enterer, verifier or enterer, approved=True) + self._shipment(completed, delivered=True) + self._esg(completed, [ + (date(2026, 8, 26), 34, 21, "Synthetic report: shea processing training held with the collective."), + (date(2026, 9, 2), 31, 19, "Synthetic report: protective gloves and aprons issued to all pickers."), + ]) + self.stdout.write(f" {COMPLETED_BATCH}: delivered, QC pass, CoA published ({coa.sha256_hash[:12]}...), EUDR approved") + + # 2. mid-flight + inflight = self._batch( + INFLIGHT_BATCH, intakes, enterer, + harvest_date=date(2026, 9, 4), weight_kg=Decimal("1500.00"), + grade=ProcessingBatch.Grade.B, status=ProcessingBatch.Status.IN_TRANSIT, + notes="Synthetic batch currently in transit - used to show work in progress.", + ) + qc2 = self._qc(inflight, enterer, verifier or enterer, spec) + self._documents(inflight, qc2, enterer, verifier or enterer, date(2026, 9, 10), + is_completed=False, coa_published=False) + self._eudr(inflight, zone, intakes, enterer, verifier or enterer, approved=False) + self._shipment(inflight, delivered=False) + self._esg(inflight, [(date(2026, 9, 6), 27, 16, "Synthetic report: intake day observations.")]) + self.stdout.write(f" {INFLIGHT_BATCH}: in transit, CoA awaiting release, EUDR complete for review") + + # 3. exception branch + failed = ProcessingBatch.objects.filter(batch_reference=FAILED_BATCH).first() + if failed: + # exactly one CAPA per demo batch, however often this command is re-run + CAPARecord.objects.filter(batch=failed).delete() + CAPARecord.objects.create( + batch=failed, + issue_description=( + "QC result outside specification: moisture 0.31% against a limit of 0.20% " + "(synthetic training record)." + ), + root_cause="Drying time shortened during a heavy-rain week at the processing site.", + corrective_action="Batch placed on hold and re-dried; re-sampled and re-tested by the laboratory.", + preventive_action="Minimum drying time added to the processing checklist; daily moisture spot checks.", + status=CAPARecord.Status.IN_PROGRESS, + assigned_to="IXG Quality (dummy owner)", + due_date=date(2026, 10, 15), + ) + self.stdout.write(f" {FAILED_BATCH}: CAPA record replaced (exception branch example)") + + buyer = self._buyer() + self.stdout.write(f" buyer: {buyer.email} ({buyer.role}) with an active Professional subscription") + + files = sorted(p.name for p in DOCS_ROOT.glob("*/*.pdf")) + self.stdout.write(self.style.SUCCESS( + f"seed_walkthrough complete: 2 demo batches + exception CAPA, {len(files)} PDF files in {DOCS_ROOT}" + ))