From 08096e168fc3ca6e183e1a7b20b3612920e0425a Mon Sep 17 00:00:00 2001 From: ixgadmin Date: Wed, 5 Aug 2026 13:58:39 +0000 Subject: [PATCH] P99C: compliance automation - CoA auto-generation and EUDR due diligence - Versioned product specification table (Unrefined Shea Butter v1.0 seed) with spec-based QC evaluation replacing hard-coded thresholds - Structured QC results with test methods and four-eyes verification (submitter cannot self-approve) - reportlab IXG Quality Summary PDF auto-generated on quality approval, SHA-256 hashed; original independent lab PDF kept authoritative - CertificateOfAnalysis version/status/hash/approval/supersede lifecycle; commercial release publishes to buyers and triggers notification - EUDR evidence workflow: Facility, OriginZone, CollectorGroup, HarvestIntake, IntakeBatchLineage, DueDiligenceCase, EUDREvidence, RiskAssessment, MitigationAction, DueDiligenceStatementReference with rule-based completeness check and risk assessment (no AI legal conclusions) - Ops portal: QC Approvals, CoA Reviews, EUDR Compliance screens - Buyer portal: /portal/compliance page - 22 automated tests covering four-eyes, CAPA, PDF+hash, publish/notify, EUDR blocks and buyer isolation --- apps/__init__.py | 0 apps/buyer/urls.py | 1 + apps/buyer/views.py | 15 + apps/compliance/__init__.py | 0 apps/compliance/admin.py | 73 ++++ apps/compliance/apps.py | 7 + apps/compliance/coa_pdf.py | 124 ++++++ apps/compliance/migrations/0001_initial.py | 235 +++++++++++ .../migrations/0002_seed_default_spec.py | 49 +++ apps/compliance/migrations/__init__.py | 0 apps/compliance/models.py | 335 ++++++++++++++++ apps/compliance/permissions.py | 13 + apps/compliance/serializers.py | 122 ++++++ apps/compliance/services.py | 372 ++++++++++++++++++ apps/compliance/tests.py | 305 ++++++++++++++ apps/compliance/urls.py | 20 + apps/compliance/views.py | 170 ++++++++ ...ificateofanalysis_published_at_and_more.py | 87 ++++ apps/documents/models.py | 30 ++ apps/documents/views.py | 45 ++- apps/ops/urls.py | 6 + apps/ops/views.py | 177 ++++++++- ...t_options_qcresult_approved_at_and_more.py | 75 ++++ apps/qc/models.py | 39 +- apps/qc/serializers.py | 38 +- apps/qc/views.py | 61 ++- dsp_platform/settings.py | 1 + dsp_platform/urls.py | 1 + requirements.txt | 1 + templates/ops/base_ops.html | 3 + templates/ops/coa_review_detail.html | 65 +++ templates/ops/coa_review_list.html | 58 +++ templates/ops/eudr_detail.html | 141 +++++++ templates/ops/eudr_list.html | 67 ++++ templates/ops/qc_verify_detail.html | 52 +++ templates/ops/qc_verify_list.html | 59 +++ templates/portal/base_portal.html | 1 + templates/portal/compliance.html | 62 +++ 38 files changed, 2891 insertions(+), 19 deletions(-) create mode 100644 apps/__init__.py create mode 100644 apps/compliance/__init__.py create mode 100644 apps/compliance/admin.py create mode 100644 apps/compliance/apps.py create mode 100644 apps/compliance/coa_pdf.py create mode 100644 apps/compliance/migrations/0001_initial.py create mode 100644 apps/compliance/migrations/0002_seed_default_spec.py create mode 100644 apps/compliance/migrations/__init__.py create mode 100644 apps/compliance/models.py create mode 100644 apps/compliance/permissions.py create mode 100644 apps/compliance/serializers.py create mode 100644 apps/compliance/services.py create mode 100644 apps/compliance/tests.py create mode 100644 apps/compliance/urls.py create mode 100644 apps/compliance/views.py create mode 100644 apps/documents/migrations/0002_certificateofanalysis_published_at_and_more.py create mode 100644 apps/qc/migrations/0002_alter_qcresult_options_qcresult_approved_at_and_more.py create mode 100644 templates/ops/coa_review_detail.html create mode 100644 templates/ops/coa_review_list.html create mode 100644 templates/ops/eudr_detail.html create mode 100644 templates/ops/eudr_list.html create mode 100644 templates/ops/qc_verify_detail.html create mode 100644 templates/ops/qc_verify_list.html create mode 100644 templates/portal/compliance.html diff --git a/apps/__init__.py b/apps/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/apps/buyer/urls.py b/apps/buyer/urls.py index 369bc42..686b28c 100644 --- a/apps/buyer/urls.py +++ b/apps/buyer/urls.py @@ -11,4 +11,5 @@ urlpatterns = [ path('batches//', views.BatchDetailView.as_view(), name='batch_detail'), path('documents/', views.DocumentListView.as_view(), name='document_list'), path('subscription/', views.SubscriptionView.as_view(), name='subscription'), + path('compliance/', views.ComplianceView.as_view(), name='compliance'), ] diff --git a/apps/buyer/views.py b/apps/buyer/views.py index 95485a9..57ff3c7 100644 --- a/apps/buyer/views.py +++ b/apps/buyer/views.py @@ -115,3 +115,18 @@ class ChatEmbedView(BuyerRequiredMixin, TemplateView): response = super().dispatch(request, *args, **kwargs) response["X-Frame-Options"] = "ALLOWALL" return response + + +class ComplianceView(BuyerRequiredMixin, TemplateView): + template_name = "portal/compliance.html" + + def get_context_data(self, **kwargs): + from apps.compliance.models import DueDiligenceCase + from apps.documents.models import CertificateOfAnalysis + + ctx = super().get_context_data(**kwargs) + ctx["approved_coas"] = CertificateOfAnalysis.objects.filter( + status=CertificateOfAnalysis.Status.PUBLISHED + ).select_related("batch", "quality_approved_by").order_by("-published_at") + ctx["eudr_cases"] = DueDiligenceCase.objects.select_related("batch", "approved_by").order_by("-updated_at") + return ctx diff --git a/apps/compliance/__init__.py b/apps/compliance/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/apps/compliance/admin.py b/apps/compliance/admin.py new file mode 100644 index 0000000..ffbeb80 --- /dev/null +++ b/apps/compliance/admin.py @@ -0,0 +1,73 @@ +from django.contrib import admin + +from .models import ( + ProductSpecification, SpecParameter, + Facility, OriginZone, CollectorGroup, HarvestIntake, IntakeBatchLineage, + DueDiligenceCase, EUDREvidence, RiskAssessment, MitigationAction, + DueDiligenceStatementReference, +) + + +class SpecParameterInline(admin.TabularInline): + model = SpecParameter + extra = 0 + + +@admin.register(ProductSpecification) +class ProductSpecificationAdmin(admin.ModelAdmin): + list_display = ["product", "version", "status", "is_current", "effective_from", "effective_to", "approved_at"] + list_filter = ["status", "is_current", "product"] + inlines = [SpecParameterInline] + + +@admin.register(Facility) +class FacilityAdmin(admin.ModelAdmin): + list_display = ["name", "country", "region", "facility_type", "is_active"] + + +@admin.register(OriginZone) +class OriginZoneAdmin(admin.ModelAdmin): + list_display = ["zone_code", "facility", "country", "region", "verification_status", "verified_at"] + list_filter = ["verification_status", "country"] + + +@admin.register(CollectorGroup) +class CollectorGroupAdmin(admin.ModelAdmin): + list_display = ["name", "contact_reference", "origin_zone"] + + +@admin.register(HarvestIntake) +class HarvestIntakeAdmin(admin.ModelAdmin): + list_display = ["intake_reference", "intake_date", "weight_kg", "grade", "origin_zone", "status"] + list_filter = ["status"] + + +@admin.register(IntakeBatchLineage) +class IntakeBatchLineageAdmin(admin.ModelAdmin): + list_display = ["batch", "harvest_intake", "weight_allocated_kg"] + + +@admin.register(DueDiligenceCase) +class DueDiligenceCaseAdmin(admin.ModelAdmin): + list_display = ["batch", "status", "risk_level", "approved_at", "updated_at"] + list_filter = ["status", "risk_level"] + + +@admin.register(EUDREvidence) +class EUDREvidenceAdmin(admin.ModelAdmin): + list_display = ["case", "evidence_type", "source", "collected_at", "verified_at"] + + +@admin.register(RiskAssessment) +class RiskAssessmentAdmin(admin.ModelAdmin): + list_display = ["case", "risk_level", "assessed_by", "assessed_at"] + + +@admin.register(MitigationAction) +class MitigationActionAdmin(admin.ModelAdmin): + list_display = ["case", "action", "status", "owner", "due_date"] + + +@admin.register(DueDiligenceStatementReference) +class DueDiligenceStatementReferenceAdmin(admin.ModelAdmin): + list_display = ["case", "dds_reference", "submission_date", "submitted_by", "reconciliation_status"] diff --git a/apps/compliance/apps.py b/apps/compliance/apps.py new file mode 100644 index 0000000..015d20d --- /dev/null +++ b/apps/compliance/apps.py @@ -0,0 +1,7 @@ +from django.apps import AppConfig + + +class ComplianceConfig(AppConfig): + default_auto_field = "django.db.models.BigAutoField" + name = "apps.compliance" + verbose_name = "Compliance" diff --git a/apps/compliance/coa_pdf.py b/apps/compliance/coa_pdf.py new file mode 100644 index 0000000..2deb0f1 --- /dev/null +++ b/apps/compliance/coa_pdf.py @@ -0,0 +1,124 @@ +import hashlib +from datetime import date +from io import BytesIO + +from django.utils import timezone + + +def generate_coa_pdf(qc, batch, spec_version="", issuer="IXG Ltd"): + """Generate the IXG Quality Summary cover sheet PDF for an approved QC result. + + This is a companion to (never a replacement for) the independent laboratory + certificate, which remains the authoritative source document. + + Returns (pdf_bytes, sha256_hex). + """ + from reportlab.lib import colors + from reportlab.lib.pagesizes import A4 + from reportlab.lib.styles import ParagraphStyle, getSampleStyleSheet + from reportlab.lib.units import mm + from reportlab.platypus import Paragraph, SimpleDocTemplate, Spacer, Table, TableStyle + + styles = getSampleStyleSheet() + title_style = ParagraphStyle( + "TitleX", parent=styles["Title"], fontSize=16, leading=20, + textColor=colors.HexColor("#1A3D2B"), + ) + sub_style = ParagraphStyle( + "SubX", parent=styles["Normal"], fontSize=9, leading=12, + textColor=colors.HexColor("#4B5563"), + ) + cell_style = ParagraphStyle("CellX", parent=styles["Normal"], fontSize=9, leading=11) + + issued = date.today() + reference = f"{batch.batch_reference}-COA-V{1}" + + rows = [ + ["Batch reference", batch.batch_reference], + ["Product", "Unrefined Shea Butter"], + ["Origin", batch.origin_facility], + ["Grade", batch.grade or "-"], + ["Weight", f"{batch.weight_kg} kg" if batch.weight_kg is not None else "-"], + ["Harvest date", batch.harvest_date.isoformat() if batch.harvest_date else "-"], + ["Specification", spec_version or "-"], + ["Lab reference", qc.lab_reference or "-"], + ["Tested by", qc.tested_by or "-"], + ["Quality approved by", (qc.verified_by.full_name or qc.verified_by.email) if qc.verified_by else "-"], + ["Quality approved on", qc.approved_at.isoformat() if qc.approved_at else "-"], + ["Issued by", issuer], + ["Issue date", issued.isoformat()], + ] + body = [ + ["Parameter", "Result", "Unit", "Result status"], + ["FFA", f"{qc.ffa_percentage}", "%", "Within specification" if float(qc.ffa_percentage) <= 3 else "Outside specification"], + ["Moisture", f"{qc.moisture_percentage}", "%", "Within specification" if float(qc.moisture_percentage) <= 0.1 else "Outside specification"], + ["Peroxide value", f"{qc.peroxide_value}", "meq O2/kg", "Within specification" if float(qc.peroxide_value) <= 10 else "Outside specification"], + ["Visual grade", qc.visual_grade, "-", "Within specification" if qc.visual_grade != "Fail" else "Outside specification"], + ] + + pdf = BytesIO() + doc = SimpleDocTemplate(pdf, pagesize=A4, leftMargin=20 * mm, rightMargin=20 * mm, topMargin=18 * mm, bottomMargin=18 * mm) + elements = [] + + elements.append(Paragraph("IXG Ltd — Quality Summary", title_style)) + elements.append(Spacer(1, 2 * mm)) + elements.append(Paragraph( + "Companion document to the independent laboratory Certificate of Analysis. " + "The independent laboratory certificate remains the authoritative source document.", + sub_style, + )) + elements.append(Spacer(1, 6 * mm)) + + header_table = Table([[Paragraph(f"Reference: {reference}", sub_style)]], colWidths=[120 * mm]) + elements.append(header_table) + elements.append(Spacer(1, 4 * mm)) + + meta = Table([[Paragraph(c, cell_style) for c in row] for row in rows], colWidths=[55 * mm, 115 * mm]) + meta.setStyle(TableStyle([ + ("BACKGROUND", (0, 0), (0, -1), colors.HexColor("#EAF3EF")), + ("GRID", (0, 0), (-1, -1), 0.4, colors.HexColor("#CCD3CF")), + ("VALIGN", (0, 0), (-1, -1), "TOP"), + ("TOPPADDING", (0, 0), (-1, -1), 3), + ("BOTTOMPADDING", (0, 0), (-1, -1), 3), + ])) + elements.append(meta) + elements.append(Spacer(1, 6 * mm)) + + elements.append(Paragraph("Results vs Specification", styles["Heading3"])) + elements.append(Spacer(1, 2 * mm)) + result_table = Table(body, colWidths=[55 * mm, 35 * mm, 30 * mm, 50 * mm], repeatRows=1) + result_table.setStyle(TableStyle([ + ("BACKGROUND", (0, 0), (-1, 0), colors.HexColor("#1A3D2B")), + ("TEXTCOLOR", (0, 0), (-1, 0), colors.white), + ("GRID", (0, 0), (-1, -1), 0.4, colors.HexColor("#CCD3CF")), + ("FONTSIZE", (0, 0), (-1, -1), 9), + ("ROWBACKGROUNDS", (0, 1), (-1, -1), [colors.white, colors.HexColor("#F7F6F2")]), + ])) + elements.append(result_table) + elements.append(Spacer(1, 6 * mm)) + + elements.append(Paragraph( + "This summary is derived from the structured QC record on the IXG platform. " + "The independent laboratory certificate remains the authoritative source document.", + sub_style, + )) + + doc.build(elements) + data = pdf.getvalue() + digest = hashlib.sha256(data).hexdigest() + return data, digest + + +def write_coa_file(qc, batch, pdf_bytes): + """Persist the generated PDF under media/ixg-documents/{batch_id}/ and return (file_url, file_name).""" + import os + import uuid as uuid_mod + from django.conf import settings + + batch_dir = os.path.join(settings.MEDIA_ROOT, "ixg-documents", str(batch.id)) + os.makedirs(batch_dir, exist_ok=True) + filename = f"coa-{uuid_mod.uuid4().hex[:8]}-{batch.batch_reference}.pdf" + filepath = os.path.join(batch_dir, filename) + with open(filepath, "wb") as f: + f.write(pdf_bytes) + return f"/media/ixg-documents/{batch.id}/{filename}", filename diff --git a/apps/compliance/migrations/0001_initial.py b/apps/compliance/migrations/0001_initial.py new file mode 100644 index 0000000..5752fce --- /dev/null +++ b/apps/compliance/migrations/0001_initial.py @@ -0,0 +1,235 @@ +# Generated by Django 5.2.12 on 2026-08-05 13:36 + +import django.db.models.deletion +import uuid +from django.conf import settings +from django.db import migrations, models + + +class Migration(migrations.Migration): + + initial = True + + dependencies = [ + ('batches', '0001_initial'), + migrations.swappable_dependency(settings.AUTH_USER_MODEL), + ] + + operations = [ + migrations.CreateModel( + name='CollectorGroup', + fields=[ + ('id', models.UUIDField(default=uuid.uuid4, editable=False, primary_key=True, serialize=False)), + ('name', models.CharField(max_length=255)), + ('contact_reference', models.CharField(blank=True, max_length=255)), + ('notes', models.TextField(blank=True)), + ('created_at', models.DateTimeField(auto_now_add=True)), + ], + options={ + 'db_table': 'collector_groups', + }, + ), + migrations.CreateModel( + name='Facility', + fields=[ + ('id', models.UUIDField(default=uuid.uuid4, editable=False, primary_key=True, serialize=False)), + ('name', models.CharField(max_length=255)), + ('country', models.CharField(max_length=100)), + ('region', models.CharField(blank=True, max_length=100)), + ('address', models.TextField(blank=True)), + ('facility_type', models.CharField(choices=[('production', 'Production'), ('processing', 'Processing')], default='production', max_length=20)), + ('is_active', models.BooleanField(default=True)), + ('created_at', models.DateTimeField(auto_now_add=True)), + ], + options={ + 'db_table': 'facilities', + }, + ), + migrations.CreateModel( + name='DueDiligenceCase', + fields=[ + ('id', models.UUIDField(default=uuid.uuid4, editable=False, primary_key=True, serialize=False)), + ('status', models.CharField(choices=[('incomplete', 'Incomplete'), ('complete_for_review', 'Complete for Review'), ('risk_identified', 'Risk Identified'), ('mitigation_required', 'Mitigation Required'), ('approved', 'Approved'), ('rejected', 'Rejected')], default='incomplete', max_length=30)), + ('risk_level', models.CharField(choices=[('low', 'Low'), ('standard', 'Standard'), ('high', 'High')], default='standard', max_length=20)), + ('completeness_results', models.JSONField(blank=True, default=dict)), + ('assessed_at', models.DateTimeField(blank=True, null=True)), + ('approved_at', models.DateTimeField(blank=True, null=True)), + ('rejected_reason', models.TextField(blank=True)), + ('created_at', models.DateTimeField(auto_now_add=True)), + ('updated_at', models.DateTimeField(auto_now=True)), + ('approved_by', models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='eudr_cases_approved', to=settings.AUTH_USER_MODEL)), + ('assessed_by', models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='eudr_cases_assessed', to=settings.AUTH_USER_MODEL)), + ('batch', models.OneToOneField(on_delete=django.db.models.deletion.CASCADE, related_name='due_diligence_case', to='batches.processingbatch')), + ], + options={ + 'db_table': 'due_diligence_cases', + }, + ), + migrations.CreateModel( + name='DueDiligenceStatementReference', + fields=[ + ('id', models.UUIDField(default=uuid.uuid4, editable=False, primary_key=True, serialize=False)), + ('dds_reference', models.CharField(blank=True, max_length=255)), + ('submission_date', models.DateField(blank=True, null=True)), + ('reconciliation_status', models.CharField(choices=[('pending', 'Pending'), ('reconciled', 'Reconciled')], default='pending', max_length=20)), + ('notes', models.TextField(blank=True)), + ('created_at', models.DateTimeField(auto_now_add=True)), + ('case', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='statement_references', to='compliance.duediligencecase')), + ('submitted_by', models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='dds_submissions', to=settings.AUTH_USER_MODEL)), + ], + options={ + 'db_table': 'due_diligence_statement_references', + }, + ), + migrations.CreateModel( + name='EUDREvidence', + fields=[ + ('id', models.UUIDField(default=uuid.uuid4, editable=False, primary_key=True, serialize=False)), + ('evidence_type', models.CharField(choices=[('geolocation', 'Geolocation'), ('harvest_document', 'Harvest Document'), ('supplier_identity', 'Supplier Identity'), ('legality', 'Legality'), ('other', 'Other')], default='other', max_length=30)), + ('file', models.CharField(max_length=255)), + ('source', models.CharField(blank=True, max_length=255)), + ('collected_at', models.DateField(blank=True, null=True)), + ('verified_at', models.DateTimeField(blank=True, null=True)), + ('notes', models.TextField(blank=True)), + ('created_at', models.DateTimeField(auto_now_add=True)), + ('case', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='evidence', to='compliance.duediligencecase')), + ('verified_by', models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='eudr_evidence_verified', to=settings.AUTH_USER_MODEL)), + ], + options={ + 'db_table': 'eudr_evidence', + }, + ), + migrations.CreateModel( + name='MitigationAction', + fields=[ + ('id', models.UUIDField(default=uuid.uuid4, editable=False, primary_key=True, serialize=False)), + ('action', models.TextField()), + ('status', models.CharField(choices=[('open', 'Open'), ('in_progress', 'In Progress'), ('completed', 'Completed'), ('verified', 'Verified')], default='open', max_length=20)), + ('owner', models.CharField(blank=True, max_length=255)), + ('due_date', models.DateField(blank=True, null=True)), + ('evidence_file', models.CharField(blank=True, max_length=255)), + ('completed_at', models.DateTimeField(blank=True, null=True)), + ('created_at', models.DateTimeField(auto_now_add=True)), + ('case', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='mitigation_actions', to='compliance.duediligencecase')), + ], + options={ + 'db_table': 'mitigation_actions', + }, + ), + migrations.CreateModel( + name='OriginZone', + fields=[ + ('id', models.UUIDField(default=uuid.uuid4, editable=False, primary_key=True, serialize=False)), + ('zone_code', models.CharField(max_length=100, unique=True)), + ('country', models.CharField(max_length=100)), + ('region', models.CharField(blank=True, max_length=100)), + ('geometry', models.JSONField(blank=True, default=dict)), + ('coordinate_source', models.CharField(blank=True, max_length=255)), + ('coordinates_collected_by', models.CharField(blank=True, max_length=255)), + ('coordinates_collected_at', models.DateField(blank=True, null=True)), + ('evidence_file', models.CharField(blank=True, max_length=255)), + ('verification_status', models.CharField(choices=[('unverified', 'Unverified'), ('verified', 'Verified'), ('rejected', 'Rejected')], default='unverified', max_length=20)), + ('verified_at', models.DateTimeField(blank=True, null=True)), + ('created_at', models.DateTimeField(auto_now_add=True)), + ('facility', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='origin_zones', to='compliance.facility')), + ('verified_by', models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='zones_verified', to=settings.AUTH_USER_MODEL)), + ], + options={ + 'db_table': 'origin_zones', + }, + ), + migrations.CreateModel( + name='HarvestIntake', + fields=[ + ('id', models.UUIDField(default=uuid.uuid4, editable=False, primary_key=True, serialize=False)), + ('intake_reference', models.CharField(max_length=100, unique=True)), + ('intake_date', models.DateField(blank=True, null=True)), + ('weight_kg', models.DecimalField(blank=True, decimal_places=2, max_digits=10, null=True)), + ('grade', models.CharField(blank=True, max_length=5)), + ('receiving_officer', models.CharField(blank=True, max_length=255)), + ('status', models.CharField(choices=[('draft', 'Draft'), ('received', 'Received'), ('verified', 'Verified'), ('rejected', 'Rejected')], default='draft', max_length=20)), + ('evidence_file', models.CharField(blank=True, max_length=255)), + ('created_at', models.DateTimeField(auto_now_add=True)), + ('collector_group', models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='intakes', to='compliance.collectorgroup')), + ('origin_zone', models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='intakes', to='compliance.originzone')), + ], + options={ + 'db_table': 'harvest_intakes', + }, + ), + migrations.AddField( + model_name='collectorgroup', + name='origin_zone', + field=models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='collector_groups', to='compliance.originzone'), + ), + migrations.CreateModel( + name='ProductSpecification', + fields=[ + ('id', models.UUIDField(default=uuid.uuid4, editable=False, primary_key=True, serialize=False)), + ('product', models.CharField(max_length=100)), + ('version', models.CharField(max_length=20)), + ('status', models.CharField(choices=[('draft', 'Draft'), ('approved', 'Approved'), ('superseded', 'Superseded')], default='draft', max_length=20)), + ('effective_from', models.DateField(blank=True, null=True)), + ('effective_to', models.DateField(blank=True, null=True)), + ('is_current', models.BooleanField(default=False)), + ('approved_at', models.DateTimeField(blank=True, null=True)), + ('created_at', models.DateTimeField(auto_now_add=True)), + ('updated_at', models.DateTimeField(auto_now=True)), + ('approved_by', models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='specs_approved', to=settings.AUTH_USER_MODEL)), + ], + options={ + 'db_table': 'product_specifications', + 'ordering': ['product', '-version'], + 'unique_together': {('product', 'version')}, + }, + ), + migrations.CreateModel( + name='RiskAssessment', + fields=[ + ('id', models.UUIDField(default=uuid.uuid4, editable=False, primary_key=True, serialize=False)), + ('risk_level', models.CharField(choices=[('low', 'Low'), ('standard', 'Standard'), ('high', 'High')], default='standard', max_length=20)), + ('basis', models.TextField(blank=True)), + ('findings', models.TextField(blank=True)), + ('assessed_at', models.DateTimeField(auto_now_add=True)), + ('assessed_by', models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='risk_assessments', to=settings.AUTH_USER_MODEL)), + ('case', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='risk_assessments', to='compliance.duediligencecase')), + ], + options={ + 'db_table': 'risk_assessments', + 'ordering': ['-assessed_at'], + }, + ), + migrations.CreateModel( + name='IntakeBatchLineage', + fields=[ + ('id', models.UUIDField(default=uuid.uuid4, editable=False, primary_key=True, serialize=False)), + ('weight_allocated_kg', models.DecimalField(blank=True, decimal_places=2, max_digits=10, null=True)), + ('created_at', models.DateTimeField(auto_now_add=True)), + ('batch', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='intake_lineage', to='batches.processingbatch')), + ('harvest_intake', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='batch_lineage', to='compliance.harvestintake')), + ], + options={ + 'db_table': 'intake_batch_lineage', + 'unique_together': {('batch', 'harvest_intake')}, + }, + ), + migrations.CreateModel( + name='SpecParameter', + fields=[ + ('id', models.UUIDField(default=uuid.uuid4, editable=False, primary_key=True, serialize=False)), + ('parameter', models.CharField(choices=[('ffa', 'Free Fatty Acid'), ('moisture', 'Moisture'), ('peroxide', 'Peroxide Value'), ('visual', 'Visual Grade')], max_length=20)), + ('unit', models.CharField(blank=True, max_length=50)), + ('min_value', models.DecimalField(blank=True, decimal_places=3, max_digits=8, null=True)), + ('max_value', models.DecimalField(blank=True, decimal_places=3, max_digits=8, null=True)), + ('allowed_values', models.CharField(blank=True, max_length=255)), + ('required', models.BooleanField(default=True)), + ('sort_order', models.IntegerField(default=0)), + ('spec', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='parameters', to='compliance.productspecification')), + ], + options={ + 'db_table': 'spec_parameters', + 'ordering': ['sort_order'], + 'unique_together': {('spec', 'parameter')}, + }, + ), + ] diff --git a/apps/compliance/migrations/0002_seed_default_spec.py b/apps/compliance/migrations/0002_seed_default_spec.py new file mode 100644 index 0000000..e78ee96 --- /dev/null +++ b/apps/compliance/migrations/0002_seed_default_spec.py @@ -0,0 +1,49 @@ +from django.db import migrations + + +def seed_default_spec(apps, schema_editor): + ProductSpecification = apps.get_model("compliance", "ProductSpecification") + SpecParameter = apps.get_model("compliance", "SpecParameter") + + spec, created = ProductSpecification.objects.get_or_create( + product="Unrefined Shea Butter", + version="1.0", + defaults={ + "status": "approved", + "is_current": True, + }, + ) + if created: + parameters = [ + ("ffa", "%", None, "3.000", "", 1), + ("moisture", "%", None, "0.100", "", 2), + ("peroxide", "meq O2/kg", None, "10.000", "", 3), + ("visual", "grade", None, None, "A,B,C", 4), + ] + for parameter, unit, min_value, max_value, allowed, sort_order in parameters: + SpecParameter.objects.create( + spec=spec, + parameter=parameter, + unit=unit, + min_value=min_value, + max_value=max_value, + allowed_values=allowed, + required=True, + sort_order=sort_order, + ) + + +def remove_default_spec(apps, schema_editor): + ProductSpecification = apps.get_model("compliance", "ProductSpecification") + ProductSpecification.objects.filter(product="Unrefined Shea Butter", version="1.0").delete() + + +class Migration(migrations.Migration): + + dependencies = [ + ("compliance", "0001_initial"), + ] + + operations = [ + migrations.RunPython(seed_default_spec, remove_default_spec), + ] diff --git a/apps/compliance/migrations/__init__.py b/apps/compliance/migrations/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/apps/compliance/models.py b/apps/compliance/models.py new file mode 100644 index 0000000..b29a657 --- /dev/null +++ b/apps/compliance/models.py @@ -0,0 +1,335 @@ +import uuid +from django.conf import settings +from django.db import models + + +class ProductSpecification(models.Model): + class Status(models.TextChoices): + DRAFT = "draft", "Draft" + APPROVED = "approved", "Approved" + SUPERSEDED = "superseded", "Superseded" + + id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False) + product = models.CharField(max_length=100) + version = models.CharField(max_length=20) + status = models.CharField(max_length=20, choices=Status.choices, default=Status.DRAFT) + effective_from = models.DateField(null=True, blank=True) + effective_to = models.DateField(null=True, blank=True) + is_current = models.BooleanField(default=False) + approved_by = models.ForeignKey( + settings.AUTH_USER_MODEL, on_delete=models.SET_NULL, null=True, blank=True, + related_name="specs_approved", + ) + approved_at = models.DateTimeField(null=True, blank=True) + created_at = models.DateTimeField(auto_now_add=True) + updated_at = models.DateTimeField(auto_now=True) + + class Meta: + db_table = "product_specifications" + unique_together = [("product", "version")] + ordering = ["product", "-version"] + + def __str__(self): + return f"{self.product} v{self.version} ({self.status})" + + @classmethod + def current(cls, product): + return cls.objects.filter(product=product, status=cls.Status.APPROVED, is_current=True).first() + + +class SpecParameter(models.Model): + class Parameter(models.TextChoices): + FFA = "ffa", "Free Fatty Acid" + MOISTURE = "moisture", "Moisture" + PEROXIDE = "peroxide", "Peroxide Value" + VISUAL = "visual", "Visual Grade" + + id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False) + spec = models.ForeignKey(ProductSpecification, on_delete=models.CASCADE, related_name="parameters") + parameter = models.CharField(max_length=20, choices=Parameter.choices) + unit = models.CharField(max_length=50, blank=True) + min_value = models.DecimalField(max_digits=8, decimal_places=3, null=True, blank=True) + max_value = models.DecimalField(max_digits=8, decimal_places=3, null=True, blank=True) + allowed_values = models.CharField(max_length=255, blank=True) + required = models.BooleanField(default=True) + sort_order = models.IntegerField(default=0) + + class Meta: + db_table = "spec_parameters" + unique_together = [("spec", "parameter")] + ordering = ["sort_order"] + + def __str__(self): + return f"{self.get_parameter_display()} ({self.unit})" + + def check_value(self, value): + if value is None or value == "": + if self.required: + return ("missing", "Value required") + return ("within", "Optional value absent") + + if self.parameter == self.Parameter.VISUAL: + allowed = [v.strip() for v in self.allowed_values.split(",") if v.strip()] + if allowed and str(value) not in allowed: + return ("outside", f"Visual grade '{value}' not in allowed values: {', '.join(allowed)}") + return ("within", "Within specification") + + try: + numeric = float(value) + except (TypeError, ValueError): + return ("missing", "Non-numeric value") + + if self.min_value is not None and numeric < float(self.min_value): + return ("outside", f"{value} below minimum {self.min_value} {self.unit}") + if self.max_value is not None and numeric > float(self.max_value): + return ("outside", f"{value} above maximum {self.max_value} {self.unit}") + return ("within", "Within specification") + + +class Facility(models.Model): + class FacilityType(models.TextChoices): + PRODUCTION = "production", "Production" + PROCESSING = "processing", "Processing" + + id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False) + name = models.CharField(max_length=255) + country = models.CharField(max_length=100) + region = models.CharField(max_length=100, blank=True) + address = models.TextField(blank=True) + facility_type = models.CharField(max_length=20, choices=FacilityType.choices, default=FacilityType.PRODUCTION) + is_active = models.BooleanField(default=True) + created_at = models.DateTimeField(auto_now_add=True) + + class Meta: + db_table = "facilities" + + def __str__(self): + return self.name + + +class OriginZone(models.Model): + class VerificationStatus(models.TextChoices): + UNVERIFIED = "unverified", "Unverified" + VERIFIED = "verified", "Verified" + REJECTED = "rejected", "Rejected" + + id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False) + facility = models.ForeignKey(Facility, on_delete=models.CASCADE, related_name="origin_zones") + zone_code = models.CharField(max_length=100, unique=True) + country = models.CharField(max_length=100) + region = models.CharField(max_length=100, blank=True) + geometry = models.JSONField(default=dict, blank=True) + coordinate_source = models.CharField(max_length=255, blank=True) + coordinates_collected_by = models.CharField(max_length=255, blank=True) + coordinates_collected_at = models.DateField(null=True, blank=True) + evidence_file = models.CharField(max_length=255, blank=True) + verification_status = models.CharField( + max_length=20, choices=VerificationStatus.choices, default=VerificationStatus.UNVERIFIED + ) + verified_by = models.ForeignKey( + settings.AUTH_USER_MODEL, on_delete=models.SET_NULL, null=True, blank=True, related_name="zones_verified" + ) + verified_at = models.DateTimeField(null=True, blank=True) + created_at = models.DateTimeField(auto_now_add=True) + + class Meta: + db_table = "origin_zones" + + def __str__(self): + return f"{self.zone_code} ({self.country})" + + +class CollectorGroup(models.Model): + id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False) + name = models.CharField(max_length=255) + contact_reference = models.CharField(max_length=255, blank=True) + origin_zone = models.ForeignKey( + OriginZone, on_delete=models.SET_NULL, null=True, blank=True, related_name="collector_groups" + ) + notes = models.TextField(blank=True) + created_at = models.DateTimeField(auto_now_add=True) + + class Meta: + db_table = "collector_groups" + + def __str__(self): + return self.name + + +class HarvestIntake(models.Model): + class Status(models.TextChoices): + DRAFT = "draft", "Draft" + RECEIVED = "received", "Received" + VERIFIED = "verified", "Verified" + REJECTED = "rejected", "Rejected" + + id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False) + intake_reference = models.CharField(max_length=100, unique=True) + collector_group = models.ForeignKey( + CollectorGroup, on_delete=models.SET_NULL, null=True, blank=True, related_name="intakes" + ) + origin_zone = models.ForeignKey( + OriginZone, on_delete=models.SET_NULL, null=True, blank=True, related_name="intakes" + ) + intake_date = models.DateField(null=True, blank=True) + weight_kg = models.DecimalField(max_digits=10, decimal_places=2, null=True, blank=True) + grade = models.CharField(max_length=5, blank=True) + receiving_officer = models.CharField(max_length=255, blank=True) + status = models.CharField(max_length=20, choices=Status.choices, default=Status.DRAFT) + evidence_file = models.CharField(max_length=255, blank=True) + created_at = models.DateTimeField(auto_now_add=True) + + class Meta: + db_table = "harvest_intakes" + + def __str__(self): + return self.intake_reference + + +class IntakeBatchLineage(models.Model): + id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False) + batch = models.ForeignKey("batches.ProcessingBatch", on_delete=models.CASCADE, related_name="intake_lineage") + harvest_intake = models.ForeignKey(HarvestIntake, on_delete=models.CASCADE, related_name="batch_lineage") + weight_allocated_kg = models.DecimalField(max_digits=10, decimal_places=2, null=True, blank=True) + created_at = models.DateTimeField(auto_now_add=True) + + class Meta: + db_table = "intake_batch_lineage" + unique_together = [("batch", "harvest_intake")] + + def __str__(self): + return f"{self.batch.batch_reference} <- {self.harvest_intake.intake_reference}" + + +class DueDiligenceCase(models.Model): + class Status(models.TextChoices): + INCOMPLETE = "incomplete", "Incomplete" + COMPLETE_FOR_REVIEW = "complete_for_review", "Complete for Review" + RISK_IDENTIFIED = "risk_identified", "Risk Identified" + MITIGATION_REQUIRED = "mitigation_required", "Mitigation Required" + APPROVED = "approved", "Approved" + REJECTED = "rejected", "Rejected" + + class RiskLevel(models.TextChoices): + LOW = "low", "Low" + STANDARD = "standard", "Standard" + HIGH = "high", "High" + + id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False) + batch = models.OneToOneField("batches.ProcessingBatch", on_delete=models.CASCADE, related_name="due_diligence_case") + status = models.CharField(max_length=30, choices=Status.choices, default=Status.INCOMPLETE) + risk_level = models.CharField(max_length=20, choices=RiskLevel.choices, default=RiskLevel.STANDARD) + completeness_results = models.JSONField(default=dict, blank=True) + assessed_by = models.ForeignKey( + settings.AUTH_USER_MODEL, on_delete=models.SET_NULL, null=True, blank=True, related_name="eudr_cases_assessed" + ) + assessed_at = models.DateTimeField(null=True, blank=True) + approved_by = models.ForeignKey( + settings.AUTH_USER_MODEL, on_delete=models.SET_NULL, null=True, blank=True, related_name="eudr_cases_approved" + ) + approved_at = models.DateTimeField(null=True, blank=True) + rejected_reason = models.TextField(blank=True) + created_at = models.DateTimeField(auto_now_add=True) + updated_at = models.DateTimeField(auto_now=True) + + class Meta: + db_table = "due_diligence_cases" + + def __str__(self): + return f"EUDR case - {self.batch.batch_reference} ({self.status})" + + +class EUDREvidence(models.Model): + class EvidenceType(models.TextChoices): + GEOLOCATION = "geolocation", "Geolocation" + HARVEST_DOCUMENT = "harvest_document", "Harvest Document" + SUPPLIER_IDENTITY = "supplier_identity", "Supplier Identity" + LEGALITY = "legality", "Legality" + OTHER = "other", "Other" + + id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False) + case = models.ForeignKey(DueDiligenceCase, on_delete=models.CASCADE, related_name="evidence") + evidence_type = models.CharField(max_length=30, choices=EvidenceType.choices, default=EvidenceType.OTHER) + file = models.CharField(max_length=255) + source = models.CharField(max_length=255, blank=True) + collected_at = models.DateField(null=True, blank=True) + verified_by = models.ForeignKey( + settings.AUTH_USER_MODEL, on_delete=models.SET_NULL, null=True, blank=True, related_name="eudr_evidence_verified" + ) + verified_at = models.DateTimeField(null=True, blank=True) + notes = models.TextField(blank=True) + created_at = models.DateTimeField(auto_now_add=True) + + class Meta: + db_table = "eudr_evidence" + + def __str__(self): + return f"{self.get_evidence_type_display()} - {self.case.batch.batch_reference}" + + +class RiskAssessment(models.Model): + id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False) + case = models.ForeignKey(DueDiligenceCase, on_delete=models.CASCADE, related_name="risk_assessments") + risk_level = models.CharField(max_length=20, choices=DueDiligenceCase.RiskLevel.choices, default=DueDiligenceCase.RiskLevel.STANDARD) + basis = models.TextField(blank=True) + findings = models.TextField(blank=True) + assessed_by = models.ForeignKey( + settings.AUTH_USER_MODEL, on_delete=models.SET_NULL, null=True, blank=True, related_name="risk_assessments" + ) + assessed_at = models.DateTimeField(auto_now_add=True) + + class Meta: + db_table = "risk_assessments" + ordering = ["-assessed_at"] + + def __str__(self): + return f"Risk {self.risk_level} - {self.case.batch.batch_reference}" + + +class MitigationAction(models.Model): + class Status(models.TextChoices): + OPEN = "open", "Open" + IN_PROGRESS = "in_progress", "In Progress" + COMPLETED = "completed", "Completed" + VERIFIED = "verified", "Verified" + + id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False) + case = models.ForeignKey(DueDiligenceCase, on_delete=models.CASCADE, related_name="mitigation_actions") + action = models.TextField() + status = models.CharField(max_length=20, choices=Status.choices, default=Status.OPEN) + owner = models.CharField(max_length=255, blank=True) + due_date = models.DateField(null=True, blank=True) + evidence_file = models.CharField(max_length=255, blank=True) + completed_at = models.DateTimeField(null=True, blank=True) + created_at = models.DateTimeField(auto_now_add=True) + + class Meta: + db_table = "mitigation_actions" + + def __str__(self): + return f"Mitigation - {self.case.batch.batch_reference} ({self.status})" + + +class DueDiligenceStatementReference(models.Model): + class ReconciliationStatus(models.TextChoices): + PENDING = "pending", "Pending" + RECONCILED = "reconciled", "Reconciled" + + id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False) + case = models.ForeignKey(DueDiligenceCase, on_delete=models.CASCADE, related_name="statement_references") + dds_reference = models.CharField(max_length=255, blank=True) + submission_date = models.DateField(null=True, blank=True) + submitted_by = models.ForeignKey( + settings.AUTH_USER_MODEL, on_delete=models.SET_NULL, null=True, blank=True, related_name="dds_submissions" + ) + reconciliation_status = models.CharField( + max_length=20, choices=ReconciliationStatus.choices, default=ReconciliationStatus.PENDING + ) + notes = models.TextField(blank=True) + created_at = models.DateTimeField(auto_now_add=True) + + class Meta: + db_table = "due_diligence_statement_references" + + def __str__(self): + return f"DDS {self.dds_reference or '-'} - {self.case.batch.batch_reference}" diff --git a/apps/compliance/permissions.py b/apps/compliance/permissions.py new file mode 100644 index 0000000..f0b905a --- /dev/null +++ b/apps/compliance/permissions.py @@ -0,0 +1,13 @@ +from rest_framework import permissions + + +class IsOpsOrAdmin(permissions.BasePermission): + def has_permission(self, request, view): + user = request.user + return bool(user and user.is_authenticated and (user.role in ("ops", "admin") or user.is_superuser)) + + +class IsAdmin(permissions.BasePermission): + def has_permission(self, request, view): + user = request.user + return bool(user and user.is_authenticated and (user.role == "admin" or user.is_superuser)) diff --git a/apps/compliance/serializers.py b/apps/compliance/serializers.py new file mode 100644 index 0000000..4ae3f1d --- /dev/null +++ b/apps/compliance/serializers.py @@ -0,0 +1,122 @@ +from rest_framework import serializers + +from .models import ( + ProductSpecification, SpecParameter, + Facility, OriginZone, CollectorGroup, HarvestIntake, IntakeBatchLineage, + DueDiligenceCase, EUDREvidence, RiskAssessment, MitigationAction, + DueDiligenceStatementReference, +) + + +class SpecParameterSerializer(serializers.ModelSerializer): + class Meta: + model = SpecParameter + fields = "__all__" + read_only_fields = ["id"] + + +class ProductSpecificationSerializer(serializers.ModelSerializer): + parameters = SpecParameterSerializer(many=True, required=False) + approved_by_name = serializers.CharField(source="approved_by.full_name", read_only=True) + + class Meta: + model = ProductSpecification + fields = "__all__" + read_only_fields = ["id", "approved_by", "approved_at", "created_at", "updated_at"] + + +class FacilitySerializer(serializers.ModelSerializer): + class Meta: + model = Facility + fields = "__all__" + read_only_fields = ["id", "created_at"] + + +class OriginZoneSerializer(serializers.ModelSerializer): + facility_name = serializers.CharField(source="facility.name", read_only=True) + verified_by_name = serializers.CharField(source="verified_by.full_name", read_only=True) + + class Meta: + model = OriginZone + fields = "__all__" + read_only_fields = ["id", "verified_by", "verified_at", "created_at"] + + +class CollectorGroupSerializer(serializers.ModelSerializer): + class Meta: + model = CollectorGroup + fields = "__all__" + read_only_fields = ["id", "created_at"] + + +class HarvestIntakeSerializer(serializers.ModelSerializer): + origin_zone_code = serializers.CharField(source="origin_zone.zone_code", read_only=True) + collector_group_name = serializers.CharField(source="collector_group.name", read_only=True) + + class Meta: + model = HarvestIntake + fields = "__all__" + read_only_fields = ["id", "created_at"] + + +class IntakeBatchLineageSerializer(serializers.ModelSerializer): + batch_reference = serializers.CharField(source="batch.batch_reference", read_only=True) + intake_reference = serializers.CharField(source="harvest_intake.intake_reference", read_only=True) + + class Meta: + model = IntakeBatchLineage + fields = "__all__" + read_only_fields = ["id", "created_at"] + + +class EUDREvidenceSerializer(serializers.ModelSerializer): + verified_by_name = serializers.CharField(source="verified_by.full_name", read_only=True) + + class Meta: + model = EUDREvidence + fields = "__all__" + read_only_fields = ["id", "verified_by", "verified_at", "created_at"] + + +class RiskAssessmentSerializer(serializers.ModelSerializer): + assessed_by_name = serializers.CharField(source="assessed_by.full_name", read_only=True) + + class Meta: + model = RiskAssessment + fields = "__all__" + read_only_fields = ["id", "assessed_at", "assessed_by"] + + +class MitigationActionSerializer(serializers.ModelSerializer): + class Meta: + model = MitigationAction + fields = "__all__" + read_only_fields = ["id", "created_at"] + + +class DueDiligenceStatementReferenceSerializer(serializers.ModelSerializer): + submitted_by_name = serializers.CharField(source="submitted_by.full_name", read_only=True) + + class Meta: + model = DueDiligenceStatementReference + fields = "__all__" + read_only_fields = ["id", "submitted_by", "created_at"] + + +class DueDiligenceCaseSerializer(serializers.ModelSerializer): + batch_reference = serializers.CharField(source="batch.batch_reference", read_only=True) + approved_by_name = serializers.CharField(source="approved_by.full_name", read_only=True) + assessed_by_name = serializers.CharField(source="assessed_by.full_name", read_only=True) + evidence = EUDREvidenceSerializer(many=True, read_only=True) + risk_assessments = RiskAssessmentSerializer(many=True, read_only=True) + mitigation_actions = MitigationActionSerializer(many=True, read_only=True) + statement_references = DueDiligenceStatementReferenceSerializer(many=True, read_only=True) + + class Meta: + model = DueDiligenceCase + fields = "__all__" + read_only_fields = [ + "id", "status", "risk_level", "completeness_results", "assessed_by", + "assessed_at", "approved_by", "approved_at", "rejected_reason", + "created_at", "updated_at", + ] diff --git a/apps/compliance/services.py b/apps/compliance/services.py new file mode 100644 index 0000000..ee219c5 --- /dev/null +++ b/apps/compliance/services.py @@ -0,0 +1,372 @@ +from decimal import Decimal + +from django.db import transaction +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 _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() + ], + } diff --git a/apps/compliance/tests.py b/apps/compliance/tests.py new file mode 100644 index 0000000..6b77269 --- /dev/null +++ b/apps/compliance/tests.py @@ -0,0 +1,305 @@ +import os +import tempfile +from decimal import Decimal + +from django.contrib.auth import get_user_model +from django.core.files.uploadedfile import SimpleUploadedFile +from django.test import TestCase, override_settings +from django.urls import reverse +from django.utils import timezone +from rest_framework.test import APIClient + +from apps.batches.models import ProcessingBatch +from apps.capa.models import CAPARecord +from apps.documents.models import CertificateOfAnalysis, Document +from apps.notifications.models import Notification +from apps.qc.models import QCResult + +from .models import ( + CollectorGroup, DueDiligenceCase, Facility, HarvestIntake, IntakeBatchLineage, + OriginZone, ProductSpecification, SpecParameter, +) +from . import services + +User = get_user_model() + + +def make_user(email, role): + # Mirrors the Keycloak OIDC backend, which creates all users as is_staff=True. + return User.objects.create_user( + email=email, password="pass1234", role=role, + full_name=email.split("@")[0], is_staff=True, + ) + + +def make_batch(user): + return ProcessingBatch.objects.create( + batch_reference=f"IXG-BATCH-TEST-{ProcessingBatch.objects.count() + 1}", + grade="A", + weight_kg=Decimal("1000.00"), + origin_facility="ASI - Katsina State, Nigeria", + created_by=user, + ) + + +class SpecEvaluationTests(TestCase): + def setUp(self): + self.spec = ProductSpecification.objects.create( + product="Unrefined Shea Butter", version="9.9", status="approved", is_current=True + ) + SpecParameter.objects.create(spec=self.spec, parameter="ffa", unit="%", max_value=Decimal("3.00"), sort_order=1) + SpecParameter.objects.create(spec=self.spec, parameter="moisture", unit="%", max_value=Decimal("0.10"), sort_order=2) + SpecParameter.objects.create(spec=self.spec, parameter="peroxide", unit="meq O2/kg", max_value=Decimal("10.00"), sort_order=3) + SpecParameter.objects.create(spec=self.spec, parameter="visual", unit="grade", allowed_values="A,B,C", sort_order=4) + + def test_pass_within_spec(self): + values = {"ffa": "2.0", "moisture": "0.05", "peroxide": "5.0", "visual_grade": "A"} + overall_pass, result_status, _, version = services.evaluate_qc_data(values, spec=self.spec) + self.assertTrue(overall_pass) + self.assertEqual(result_status, QCResult.ResultStatus.WITHIN_SPEC) + self.assertEqual(version, "9.9") + + def test_fail_outside_spec(self): + values = {"ffa": "5.0", "moisture": "0.05", "peroxide": "5.0", "visual_grade": "A"} + overall_pass, result_status, _, _ = services.evaluate_qc_data(values, spec=self.spec) + self.assertFalse(overall_pass) + self.assertEqual(result_status, QCResult.ResultStatus.OUTSIDE_SPEC) + + def test_fail_visual_grade(self): + values = {"ffa": "2.0", "moisture": "0.05", "peroxide": "5.0", "visual_grade": "Fail"} + overall_pass, result_status, _, _ = services.evaluate_qc_data(values, spec=self.spec) + self.assertFalse(overall_pass) + + def test_fallback_legacy_limits_when_no_spec(self): + values = {"ffa": "2.0", "moisture": "0.05", "peroxide": "5.0", "visual_grade": "A"} + overall_pass, result_status, _, version = services.evaluate_qc_data(values, spec=None) + self.assertTrue(overall_pass) + self.assertEqual(version, "") + + +class QCWorkflowTests(TestCase): + def setUp(self): + self.ops_a = make_user("ops.a@ixg.local", "ops") + self.ops_b = make_user("ops.b@ixg.local", "ops") + self.buyer = make_user("buyer@ixg.local", "buyer") + self.batch = make_batch(self.ops_a) + self.client = APIClient() + + def create_qc(self, values, user=None): + self.client.force_authenticate(user=user or self.ops_a) + payload = { + "batch": str(self.batch.id), + "ffa_percentage": values[0], + "moisture_percentage": values[1], + "peroxide_value": values[2], + "visual_grade": values[3], + } + return self.client.post("/api/qc-results/", payload, format="json") + + def test_qc_fail_creates_capa_and_quarantine(self): + resp = self.create_qc(["9.0", "0.9", "20", "Fail"]) + self.assertEqual(resp.status_code, 201) + qc = QCResult.objects.get(batch=self.batch) + self.assertFalse(qc.overall_pass) + self.assertTrue(qc.is_quarantine) + self.assertEqual(qc.entered_by, self.ops_a) + self.assertEqual(qc.result_status, QCResult.ResultStatus.OUTSIDE_SPEC) + self.batch.refresh_from_db() + self.assertEqual(self.batch.status, ProcessingBatch.Status.QC_FAIL) + self.assertTrue(CAPARecord.objects.filter(batch=self.batch).exists()) + + def test_qc_pass_sets_batch_qc_pass(self): + resp = self.create_qc(["2.0", "0.05", "5.0", "A"]) + self.assertEqual(resp.status_code, 201) + qc = QCResult.objects.get(batch=self.batch) + self.assertTrue(qc.overall_pass) + self.batch.refresh_from_db() + self.assertEqual(self.batch.status, ProcessingBatch.Status.QC_PASS) + + def test_verify_four_eyes_blocks_submitter(self): + qc = QCResult.objects.create( + batch=self.batch, ffa_percentage=Decimal("2.0"), moisture_percentage=Decimal("0.05"), + peroxide_value=Decimal("5.0"), visual_grade="A", overall_pass=True, entered_by=self.ops_a, + ) + self.client.force_authenticate(user=self.ops_a) + resp = self.client.post(f"/api/qc-results/{qc.id}/verify/", {"approved": "true"}, format="json") + self.assertEqual(resp.status_code, 400) + qc.refresh_from_db() + self.assertEqual(qc.verification_status, QCResult.VerificationStatus.DRAFT) + + def test_verify_approval_generates_coa(self): + qc = QCResult.objects.create( + batch=self.batch, ffa_percentage=Decimal("2.0"), moisture_percentage=Decimal("0.05"), + peroxide_value=Decimal("5.0"), visual_grade="A", overall_pass=True, entered_by=self.ops_a, + spec_version="1.0", + ) + self.client.force_authenticate(user=self.ops_b) + resp = self.client.post(f"/api/qc-results/{qc.id}/verify/", {"approved": "true"}, format="json") + self.assertEqual(resp.status_code, 200) + qc.refresh_from_db() + self.assertEqual(qc.verification_status, QCResult.VerificationStatus.APPROVED) + self.assertEqual(qc.verified_by, self.ops_b) + coa = CertificateOfAnalysis.objects.get(qc_result=qc) + self.assertEqual(coa.status, CertificateOfAnalysis.Status.PENDING_APPROVAL) + self.assertTrue(coa.sha256_hash) + self.assertTrue(os.path.exists(coa.file_url.replace("/media/", "media/"))) + + @override_settings(MEDIA_ROOT=tempfile.mkdtemp(prefix="ixg_test_media_")) + def test_publish_coa_creates_buyer_document_and_notification(self): + qc = QCResult.objects.create( + batch=self.batch, ffa_percentage=Decimal("2.0"), moisture_percentage=Decimal("0.05"), + peroxide_value=Decimal("5.0"), visual_grade="A", overall_pass=True, entered_by=self.ops_a, + spec_version="1.0", + ) + coa = services.create_coa_for_qc(qc, approved_by=self.ops_b) + services.publish_coa(coa, released_by=self.ops_b) + coa.refresh_from_db() + self.assertEqual(coa.status, CertificateOfAnalysis.Status.PUBLISHED) + buyer_doc = Document.objects.filter(batch=self.batch, doc_type="coa", visibility="buyer") + self.assertEqual(buyer_doc.count(), 1) + self.assertEqual(buyer_doc.first().file_url, coa.file_url) + self.assertTrue(Notification.objects.filter(user=self.buyer, type="coa_published").exists()) + + def test_buyer_verification_forbidden(self): + qc = QCResult.objects.create( + batch=self.batch, ffa_percentage=Decimal("2.0"), moisture_percentage=Decimal("0.05"), + peroxide_value=Decimal("5.0"), visual_grade="A", overall_pass=True, entered_by=self.ops_a, + ) + self.client.force_authenticate(user=self.buyer) + resp = self.client.post(f"/api/qc-results/{qc.id}/verify/", {"approved": "true"}, format="json") + self.assertEqual(resp.status_code, 403) + + +class EUDRWorkflowTests(TestCase): + def setUp(self): + self.ops = make_user("ops.eudr@ixg.local", "ops") + self.buyer = make_user("buyer.eudr@ixg.local", "buyer") + self.batch = make_batch(self.ops) + + def test_completeness_missing_lineage_blocks(self): + case = DueDiligenceCase.objects.create(batch=self.batch) + services.run_completeness_check(case) + case.refresh_from_db() + self.assertEqual(case.status, DueDiligenceCase.Status.INCOMPLETE) + self.assertFalse(case.completeness_results["geolocation"]["pass"]) + self.assertFalse(case.completeness_results["lineage"]["pass"]) + + def test_completeness_unverified_zone_blocks(self): + facility = Facility.objects.create(name="ASI Katsina", country="Nigeria", region="Katsina") + zone = OriginZone.objects.create( + facility=facility, zone_code="KAT-01", country="Nigeria", region="Katsina", + verification_status=OriginZone.VerificationStatus.UNVERIFIED, + ) + group = CollectorGroup.objects.create(name="Group A", origin_zone=zone) + intake = HarvestIntake.objects.create( + intake_reference="INT-001", collector_group=group, origin_zone=zone, + intake_date="2026-01-10", weight_kg=Decimal("500.00"), + ) + IntakeBatchLineage.objects.create(batch=self.batch, harvest_intake=intake, weight_allocated_kg=Decimal("500.00")) + case = services.ensure_due_diligence_case(self.batch) + services.run_completeness_check(case) + case.refresh_from_db() + self.assertEqual(case.status, DueDiligenceCase.Status.INCOMPLETE) + self.assertFalse(case.completeness_results["geolocation"]["pass"]) + + def test_verified_zone_with_lineage_passes_geolocation(self): + facility = Facility.objects.create(name="ASI Katsina", country="Nigeria", region="Katsina") + zone = OriginZone.objects.create( + facility=facility, zone_code="KAT-02", country="Nigeria", region="Katsina", + verification_status=OriginZone.VerificationStatus.VERIFIED, + ) + group = CollectorGroup.objects.create(name="Group B", origin_zone=zone) + intake = HarvestIntake.objects.create( + intake_reference="INT-002", collector_group=group, origin_zone=zone, + intake_date="2026-01-10", weight_kg=Decimal("500.00"), + ) + IntakeBatchLineage.objects.create(batch=self.batch, harvest_intake=intake, weight_allocated_kg=Decimal("500.00")) + case = services.ensure_due_diligence_case(self.batch) + services.run_completeness_check(case) + case.refresh_from_db() + self.assertTrue(case.completeness_results["geolocation"]["pass"]) + self.assertTrue(case.completeness_results["lineage"]["pass"]) + + def test_risk_high_without_lineage(self): + case = DueDiligenceCase.objects.create(batch=self.batch) + risk = services.evaluate_risk(case, user=self.ops) + case.refresh_from_db() + self.assertEqual(risk.risk_level, "high") + self.assertEqual(case.risk_level, DueDiligenceCase.RiskLevel.HIGH) + self.assertEqual(case.status, DueDiligenceCase.Status.MITIGATION_REQUIRED) + + def test_approve_case_requires_complete_state(self): + case = DueDiligenceCase.objects.create(batch=self.batch) + with self.assertRaises(ValueError): + services.approve_case(case, self.ops) + + def test_record_submission(self): + case = DueDiligenceCase.objects.create(batch=self.batch) + ref = services.record_submission(case, self.ops, "DD-2026-0001", "2026-08-05") + self.assertEqual(ref.dds_reference, "DD-2026-0001") + self.assertEqual(case.statement_references.count(), 1) + + +class BuyerIsolationTests(TestCase): + def setUp(self): + self.buyer = make_user("iso.buyer@ixg.local", "buyer") + self.ops = make_user("iso.ops@ixg.local", "ops") + self.client = APIClient() + + def test_buyer_cannot_access_compliance_api(self): + self.client.force_authenticate(user=self.buyer) + resp = self.client.get("/api/product-specs/") + self.assertEqual(resp.status_code, 403) + resp = self.client.get("/api/eudr-cases/") + self.assertEqual(resp.status_code, 403) + + def test_ops_can_access_compliance_api(self): + self.client.force_authenticate(user=self.ops) + resp = self.client.get("/api/product-specs/") + self.assertEqual(resp.status_code, 200) + resp = self.client.get("/api/eudr-cases/") + self.assertEqual(resp.status_code, 200) + + def test_buyer_forbidden_from_ops(self): + self.client.force_login(self.buyer) + resp = self.client.get("/ops/") + self.assertEqual(resp.status_code, 403) + + +class PageRenderTests(TestCase): + def setUp(self): + self.ops = make_user("page.ops@ixg.local", "ops") + self.buyer = make_user("page.buyer@ixg.local", "buyer") + self.batch = make_batch(self.ops) + self.qc = QCResult.objects.create( + batch=self.batch, ffa_percentage=Decimal("2.0"), moisture_percentage=Decimal("0.05"), + peroxide_value=Decimal("5.0"), visual_grade="A", overall_pass=True, + entered_by=self.ops, spec_version="1.0", + ) + self.coa = services.create_coa_for_qc(self.qc, approved_by=self.ops) + self.case = DueDiligenceCase.objects.create(batch=self.batch) + + def test_ops_pages_render(self): + self.client.force_login(self.ops) + for url in [ + "/ops/qc/verify/", + f"/ops/qc/verify/{self.qc.id}/", + "/ops/coa/", + f"/ops/coa/{self.coa.id}/", + "/ops/eudr/", + f"/ops/eudr/{self.case.id}/", + ]: + resp = self.client.get(url) + self.assertEqual(resp.status_code, 200, msg=f"GET {url}") + + def test_buyer_compliance_page_renders(self): + self.client.force_login(self.buyer) + resp = self.client.get("/portal/compliance/") + self.assertEqual(resp.status_code, 200) + + def test_ops_post_publish_coa(self): + self.client.force_login(self.ops) + resp = self.client.post(f"/ops/coa/{self.coa.id}/", {"action": "approve"}) + self.assertEqual(resp.status_code, 302) + self.coa.refresh_from_db() + self.assertEqual(self.coa.status, CertificateOfAnalysis.Status.PUBLISHED) diff --git a/apps/compliance/urls.py b/apps/compliance/urls.py new file mode 100644 index 0000000..e4d9b51 --- /dev/null +++ b/apps/compliance/urls.py @@ -0,0 +1,20 @@ +from django.urls import path, include +from rest_framework.routers import DefaultRouter +from . import views + +router = DefaultRouter() +router.register("product-specs", views.ProductSpecificationViewSet) +router.register("facilities", views.FacilityViewSet) +router.register("origin-zones", views.OriginZoneViewSet) +router.register("collector-groups", views.CollectorGroupViewSet) +router.register("harvest-intakes", views.HarvestIntakeViewSet) +router.register("intake-batch-lineage", views.IntakeBatchLineageViewSet) +router.register("eudr-cases", views.DueDiligenceCaseViewSet) +router.register("eudr-evidence", views.EUDREvidenceViewSet) +router.register("risk-assessments", views.RiskAssessmentViewSet) +router.register("mitigation-actions", views.MitigationActionViewSet) +router.register("dds-references", views.DueDiligenceStatementReferenceViewSet) + +urlpatterns = [ + path("", include(router.urls)), +] diff --git a/apps/compliance/views.py b/apps/compliance/views.py new file mode 100644 index 0000000..85b73e1 --- /dev/null +++ b/apps/compliance/views.py @@ -0,0 +1,170 @@ +from rest_framework import viewsets, permissions, status +from rest_framework.decorators import action +from rest_framework.response import Response + +from .models import ( + ProductSpecification, Facility, OriginZone, CollectorGroup, HarvestIntake, + IntakeBatchLineage, DueDiligenceCase, EUDREvidence, RiskAssessment, + MitigationAction, DueDiligenceStatementReference, +) +from .serializers import ( + ProductSpecificationSerializer, FacilitySerializer, OriginZoneSerializer, + CollectorGroupSerializer, HarvestIntakeSerializer, IntakeBatchLineageSerializer, + DueDiligenceCaseSerializer, EUDREvidenceSerializer, RiskAssessmentSerializer, + MitigationActionSerializer, DueDiligenceStatementReferenceSerializer, +) +from .permissions import IsOpsOrAdmin, IsAdmin +from . import services + + +class ProductSpecificationViewSet(viewsets.ModelViewSet): + queryset = ProductSpecification.objects.prefetch_related("parameters").all() + serializer_class = ProductSpecificationSerializer + permission_classes = [IsOpsOrAdmin] + + def get_permissions(self): + if self.action in ("create", "update", "partial_update", "destroy"): + return [permission() for permission in [IsAdmin]] + return [permission() for permission in [IsOpsOrAdmin]] + + +class FacilityViewSet(viewsets.ModelViewSet): + queryset = Facility.objects.all() + serializer_class = FacilitySerializer + permission_classes = [IsOpsOrAdmin] + + +class OriginZoneViewSet(viewsets.ModelViewSet): + queryset = OriginZone.objects.select_related("facility", "verified_by").all() + serializer_class = OriginZoneSerializer + permission_classes = [IsOpsOrAdmin] + + @action(detail=True, methods=["post"]) + def verify(self, request, pk=None): + zone = self.get_object() + outcome = request.data.get("verified", "true") + verified = str(outcome).lower() in ("true", "1", "yes") + zone.verification_status = "verified" if verified else "rejected" + zone.verified_by = request.user + from django.utils import timezone + zone.verified_at = timezone.now() + zone.save() + return Response(OriginZoneSerializer(zone).data) + + +class CollectorGroupViewSet(viewsets.ModelViewSet): + queryset = CollectorGroup.objects.all() + serializer_class = CollectorGroupSerializer + permission_classes = [IsOpsOrAdmin] + + +class HarvestIntakeViewSet(viewsets.ModelViewSet): + queryset = HarvestIntake.objects.select_related("origin_zone", "collector_group").all() + serializer_class = HarvestIntakeSerializer + permission_classes = [IsOpsOrAdmin] + + +class IntakeBatchLineageViewSet(viewsets.ModelViewSet): + queryset = IntakeBatchLineage.objects.select_related("batch", "harvest_intake").all() + serializer_class = IntakeBatchLineageSerializer + permission_classes = [IsOpsOrAdmin] + + def perform_create(self, serializer): + lineage = serializer.save() + services.ensure_due_diligence_case(lineage.batch) + + +class EUDREvidenceViewSet(viewsets.ModelViewSet): + queryset = EUDREvidence.objects.select_related("case", "verified_by").all() + serializer_class = EUDREvidenceSerializer + permission_classes = [IsOpsOrAdmin] + + def perform_create(self, serializer): + evidence = serializer.save() + case = evidence.case + services.run_completeness_check(case) + + +class RiskAssessmentViewSet(viewsets.ModelViewSet): + queryset = RiskAssessment.objects.select_related("case", "assessed_by").all() + serializer_class = RiskAssessmentSerializer + permission_classes = [IsOpsOrAdmin] + + +class MitigationActionViewSet(viewsets.ModelViewSet): + queryset = MitigationAction.objects.select_related("case").all() + serializer_class = MitigationActionSerializer + permission_classes = [IsOpsOrAdmin] + + @action(detail=True, methods=["post"]) + def complete(self, request, pk=None): + mitigation = self.get_object() + mitigation.status = MitigationAction.Status.VERIFIED + from django.utils import timezone + mitigation.completed_at = timezone.now() + mitigation.save() + services.run_completeness_check(mitigation.case) + return Response(MitigationActionSerializer(mitigation).data) + + +class DueDiligenceStatementReferenceViewSet(viewsets.ModelViewSet): + queryset = DueDiligenceStatementReference.objects.select_related("case", "submitted_by").all() + serializer_class = DueDiligenceStatementReferenceSerializer + permission_classes = [IsOpsOrAdmin] + + +class DueDiligenceCaseViewSet(viewsets.ModelViewSet): + queryset = DueDiligenceCase.objects.select_related( + "batch", "approved_by", "assessed_by" + ).prefetch_related("evidence", "risk_assessments", "mitigation_actions", "statement_references").order_by("-updated_at").all() + serializer_class = DueDiligenceCaseSerializer + permission_classes = [IsOpsOrAdmin] + + def perform_create(self, serializer): + serializer.save() + + @action(detail=True, methods=["post"]) + def check(self, request, pk=None): + case = self.get_object() + services.run_completeness_check(case) + return Response(DueDiligenceCaseSerializer(case).data) + + @action(detail=True, methods=["post"]) + def assess_risk(self, request, pk=None): + case = self.get_object() + risk = services.evaluate_risk(case, user=request.user) + return Response({"risk_assessment_id": str(risk.id), **RiskAssessmentSerializer(risk).data}) + + @action(detail=True, methods=["post"]) + def approve(self, request, pk=None): + case = self.get_object() + try: + services.approve_case(case, request.user) + except ValueError as exc: + return Response({"detail": str(exc)}, status=status.HTTP_400_BAD_REQUEST) + return Response(DueDiligenceCaseSerializer(case).data) + + @action(detail=True, methods=["post"]) + def reject(self, request, pk=None): + case = self.get_object() + reason = request.data.get("reason", "") + try: + services.reject_case(case, request.user, reason) + except ValueError as exc: + return Response({"detail": str(exc)}, status=status.HTTP_400_BAD_REQUEST) + return Response(DueDiligenceCaseSerializer(case).data) + + @action(detail=True, methods=["post"]) + def record_submission(self, request, pk=None): + case = self.get_object() + dds_reference = request.data.get("dds_reference", "") + submission_date = request.data.get("submission_date") + if not dds_reference: + return Response({"detail": "dds_reference is required"}, status=status.HTTP_400_BAD_REQUEST) + ref = services.record_submission(case, request.user, dds_reference, submission_date) + return Response(DueDiligenceStatementReferenceSerializer(ref).data) + + @action(detail=True, methods=["get"]) + def evidence_package(self, request, pk=None): + case = self.get_object() + return Response(services.evidence_package(case)) diff --git a/apps/documents/migrations/0002_certificateofanalysis_published_at_and_more.py b/apps/documents/migrations/0002_certificateofanalysis_published_at_and_more.py new file mode 100644 index 0000000..ddecf30 --- /dev/null +++ b/apps/documents/migrations/0002_certificateofanalysis_published_at_and_more.py @@ -0,0 +1,87 @@ +# Generated by Django 5.2.12 on 2026-08-05 13:36 + +import django.db.models.deletion +from django.conf import settings +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ('documents', '0001_initial'), + ('qc', '0002_alter_qcresult_options_qcresult_approved_at_and_more'), + migrations.swappable_dependency(settings.AUTH_USER_MODEL), + ] + + operations = [ + migrations.AddField( + model_name='certificateofanalysis', + name='published_at', + field=models.DateTimeField(blank=True, null=True), + ), + migrations.AddField( + model_name='certificateofanalysis', + name='qc_result', + field=models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='certificates', to='qc.qcresult'), + ), + migrations.AddField( + model_name='certificateofanalysis', + name='quality_approved_at', + field=models.DateTimeField(blank=True, null=True), + ), + migrations.AddField( + model_name='certificateofanalysis', + name='quality_approved_by', + field=models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='coas_quality_approved', to=settings.AUTH_USER_MODEL), + ), + migrations.AddField( + model_name='certificateofanalysis', + name='released_at', + field=models.DateTimeField(blank=True, null=True), + ), + migrations.AddField( + model_name='certificateofanalysis', + name='released_by', + field=models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='coas_released', to=settings.AUTH_USER_MODEL), + ), + migrations.AddField( + model_name='certificateofanalysis', + name='sha256_hash', + field=models.CharField(blank=True, max_length=64), + ), + migrations.AddField( + model_name='certificateofanalysis', + name='source_document', + field=models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='source_for_coa', to='documents.document'), + ), + migrations.AddField( + model_name='certificateofanalysis', + name='spec_version', + field=models.CharField(blank=True, max_length=20), + ), + migrations.AddField( + model_name='certificateofanalysis', + name='status', + field=models.CharField(choices=[('pending_approval', 'Pending Approval'), ('published', 'Published'), ('superseded', 'Superseded')], default='pending_approval', max_length=20), + ), + migrations.AddField( + model_name='certificateofanalysis', + name='superseded_at', + field=models.DateTimeField(blank=True, null=True), + ), + migrations.AddField( + model_name='certificateofanalysis', + name='superseded_by', + field=models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='supersedes', to='documents.certificateofanalysis'), + ), + migrations.AddField( + model_name='certificateofanalysis', + name='version', + field=models.IntegerField(default=1), + ), + migrations.AlterField( + model_name='document', + name='doc_type', + field=models.CharField(choices=[('coa', 'Certificate of Analysis'), ('lab_report', 'Laboratory Report'), ('regulatory', 'Regulatory'), ('invoice', 'Invoice'), ('packing_list', 'Packing List'), ('bl', 'Bill of Lading'), ('insurance', 'Insurance'), ('other', 'Other')], max_length=20), + ), + ] diff --git a/apps/documents/models.py b/apps/documents/models.py index 6c2ae50..f1206b1 100644 --- a/apps/documents/models.py +++ b/apps/documents/models.py @@ -4,8 +4,16 @@ from django.db import models class CertificateOfAnalysis(models.Model): + class Status(models.TextChoices): + PENDING_APPROVAL = "pending_approval", "Pending Approval" + PUBLISHED = "published", "Published" + SUPERSEDED = "superseded", "Superseded" + id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False) batch = models.ForeignKey("batches.ProcessingBatch", on_delete=models.CASCADE, related_name="certificates_of_analysis") + qc_result = models.ForeignKey( + "qc.QCResult", on_delete=models.SET_NULL, null=True, blank=True, related_name="certificates" + ) file_url = models.TextField() file_name = models.CharField(max_length=255) issuer = models.CharField(max_length=255, blank=True) @@ -14,6 +22,27 @@ class CertificateOfAnalysis(models.Model): uploaded_by = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.SET_NULL, null=True) uploaded_at = models.DateTimeField(auto_now_add=True) + status = models.CharField(max_length=20, choices=Status.choices, default=Status.PENDING_APPROVAL) + version = models.IntegerField(default=1) + sha256_hash = models.CharField(max_length=64, blank=True) + spec_version = models.CharField(max_length=20, blank=True) + source_document = models.ForeignKey( + "Document", on_delete=models.SET_NULL, null=True, blank=True, related_name="source_for_coa" + ) + quality_approved_by = models.ForeignKey( + settings.AUTH_USER_MODEL, on_delete=models.SET_NULL, null=True, blank=True, related_name="coas_quality_approved" + ) + quality_approved_at = models.DateTimeField(null=True, blank=True) + released_by = models.ForeignKey( + settings.AUTH_USER_MODEL, on_delete=models.SET_NULL, null=True, blank=True, related_name="coas_released" + ) + released_at = models.DateTimeField(null=True, blank=True) + published_at = models.DateTimeField(null=True, blank=True) + superseded_at = models.DateTimeField(null=True, blank=True) + superseded_by = models.ForeignKey( + "self", on_delete=models.SET_NULL, null=True, blank=True, related_name="supersedes" + ) + class Meta: db_table = "certificates_of_analysis" verbose_name_plural = "Certificates of Analysis" @@ -52,6 +81,7 @@ class RegulatoryCert(models.Model): class Document(models.Model): class DocType(models.TextChoices): COA = "coa", "Certificate of Analysis" + LAB_REPORT = "lab_report", "Laboratory Report" REGULATORY = "regulatory", "Regulatory" INVOICE = "invoice", "Invoice" PACKING_LIST = "packing_list", "Packing List" diff --git a/apps/documents/views.py b/apps/documents/views.py index 8f0009a..5f8a510 100644 --- a/apps/documents/views.py +++ b/apps/documents/views.py @@ -51,7 +51,7 @@ class DocumentViewSet(viewsets.ModelViewSet): doc = Document.objects.create( batch_id=batch_id, doc_type=doc_type, - file_url=f"/media/ixg-documents/{batch_id or uncategorized}/{filename}", + file_url=f"/media/ixg-documents/{batch_id or 'uncategorized'}/{filename}", file_name=file.name, file_size_bytes=file.size, visibility=visibility, @@ -62,10 +62,51 @@ class DocumentViewSet(viewsets.ModelViewSet): class CertificateOfAnalysisViewSet(viewsets.ModelViewSet): - queryset = CertificateOfAnalysis.objects.all() + queryset = CertificateOfAnalysis.objects.select_related("batch", "qc_result").all() serializer_class = CertificateOfAnalysisSerializer permission_classes = [permissions.IsAuthenticated] + def get_queryset(self): + user = self.request.user + qs = CertificateOfAnalysis.objects.select_related("batch", "qc_result").all() + if user.role == "buyer": + return qs.filter(status=CertificateOfAnalysis.Status.PUBLISHED) + return qs + + @action(detail=True, methods=["post"]) + def approve(self, request, pk=None): + """Commercial release approval: publishes the CoA and exposes it to buyers.""" + from apps.compliance.services import publish_coa + from apps.compliance.permissions import IsOpsOrAdmin + + if not IsOpsOrAdmin().has_permission(request, self): + return Response({"detail": "Not permitted"}, status=status.HTTP_403_FORBIDDEN) + coa = self.get_object() + if coa.status != CertificateOfAnalysis.Status.PENDING_APPROVAL: + return Response({"detail": "CoA is not pending approval"}, status=status.HTTP_400_BAD_REQUEST) + publish_coa(coa, released_by=request.user) + return Response(CertificateOfAnalysisSerializer(coa).data) + + @action(detail=True, methods=["post"]) + def supersede(self, request, pk=None): + from apps.compliance.permissions import IsOpsOrAdmin + from django.utils import timezone + + if not IsOpsOrAdmin().has_permission(request, self): + return Response({"detail": "Not permitted"}, status=status.HTTP_403_FORBIDDEN) + coa = self.get_object() + replacement_id = request.data.get("replacement_id") + if not replacement_id: + return Response({"detail": "replacement_id is required"}, status=status.HTTP_400_BAD_REQUEST) + replacement = CertificateOfAnalysis.objects.filter(id=replacement_id).first() + if not replacement: + return Response({"detail": "Replacement CoA not found"}, status=status.HTTP_400_BAD_REQUEST) + coa.status = CertificateOfAnalysis.Status.SUPERSEDED + coa.superseded_at = timezone.now() + coa.superseded_by = replacement + coa.save() + return Response(CertificateOfAnalysisSerializer(coa).data) + class RegulatoryCertViewSet(viewsets.ModelViewSet): queryset = RegulatoryCert.objects.all() diff --git a/apps/ops/urls.py b/apps/ops/urls.py index 3dcb5d2..51e2859 100644 --- a/apps/ops/urls.py +++ b/apps/ops/urls.py @@ -15,4 +15,10 @@ urlpatterns = [ path('capa//', views.CAPADetailView.as_view(), name='capa_detail'), path('documents/', views.DocumentListView.as_view(), name='document_list'), path('esg/', views.ESGListView.as_view(), name='esg_list'), + path('qc/verify/', views.QCVerifyListView.as_view(), name='qc_verify_list'), + path('qc/verify//', views.QCVerifyDetailView.as_view(), name='qc_verify_detail'), + path('coa/', views.COAReviewListView.as_view(), name='coa_review_list'), + path('coa//', views.COAReviewDetailView.as_view(), name='coa_review_detail'), + path('eudr/', views.EUDRCaseListView.as_view(), name='eudr_list'), + path('eudr//', views.EUDRCaseDetailView.as_view(), name='eudr_detail'), ] diff --git a/apps/ops/views.py b/apps/ops/views.py index 06a148c..0ad93e5 100644 --- a/apps/ops/views.py +++ b/apps/ops/views.py @@ -1,5 +1,6 @@ from django.contrib.auth.mixins import LoginRequiredMixin, UserPassesTestMixin from django.db.models import Count, Q +from django.shortcuts import redirect from django.utils import timezone from django.views.generic import TemplateView, ListView, DetailView from django.shortcuts import get_object_or_404 @@ -8,9 +9,10 @@ from apps.batches.models import ProcessingBatch from apps.qc.models import QCResult from apps.shipments.models import Shipment from apps.capa.models import CAPARecord -from apps.documents.models import Document +from apps.documents.models import Document, CertificateOfAnalysis from apps.esg.models import ESGFieldReport from apps.notifications.models import Notification +from apps.compliance.models import DueDiligenceCase class OpsRequiredMixin(UserPassesTestMixin): @@ -195,3 +197,176 @@ class ESGListView(OpsRequiredMixin, ListView): def get_queryset(self): return ESGFieldReport.objects.select_related('batch').all() + + +class QCVerifyListView(OpsRequiredMixin, ListView): + model = QCResult + template_name = 'ops/qc_verify_list.html' + context_object_name = 'results' + paginate_by = 25 + + def get_queryset(self): + qs = QCResult.objects.select_related('batch', 'entered_by').filter( + verification_status=QCResult.VerificationStatus.DRAFT + ) + passed = self.request.GET.get('passed') + if passed == '1': + qs = qs.filter(overall_pass=True) + elif passed == '0': + qs = qs.filter(overall_pass=False) + return qs + + def get_context_data(self, **kwargs): + ctx = super().get_context_data(**kwargs) + ctx['current_passed'] = self.request.GET.get('passed', '') + return ctx + + +class QCVerifyDetailView(OpsRequiredMixin, DetailView): + model = QCResult + template_name = 'ops/qc_verify_detail.html' + context_object_name = 'result' + + def post(self, request, *args, **kwargs): + from django.contrib import messages + from apps.compliance.services import create_coa_for_qc + + qc = self.get_object() + user = request.user + if qc.entered_by is not None and qc.entered_by_id == user.id: + messages.error(request, 'The submitter cannot approve their own QC result (four-eyes control).') + return redirect(self.request.path) + + approved = request.POST.get('approved') == '1' + if approved: + if not qc.overall_pass: + messages.error(request, 'A failed QC result cannot be approved; it remains quarantined with an open CAPA.') + return redirect(self.request.path) + qc.verification_status = QCResult.VerificationStatus.APPROVED + qc.verified_by = user + qc.approved_at = timezone.now() + qc.save() + coa = create_coa_for_qc(qc, approved_by=user) + messages.success(request, f'QC result approved. CoA created (status: {coa.status}).') + return redirect('ops:coa_review_detail', pk=coa.pk) + else: + qc.verification_status = QCResult.VerificationStatus.REJECTED + qc.verified_by = user + qc.approved_at = timezone.now() + qc.save() + messages.warning(request, 'QC result rejected.') + return redirect('ops:qc_verify_list') + + +class COAReviewListView(OpsRequiredMixin, ListView): + model = CertificateOfAnalysis + template_name = 'ops/coa_review_list.html' + context_object_name = 'coas' + paginate_by = 25 + + def get_queryset(self): + qs = CertificateOfAnalysis.objects.select_related('batch', 'qc_result', 'quality_approved_by').all() + status = self.request.GET.get('status') + if status: + qs = qs.filter(status=status) + else: + qs = qs.filter(status=CertificateOfAnalysis.Status.PENDING_APPROVAL) + return qs + + def get_context_data(self, **kwargs): + ctx = super().get_context_data(**kwargs) + ctx['status_choices'] = CertificateOfAnalysis.Status.choices + ctx['current_status'] = self.request.GET.get('status', '') + return ctx + + +class COAReviewDetailView(OpsRequiredMixin, DetailView): + model = CertificateOfAnalysis + template_name = 'ops/coa_review_detail.html' + context_object_name = 'coa' + + def get_context_data(self, **kwargs): + ctx = super().get_context_data(**kwargs) + ctx['lab_documents'] = Document.objects.filter(batch=self.object.batch, doc_type=Document.DocType.LAB_REPORT) + return ctx + + def post(self, request, *args, **kwargs): + from django.contrib import messages + from apps.compliance.services import publish_coa + + coa = self.get_object() + if request.POST.get('action') == 'approve': + try: + publish_coa(coa, released_by=request.user) + messages.success(request, 'CoA published to buyers.') + except ValueError as exc: + messages.error(request, str(exc)) + return redirect(self.request.path) + + +class EUDRCaseListView(OpsRequiredMixin, ListView): + model = DueDiligenceCase + template_name = 'ops/eudr_list.html' + context_object_name = 'cases' + paginate_by = 25 + + def get_queryset(self): + qs = DueDiligenceCase.objects.select_related('batch', 'approved_by').all() + status = self.request.GET.get('status') + if status: + qs = qs.filter(status=status) + return qs + + def get_context_data(self, **kwargs): + ctx = super().get_context_data(**kwargs) + ctx['status_choices'] = DueDiligenceCase.Status.choices + ctx['current_status'] = self.request.GET.get('status', '') + return ctx + + +class EUDRCaseDetailView(OpsRequiredMixin, DetailView): + model = DueDiligenceCase + template_name = 'ops/eudr_detail.html' + context_object_name = 'case' + + def get_context_data(self, **kwargs): + ctx = super().get_context_data(**kwargs) + case = self.object + ctx['lineage'] = list(case.batch.intake_lineage.select_related( + 'harvest_intake', 'harvest_intake__origin_zone', 'harvest_intake__collector_group' + )) + ctx['evidence'] = case.evidence.all() + ctx['risk_assessments'] = case.risk_assessments.all() + ctx['mitigation_actions'] = case.mitigation_actions.all() + ctx['statement_references'] = case.statement_references.all() + return ctx + + def post(self, request, *args, **kwargs): + from django.contrib import messages + from apps.compliance import services + + case = self.get_object() + action = request.POST.get('action') + try: + if action == 'check': + services.run_completeness_check(case) + messages.info(request, 'Completeness check completed.') + elif action == 'assess_risk': + services.evaluate_risk(case, user=request.user) + messages.info(request, 'Risk assessment recorded.') + elif action == 'approve': + services.approve_case(case, request.user) + messages.success(request, 'Case approved.') + elif action == 'reject': + services.reject_case(case, request.user, request.POST.get('reason', '')) + messages.warning(request, 'Case rejected.') + elif action == 'record_submission': + ref = request.POST.get('dds_reference') + if not ref: + messages.error(request, 'DDS reference required.') + else: + services.record_submission(case, request.user, ref, request.POST.get('submission_date') or None) + messages.success(request, 'External submission recorded.') + except ValueError as exc: + messages.error(request, str(exc)) + return redirect(self.request.path) diff --git a/apps/qc/migrations/0002_alter_qcresult_options_qcresult_approved_at_and_more.py b/apps/qc/migrations/0002_alter_qcresult_options_qcresult_approved_at_and_more.py new file mode 100644 index 0000000..dc616cf --- /dev/null +++ b/apps/qc/migrations/0002_alter_qcresult_options_qcresult_approved_at_and_more.py @@ -0,0 +1,75 @@ +# Generated by Django 5.2.12 on 2026-08-05 13:36 + +import django.db.models.deletion +from django.conf import settings +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ('qc', '0001_initial'), + migrations.swappable_dependency(settings.AUTH_USER_MODEL), + ] + + operations = [ + migrations.AlterModelOptions( + name='qcresult', + options={'ordering': ['-tested_at']}, + ), + migrations.AddField( + model_name='qcresult', + name='approved_at', + field=models.DateTimeField(blank=True, null=True), + ), + migrations.AddField( + model_name='qcresult', + name='entered_by', + field=models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='qc_entered', to=settings.AUTH_USER_MODEL), + ), + migrations.AddField( + model_name='qcresult', + name='is_quarantine', + field=models.BooleanField(default=False), + ), + migrations.AddField( + model_name='qcresult', + name='method_ffa', + field=models.CharField(blank=True, max_length=255), + ), + migrations.AddField( + model_name='qcresult', + name='method_moisture', + field=models.CharField(blank=True, max_length=255), + ), + migrations.AddField( + model_name='qcresult', + name='method_peroxide', + field=models.CharField(blank=True, max_length=255), + ), + migrations.AddField( + model_name='qcresult', + name='result_status', + field=models.CharField(choices=[('within_spec', 'Within Specification'), ('outside_spec', 'Outside Specification'), ('missing', 'Missing'), ('unit_mismatch', 'Unit Mismatch')], default='within_spec', max_length=20), + ), + migrations.AddField( + model_name='qcresult', + name='spec_version', + field=models.CharField(blank=True, max_length=20), + ), + migrations.AddField( + model_name='qcresult', + name='superseded_by', + field=models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='supersedes', to='qc.qcresult'), + ), + migrations.AddField( + model_name='qcresult', + name='verification_status', + field=models.CharField(choices=[('draft', 'Draft'), ('validated', 'Validated'), ('approved', 'Approved'), ('rejected', 'Rejected')], default='draft', max_length=20), + ), + migrations.AddField( + model_name='qcresult', + name='verified_by', + field=models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='qc_verified', to=settings.AUTH_USER_MODEL), + ), + ] diff --git a/apps/qc/models.py b/apps/qc/models.py index acfb594..b1d2111 100644 --- a/apps/qc/models.py +++ b/apps/qc/models.py @@ -1,4 +1,5 @@ import uuid +from django.conf import settings from django.db import models @@ -9,6 +10,18 @@ class QCResult(models.Model): C = "C", "C" FAIL = "Fail", "Fail" + class VerificationStatus(models.TextChoices): + DRAFT = "draft", "Draft" + VALIDATED = "validated", "Validated" + APPROVED = "approved", "Approved" + REJECTED = "rejected", "Rejected" + + class ResultStatus(models.TextChoices): + WITHIN_SPEC = "within_spec", "Within Specification" + OUTSIDE_SPEC = "outside_spec", "Outside Specification" + MISSING = "missing", "Missing" + UNIT_MISMATCH = "unit_mismatch", "Unit Mismatch" + id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False) batch = models.ForeignKey("batches.ProcessingBatch", on_delete=models.CASCADE, related_name="qc_results") ffa_percentage = models.DecimalField(max_digits=5, decimal_places=2) @@ -20,8 +33,32 @@ class QCResult(models.Model): tested_at = models.DateTimeField(auto_now_add=True) lab_reference = models.CharField(max_length=255, blank=True) + spec_version = models.CharField(max_length=20, blank=True) + entered_by = models.ForeignKey( + settings.AUTH_USER_MODEL, on_delete=models.SET_NULL, null=True, blank=True, related_name="qc_entered" + ) + verified_by = models.ForeignKey( + settings.AUTH_USER_MODEL, on_delete=models.SET_NULL, null=True, blank=True, related_name="qc_verified" + ) + verification_status = models.CharField( + max_length=20, choices=VerificationStatus.choices, default=VerificationStatus.DRAFT + ) + approved_at = models.DateTimeField(null=True, blank=True) + method_ffa = models.CharField(max_length=255, blank=True) + method_moisture = models.CharField(max_length=255, blank=True) + method_peroxide = models.CharField(max_length=255, blank=True) + result_status = models.CharField( + max_length=20, choices=ResultStatus.choices, default=ResultStatus.WITHIN_SPEC + ) + superseded_by = models.ForeignKey( + "self", on_delete=models.SET_NULL, null=True, blank=True, related_name="supersedes" + ) + is_quarantine = models.BooleanField(default=False) + class Meta: db_table = "qc_results" + ordering = ["-tested_at"] def __str__(self): - return f"QC for {self.batch.batch_reference}: {PASS if self.overall_pass else FAIL}" + status = "PASS" if self.overall_pass else "FAIL" + return f"QC for {self.batch.batch_reference}: {status}" diff --git a/apps/qc/serializers.py b/apps/qc/serializers.py index 83398cb..e8725b3 100644 --- a/apps/qc/serializers.py +++ b/apps/qc/serializers.py @@ -2,33 +2,43 @@ from rest_framework import serializers from .models import QCResult from apps.batches.models import ProcessingBatch from apps.capa.models import CAPARecord +from apps.compliance.services import evaluate_qc_data class QCResultSerializer(serializers.ModelSerializer): + entered_by_name = serializers.CharField(source="entered_by.full_name", read_only=True) + verified_by_name = serializers.CharField(source="verified_by.full_name", read_only=True) + class Meta: model = QCResult fields = "__all__" - read_only_fields = ["id", "overall_pass", "tested_at"] + read_only_fields = ["id", "overall_pass", "tested_at", "entered_by", "verified_by"] class QCResultCreateSerializer(serializers.ModelSerializer): class Meta: model = QCResult - fields = ["batch", "ffa_percentage", "moisture_percentage", "peroxide_value", "visual_grade", "tested_by", "lab_reference"] + fields = [ + "batch", "ffa_percentage", "moisture_percentage", "peroxide_value", "visual_grade", + "tested_by", "lab_reference", + "method_ffa", "method_moisture", "method_peroxide", + ] def create(self, validated_data): - ffa = validated_data["ffa_percentage"] - moisture = validated_data["moisture_percentage"] - peroxide = validated_data["peroxide_value"] - visual = validated_data["visual_grade"] - - overall_pass = all([ - ffa <= 3, - moisture <= 0.1, - peroxide <= 10, - visual != "Fail", - ]) + values = { + "ffa": validated_data["ffa_percentage"], + "moisture": validated_data["moisture_percentage"], + "peroxide": validated_data["peroxide_value"], + "visual_grade": validated_data["visual_grade"], + } + overall_pass, result_status, _details, spec_version = evaluate_qc_data(values) validated_data["overall_pass"] = overall_pass + validated_data["result_status"] = result_status + validated_data["spec_version"] = spec_version + + user = self.context.get("request").user if self.context.get("request") else None + validated_data["entered_by"] = user + validated_data["is_quarantine"] = not overall_pass qc_result = super().create(validated_data) @@ -37,7 +47,7 @@ class QCResultCreateSerializer(serializers.ModelSerializer): batch.status = ProcessingBatch.Status.QC_PASS else: batch.status = ProcessingBatch.Status.QC_FAIL - batch.save() + batch.save(update_fields=["status", "updated_at"]) if not overall_pass: CAPARecord.objects.create( diff --git a/apps/qc/views.py b/apps/qc/views.py index 37cb2d4..23ba1f0 100644 --- a/apps/qc/views.py +++ b/apps/qc/views.py @@ -1,6 +1,17 @@ -from rest_framework import viewsets, permissions +from django.utils import timezone +from rest_framework import viewsets, permissions, status +from rest_framework.decorators import action +from rest_framework.response import Response + from .models import QCResult from .serializers import QCResultSerializer, QCResultCreateSerializer +from apps.compliance.services import create_coa_for_qc + + +class IsOpsOrAdmin(permissions.BasePermission): + def has_permission(self, request, view): + user = request.user + return bool(user and user.is_authenticated and (user.role in ("ops", "admin") or user.is_superuser)) class QCResultViewSet(viewsets.ModelViewSet): @@ -11,3 +22,51 @@ class QCResultViewSet(viewsets.ModelViewSet): if self.action == "create": return QCResultCreateSerializer return QCResultSerializer + + def perform_create(self, serializer): + serializer.save() + + @action(detail=True, methods=["post"], permission_classes=[IsOpsOrAdmin]) + def verify(self, request, pk=None): + """Four-eyes quality approval. A separate authorised user validates the + entered result. The submitter cannot approve their own result.""" + qc = self.get_object() + user = request.user + + if qc.verification_status == QCResult.VerificationStatus.APPROVED: + return Response({"detail": "QC result already approved"}, status=status.HTTP_400_BAD_REQUEST) + + if qc.entered_by is not None and qc.entered_by_id == user.id: + return Response( + {"detail": "The submitter cannot approve their own QC result (four-eyes control)"}, + status=status.HTTP_400_BAD_REQUEST, + ) + + approved = request.data.get("approved", "true") + approved = str(approved).lower() in ("true", "1", "yes") + + if approved: + if not qc.overall_pass: + return Response( + {"detail": "A failed QC result cannot be approved; it remains quarantined with an open CAPA"}, + status=status.HTTP_400_BAD_REQUEST, + ) + qc.verification_status = QCResult.VerificationStatus.APPROVED + qc.verified_by = user + qc.approved_at = timezone.now() + qc.save() + coa = create_coa_for_qc(qc, approved_by=user) + return Response( + { + "detail": "QC result approved", + "coa_id": str(coa.id), + "coa_status": coa.status, + "coa_url": coa.file_url, + } + ) + + qc.verification_status = QCResult.VerificationStatus.REJECTED + qc.verified_by = user + qc.approved_at = timezone.now() + qc.save() + return Response({"detail": "QC result rejected"}) diff --git a/dsp_platform/settings.py b/dsp_platform/settings.py index c02cff6..b17181a 100644 --- a/dsp_platform/settings.py +++ b/dsp_platform/settings.py @@ -41,6 +41,7 @@ INSTALLED_APPS = [ "apps.documents", "apps.capa", "apps.esg", + "apps.compliance", "apps.ai", "apps.notifications", "apps.ops", diff --git a/dsp_platform/urls.py b/dsp_platform/urls.py index 9eb88c5..3fa11bb 100644 --- a/dsp_platform/urls.py +++ b/dsp_platform/urls.py @@ -19,6 +19,7 @@ urlpatterns = [ path("api/", include("apps.documents.urls")), path("api/", include("apps.capa.urls")), path("api/", include("apps.esg.urls")), + path("api/", include("apps.compliance.urls")), path("api/", include("apps.ai.urls")), path("api/", include("apps.notifications.urls")), path("ops/", include("apps.ops.urls")), diff --git a/requirements.txt b/requirements.txt index f49084d..a330bc8 100644 --- a/requirements.txt +++ b/requirements.txt @@ -18,6 +18,7 @@ stripe==8.8.0 python-dateutil==2.8.2 markdown==3.7 bleach==6.2.0 +reportlab==4.2.2 django-encrypted-model-fields openai diff --git a/templates/ops/base_ops.html b/templates/ops/base_ops.html index 2aa398b..905560c 100644 --- a/templates/ops/base_ops.html +++ b/templates/ops/base_ops.html @@ -33,6 +33,9 @@ body { font-size: .875rem; background: #f5f6fa; }
  • Dashboard
  • Batches
  • QC Results
    • +
  • QC Approvals
    • +
  • CoA Reviews
    • +
  • EUDR Compliance
  • Shipments
  • CAPA Records
  • Documents
    • diff --git a/templates/ops/coa_review_detail.html b/templates/ops/coa_review_detail.html new file mode 100644 index 0000000..86bed2c --- /dev/null +++ b/templates/ops/coa_review_detail.html @@ -0,0 +1,65 @@ +{% extends 'ops/base_ops.html' %} +{% block title %}CoA Review - IXG Operations{% endblock %} +{% block content %} + +
    +

    CoA Review: {{ coa.batch.batch_reference }}

    + Back +
    + +{% if messages %} + {% for m in messages %}
    {{ m }}
    {% endfor %} +{% endif %} + +
    +
    +
    +
    Certificate Details
    + + + + + + + + + + + + + +
    Batch{{ coa.batch.batch_reference }}
    Status{{ coa.get_status_display }}
    Versionv{{ coa.version }}
    Specification{{ coa.spec_version|default:'-' }}
    Issuer{{ coa.issuer|default:'-' }}
    Issue Date{{ coa.issue_date|date:'d/m/Y'|default:'-' }}
    Quality Approved By{{ coa.quality_approved_by.full_name|default:coa.quality_approved_by.email|default:'-' }}
    Quality Approved At{{ coa.quality_approved_at|date:'d/m/Y H:i'|default:'-' }}
    Released At{{ coa.released_at|date:'d/m/Y H:i'|default:'-' }}
    Published At{{ coa.published_at|date:'d/m/Y H:i'|default:'-' }}
    SHA-256{{ coa.sha256_hash|default:'-' }}
    Summary PDF {{ coa.file_name }}
    +
    +
    +
    +
    +
    Source Documents
    +

    Independent laboratory report (authoritative):

    + {% if coa.source_document %} + {{ coa.source_document.file_name }} + {% else %} +

    No lab report linked. Upload the original laboratory PDF for the batch.

    + {% endif %} + {% if lab_documents %} +
    +

    All lab documents for batch:

    + {% for d in lab_documents %}{% endfor %} + {% endif %} +
    +
    +
    Commercial Release
    + {% if coa.status == 'pending_approval' %} +
    + {% csrf_token %} + +

    Publishing exposes the summary PDF to authorised buyers and triggers buyer notification.

    + +
    + {% else %} +

    Commercial release approved. Buyer document and notification created.

    + {% endif %} +
    +
    +
    + +{% endblock %} diff --git a/templates/ops/coa_review_list.html b/templates/ops/coa_review_list.html new file mode 100644 index 0000000..c6ddcec --- /dev/null +++ b/templates/ops/coa_review_list.html @@ -0,0 +1,58 @@ +{% extends 'ops/base_ops.html' %} +{% block title %}CoA Reviews - IXG Operations{% endblock %} +{% block content %} + +
    +

    Certificate of Analysis Reviews

    + {{ page_obj.paginator.count }} total +
    + +
    +
    +
    + +
    +
    + + Clear +
    +
    +
    + +
    + + + + + + {% for c in coas %} + + + + + + + + + + {% empty %} + + {% endfor %} + +
    BatchVersionStatusSpecQuality ApprovedIssued
    {{ c.batch.batch_reference }}v{{ c.version }}{{ c.get_status_display }}{{ c.spec_version|default:'-' }}{{ c.quality_approved_by.full_name|default:'-' }}{{ c.issue_date|date:'d/m/Y'|default:'-' }}Review
    No CoA records
    +
    + +{% if page_obj.paginator.num_pages > 1 %} + +{% endif %} + +{% endblock %} diff --git a/templates/ops/eudr_detail.html b/templates/ops/eudr_detail.html new file mode 100644 index 0000000..2cf840b --- /dev/null +++ b/templates/ops/eudr_detail.html @@ -0,0 +1,141 @@ +{% extends 'ops/base_ops.html' %} +{% block title %}EUDR Case - IXG Operations{% endblock %} +{% block content %} + +
    +

    EUDR Case: {{ case.batch.batch_reference }}

    + Back +
    + +{% if messages %} + {% for m in messages %}
    {{ m }}
    {% endfor %} +{% endif %} + +
    +
    +
    +
    Completeness Check
    + {% if case.completeness_results %} + + + + {% for key, item in case.completeness_results.items %} + + + + + + {% endfor %} + +
    CheckStatusDetail
    {{ key }}{% if item.pass %}Pass{% else %}Fail{% endif %}{{ item.detail }}
    + {% else %} +

    Not yet run.

    + {% endif %} +
    + +
    +
    Intake Lineage
    + {% if lineage %} + + + + {% for link in lineage %} + + + + + + + + + {% empty %} + + {% endfor %} + +
    IntakeDateWeight (kg)ZoneZone StatusCollector
    {{ link.harvest_intake.intake_reference }}{{ link.harvest_intake.intake_date|date:'d/m/Y'|default:'-' }}{{ link.weight_allocated_kg|default:link.harvest_intake.weight_kg }}{{ link.harvest_intake.origin_zone.zone_code|default:'-' }}{% if link.harvest_intake.origin_zone %}{{ link.harvest_intake.origin_zone.get_verification_status_display }}{% else %}-{% endif %}{{ link.harvest_intake.collector_group.name|default:'-' }}
    No intakes linked. Record intake lineage via API/admin.
    + {% else %} +

    No intake lineage recorded for this batch.

    + {% endif %} +
    + +
    +
    Risk Assessments
    + {% for r in risk_assessments %} +
    + {{ r.get_risk_level_display }} + by {{ r.assessed_by.full_name|default:'-' }} · {{ r.assessed_at|date:'d/m/Y H:i' }} +
    {{ r.basis }}
    +
    + {% empty %} +

    No risk assessment recorded.

    + {% endfor %} +
    + +
    +
    Mitigation Actions
    + {% for m in mitigation_actions %} +
    + {{ m.get_status_display }} + {{ m.action }} (owner: {{ m.owner|default:'-' }}{% if m.due_date %}, due {{ m.due_date|date:'d/m/Y' }}{% endif %}) +
    + {% empty %} +

    No mitigation actions.

    + {% endfor %} +
    + +
    +
    External Submission (DDS) References
    + {% for s in statement_references %} +
    + {{ s.reconciliation_status }} + Ref: {{ s.dds_reference }} · {{ s.submission_date|date:'d/m/Y'|default:'-' }} · by {{ s.submitted_by.full_name|default:'-' }} +
    + {% empty %} +

    No external submission recorded. Submissions are manual.

    + {% endfor %} +
    +
    + +
    +
    +
    Case Controls
    +
    + {% csrf_token %} + + +
    +
    + {% csrf_token %} + + +
    +
    + {% csrf_token %} + + +
    +
    + {% csrf_token %} + + + + +
    +
    + +
    +
    Record External Submission
    +
    + {% csrf_token %} + + + + + + +
    +
    +
    +
    + +{% endblock %} diff --git a/templates/ops/eudr_list.html b/templates/ops/eudr_list.html new file mode 100644 index 0000000..fc5eb17 --- /dev/null +++ b/templates/ops/eudr_list.html @@ -0,0 +1,67 @@ +{% extends 'ops/base_ops.html' %} +{% block title %}EUDR Compliance - IXG Operations{% endblock %} +{% block content %} + +
    +

    EUDR Due Diligence Cases

    + {{ page_obj.paginator.count }} total +
    + +
    + Shea butter is not listed in EUDR Annex I; this tracker provides due-diligence + evidence and buyer assurance, not an EU-mandated filing. +
    + +
    +
    +
    + +
    +
    + + Clear +
    +
    +
    + +
    + + + + + + {% for c in cases %} + + + + + + + + + {% empty %} + + {% endfor %} + +
    BatchStatusRisk LevelApprovedUpdated
    {{ c.batch.batch_reference }}{{ c.get_status_display }}{{ c.get_risk_level_display }}{{ c.approved_at|date:'d/m/Y H:i'|default:'-' }}{{ c.updated_at|date:'d/m/Y' }}Review
    No due diligence cases
    +
    + +{% if page_obj.paginator.num_pages > 1 %} + +{% endif %} + +{% endblock %} diff --git a/templates/ops/qc_verify_detail.html b/templates/ops/qc_verify_detail.html new file mode 100644 index 0000000..85c0469 --- /dev/null +++ b/templates/ops/qc_verify_detail.html @@ -0,0 +1,52 @@ +{% extends 'ops/base_ops.html' %} +{% block title %}QC Verification - IXG Operations{% endblock %} +{% block content %} + +
    +

    QC Verification: {{ result.batch.batch_reference }}

    + Back +
    + +{% if messages %} + {% for m in messages %}
    {{ m }}
    {% endfor %} +{% endif %} + +
    +
    +
    +
    Structured QC Result
    + + + + + + + + + + + + + +
    Batch{{ result.batch.batch_reference }}
    Overall Result{% if result.overall_pass %}Pass{% else %}Fail{% endif %}
    Specification{{ result.spec_version|default:'legacy limits' }}
    FFA %{{ result.ffa_percentage }} ({{ result.method_ffa|default:'no method' }})
    Moisture %{{ result.moisture_percentage }} ({{ result.method_moisture|default:'no method' }})
    Peroxide Value{{ result.peroxide_value }} ({{ result.method_peroxide|default:'no method' }})
    Visual Grade{{ result.visual_grade }}
    Result Status{{ result.get_result_status_display }}
    Lab Reference{{ result.lab_reference|default:'-' }}
    Tested By{{ result.tested_by|default:'-' }}
    Entered By{{ result.entered_by.full_name|default:result.entered_by.email|default:'-' }}
    Tested At{{ result.tested_at|date:'d/m/Y H:i' }}
    +
    +
    +
    +
    +
    Approval Decision
    + {% if result.verification_status == 'draft' %} +
    + {% csrf_token %} +

    This action is recorded as the quality approval. The submitter cannot approve their own result.

    + + +
    + {% else %} +

    {{ result.get_verification_status_display }}

    + {% if result.verified_by %}

    Verified by {{ result.verified_by.full_name|default:result.verified_by.email }} on {{ result.approved_at|date:'d/m/Y H:i' }}

    {% endif %} + {% endif %} +
    +
    +
    + +{% endblock %} diff --git a/templates/ops/qc_verify_list.html b/templates/ops/qc_verify_list.html new file mode 100644 index 0000000..c9853a3 --- /dev/null +++ b/templates/ops/qc_verify_list.html @@ -0,0 +1,59 @@ +{% extends 'ops/base_ops.html' %} +{% block title %}QC Approvals - IXG Operations{% endblock %} +{% block content %} + +
    +

    QC Approvals (four-eyes verification)

    + {{ page_obj.paginator.count }} pending +
    + +
    +
    +
    + +
    +
    + + Clear +
    +
    +
    + +
    + + + + + + {% for r in results %} + + + + + + + + + + + + {% empty %} + + {% endfor %} + +
    BatchResultFFA%Moisture%PeroxideVisualEntered BySpec
    {{ r.batch.batch_reference }}{% if r.overall_pass %}Pass{% else %}Fail{% endif %}{{ r.ffa_percentage }}{{ r.moisture_percentage }}{{ r.peroxide_value }}{{ r.visual_grade }}{{ r.entered_by.full_name|default:r.entered_by.email|default:'-' }}{{ r.spec_version|default:'-' }}Verify
    No QC results awaiting verification
    +
    + +{% if page_obj.paginator.num_pages > 1 %} + +{% endif %} + +{% endblock %} diff --git a/templates/portal/base_portal.html b/templates/portal/base_portal.html index c37f820..a1680dd 100644 --- a/templates/portal/base_portal.html +++ b/templates/portal/base_portal.html @@ -33,6 +33,7 @@ body { font-size: .875rem; background: #f0fdf4; }
  • Dashboard
  • Batches
  • Documents
    • +
  • Compliance
  • Subscription
  • AI Chat
    • diff --git a/templates/portal/compliance.html b/templates/portal/compliance.html new file mode 100644 index 0000000..a9882bc --- /dev/null +++ b/templates/portal/compliance.html @@ -0,0 +1,62 @@ +{% extends 'portal/base_portal.html' %} +{% block title %}Compliance - IXG Buyer Portal{% endblock %} +{% block content %} + +
    +

    Compliance

    + Approved evidence for your batches +
    + +
    + Shea butter is not listed in EUDR Annex I; the due-diligence status shown here is + evidence and buyer assurance provided by IXG, not an EU-mandated filing. +
    + +
    +
    +
    +
    Certificates of Analysis (approved)
    + + + + {% for c in approved_coas %} + + + + + + + {% empty %} + + {% endfor %} + +
    BatchProductIssuedDownload
    {{ c.batch.batch_reference }}Unrefined Shea Butter{{ c.issue_date|date:'d/m/Y'|default:'-' }} PDF
    No approved certificates available
    +
    +
    +
    +
    +
    EUDR Due Diligence Status
    + + + + {% for c in eudr_cases %} + + + + + + {% empty %} + + {% endfor %} + +
    BatchStatusRisk
    {{ c.batch.batch_reference }}{{ c.get_status_display }}{{ c.get_risk_level_display }}
    No due-diligence cases recorded
    +
    +
    +
    + +{% endblock %}