ixgadmin/ixg_platform
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add seed_walkthrough management command

Browse source 
Builds one coherent, fully fictional batch chain for ops training, so the same
batch reference appears on every screen a trainer walks through: origin zone and
collector groups, harvest intakes and lineage, processing batch, QC result and
four-eyes verification, certificate of analysis with its supporting documents,
EUDR due-diligence case, shipment and ESG report.

Three batches on purpose: a completed happy path, one mid-flight, and one QC
failure with a CAPA so the exception branch has something to show.

Strictly synthetic data - no real buyer, supplier, collector or producer data.
This commit is contained in:
ixgadmin 2026-09-16 16:39:07 +00:00
parent c568a34dc0
commit 4667637265
3 changed files with 635 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

0
apps/core/management/__init__.py Normal file
View file

0
apps/core/management/commands/__init__.py Normal file
View file

635
apps/core/management/commands/seed_walkthrough.py Normal file
View file

@ -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}"
))