"""The Intake aggregate: the starter's self-service form for one engagement (hr_intake, intake_id = person_id|n).
* one open link at a time: issuing again revokes the previous one (the platform's door does it: one link per
person per grant), so a reissue makes the old link stop working;
* the fields asked come from the localisation pack's intake-field rows for the worker type; each lands in the
Sensitive aggregate, never on the Person, never echoed back;
* 'complete' and 'missing' are derived from what the Sensitive aggregate holds — the manager sees counts and
field NAMES ("details: 5 of 6 done"), never a value.
"""
from dataclasses import dataclass, field
from datetime import datetime
from typing import Optional
from brick_hr.domain.errors import Incomplete
def intake_id(person_id: int, n: int) -> str:
return f"{int(person_id)}|{int(n)}"
def parse_intake_id(iid: str) -> tuple:
try:
pid, n = str(iid).split("|")
return int(pid), int(n)
except ValueError:
raise Incomplete(f"{iid!r} is not an intake id (<person_id>|<n>)") from None
@dataclass
class Intake:
intake_id: str
person_id: int
n: int
worker_type: str
pack: str
link_ref: Optional[str] = None
issued_at: Optional[datetime] = None
expires_at: Optional[datetime] = None
submitted_at: Optional[datetime] = None
missing_fields: list = field(default_factory=list)
chases: int = 0
version: int = 0
def issue(self, link_ref: str, issued_at: datetime, expires_at: Optional[datetime]) -> None:
self.link_ref, self.issued_at, self.expires_at = link_ref, issued_at, expires_at
self.submitted_at = None if self.missing_fields else self.submitted_at
def field_values(fields: list, form: dict) -> dict:
"""The submitted form → {sensitive_kind: {field: value}} for the TEXT fields that were filled in."""
out: dict = {}
for f in fields:
if f.get("input", "text") != "text":
continue
v = (form.get(f["field"]) or "").strip()
if v:
out.setdefault(f["sensitive_kind"], {})[f["field"]] = v
return out
def status(fields: list, held: dict) -> dict:
"""complete / missing by field NAME, from what the Sensitive aggregate holds ({kind: value dict, files}).
`held` is {kind: {"value": {...}, "has_file": bool}}. Never returns a value."""
complete, missing = [], []
for f in fields:
h = held.get(f["sensitive_kind"])
if f.get("input") == "file":
ok = bool(h and h.get("has_file"))
else:
val = (h or {}).get("value") or {}
ok = bool(val.get(f["field"]) or (len(val) == 1 and "value" in val and f["field"] == f["sensitive_kind"]))
(complete if ok else missing).append(f["field"])
required = [f["field"] for f in fields if f.get("required", True)]
done = len([c for c in complete if c in required])
return {"done": done, "total": len(required), "complete": complete,
"missing": [m for m in missing if m in required], "label": f"details: {done} of {len(required)} done"}
__all__ = ["Intake", "intake_id", "parse_intake_id", "field_values", "status"]