ippadmin/dre
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
dre/backend/staff.py
T
root 7a62b0b340 Backend Aug 22-25: AI analysis, welcome-packet templating, LetterStream, DocuSeal, staff RBAC + tier gate 
- analysis.py: deterministic claim scorer + /analyze /approve /letter /advance-tier endpoints (auto-runs on intake)
- packet.py + packet_fields.json: welcome-packet templating engine (6 onboarding docs, field catalog)
- letterstream.py + letters.py: certified-mail send pipeline + letter lifecycle (webhook verified)
- docuseal.py: DocuSeal signing integration
- staff.py/models.py/schema.sql/auth.py: approval actor from staff key, tier gate (APPROVED+ACTIVE+onboarding docs), onboarding_docs table
- frontend/: dependency-free static portal (intake, magic-link login/verify, dashboard)
- landing-mockups/: 4 design-stance mockups + favicons
- legal/: aup/privacy/sms-terms/terms HTML
- docs/: letter-queue scope, letterstream API contract, 6 welcome-packet templates
- review-dre-landing-2026-08-21.md: 3-variant landing feedback sprint
- compliance/DRE_Compliance_Manual.md: updated

Source synced from deployed /opt/dre-portal/app/ (was 4 days ahead of git).
2026-08-26 02:26:33 -04:00

862 lines
41 KiB
Python
Raw Permalink Blame History

"""Internal staff endpoints (staff-key auth): claims list/detail/patch, notes, documents, stats, audit."""
from __future__ import annotations
import hashlib
import html
import json
import logging
import os
import uuid
from fastapi import APIRouter, Depends, File, HTTPException, Request, UploadFile, status
from fastapi.responses import JSONResponse
from pydantic import ValidationError
from . import auth as authmod
from . import db
from .db import get_conn, get_upload_dir, new_uuid, utcnow_iso
from .models import StaffApproval, StaffClaimPatch, StaffLetterUpdate, StaffNoteCreate, StaffOnboardingUpdate
from .claims import ( # noqa: E402
MAX_FILE_BYTES, ALLOWED_EXT, MAGIC_BYTES, EXT_TO_KIND, STATUS_LABELS,
TIER_STEPS, _money, _err,
)
from . import analysis
from . import packet
from . import docuseal
logger = logging.getLogger("dre.staff")
router = APIRouter()
def _json_or_none(raw: str | None):
"""Parse a JSON TEXT column safely, returning None on empty/invalid."""
if not raw:
return None
try:
import json
return json.loads(raw)
except (ValueError, TypeError):
return None
def _tier_advance_gate(conn, row) -> str | None:
"""Return a human-readable blocker if recovery may NOT advance tier, else None.
Rules: the claim must be approved & accepted (approval APPROVED + status
ACTIVE) AND all onboarding paperwork must be returned before escalation.
"""
if row["approval_status"] != "APPROVED":
return "Cannot advance tier until the claim is approved and accepted."
if row["status"] != "ACTIVE":
return "Cannot advance tier until the claim is accepted (status ACTIVE)."
ob = packet.onboarding_status(conn, row["id"])
if not ob["complete"]:
outstanding = ", ".join(i["title"] for i in ob["outstanding"])
return f"Cannot advance tier until all onboarding paperwork is returned. Outstanding: {outstanding}."
return None
# ---------------------------------------------------------------
# GET /api/staff/me (acting staff identity)
# ---------------------------------------------------------------
@router.get("/api/staff/me")
async def staff_me(staff_name: str = Depends(authmod.require_staff)):
return {"name": staff_name}
# ---------------------------------------------------------------
# POST /api/staff/claims/{claim_number}/onboarding/send
# Create DocuSeal signature requests for all pending onboarding docs,
# prefilled from the client's submission, and email the signing links.
# NOTE: registered BEFORE the {doc_key} route below so "send" is not
# swallowed as a doc_key path segment.
# ---------------------------------------------------------------
@router.post("/api/staff/claims/{claim_number}/onboarding/send")
async def staff_send_onboarding(claim_number: str, request: Request,
staff_name: str = Depends(authmod.require_staff)):
try:
body = await request.json()
except Exception:
body = {}
send_email = bool(body.get("send_email", True))
with get_conn() as conn:
row = conn.execute("SELECT id FROM claims WHERE claim_number = ?", (claim_number,)).fetchone()
if row is None:
return _err("not_found", "Claim not found.", status.HTTP_404_NOT_FOUND)
claim_id = row["id"]
packet.ensure_onboarding_docs(conn, claim_id)
client = conn.execute(
"SELECT email, contact_name FROM clients cl JOIN claims c ON c.client_id = cl.id WHERE c.id = ?",
(claim_id,),
).fetchone()
if client is None:
return _err("validation_error", "Claim has no linked client.", status.HTTP_422_UNPROCESSABLE_ENTITY)
if not client["email"]:
return _err("validation_error", "Client has no email on file.", status.HTTP_422_UNPROCESSABLE_ENTITY)
values, _vals_row = packet.merged_values(conn, claim_number)
pending = conn.execute(
"SELECT doc_key FROM onboarding_docs WHERE claim_id = ? AND status != 'RECEIVED' ORDER BY rowid",
(claim_id,),
).fetchall()
if not pending:
return {"claim_number": claim_number, "sent": [], "errors": [],
"message": "No pending onboarding documents to send."}
title_by_key = dict(packet.ONBOARDING_DOCS)
sent = []
errors = []
for p in pending:
doc_key = p["doc_key"]
try:
sub = docuseal.create_submission(
doc_key, client["email"], client["contact_name"] or client["email"],
values, send_email=send_email,
)
except docuseal.DocuSealError as exc:
errors.append({"doc_key": doc_key, "error": str(exc)})
continue
now = utcnow_iso()
conn.execute(
"UPDATE onboarding_docs SET docuseal_submission_id = ?, docuseal_submitter_id = ?, "
"docuseal_slug = ?, docuseal_embed_src = ?, docuseal_status = ?, docuseal_sent_at = ? "
"WHERE claim_id = ? AND doc_key = ?",
(str(sub.get("submission_id")), str(sub.get("id")), sub.get("slug"),
sub.get("embed_src"), sub.get("status"), sub.get("sent_at") or now,
claim_id, doc_key),
)
sent.append({
"doc_key": doc_key,
"title": title_by_key.get(doc_key, doc_key),
"submission_id": sub.get("submission_id"),
"status": sub.get("status"),
"signing_url": docuseal.signing_url(sub.get("slug"), sub.get("embed_src")),
"sent_email": send_email,
})
conn.execute(
"INSERT INTO audit_log (id, entity_type, entity_id, action, field, old_value, new_value, actor, created_at) "
"VALUES (?, 'claim', ?, 'onboarding_send', NULL, NULL, ?, ?, ?)",
(new_uuid(), claim_id, f"{len(sent)} doc(s) sent", staff_name, utcnow_iso()),
)
conn.commit()
return {"claim_number": claim_number, "sent": sent, "errors": errors}
# ---------------------------------------------------------------
# POST /api/staff/claims/{claim_number}/onboarding/{doc_key}
# Mark a welcome-packet document received (or not).
# ---------------------------------------------------------------
@router.post("/api/staff/claims/{claim_number}/onboarding/{doc_key}")
async def staff_set_onboarding(claim_number: str, doc_key: str, request: Request,
staff_name: str = Depends(authmod.require_staff)):
if doc_key not in [k for k, _t in packet.ONBOARDING_DOCS]:
return _err("validation_error", "Unknown document key.", status.HTTP_422_UNPROCESSABLE_ENTITY)
try:
body = await request.json()
except Exception:
return _err("validation_error", "Invalid JSON body.", status.HTTP_422_UNPROCESSABLE_ENTITY)
try:
upd = StaffOnboardingUpdate.model_validate(body)
except ValidationError as exc:
parts = [f"{'.'.join(str(x) for x in e['loc'])}: {e['msg']}" for e in exc.errors()]
return _err("validation_error", "; ".join(parts), status.HTTP_422_UNPROCESSABLE_ENTITY)
with get_conn() as conn:
row = conn.execute("SELECT id FROM claims WHERE claim_number = ?", (claim_number,)).fetchone()
if row is None:
return _err("not_found", "Claim not found.", status.HTTP_404_NOT_FOUND)
packet.ensure_onboarding_docs(conn, row["id"])
now = utcnow_iso()
new_status = "RECEIVED" if upd.received else "PENDING"
received_by = (upd.received_by or staff_name) if upd.received else None
received_at = now if upd.received else None
conn.execute(
"UPDATE onboarding_docs SET status = ?, received_by = ?, received_at = ? "
"WHERE claim_id = ? AND doc_key = ?",
(new_status, received_by, received_at, row["id"], doc_key),
)
conn.execute(
"INSERT INTO audit_log (id, entity_type, entity_id, action, field, old_value, new_value, actor, created_at) "
"VALUES (?, 'claim', ?, 'onboarding_doc', ?, NULL, ?, ?, ?)",
(new_uuid(), row["id"], doc_key, new_status, staff_name, now),
)
conn.commit()
return await staff_get_claim(claim_number)
# ---------------------------------------------------------------
# GET /api/staff/claims
# ---------------------------------------------------------------
@router.get("/api/staff/claims")
async def staff_list_claims(request: Request, _staff=Depends(authmod.require_staff)):
status_filter = request.query_params.get("status")
tier_filter = request.query_params.get("tier")
q = request.query_params.get("q", "").strip()
limit = min(int(request.query_params.get("limit", "50")), 200)
offset = max(int(request.query_params.get("offset", "0")), 0)
where = []
params: list = []
if status_filter:
where.append("c.status = ?")
params.append(status_filter)
if tier_filter:
where.append("c.tier = ?")
params.append(tier_filter)
if q:
where.append("(c.claim_number LIKE ? OR cl.company_name LIKE ? OR d.name LIKE ?)")
like = f"%{q}%"
params.extend([like, like, like])
where_sql = ("WHERE " + " AND ".join(where)) if where else ""
sql = (
"SELECT c.id AS claim_id, c.claim_number, c.status, c.tier, c.amount_cents, c.created_at, c.date_resolved, "
"c.date_assigned, c.analysis_score, c.approval_status, c.recommended_tier, "
"cl.company_name, cl.client_number, d.name AS debtor_name, d.business_type "
"FROM claims c JOIN clients cl ON cl.id = c.client_id JOIN debtors d ON d.id = c.debtor_id "
f"{where_sql} ORDER BY c.created_at DESC LIMIT ? OFFSET ?"
)
params.extend([limit, offset])
with get_conn() as conn:
rows = conn.execute(sql, tuple(params)).fetchall()
total = conn.execute(f"SELECT COUNT(*) AS n FROM claims c JOIN clients cl ON cl.id=c.client_id JOIN debtors d ON d.id=c.debtor_id {where_sql}", tuple(params[:-2])).fetchone()["n"]
claims = []
for r in rows:
ob = packet.onboarding_status(conn, r["claim_id"])
claims.append({
"claim_number": r["claim_number"],
"status": r["status"],
"status_label": STATUS_LABELS.get(r["status"], r["status"]),
"tier": r["tier"],
"tier_label": analysis.TIER_LABELS.get(r["tier"], r["tier"]),
"amount_cents": r["amount_cents"],
"amount_display": _money(r["amount_cents"]),
"analysis_score": r["analysis_score"],
"approval_status": r["approval_status"],
"recommended_tier": r["recommended_tier"],
"company_name": r["company_name"],
"client_number": r["client_number"],
"debtor_name": r["debtor_name"],
"business_type": r["business_type"],
"created_at": r["created_at"],
"date_assigned": r["date_assigned"],
"date_resolved": r["date_resolved"],
"onboarding": {
"received": ob["received"],
"total": ob["total"],
"complete": ob["complete"],
},
})
return {
"claims": claims,
"total": total,
"limit": limit,
"offset": offset,
}
# ---------------------------------------------------------------
# GET /api/staff/claims/{claim_number}
# ---------------------------------------------------------------
@router.get("/api/staff/claims/{claim_number}")
async def staff_get_claim(claim_number: str, _staff=Depends(authmod.require_staff)):
with get_conn() as conn:
row = conn.execute(
"SELECT c.*, cl.company_name, cl.contact_name, cl.email AS client_email, cl.phone AS client_phone, "
"cl.client_number, d.name AS debtor_name, d.business_type, d.contact_email AS debtor_email, "
"d.contact_phone AS debtor_phone, d.physical_address AS debtor_address "
"FROM claims c JOIN clients cl ON cl.id = c.client_id JOIN debtors d ON d.id = c.debtor_id "
"WHERE c.claim_number = ?",
(claim_number,),
).fetchone()
if row is None:
return _err("not_found", "Claim not found.", status.HTTP_404_NOT_FOUND)
docs = conn.execute(
"SELECT id, original_name, size_bytes, mime_type, uploaded_by, created_at FROM documents WHERE claim_id = ?",
(row["id"],),
).fetchall()
notes = conn.execute(
"SELECT author_type, author_name, subject, content, visibility, created_at FROM case_notes "
"WHERE claim_id = ? ORDER BY created_at ASC",
(row["id"],),
).fetchall()
audit = conn.execute(
"SELECT entity_type, entity_id, action, field, old_value, new_value, actor, reason, created_at "
"FROM audit_log WHERE entity_type IN ('claim','document','note') AND entity_id = ? "
"ORDER BY created_at ASC",
(row["id"],),
).fetchall()
onboarding = packet.onboarding_status(conn, row["id"])
return {
"claim_number": row["claim_number"],
"status": row["status"],
"status_label": STATUS_LABELS.get(row["status"], row["status"]),
"tier": row["tier"],
"tier_step": TIER_STEPS.get(row["tier"], 1),
"tier_label": analysis.TIER_LABELS.get(row["tier"], row["tier"]),
"tier_description": analysis.TIER_DESCRIPTIONS.get(row["tier"], ""),
"next_tier": analysis.next_tier(row["tier"]),
"amount_cents": row["amount_cents"],
"amount_display": _money(row["amount_cents"]),
"description": row["description"],
"client_reference": row["client_reference"],
"invoice_date": row["invoice_date"],
"date_assigned": row["date_assigned"],
"date_resolved": row["date_resolved"],
"created_at": row["created_at"],
"analysis": {
"score": row["analysis_score"],
"summary": row["analysis_summary"],
"components": _json_or_none(row["analysis_components"]),
"at": row["analysis_at"],
"recommended_tier": row["recommended_tier"],
},
"approval": {
"status": row["approval_status"],
"decision_by": row["approval_decision_by"],
"decision_at": row["approval_decision_at"],
},
"letter": {
"subject": row["letter_subject"],
"body": row["letter_body"],
"tier": row["letter_tier"],
"updated_at": row["letter_updated_at"],
},
"onboarding": onboarding,
"client": {
"client_number": row["client_number"],
"company_name": row["company_name"],
"contact_name": row["contact_name"],
"email": row["client_email"],
"phone": row["client_phone"],
},
"debtor": {
"name": row["debtor_name"],
"business_type": row["business_type"],
"contact_email": row["debtor_email"],
"contact_phone": row["debtor_phone"],
"physical_address": row["debtor_address"],
},
"documents": [
{
"id": d["id"], "original_name": d["original_name"], "size_bytes": d["size_bytes"],
"mime_type": d["mime_type"], "uploaded_by": d["uploaded_by"], "created_at": d["created_at"],
} for d in docs
],
"notes": [
{
"author_type": n["author_type"], "author_name": n["author_name"], "subject": n["subject"],
"content": html.escape(n["content"]), "visibility": n["visibility"], "created_at": n["created_at"],
} for n in notes
],
"audit": [
{
"entity_type": a["entity_type"], "entity_id": a["entity_id"], "action": a["action"],
"field": a["field"], "old_value": a["old_value"], "new_value": a["new_value"],
"actor": a["actor"], "reason": a["reason"], "created_at": a["created_at"],
} for a in audit
],
}
# ---------------------------------------------------------------
# PATCH /api/staff/claims/{claim_number}
# ---------------------------------------------------------------
@router.patch("/api/staff/claims/{claim_number}")
async def staff_patch_claim(claim_number: str, request: Request, staff_name: str = Depends(authmod.require_staff)):
try:
body = await request.json()
except Exception:
return _err("validation_error", "Invalid JSON body.", status.HTTP_422_UNPROCESSABLE_ENTITY)
try:
patch = StaffClaimPatch.model_validate(body)
except ValidationError as exc:
parts = [f"{'.'.join(str(x) for x in e['loc'])}: {e['msg']}" for e in exc.errors()]
return _err("validation_error", "; ".join(parts), status.HTTP_422_UNPROCESSABLE_ENTITY)
with get_conn() as conn:
row = conn.execute("SELECT * FROM claims WHERE claim_number = ?", (claim_number,)).fetchone()
if row is None:
return _err("not_found", "Claim not found.", status.HTTP_404_NOT_FOUND)
claim_id = row["id"]
now = utcnow_iso()
updates: list[tuple] = []
new_status = row["status"]
new_tier = row["tier"]
date_assigned = row["date_assigned"]
date_resolved = row["date_resolved"]
# Status change
if patch.status is not None and patch.status != row["status"]:
conn.execute(
"INSERT INTO audit_log (id, entity_type, entity_id, action, field, old_value, new_value, actor, reason, created_at) "
"VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
(new_uuid(), "claim", claim_id, "status_change", "status", row["status"], patch.status, staff_name, patch.reason, now),
)
new_status = patch.status
# set date_assigned on first move out of NEW
if row["status"] == "NEW" and patch.status != "NEW" and not date_assigned:
date_assigned = now
# set date_resolved on resolved states
if patch.status in ("SETTLED", "CLOSED", "WRITE_OFF", "REJECTED") and not date_resolved:
date_resolved = now
# auto-add SYSTEM/SHARED note so client sees status change
conn.execute(
"INSERT INTO case_notes (id, claim_id, author_type, author_name, subject, content, visibility, twentycrm_id, created_at) "
"VALUES (?, ?, 'SYSTEM', 'System', NULL, ?, 'SHARED', NULL, ?)",
(new_uuid(), claim_id, f"Status updated to: {STATUS_LABELS.get(patch.status, patch.status)}", now),
)
# Tier change
if patch.tier is not None and patch.tier != row["tier"]:
# Gate escalation: recovery may not escalate until the claim is
# approved & accepted AND all onboarding paperwork is returned.
try:
escalating = analysis.TIER_ORDER.index(patch.tier) > analysis.TIER_ORDER.index(row["tier"])
except ValueError:
escalating = False
if escalating:
gate = _tier_advance_gate(conn, row)
if gate is not None:
return _err("conflict", gate, status.HTTP_409_CONFLICT)
conn.execute(
"INSERT INTO audit_log (id, entity_type, entity_id, action, field, old_value, new_value, actor, reason, created_at) "
"VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
(new_uuid(), "claim", claim_id, "status_change", "tier", row["tier"], patch.tier, staff_name, patch.reason, now),
)
new_tier = patch.tier
conn.execute(
"UPDATE claims SET status = ?, tier = ?, date_assigned = ?, date_resolved = ?, updated_at = ? WHERE id = ?",
(new_status, new_tier, date_assigned, date_resolved, now, claim_id),
)
conn.commit()
# Return updated detail
return await staff_get_claim_inner(claim_number)
async def staff_get_claim_inner(claim_number: str):
"""Reuse detail query without re-checking staff auth."""
with get_conn() as conn:
row = conn.execute(
"SELECT c.*, cl.company_name, cl.contact_name, cl.email AS client_email, cl.phone AS client_phone, "
"cl.client_number, d.name AS debtor_name, d.business_type "
"FROM claims c JOIN clients cl ON cl.id = c.client_id JOIN debtors d ON d.id = c.debtor_id "
"WHERE c.claim_number = ?",
(claim_number,),
).fetchone()
if row is None:
return JSONResponse(status_code=status.HTTP_404_NOT_FOUND,
content={"error": {"code": "not_found", "message": "Claim not found."}})
return {
"claim_number": row["claim_number"],
"status": row["status"],
"status_label": STATUS_LABELS.get(row["status"], row["status"]),
"tier": row["tier"],
"tier_step": TIER_STEPS.get(row["tier"], 1),
"amount_cents": row["amount_cents"],
"amount_display": _money(row["amount_cents"]),
"date_assigned": row["date_assigned"],
"date_resolved": row["date_resolved"],
"updated_at": row["updated_at"],
}
# ---------------------------------------------------------------
# POST /api/staff/claims/{claim_number}/analyze (AI score)
# ---------------------------------------------------------------
@router.post("/api/staff/claims/{claim_number}/analyze")
async def staff_analyze_claim(claim_number: str, _staff=Depends(authmod.require_staff)):
with get_conn() as conn:
row = conn.execute(
"SELECT id FROM claims WHERE claim_number = ?", (claim_number,)
).fetchone()
if row is None:
return _err("not_found", "Claim not found.", status.HTTP_404_NOT_FOUND)
analysis.analyze_and_store(conn, row["id"], actor="staff")
conn.commit()
return await staff_get_claim(claim_number)
# ---------------------------------------------------------------
# POST /api/staff/claims/{claim_number}/approve (accept / reject)
# ---------------------------------------------------------------
@router.post("/api/staff/claims/{claim_number}/approve")
async def staff_approve_claim(claim_number: str, request: Request, staff_name: str = Depends(authmod.require_staff)):
try:
body = await request.json()
except Exception:
return _err("validation_error", "Invalid JSON body.", status.HTTP_422_UNPROCESSABLE_ENTITY)
try:
approval = StaffApproval.model_validate(body)
except ValidationError as exc:
parts = [f"{'.'.join(str(x) for x in e['loc'])}: {e['msg']}" for e in exc.errors()]
return _err("validation_error", "; ".join(parts), status.HTTP_422_UNPROCESSABLE_ENTITY)
with get_conn() as conn:
row = conn.execute("SELECT * FROM claims WHERE claim_number = ?", (claim_number,)).fetchone()
if row is None:
return _err("not_found", "Claim not found.", status.HTTP_404_NOT_FOUND)
if row["analysis_score"] is None:
return _err("conflict", "Claim has not been analyzed yet. Run analysis first.", status.HTTP_409_CONFLICT)
claim_id = row["id"]
actor = approval.staff_name or staff_name
now = utcnow_iso()
if approval.decision == "APPROVE":
new_approval = "APPROVED"
new_status = "ACTIVE"
# Ensure date_assigned is set on first move out of NEW
date_assigned = row["date_assigned"] or now
conn.execute(
"UPDATE claims SET approval_status = ?, approval_decision_by = ?, approval_decision_at = ?, "
"status = ?, date_assigned = ?, updated_at = ? WHERE id = ?",
(new_approval, actor, now, new_status, date_assigned, now, claim_id),
)
conn.execute(
"INSERT INTO case_notes (id, claim_id, author_type, author_name, subject, content, visibility, twentycrm_id, created_at) "
"VALUES (?, ?, 'SYSTEM', 'System', NULL, ?, 'SHARED', NULL, ?)",
(new_uuid(), claim_id, f"Claim approved by {actor} and moved to In Progress.", now),
)
# Seed the onboarding paperwork checklist on approval.
packet.ensure_onboarding_docs(conn, claim_id)
else: # REJECT
new_approval = "REJECTED"
new_status = "REJECTED"
conn.execute(
"UPDATE claims SET approval_status = ?, approval_decision_by = ?, approval_decision_at = ?, "
"status = ?, date_resolved = ?, updated_at = ? WHERE id = ?",
(new_approval, actor, now, new_status, now, now, claim_id),
)
conn.execute(
"INSERT INTO case_notes (id, claim_id, author_type, author_name, subject, content, visibility, twentycrm_id, created_at) "
"VALUES (?, ?, 'SYSTEM', 'System', NULL, ?, 'SHARED', NULL, ?)",
(new_uuid(), claim_id, f"Claim rejected by {actor}.", now),
)
conn.execute(
"INSERT INTO audit_log (id, entity_type, entity_id, action, field, old_value, new_value, actor, reason, created_at) "
"VALUES (?, 'claim', ?, 'status_change', 'approval_status', ?, ?, ?, ?, ?)",
(new_uuid(), claim_id, row["approval_status"], new_approval, actor, approval.reason, now),
)
conn.commit()
return await staff_get_claim(claim_number)
# ---------------------------------------------------------------
# POST /api/staff/claims/{claim_number}/letter (generate recommended letter)
# Optional JSON body: {"tier": "TIER_2"} to generate for an explicit tier.
# Defaults to the claim's current tier when omitted.
# ---------------------------------------------------------------
@router.post("/api/staff/claims/{claim_number}/letter")
async def staff_generate_letter(claim_number: str, request: Request, _staff=Depends(authmod.require_staff)):
try:
body = await request.json()
except Exception:
body = {}
override_tier = (body or {}).get("tier")
if override_tier is not None and override_tier not in analysis.TIER_ORDER:
return _err("validation_error", "tier must be one of: " + ", ".join(analysis.TIER_ORDER), status.HTTP_422_UNPROCESSABLE_ENTITY)
with get_conn() as conn:
row = conn.execute(
"SELECT c.id, c.claim_number, c.tier, c.amount_cents, c.client_reference, c.invoice_date, "
"cl.company_name, d.name AS debtor_name, d.physical_address "
"FROM claims c JOIN clients cl ON cl.id = c.client_id JOIN debtors d ON d.id = c.debtor_id "
"WHERE c.claim_number = ?",
(claim_number,),
).fetchone()
if row is None:
return _err("not_found", "Claim not found.", status.HTTP_404_NOT_FOUND)
letter = analysis.recommend_letter(
tier=override_tier or row["tier"],
client_name=row["company_name"],
debtor_name=row["debtor_name"],
amount_display=_money(row["amount_cents"]),
invoice_ref=row["client_reference"],
claim_number=row["claim_number"],
debtor_address=row["physical_address"],
)
now = utcnow_iso()
conn.execute(
"UPDATE claims SET letter_subject = ?, letter_body = ?, letter_tier = ?, letter_updated_at = ?, updated_at = ? WHERE id = ?",
(letter["subject"], letter["body"], letter["tier"], now, now, row["id"]),
)
conn.execute(
"INSERT INTO audit_log (id, entity_type, entity_id, action, field, old_value, new_value, actor, created_at) "
"VALUES (?, 'claim', ?, 'letter_generate', 'letter', NULL, ?, 'staff', ?)",
(new_uuid(), row["id"], letter["tier"], now),
)
conn.commit()
return await staff_get_claim(claim_number)
# ---------------------------------------------------------------
# PUT /api/staff/claims/{claim_number}/letter (save customized letter)
# ---------------------------------------------------------------
@router.put("/api/staff/claims/{claim_number}/letter")
async def staff_save_letter(claim_number: str, request: Request, staff_name: str = Depends(authmod.require_staff)):
try:
body = await request.json()
except Exception:
return _err("validation_error", "Invalid JSON body.", status.HTTP_422_UNPROCESSABLE_ENTITY)
try:
letter = StaffLetterUpdate.model_validate(body)
except ValidationError as exc:
parts = [f"{'.'.join(str(x) for x in e['loc'])}: {e['msg']}" for e in exc.errors()]
return _err("validation_error", "; ".join(parts), status.HTTP_422_UNPROCESSABLE_ENTITY)
actor = letter.staff_name or staff_name
with get_conn() as conn:
row = conn.execute("SELECT id FROM claims WHERE claim_number = ?", (claim_number,)).fetchone()
if row is None:
return _err("not_found", "Claim not found.", status.HTTP_404_NOT_FOUND)
now = utcnow_iso()
conn.execute(
"UPDATE claims SET letter_subject = ?, letter_body = ?, letter_updated_at = ?, updated_at = ? WHERE id = ?",
(letter.subject, letter.body, now, now, row["id"]),
)
conn.execute(
"INSERT INTO audit_log (id, entity_type, entity_id, action, field, old_value, new_value, actor, created_at) "
"VALUES (?, 'claim', ?, 'letter_edit', 'letter_body', NULL, 'edited', ?, ?)",
(new_uuid(), row["id"], actor, now),
)
conn.commit()
return await staff_get_claim(claim_number)
# ---------------------------------------------------------------
# POST /api/staff/claims/{claim_number}/advance-tier (manual escalation)
# ---------------------------------------------------------------
@router.post("/api/staff/claims/{claim_number}/advance-tier")
async def staff_advance_tier(claim_number: str, request: Request, staff_name: str = Depends(authmod.require_staff)):
try:
body = await request.json()
except Exception:
body = {}
actor = (body or {}).get("staff_name") or staff_name
reason = (body or {}).get("reason")
with get_conn() as conn:
row = conn.execute("SELECT * FROM claims WHERE claim_number = ?", (claim_number,)).fetchone()
if row is None:
return _err("not_found", "Claim not found.", status.HTTP_404_NOT_FOUND)
nxt = analysis.next_tier(row["tier"])
if nxt is None:
return _err("conflict", "Claim is already at the final tier (Tier 4 - Legal Action).", status.HTTP_409_CONFLICT)
gate = _tier_advance_gate(conn, row)
if gate is not None:
return _err("conflict", gate, status.HTTP_409_CONFLICT)
now = utcnow_iso()
conn.execute(
"UPDATE claims SET tier = ?, updated_at = ? WHERE id = ?",
(nxt, now, row["id"]),
)
conn.execute(
"INSERT INTO audit_log (id, entity_type, entity_id, action, field, old_value, new_value, actor, reason, created_at) "
"VALUES (?, 'claim', ?, 'status_change', 'tier', ?, ?, ?, ?, ?)",
(new_uuid(), row["id"], row["tier"], nxt, actor, reason, now),
)
conn.execute(
"INSERT INTO case_notes (id, claim_id, author_type, author_name, subject, content, visibility, twentycrm_id, created_at) "
"VALUES (?, ?, 'SYSTEM', 'System', NULL, ?, 'SHARED', NULL, ?)",
(new_uuid(), row["id"],
f"Case escalated from {analysis.TIER_LABELS.get(row['tier'], row['tier'])} to {analysis.TIER_LABELS.get(nxt, nxt)}.", now),
)
conn.commit()
return await staff_get_claim(claim_number)
# ---------------------------------------------------------------
# POST /api/staff/claims/{claim_number}/notes
# ---------------------------------------------------------------
@router.post("/api/staff/claims/{claim_number}/notes")
async def staff_add_note(claim_number: str, request: Request, staff_name: str = Depends(authmod.require_staff)):
try:
body = await request.json()
except Exception:
return _err("validation_error", "Invalid JSON body.", status.HTTP_422_UNPROCESSABLE_ENTITY)
try:
note = StaffNoteCreate.model_validate(body)
except ValidationError as exc:
parts = [f"{'.'.join(str(x) for x in e['loc'])}: {e['msg']}" for e in exc.errors()]
return _err("validation_error", "; ".join(parts), status.HTTP_422_UNPROCESSABLE_ENTITY)
actor = note.author_name or staff_name
with get_conn() as conn:
row = conn.execute("SELECT id FROM claims WHERE claim_number = ?", (claim_number,)).fetchone()
if row is None:
return _err("not_found", "Claim not found.", status.HTTP_404_NOT_FOUND)
claim_id = row["id"]
now = utcnow_iso()
note_id = new_uuid()
conn.execute(
"INSERT INTO case_notes (id, claim_id, author_type, author_name, subject, content, visibility, twentycrm_id, created_at) "
"VALUES (?, ?, 'STAFF', ?, NULL, ?, ?, NULL, ?)",
(note_id, claim_id, actor, note.content, note.visibility, now),
)
conn.execute(
"INSERT INTO audit_log (id, entity_type, entity_id, action, actor, created_at) VALUES (?, ?, ?, ?, ?, ?)",
(new_uuid(), "note", note_id, "note_add", f"staff:{actor}", now),
)
conn.commit()
return {
"id": note_id,
"author_type": "STAFF",
"author_name": actor,
"content": html.escape(note.content),
"visibility": note.visibility,
"created_at": now,
}
# ---------------------------------------------------------------
# POST /api/staff/claims/{claim_number}/documents
# ---------------------------------------------------------------
@router.post("/api/staff/claims/{claim_number}/documents")
async def staff_upload_document(claim_number: str, request: Request, file: UploadFile = File(...),
_staff=Depends(authmod.require_staff)):
cl = request.headers.get("content-length")
if cl and int(cl) > MAX_FILE_BYTES + 4096:
return _err("payload_too_large", "File exceeds 20 MB limit.", status.HTTP_413_REQUEST_ENTITY_TOO_LARGE)
ext = os.path.splitext(file.filename or "")[1].lower()
if ext not in ALLOWED_EXT:
return _err("unsupported_media_type", "File type not allowed.", status.HTTP_415_UNSUPPORTED_MEDIA_TYPE)
kind = EXT_TO_KIND.get(ext)
if not kind:
return _err("unsupported_media_type", "File type not allowed.", status.HTTP_415_UNSUPPORTED_MEDIA_TYPE)
with get_conn() as conn:
row = conn.execute("SELECT id FROM claims WHERE claim_number = ?", (claim_number,)).fetchone()
if row is None:
return _err("not_found", "Claim not found.", status.HTTP_404_NOT_FOUND)
claim_id = row["id"]
count = conn.execute("SELECT COUNT(*) AS n FROM documents WHERE claim_id = ?", (claim_id,)).fetchone()["n"]
if count >= 50:
return _err("conflict", "Document limit reached for this claim.", status.HTTP_409_CONFLICT)
upload_dir = os.path.join(get_upload_dir(), claim_id)
os.makedirs(upload_dir, exist_ok=True)
stored_uuid = str(uuid.uuid4())
stored_name = f"{stored_uuid}{ext}"
stored_path = os.path.join(upload_dir, stored_name)
sha = hashlib.sha256()
total = 0
magic_seen = False
with open(stored_path, "wb") as f:
while True:
chunk = await file.read(64 * 1024)
if not chunk:
break
total += len(chunk)
if total > MAX_FILE_BYTES:
f.close()
os.remove(stored_path)
return _err("payload_too_large", "File exceeds 20 MB limit.", status.HTTP_413_REQUEST_ENTITY_TOO_LARGE)
if not magic_seen:
sigs = MAGIC_BYTES.get(kind, ())
if sigs and any(chunk.startswith(s) for s in sigs):
magic_seen = True
elif sigs:
f.close()
os.remove(stored_path)
return _err("unsupported_media_type", "File content does not match extension.", status.HTTP_415_UNSUPPORTED_MEDIA_TYPE)
sha.update(chunk)
f.write(chunk)
os.chmod(stored_path, 0o640)
now = utcnow_iso()
doc_id = new_uuid()
mime = file.content_type or "application/octet-stream"
conn.execute(
"INSERT INTO documents (id, claim_id, original_name, stored_path, mime_type, size_bytes, sha256, uploaded_by, twentycrm_id, created_at) "
"VALUES (?, ?, ?, ?, ?, ?, ?, 'STAFF', NULL, ?)",
(doc_id, claim_id, os.path.basename(file.filename or "file"), stored_path, mime, total, sha.hexdigest(), now),
)
conn.execute(
"INSERT INTO audit_log (id, entity_type, entity_id, action, actor, created_at) VALUES (?, ?, ?, ?, ?, ?)",
(new_uuid(), "document", doc_id, "upload", "staff", now),
)
conn.commit()
return {
"id": doc_id,
"original_name": os.path.basename(file.filename or "file"),
"size_bytes": total,
"mime_type": mime,
"uploaded_by": "STAFF",
"created_at": now,
}
# ---------------------------------------------------------------
# GET /api/staff/stats
# ---------------------------------------------------------------
@router.get("/api/staff/stats")
async def staff_stats(_staff=Depends(authmod.require_staff)):
with get_conn() as conn:
total = conn.execute("SELECT COUNT(*) AS n FROM claims").fetchone()["n"]
by_status = {}
for r in conn.execute("SELECT status, COUNT(*) AS n FROM claims GROUP BY status").fetchall():
by_status[r["status"]] = r["n"]
by_tier = {}
for r in conn.execute("SELECT tier, COUNT(*) AS n FROM claims GROUP BY tier").fetchall():
by_tier[r["tier"]] = r["n"]
total_amount = conn.execute("SELECT COALESCE(SUM(amount_cents),0) AS s FROM claims").fetchone()["s"]
# recovered = claims in SETTLED/CLOSED
recovered = conn.execute(
"SELECT COALESCE(SUM(amount_cents),0) AS s FROM claims WHERE status IN ('SETTLED','CLOSED')"
).fetchone()["s"]
open_amount = total_amount - recovered
# aging: claims not resolved, by age
now_iso = utcnow_iso()
over_30 = conn.execute(
"SELECT COUNT(*) AS n FROM claims WHERE date_resolved IS NULL AND created_at < ?",
(over_30_cutoff(),),
).fetchone()["n"]
over_60 = conn.execute(
"SELECT COUNT(*) AS n FROM claims WHERE date_resolved IS NULL AND created_at < ?",
(over_60_cutoff(),),
).fetchone()["n"]
return {
"total_claims": total,
"by_status": by_status,
"by_tier": by_tier,
"total_amount_cents": total_amount,
"total_amount_display": _money(total_amount),
"recovered_amount_cents": recovered,
"recovered_amount_display": _money(recovered),
"open_amount_cents": open_amount,
"open_amount_display": _money(open_amount),
"aging": {"over_30_days": over_30, "over_60_days": over_60},
}
def over_30_cutoff() -> str:
from datetime import datetime, timedelta, timezone
return (datetime.now(timezone.utc) - timedelta(days=30)).strftime("%Y-%m-%dT%H:%M:%SZ")
def over_60_cutoff() -> str:
from datetime import datetime, timedelta, timezone
return (datetime.now(timezone.utc) - timedelta(days=60)).strftime("%Y-%m-%dT%H:%M:%SZ")
# ---------------------------------------------------------------
# GET /api/staff/audit
# ---------------------------------------------------------------
@router.get("/api/staff/audit")
async def staff_audit(request: Request, _staff=Depends(authmod.require_staff)):
entity_type = request.query_params.get("entity_type")
entity_id = request.query_params.get("entity_id")
limit = min(int(request.query_params.get("limit", "100")), 500)
offset = max(int(request.query_params.get("offset", "0")), 0)
where = []
params: list = []
if entity_type:
where.append("entity_type = ?")
params.append(entity_type)
if entity_id:
where.append("entity_id = ?")
params.append(entity_id)
where_sql = ("WHERE " + " AND ".join(where)) if where else ""
with get_conn() as conn:
rows = conn.execute(
f"SELECT * FROM audit_log {where_sql} ORDER BY created_at DESC LIMIT ? OFFSET ?",
tuple(params + [limit, offset]),
).fetchall()
total = conn.execute(f"SELECT COUNT(*) AS n FROM audit_log {where_sql}", tuple(params)).fetchone()["n"]
return {
"audit": [
{
"id": r["id"], "entity_type": r["entity_type"], "entity_id": r["entity_id"],
"action": r["action"], "field": r["field"], "old_value": r["old_value"],
"new_value": r["new_value"], "actor": r["actor"], "reason": r["reason"],
"created_at": r["created_at"],
} for r in rows
],
"total": total,
"limit": limit,
"offset": offset,
}
Reference in New Issue View Git Blame Copy Permalink