ixgadmin
6a3a902045
The ops lineage screen refuses to allocate more of an intake than was received, counting every batch, but POST /api/intake-batch-lineage/ accepted 9999 kg against a 100 kg intake. Move the rule into a shared helper so the screen and the serializer refuse the same allocations.
410 lines
15 KiB
Python
410 lines
15 KiB
Python
from decimal import Decimal
|
|
|
|
from django.db import transaction
|
|
from django.db.models import Sum
|
|
from django.utils import timezone
|
|
|
|
from apps.batches.models import ProcessingBatch
|
|
from apps.documents.models import CertificateOfAnalysis, Document
|
|
from apps.notifications.models import Notification
|
|
from apps.qc.models import QCResult
|
|
|
|
DEFAULT_PRODUCT = "Unrefined Shea Butter"
|
|
|
|
# Legacy fallback thresholds used only when no approved product specification exists.
|
|
FALLBACK_LIMITS = {
|
|
"ffa": {"max": Decimal("3.00")},
|
|
"moisture": {"max": Decimal("0.10")},
|
|
"peroxide": {"max": Decimal("10.00")},
|
|
}
|
|
|
|
|
|
# Sentinel: default behaviour looks up the current approved spec; an explicit
|
|
# None forces legacy fallback limits (used by tests and edge cases).
|
|
_USE_CURRENT = object()
|
|
|
|
|
|
def evaluate_qc_data(values, spec=_USE_CURRENT):
|
|
"""Evaluate a QC result against a specification (or legacy fallback limits).
|
|
|
|
Returns (overall_pass, result_status, parameter_details, spec_version).
|
|
"""
|
|
from apps.compliance.models import ProductSpecification, SpecParameter
|
|
|
|
values = dict(values)
|
|
if "visual" not in values and "visual_grade" in values:
|
|
values["visual"] = values["visual_grade"]
|
|
|
|
if spec is _USE_CURRENT:
|
|
spec = ProductSpecification.current(DEFAULT_PRODUCT)
|
|
spec_version = spec.version if spec else ""
|
|
|
|
details = {}
|
|
outside = False
|
|
missing = False
|
|
|
|
if spec:
|
|
for param in spec.parameters.all():
|
|
value = values.get(param.parameter)
|
|
status, detail = param.check_value(value)
|
|
details[param.parameter] = {"status": status, "detail": detail, "unit": param.unit}
|
|
if status == "outside":
|
|
outside = True
|
|
elif status == "missing":
|
|
missing = True
|
|
else:
|
|
for key, limits in FALLBACK_LIMITS.items():
|
|
value = values.get(key)
|
|
try:
|
|
numeric = float(value)
|
|
except (TypeError, ValueError):
|
|
missing = True
|
|
details[key] = {"status": "missing", "detail": "Value missing", "unit": ""}
|
|
continue
|
|
ok = numeric <= float(limits["max"])
|
|
details[key] = {
|
|
"status": "within" if ok else "outside",
|
|
"detail": "Within specification" if ok else "Above maximum",
|
|
"unit": "",
|
|
}
|
|
if not ok:
|
|
outside = True
|
|
visual = values.get("visual")
|
|
if visual and visual != "Fail":
|
|
details["visual"] = {"status": "within", "detail": "Within specification", "unit": ""}
|
|
else:
|
|
outside = True
|
|
details["visual"] = {"status": "outside", "detail": "Visual grade failed", "unit": ""}
|
|
|
|
if outside:
|
|
result_status = QCResult.ResultStatus.OUTSIDE_SPEC
|
|
elif missing:
|
|
result_status = QCResult.ResultStatus.MISSING
|
|
else:
|
|
result_status = QCResult.ResultStatus.WITHIN_SPEC
|
|
|
|
overall_pass = not outside and not missing
|
|
return overall_pass, result_status, details, spec_version
|
|
|
|
|
|
def find_source_lab_document(batch):
|
|
"""Return the most recent uploaded independent laboratory report for a batch."""
|
|
return Document.objects.filter(batch=batch, doc_type=Document.DocType.LAB_REPORT).order_by("-uploaded_at").first()
|
|
|
|
|
|
def create_coa_for_qc(qc, approved_by):
|
|
"""Generate the IXG Quality Summary PDF and create the (pending) CoA record.
|
|
|
|
Only called after a separate quality approval of a passing QC result (four-eyes).
|
|
"""
|
|
from apps.compliance.coa_pdf import generate_coa_pdf, write_coa_file
|
|
|
|
if not qc.overall_pass:
|
|
raise ValueError("CoA can only be generated for a passing QC result")
|
|
if CertificateOfAnalysis.objects.filter(qc_result=qc).exists():
|
|
return CertificateOfAnalysis.objects.get(qc_result=qc)
|
|
|
|
batch = qc.batch
|
|
pdf_bytes, digest = generate_coa_pdf(qc, batch, spec_version=qc.spec_version)
|
|
file_url, file_name = write_coa_file(qc, batch, pdf_bytes)
|
|
source = find_source_lab_document(batch)
|
|
|
|
return CertificateOfAnalysis.objects.create(
|
|
batch=batch,
|
|
qc_result=qc,
|
|
file_url=file_url,
|
|
file_name=file_name,
|
|
issuer="IXG Ltd",
|
|
issue_date=timezone.localdate(),
|
|
uploaded_by=approved_by,
|
|
status=CertificateOfAnalysis.Status.PENDING_APPROVAL,
|
|
version=1,
|
|
sha256_hash=digest,
|
|
spec_version=qc.spec_version,
|
|
source_document=source,
|
|
quality_approved_by=approved_by,
|
|
quality_approved_at=timezone.now(),
|
|
)
|
|
|
|
|
|
def publish_coa(coa, released_by):
|
|
"""Commercial release approval: mark published, expose to buyers, notify buyers."""
|
|
if coa.status == CertificateOfAnalysis.Status.PUBLISHED:
|
|
return coa
|
|
|
|
with transaction.atomic():
|
|
coa.status = CertificateOfAnalysis.Status.PUBLISHED
|
|
coa.released_by = released_by
|
|
coa.released_at = timezone.now()
|
|
coa.published_at = timezone.now()
|
|
coa.save(update_fields=[
|
|
"status", "released_by", "released_at", "published_at",
|
|
])
|
|
|
|
Document.objects.create(
|
|
batch=coa.batch,
|
|
doc_type=Document.DocType.COA,
|
|
file_url=coa.file_url,
|
|
file_name=coa.file_name,
|
|
visibility=Document.Visibility.BUYER,
|
|
uploaded_by=released_by,
|
|
)
|
|
|
|
for buyer in _buyer_users():
|
|
Notification.objects.create(
|
|
user=buyer,
|
|
type="coa_published",
|
|
title=f"Certificate of Analysis published: {coa.batch.batch_reference}",
|
|
message=f"A Certificate of Analysis is now available for batch {coa.batch.batch_reference}.",
|
|
)
|
|
return coa
|
|
|
|
|
|
def _buyer_users():
|
|
from apps.users.models import User
|
|
|
|
return User.objects.filter(role=User.Roles.BUYER, is_active=True)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# EUDR evidence workflow
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def ensure_due_diligence_case(batch):
|
|
from apps.compliance.models import DueDiligenceCase
|
|
|
|
case, _ = DueDiligenceCase.objects.get_or_create(batch=batch)
|
|
return case
|
|
|
|
|
|
def intake_allocation_error(intake, weight, *, exclude_batch=None, exclude_lineage=None):
|
|
"""Return a message when ``weight`` cannot be allocated from ``intake``.
|
|
|
|
An intake is a finite quantity of shea, so the rule counts **every** batch
|
|
that has already drawn from it, not just the one being saved. ``exclude_batch``
|
|
and ``exclude_lineage`` keep the caller's own allocation out of that total.
|
|
Shared by the ops lineage formset and the lineage API serializer so the screen
|
|
and the API refuse the same allocations.
|
|
"""
|
|
from apps.compliance.models import IntakeBatchLineage
|
|
|
|
received = intake.weight_kg
|
|
if received is None:
|
|
# Records that arrived through the API can be missing their weight.
|
|
# Nothing to check against, so nothing is claimed.
|
|
return None
|
|
if weight > received:
|
|
return (
|
|
f"{intake.intake_reference} received {received} kg; "
|
|
f"{weight} kg cannot be allocated from it."
|
|
)
|
|
elsewhere_qs = IntakeBatchLineage.objects.filter(harvest_intake=intake)
|
|
if exclude_batch is not None:
|
|
elsewhere_qs = elsewhere_qs.exclude(batch=exclude_batch)
|
|
if exclude_lineage is not None:
|
|
elsewhere_qs = elsewhere_qs.exclude(pk=exclude_lineage.pk)
|
|
elsewhere = elsewhere_qs.aggregate(total=Sum("weight_allocated_kg"))["total"] or Decimal("0")
|
|
if elsewhere + weight > received:
|
|
others = ", ".join(sorted({row.batch.batch_reference for row in elsewhere_qs}))
|
|
return (
|
|
f"{intake.intake_reference} received {received} kg and "
|
|
f"{elsewhere} kg of it is already allocated to {others}. "
|
|
f"Only {received - elsewhere} kg is left."
|
|
)
|
|
return None
|
|
|
|
|
|
def _lineage(batch):
|
|
return list(batch.intake_lineage.select_related("harvest_intake", "harvest_intake__origin_zone", "harvest_intake__collector_group").all())
|
|
|
|
|
|
def run_completeness_check(case):
|
|
"""Evaluate evidence completeness for a due diligence case.
|
|
|
|
Outputs a JSON results map and the resulting case status. The system never
|
|
labels a case 'EUDR compliant' - it only reports evidence state.
|
|
"""
|
|
from apps.compliance.models import DueDiligenceCase
|
|
|
|
batch = case.batch
|
|
links = _lineage(batch)
|
|
results = {}
|
|
|
|
if links:
|
|
zone_statuses = []
|
|
for link in links:
|
|
intake = link.harvest_intake
|
|
zone = intake.origin_zone
|
|
if zone is None:
|
|
zone_statuses.append("no_zone")
|
|
elif zone.verification_status == "verified":
|
|
zone_statuses.append("verified")
|
|
elif zone.verification_status == "rejected":
|
|
zone_statuses.append("rejected")
|
|
else:
|
|
zone_statuses.append("unverified")
|
|
results["geolocation"] = {
|
|
"pass": all(s == "verified" for s in zone_statuses),
|
|
"detail": "All intakes trace to a verified origin zone" if all(s == "verified" for s in zone_statuses)
|
|
else f"Zone states: {', '.join(zone_statuses)}",
|
|
}
|
|
else:
|
|
results["geolocation"] = {"pass": False, "detail": "No intake lineage recorded for batch"}
|
|
|
|
results["lineage"] = {
|
|
"pass": len(links) > 0,
|
|
"detail": f"{len(links)} harvest intake(s) linked to batch" if links else "Batch has no intake lineage",
|
|
}
|
|
|
|
complete_intakes = all(
|
|
link.harvest_intake.intake_date and link.harvest_intake.weight_kg is not None
|
|
for link in links
|
|
) if links else False
|
|
results["intake_data"] = {
|
|
"pass": complete_intakes,
|
|
"detail": "Intake date and weight present" if complete_intakes else "Intake date/weight missing",
|
|
}
|
|
|
|
required_types = {"geolocation", "harvest_document", "supplier_identity"}
|
|
evidence_types = {e.evidence_type for e in case.evidence.all()}
|
|
evidence_pass = required_types.issubset(evidence_types)
|
|
results["evidence"] = {
|
|
"pass": evidence_pass,
|
|
"detail": "Required evidence present" if evidence_pass
|
|
else f"Missing evidence types: {', '.join(sorted(required_types - evidence_types))}",
|
|
}
|
|
|
|
risk_records = case.risk_assessments.count()
|
|
results["risk_assessment"] = {
|
|
"pass": risk_records > 0,
|
|
"detail": f"{risk_records} risk assessment(s) recorded" if risk_records else "No risk assessment recorded",
|
|
}
|
|
|
|
case.completeness_results = results
|
|
|
|
if case.status in (DueDiligenceCase.Status.APPROVED, DueDiligenceCase.Status.REJECTED):
|
|
case.save(update_fields=["completeness_results", "updated_at"])
|
|
return case
|
|
|
|
all_pass = all(r["pass"] for r in results.values())
|
|
if not all_pass:
|
|
case.status = DueDiligenceCase.Status.INCOMPLETE
|
|
else:
|
|
if case.risk_level == DueDiligenceCase.RiskLevel.HIGH or case.risk_assessments.filter(risk_level="high").exists():
|
|
case.status = DueDiligenceCase.Status.MITIGATION_REQUIRED
|
|
else:
|
|
case.status = DueDiligenceCase.Status.COMPLETE_FOR_REVIEW
|
|
case.save(update_fields=["status", "completeness_results", "updated_at"])
|
|
return case
|
|
|
|
|
|
def evaluate_risk(case, user=None):
|
|
"""Rule-based risk assessment. Transparency over inference: each finding is
|
|
grounded in a concrete, reviewable rule. No AI-derived legal conclusions."""
|
|
from apps.compliance.models import RiskAssessment
|
|
|
|
findings = []
|
|
level = "standard"
|
|
|
|
links = _lineage(case.batch)
|
|
if not links:
|
|
level = "high"
|
|
findings.append("No intake lineage recorded - risk of untraceable origin")
|
|
else:
|
|
unverified = any(
|
|
l.harvest_intake.origin_zone is None or l.harvest_intake.origin_zone.verification_status != "verified"
|
|
for l in links
|
|
)
|
|
if unverified:
|
|
level = "high"
|
|
findings.append("One or more origin zones unverified or missing")
|
|
|
|
expired_or_missing = not case.evidence.exists()
|
|
if expired_or_missing:
|
|
findings.append("Required EUDR evidence absent")
|
|
|
|
# Regulatory context (transparent, not a legal conclusion).
|
|
findings.append("Shea butter is not listed in EUDR Annex I; this tracker provides due-diligence "
|
|
"evidence and buyer assurance, not an EU-mandated filing")
|
|
|
|
risk = RiskAssessment.objects.create(
|
|
case=case,
|
|
risk_level=level,
|
|
basis="; ".join(findings),
|
|
findings="; ".join(findings),
|
|
assessed_by=user,
|
|
assessed_at=timezone.now(),
|
|
)
|
|
case.risk_level = level
|
|
case.assessed_by = user
|
|
case.assessed_at = timezone.now()
|
|
if level == "high":
|
|
case.status = "mitigation_required"
|
|
case.save(update_fields=["risk_level", "assessed_by", "assessed_at", "status", "updated_at"])
|
|
return risk
|
|
|
|
|
|
def approve_case(case, user):
|
|
"""Named compliance approval of a complete case."""
|
|
if case.status not in ("complete_for_review", "mitigation_required", "risk_identified"):
|
|
raise ValueError("Case is not ready for approval")
|
|
open_mitigations = case.mitigation_actions.exclude(status="verified").exists()
|
|
if case.risk_level == "high" and open_mitigations:
|
|
raise ValueError("High-risk case has unverified mitigation actions")
|
|
case.status = "approved"
|
|
case.approved_by = user
|
|
case.approved_at = timezone.now()
|
|
case.save(update_fields=["status", "approved_by", "approved_at", "updated_at"])
|
|
return case
|
|
|
|
|
|
def reject_case(case, user, reason):
|
|
if not reason:
|
|
raise ValueError("Rejection reason is required")
|
|
case.status = "rejected"
|
|
case.approved_by = user
|
|
case.approved_at = timezone.now()
|
|
case.rejected_reason = reason
|
|
case.save(update_fields=["status", "approved_by", "approved_at", "rejected_reason", "updated_at"])
|
|
return case
|
|
|
|
|
|
def record_submission(case, user, dds_reference, submission_date):
|
|
from apps.compliance.models import DueDiligenceStatementReference
|
|
|
|
return DueDiligenceStatementReference.objects.create(
|
|
case=case,
|
|
dds_reference=dds_reference,
|
|
submission_date=submission_date,
|
|
submitted_by=user,
|
|
)
|
|
|
|
|
|
def evidence_package(case):
|
|
"""Assemble the exportable evidence package summary."""
|
|
batch = case.batch
|
|
intakes = [l.harvest_intake for l in _lineage(batch)]
|
|
return {
|
|
"batch_reference": batch.batch_reference,
|
|
"product": DEFAULT_PRODUCT,
|
|
"weight_kg": float(batch.weight_kg) if batch.weight_kg is not None else None,
|
|
"origin_zones": [
|
|
{"zone_code": i.origin_zone.zone_code, "country": i.origin_zone.country,
|
|
"verification_status": i.origin_zone.verification_status}
|
|
for i in intakes if i.origin_zone
|
|
],
|
|
"collector_groups": [
|
|
{"name": i.collector_group.name, "contact_reference": i.collector_group.contact_reference}
|
|
for i in intakes if i.collector_group
|
|
],
|
|
"intake_count": len(intakes),
|
|
"risk_level": case.risk_level,
|
|
"status": case.status,
|
|
"approved_by": case.approved_by.full_name if case.approved_by else None,
|
|
"approved_at": case.approved_at.isoformat() if case.approved_at else None,
|
|
"dds_references": [
|
|
{"dds_reference": r.dds_reference, "submission_date": r.submission_date.isoformat() if r.submission_date else None}
|
|
for r in case.statement_references.all()
|
|
],
|
|
}
|