"""Forms for the ops portal. Both forms here share a principle: **stricter than the model.** The intake form is stricter about *completeness*. ``HarvestIntake`` allows almost everything to be blank, because records also arrive through the API, and the platform copes by flagging the gaps later: the EUDR completeness check has an "intake data" test whose whole job is to catch a missing date or weight. A portal form that let the same blanks through would simply move that problem to a screen where a person could have prevented it. So every field the checks read is required — reference, collector group, origin zone, date, weight and grade — and the fields that are genuinely optional at intake stay optional. The batch form and the lineage formset are stricter about *contradiction*. A batch's status is not a free field: recording a QC result sets it to QC Pass or QC Fail, so the form refuses to let a person type over a quality failure. And an intake's weight is finite, so the formset refuses to allocate more of it than was received, counting every batch that has already drawn from it. Both reasons are stated in the docstrings of the classes themselves, because a maintainer's instinct is otherwise to "fix" them back to match the model. """ from django import forms from django.db.models import Count, Q, Sum from django.forms import modelformset_factory from apps.batches.models import ProcessingBatch from apps.compliance import services from apps.compliance.models import ( CollectorGroup, DueDiligenceStatementReference, EUDREvidence, Facility, HarvestIntake, IntakeBatchLineage, MitigationAction, OriginZone, ) from apps.shipments.models import Shipment # The visual grades the product specification defines. GRADE_CHOICES = [("A", "Grade A"), ("B", "Grade B"), ("C", "Grade C")] class ZoneChoiceField(forms.ModelChoiceField): """Zone dropdown that says which zones are not verified yet. Geolocation in the EUDR case passes only for a verified zone, so an operator choosing one should be able to see that before it fails a check. It is a label, not a block: an intake can legitimately be recorded against a zone that is still awaiting verification. """ def label_from_instance(self, zone): where = zone.region or zone.country or "" mark = "" if zone.verification_status != OriginZone.VerificationStatus.VERIFIED: mark = f" — {zone.get_verification_status_display().lower()}" return f"{zone.zone_code}{f' ({where})' if where else ''}{mark}" class CollectorGroupChoiceField(forms.ModelChoiceField): """Group dropdown that names the zone each group draws from.""" def label_from_instance(self, group): if not group.origin_zone: return f"{group.name} — no origin zone set" return f"{group.name} — {group.origin_zone.zone_code}" class HarvestIntakeForm(forms.ModelForm): collector_group = CollectorGroupChoiceField( queryset=CollectorGroup.objects.select_related("origin_zone").order_by("name"), label="Collector group", help_text="The registered group that delivered the material.", empty_label="Select the collector group", ) origin_zone = ZoneChoiceField( queryset=OriginZone.objects.order_by("zone_code"), label="Origin zone", help_text="Where it was collected. Filled in from the group, and can be changed.", empty_label="Select the origin zone", ) grade = forms.ChoiceField( choices=GRADE_CHOICES, label="Grade", help_text="Visual grade recorded at intake, as defined by the product specification.", widget=forms.Select(attrs={"class": "form-select"}), ) class Meta: model = HarvestIntake fields = [ "intake_reference", "collector_group", "origin_zone", "intake_date", "weight_kg", "grade", "receiving_officer", "status", "evidence_file", ] labels = { "intake_reference": "Intake reference", "intake_date": "Intake date", "weight_kg": "Weight (kg)", "receiving_officer": "Receiving officer", "evidence_file": "Supporting file name", } help_texts = { "intake_reference": "Must be unique. Suggested from the last intake recorded, and can be overwritten to match the weighbridge ticket.", "intake_date": "The day the material was received.", "weight_kg": "Weight received, in kilograms.", "receiving_officer": "Who received and weighed it. Defaults to you.", "status": "Received is the normal state for a new intake; Verified is set once the record has been checked.", "evidence_file": "The name of the supporting record, typically the weighbridge ticket. Files are not uploaded by the platform yet, so this is the name it is filed under.", } widgets = { "intake_reference": forms.TextInput( attrs={"class": "form-control", "autocomplete": "off"} ), "collector_group": forms.Select(attrs={"class": "form-select"}), "origin_zone": forms.Select(attrs={"class": "form-select"}), # A native date input only accepts the ISO format. Without this the # en-gb locale renders "04/09/2026", which the browser silently # discards, leaving the field empty on an edit. "intake_date": forms.DateInput( format="%Y-%m-%d", attrs={"class": "form-control", "type": "date"} ), "weight_kg": forms.NumberInput( attrs={"class": "form-control", "step": "0.01", "min": "0.01"} ), "receiving_officer": forms.TextInput(attrs={"class": "form-control"}), "status": forms.Select(attrs={"class": "form-select"}), "evidence_file": forms.TextInput(attrs={"class": "form-control"}), } def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) # The model tolerates blanks; the portal form does not. See the module # docstring: these are the fields the completeness checks read. for name in ("collector_group", "origin_zone", "intake_date", "weight_kg"): self.fields[name].required = True self.fields["status"].required = True def clean_intake_reference(self): reference = (self.cleaned_data["intake_reference"] or "").strip() if not reference: raise forms.ValidationError("An intake reference is required.") clashes = HarvestIntake.objects.filter(intake_reference__iexact=reference) if self.instance.pk: clashes = clashes.exclude(pk=self.instance.pk) if clashes.exists(): raise forms.ValidationError( f"Intake {reference} already exists. References must be unique." ) return reference def clean_weight_kg(self): weight = self.cleaned_data["weight_kg"] if weight is None: raise forms.ValidationError("A weight is required.") if weight <= 0: raise forms.ValidationError("Weight must be greater than zero.") return weight def clean(self): cleaned = super().clean() group = cleaned.get("collector_group") zone = cleaned.get("origin_zone") if group and zone and group.origin_zone_id and group.origin_zone_id != zone.pk: # Not an error — a group can deliver from more than one area — but it # is worth surfacing, because the two records will disagree. self.add_error( "origin_zone", forms.ValidationError( f"{group.name} is registered to {group.origin_zone.zone_code}, " f"not {zone.zone_code}. Save it this way only if that is intended." ), ) return cleaned # --------------------------------------------------------------------------- # Batches # # The batch form exists for a different reason from the intake form: not to stop # incomplete records, but to stop **contradictory** ones. A batch's status is not # purely a human field — recording a QC result sets it to QC Pass or QC Fail — so # a free status dropdown would let someone quietly type over a quality failure. # The rules below are the whole point of the screen; read them before changing # them. # The statuses the platform sets from a QC result, and which a person therefore # must not be able to type. QC_SET_STATUSES = [ ProcessingBatch.Status.QC_PASS, ProcessingBatch.Status.QC_FAIL, ] # ...and of those, the one that is sticky: a failed batch cannot be moved out of # its failure by hand at all, only by a new result. QC Pass is not sticky, because # a passed batch is exactly the batch that goes to sea. STICKY_STATUSES = [ProcessingBatch.Status.QC_FAIL] # The journey statuses, which a person does move by hand. The real world is # untidy: a batch can genuinely go back to processing. JOURNEY_STATUSES = [ ProcessingBatch.Status.HARVESTED, ProcessingBatch.Status.PROCESSING, ProcessingBatch.Status.READY_TO_SHIP, ProcessingBatch.Status.IN_TRANSIT, ProcessingBatch.Status.AT_UK_PORT, ProcessingBatch.Status.CUSTOMS, ProcessingBatch.Status.DELIVERED, ] # A batch cannot be born mid-journey: it exists because material was processed. OPENING_STATUSES = [ProcessingBatch.Status.HARVESTED, ProcessingBatch.Status.PROCESSING] class BatchForm(forms.ModelForm): """Create or correct a processing batch. Two rules worth stating plainly, because they are the difference between a working screen and a dangerous one: - **QC Fail is sticky.** A batch sitting on a failed QC result cannot be moved by hand at all. Only a new result — a retest — takes it off. To let the portal move it would be to give the screen a way to make a quality failure disappear. - **QC Pass can move on.** It is offered as the current value (so saving alone changes nothing) alongside the journey statuses, because a batch that has passed QC is exactly the batch that goes to sea. """ status = forms.ChoiceField( choices=[], label="Status", help_text="Where the batch is in its journey.", widget=forms.Select(attrs={"class": "form-select"}), ) # Declared here so the required field opens on a plain instruction rather than # Django's bare "---------". grade = forms.ChoiceField( choices=[("", "Select the grade")] + GRADE_CHOICES, label="Grade", help_text="The visual grade recorded for the batch, as defined by the product specification.", widget=forms.Select(attrs={"class": "form-select"}), ) class Meta: model = ProcessingBatch fields = [ "batch_reference", "grade", "weight_kg", "harvest_date", "origin_facility", "notes", "status", ] labels = { "batch_reference": "Batch reference", "weight_kg": "Weight (kg)", "harvest_date": "Harvest date", "origin_facility": "Origin facility", "notes": "Notes", } help_texts = { "batch_reference": "Must be unique. Suggested from the last batch recorded, and can be overwritten.", "weight_kg": "The batch weight as processed, in kilograms.", "harvest_date": "The date the underlying harvest was taken.", "origin_facility": "The processing site. New batches choose from the facility register; an existing batch recorded against a site that is no longer registered keeps its own value, flagged so the drift is visible.", "notes": "Anything the next person needs to know about this batch.", } widgets = { "batch_reference": forms.TextInput( attrs={"class": "form-control", "autocomplete": "off"} ), "grade": forms.Select(attrs={"class": "form-select"}), "weight_kg": forms.NumberInput( attrs={"class": "form-control", "step": "0.01", "min": "0.01"} ), "harvest_date": forms.DateInput( format="%Y-%m-%d", attrs={"class": "form-control", "type": "date"} ), "notes": forms.Textarea(attrs={"class": "form-control", "rows": 4}), } def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) # `_state.adding`, not `pk is None`: ProcessingBatch's primary key is a # UUIDField with a default, so an unsaved instance already has a pk, and # `bool(instance.pk)` reads as "editing" for a brand-new batch. is_new = self.instance._state.adding locked = not is_new and self.instance.status in STICKY_STATUSES if locked: # Shown, not editable. `disabled` means Django validates the initial # value, so the current status is the only choice on offer. self.fields["status"].choices = [ (self.instance.status, self.instance.get_status_display()) ] self.fields["status"].disabled = True self.fields["status"].help_text = ( "Set by the QC result and cannot be changed here. A new result — " "a retest — is what moves it." ) elif not is_new: choices = list(JOURNEY_STATUSES) if self.instance.status == ProcessingBatch.Status.QC_PASS: # Offered so that saving alone leaves it alone, and so it can be # moved on to the journey statuses. choices.insert(0, ProcessingBatch.Status.QC_PASS) self.fields["status"].choices = [(s.value, s.label) for s in choices] else: self.fields["status"].choices = [ (s.value, s.label) for s in OPENING_STATUSES ] self.fields["status"].initial = ProcessingBatch.Status.HARVESTED # The origin facility comes from the register. An existing batch whose # recorded name is not in the register keeps its own value as a choice — # history is not silently rewritten — but it is labelled as the odd one # out so the drift is visible. current = "" if is_new else (self.instance.origin_facility or "").strip() registered = [ f.name for f in Facility.objects.filter(is_active=True).order_by("name") ] choices = [(name, name) for name in registered] if current and current not in registered: choices.append((current, f"{current} — not in the facility register")) self.fields["origin_facility"] = forms.ChoiceField( choices=choices, required=True, label="Origin facility", help_text=self.Meta.help_texts["origin_facility"], widget=forms.Select(attrs={"class": "form-select"}), ) self.fields["origin_facility"].initial = current or ( registered[0] if len(registered) == 1 else "" ) for name in ("grade", "weight_kg", "harvest_date"): self.fields[name].required = True def clean_batch_reference(self): reference = (self.cleaned_data["batch_reference"] or "").strip() if not reference: raise forms.ValidationError("A batch reference is required.") clashes = ProcessingBatch.objects.filter(batch_reference__iexact=reference) if not self.instance._state.adding: clashes = clashes.exclude(pk=self.instance.pk) if clashes.exists(): raise forms.ValidationError( f"Batch {reference} already exists. References must be unique." ) return reference def clean_weight_kg(self): weight = self.cleaned_data["weight_kg"] if weight is None: raise forms.ValidationError("A weight is required.") if weight <= 0: raise forms.ValidationError("Weight must be greater than zero.") return weight def clean_status(self): status = self.cleaned_data.get("status") # A locked batch's own status arrives with the form and must be allowed # through — it is only a *different* QC status that is a hand-setting. unchanged = not self.instance._state.adding and status == self.instance.status if status in QC_SET_STATUSES and not unchanged: raise forms.ValidationError( f"Batch status {ProcessingBatch.Status(status).label} is set by the " "QC result, not by hand." ) if self.instance._state.adding and status not in OPENING_STATUSES: raise forms.ValidationError( "A new batch starts at Harvested or Processing." ) return status class IntakeChoiceField(forms.ModelChoiceField): """Intake dropdown that carries the facts needed to allocate from it. An operator deciding how much of an intake can go into a batch needs the weight received and what is left of it. Both are read off annotations added by the view, so this costs no extra queries. Kept short on purpose: a native ``