"""
def render_pdf(html: str, out_path: str) -> str:
"""Render HTML -> PDF via headless Chromium. Returns the output path."""
tmp_html = out_path + ".html"
with open(tmp_html, "w", encoding="utf-8") as fh:
fh.write(html)
subprocess.run(
[
"/usr/bin/chromium", "--headless", "--no-sandbox", "--disable-gpu",
"--print-to-pdf=" + out_path, "--no-pdf-header-footer", tmp_html,
],
check=True, capture_output=True, timeout=60,
)
os.remove(tmp_html)
return out_path
# ---------------------------------------------------------------------------
# Change logging
# ---------------------------------------------------------------------------
def log_field_change(conn, claim_id: str, profile_key: str, old_value: str,
new_value: str, actor: str) -> None:
"""Write an audit_log entry. recovery_impact is recorded in `reason` so it's
queryable: recovery-impacting changes carry reason='recovery-impacting'."""
field = FIELD_CATALOG.get(profile_key, {})
reason = "recovery-impacting" if field.get("recovery_impact") else None
conn.execute(
"""
INSERT INTO audit_log (id, entity_type, entity_id, action, field,
old_value, new_value, actor, reason, created_at)
VALUES (?, 'claim', ?, 'packet_field_update', ?, ?, ?, ?, ?, ?)
""",
(
uuid.uuid4().hex, claim_id, profile_key,
old_value if old_value is not None else None,
new_value if new_value is not None else None,
actor, reason, dbmod.utcnow_iso(),
),
)
# ---------------------------------------------------------------------------
# Submission status — the "visual indicator" of what the client has submitted.
# ---------------------------------------------------------------------------
def submission_status(conn, claim_id: str, values: dict) -> dict:
"""Compute which documents the client has submitted so far.
Sources:
- `documents` table: claim substantiation uploads (uploaded_by=CLIENT).
- effective field values (pre-populated + client overrides): a required
field counts as supplied whenever it has a non-empty value.
- `tos_accepted_at` / `no_other_agency_at` / `no_prior_action_at`: checkboxes
the client confirmed during intake.
"""
docs = conn.execute(
"SELECT original_name, uploaded_by, created_at FROM documents WHERE claim_id = ? ORDER BY created_at ASC",
(claim_id,),
).fetchall()
client_docs = [d for d in docs if d["uploaded_by"] == "CLIENT"]
overrides = conn.execute(
"SELECT profile_key, value, updated_by FROM packet_field_overrides WHERE claim_id = ?",
(claim_id,),
).fetchall()
overridden_keys = {o["profile_key"] for o in overrides if (o["value"] or "").strip()}
# tos_accepted_at lives on clients; no_other_agency_at / no_prior_action_at on claims.
row = conn.execute(
"SELECT cl.tos_accepted_at, c.no_other_agency_at, c.no_prior_action_at "
"FROM claims c JOIN clients cl ON cl.id = c.client_id WHERE c.id = ?",
(claim_id,),
).fetchone()
tos = bool(row and row["tos_accepted_at"])
no_other_agency = bool(row and row["no_other_agency_at"])
no_prior_action = bool(row and row["no_prior_action_at"])
# A required field is "supplied" when its effective value is non-empty
# (either pre-populated from the submission or overridden by the client).
required = [key for key, meta in FIELD_CATALOG.items() if meta.get("required")]
supplied_fields = [k for k in required if (values.get(k) or "").strip()]
empty_required = [k for k in required if not (values.get(k) or "").strip()]
return {
"documents_submitted": len(client_docs),
"documents": [
{"name": d["original_name"], "uploaded_by": d["uploaded_by"], "at": d["created_at"]}
for d in docs
],
"fields_supplied": len(supplied_fields),
"fields_supplied_list": supplied_fields,
"fields_overridden_list": sorted(overridden_keys),
"fields_required_total": len(required),
"fields_required_missing": empty_required,
"tos_accepted": tos,
"no_other_agency_confirmed": no_other_agency,
"no_prior_action_confirmed": no_prior_action,
"complete": len(empty_required) == 0 and len(client_docs) > 0,
}
# ---------------------------------------------------------------------------
# API endpoints (client-scoped + staff)
# ---------------------------------------------------------------------------
def _claim_row_for_client(conn, claim_number: str, client_id: str):
row = conn.execute(
"SELECT id FROM claims WHERE claim_number = ? AND client_id = ?",
(claim_number, client_id),
).fetchone()
if row is None:
raise HTTPException(status_code=404, detail={"code": "not_found", "message": "Claim not found."})
return row["id"]
def _claim_row_any(conn, claim_number: str):
row = conn.execute("SELECT id FROM claims WHERE claim_number = ?", (claim_number,)).fetchone()
if row is None:
raise HTTPException(status_code=404, detail={"code": "not_found", "message": "Claim not found."})
return row["id"]
@router.get("/api/claims/{claim_number}/packet/fields")
def get_packet_fields(claim_number: str, session: dict = Depends(authmod.require_client)):
"""Return the field catalog with pre-populated + client-overridden values,
plus a submission-status summary (the visual indicator)."""
with get_conn() as conn:
claim_id = _claim_row_for_client(conn, claim_number, session["client_id"])
values, row = merged_values(conn, claim_number)
overrides = {
o["profile_key"]: o["value"]
for o in conn.execute(
"SELECT profile_key, value FROM packet_field_overrides WHERE claim_id = ?", (claim_id,)
).fetchall()
}
status_summary = submission_status(conn, claim_id, values)
fields = []
for key, meta in FIELD_CATALOG.items():
fields.append({
"key": key,
"label": meta["label"],
"type": meta["type"],
"required": meta["required"],
"recovery_impact": meta["recovery_impact"],
"options": meta.get("options"),
"value": values.get(key, ""),
"source": meta.get("source"),
"is_override": key in overrides,
})
return {"claim_number": claim_number, "fields": fields, "submission": status_summary}
@router.patch("/api/claims/{claim_number}/packet/fields")
async def update_packet_field(claim_number: str, request: Request,
session: dict = Depends(authmod.require_client)):
"""Client edits a packet field. Recovery-impacting changes are written to
audit_log so recovery staff see exactly what changed."""
try:
body = await request.json()
except Exception:
return JSONResponse(status_code=422, content={"error": {"code": "validation_error", "message": "Invalid JSON body."}})
key = body.get("key")
value = body.get("value")
if key not in FIELD_CATALOG:
return JSONResponse(status_code=422, content={"error": {"code": "validation_error", "message": "Unknown field."}})
if value is None or (isinstance(value, str) and value.strip() == ""):
value = ""
else:
value = str(value).strip()
with get_conn() as conn:
claim_id = _claim_row_for_client(conn, claim_number, session["client_id"])
old = conn.execute(
"SELECT value FROM packet_field_overrides WHERE claim_id = ? AND profile_key = ?",
(claim_id, key),
).fetchone()
old_value = old["value"] if old else None
# If no override exists yet, the document was showing the pre-populated
# submission value — capture that as the true "before" for the audit trail.
if old_value is None:
try:
values, _ = merged_values(conn, claim_number)
old_value = values.get(key) or None
except KeyError:
old_value = None
now = dbmod.utcnow_iso()
conn.execute(
"""
INSERT INTO packet_field_overrides (id, claim_id, profile_key, value, updated_by, created_at, updated_at)
VALUES (?, ?, ?, ?, 'CLIENT', ?, ?)
ON CONFLICT(claim_id, profile_key) DO UPDATE SET value=excluded.value,
updated_by=excluded.updated_by, updated_at=excluded.updated_at
""",
(uuid.uuid4().hex, claim_id, key, value, now, now),
)
# Log recovery-impacting changes only when the value actually changed.
if old_value != value:
log_field_change(conn, claim_id, key, old_value, value, "client")
conn.commit()
return {"key": key, "value": value, "saved": True}
@router.get("/api/claims/{claim_number}/packet/preview")
def get_packet_preview(claim_number: str, session: dict = Depends(authmod.require_client)):
"""Live HTML preview of the filled packet."""
with get_conn() as conn:
_claim_row_for_client(conn, claim_number, session["client_id"])
html_out = packet_html(conn, claim_number)
return HTMLResponse(content=html_out)
@router.get("/api/claims/{claim_number}/packet.pdf")
def get_packet_pdf(claim_number: str, session: dict = Depends(authmod.require_client)):
"""PDF of the filled packet (generated on demand via headless Chromium)."""
import tempfile
with get_conn() as conn:
_claim_row_for_client(conn, claim_number, session["client_id"])
html_out = packet_html(conn, claim_number)
with tempfile.NamedTemporaryFile(suffix=".pdf", delete=False) as tf:
out_path = tf.name
try:
render_pdf(html_out, out_path)
with open(out_path, "rb") as fh:
data = fh.read()
finally:
os.remove(out_path)
return Response(content=data, media_type="application/pdf",
headers={"Content-Disposition": f'inline; filename="{claim_number}-welcome-packet.pdf"'})
# --- Staff endpoints (staff key auth) ---
@router.get("/api/staff/claims/{claim_number}/packet/preview")
def staff_packet_preview(claim_number: str, _staff: None = Depends(authmod.require_staff)):
with get_conn() as conn:
_claim_row_any(conn, claim_number)
html_out = packet_html(conn, claim_number)
return HTMLResponse(content=html_out)
@router.get("/api/staff/claims/{claim_number}/packet.pdf")
def staff_packet_pdf(claim_number: str, _staff: None = Depends(authmod.require_staff)):
import tempfile
with get_conn() as conn:
_claim_row_any(conn, claim_number)
html_out = packet_html(conn, claim_number)
with tempfile.NamedTemporaryFile(suffix=".pdf", delete=False) as tf:
out_path = tf.name
try:
render_pdf(html_out, out_path)
with open(out_path, "rb") as fh:
data = fh.read()
finally:
os.remove(out_path)
return Response(content=data, media_type="application/pdf",
headers={"Content-Disposition": f'inline; filename="{claim_number}-welcome-packet.pdf"'})