"""安全问卷 · 乙方侧客户问卷 (inbound questionnaire) — service layer. When a CUSTOMER sends us a security questionnaire: intake (customer / due date / owner / original file) → add questions (manual / paste / xlsx import) → AI drafts answers by matching the answer library → human review → export the filled workbook, archiving each sent version for accountability. Answered questions can be backfilled into the answer library so the next questionnaire is faster. Naive-UTC datetime discipline (codebase-wide MySQL/TiDB convention). """ from __future__ import annotations import calendar import io import uuid from datetime import UTC, date, datetime from typing import Any from sqlalchemy import func, select from sqlalchemy.ext.asyncio import AsyncSession from app.core.exceptions import NotFoundError, ValidationError from app.core.exporting import to_cell from app.core.storage import presign_get, presign_put, put_object_bytes from app.domains.auditor import answer_library_service as lib_svc from app.domains.auditor.models import ( AuditorInboundQuestion, AuditorInboundQuestionnaire, AuditorInboundVersion, ) from app.identity.person_models import IdentityPerson STATUSES: tuple[str, ...] = ( "received", "answering", "review", "completed", "archived", "cancelled", ) EDITABLE_STATUSES: frozenset[str] = frozenset( {"received", "answering", "review"}, ) QUESTION_STATUSES: tuple[str, ...] = ("pending", "drafted", "reviewed") _SOURCE_KEY_PREFIX = "inbound_questionnaire" _MAX_IMPORT_BYTES = 20 * 1024 * 1024 # The customer's original questionnaire file — office docs / pdf / images. # Scripts / html excluded so a stored object can't be re-served as active # content via the presigned S3 GET. Mirrors the evidence-doc allowlist. ALLOWED_SOURCE_CONTENT_TYPES: frozenset[str] = frozenset({ "application/pdf", "image/jpeg", "image/png", "image/webp", "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", "application/vnd.ms-excel", "application/vnd.openxmlformats-officedocument.wordprocessingml.document", "application/msword", }) # Tall-sheet guard complements the archive and column bounds in xlsx_safety. _MAX_IMPORT_ROWS = 5000 def _naive_now() -> datetime: return datetime.now(UTC).replace(tzinfo=None) def _to_naive(dt: datetime | None) -> datetime | None: if dt is None: return None if dt.tzinfo is not None: return dt.astimezone(UTC).replace(tzinfo=None) return dt def _add_months(d: date, months: int) -> date: total = (d.month - 1) + months year = d.year + total // 12 month = total % 12 + 1 day = min(d.day, calendar.monthrange(year, month)[1]) return date(year, month, day) async def _persons_index( db: AsyncSession, ids: set[uuid.UUID], ) -> dict[uuid.UUID, IdentityPerson]: ids = {i for i in ids if i} if not ids: return {} rows = (await db.execute( select(IdentityPerson).where(IdentityPerson.id.in_(list(ids))) )).scalars().all() return {p.id: p for p in rows} def _person_brief(p: IdentityPerson | None) -> dict[str, Any] | None: if p is None: return None return {"id": str(p.id), "display_name": p.display_name, "work_email": p.work_email} # ════════════════════════════════════════════════════════════════════ # Questionnaire lifecycle # ════════════════════════════════════════════════════════════════════ async def get_questionnaire( db: AsyncSession, qid: uuid.UUID, ) -> AuditorInboundQuestionnaire: q = await db.get(AuditorInboundQuestionnaire, qid) if q is None: raise NotFoundError("inbound questionnaire not found") return q async def create_questionnaire( db: AsyncSession, *, payload: dict[str, Any], actor_id: uuid.UUID | None, ) -> AuditorInboundQuestionnaire: if not (payload.get("customer_name") or "").strip(): raise ValidationError("customer_name is required") if not (payload.get("title") or "").strip(): raise ValidationError("title is required") due = payload.get("due_date") if isinstance(due, str) and due: due = date.fromisoformat(due) q = AuditorInboundQuestionnaire( customer_name=payload["customer_name"].strip(), title=payload["title"].strip(), due_date=due or None, owner_person_id=payload.get("owner_person_id") or actor_id, status="received", # source_object_key is NOT accepted here — the only write path is the # key-validated update_meta (register source), so a create call can't # bind an unvalidated arbitrary bucket key. notes=payload.get("notes") or None, created_by_person_id=actor_id, ) db.add(q) await db.flush() return q async def update_meta( db: AsyncSession, qid: uuid.UUID, *, payload: dict[str, Any], ) -> AuditorInboundQuestionnaire: q = await get_questionnaire(db, qid) if q.status in ("archived", "cancelled"): raise ValidationError("cannot edit an archived/cancelled questionnaire") if "customer_name" in payload and payload["customer_name"]: q.customer_name = payload["customer_name"].strip() if "title" in payload and payload["title"]: q.title = payload["title"].strip() if "due_date" in payload: due = payload["due_date"] q.due_date = date.fromisoformat(due) if isinstance(due, str) and due else (due or None) if "owner_person_id" in payload: q.owner_person_id = payload["owner_person_id"] if "notes" in payload: q.notes = payload["notes"] or None if "source_object_key" in payload: key = payload["source_object_key"] or None if key: _validate_source_key(key) q.source_object_key = key q.source_filename = payload.get("source_filename") or q.source_filename await db.flush() return q async def set_status( db: AsyncSession, qid: uuid.UUID, *, status: str, ) -> AuditorInboundQuestionnaire: if status not in STATUSES: raise ValidationError(f"status must be one of {STATUSES}") q = await get_questionnaire(db, qid) q.status = status if status == "completed" and q.completed_at is None: q.completed_at = _naive_now() await db.flush() return q async def delete_questionnaire(db: AsyncSession, qid: uuid.UUID) -> None: q = await get_questionnaire(db, qid) await db.delete(q) # questions + versions cascade await db.flush() # ── source file (original customer questionnaire) upload ────────────── def _validate_source_key(object_key: str) -> None: """Reject a client-supplied ``source_object_key`` that isn't a real upload under this feature's namespace. The stored key is later handed to ``presign_get`` — a URL that bypasses Helix auth once minted. Without this guard an auditor could register an arbitrary bucket object (another domain's upload, a cloud-connector export) and exfiltrate it. Segment-exact (not substring) + no ``..``. Mirrors the evidence-doc ``_validate_key`` hardening. (Cross-binding one questionnaire's file to another is not gated — both are auditor-visible in this single-tenant domain, so it grants no additional access.) """ segments = object_key.split("/") if ".." in segments: raise ValidationError("source_object_key must not contain '..' segments") if _SOURCE_KEY_PREFIX not in segments: raise ValidationError( "source_object_key is not under the inbound-questionnaire namespace", ) def _source_relative_key(*, qid: uuid.UUID, filename: str) -> str: from app.storage_router import _safe_filename today = _naive_now().strftime("%Y/%m/%d") rand = uuid.uuid4().hex[:12] return f"{_SOURCE_KEY_PREFIX}/{today}/{qid}/{rand}-{_safe_filename(filename)}" async def presign_source_upload( db: AsyncSession, *, qid: uuid.UUID, person_id: uuid.UUID, filename: str, content_type: str, size_bytes: int, checksum_sha256: str, ) -> dict[str, Any]: if content_type not in ALLOWED_SOURCE_CONTENT_TYPES: raise ValidationError( f"unsupported content type '{content_type}' — allowed: " "pdf / image / office doc", ) await get_questionnaire(db, qid) from app.core.storage_uploads import create_scoped_upload_grant grant = await create_scoped_upload_grant( db, person_id=person_id, purpose=_SOURCE_KEY_PREFIX, relative_key=_source_relative_key(qid=qid, filename=filename), content_type=content_type, size_bytes=size_bytes, allowed_content_types=ALLOWED_SOURCE_CONTENT_TYPES, max_bytes=_MAX_IMPORT_BYTES, checksum_sha256=checksum_sha256, ) await db.commit() return await presign_put( db, object_key=grant.object_key, content_type=content_type, size_bytes=grant.declared_size_bytes, checksum_sha256=grant.checksum_sha256, ) async def presign_source_download(db: AsyncSession, qid: uuid.UUID) -> str | None: q = await get_questionnaire(db, qid) if not q.source_object_key: return None # Belt-and-suspenders: never mint a bypassing GET for a stored key that # isn't in this questionnaire's namespace (guards a poisoned row). _validate_source_key(q.source_object_key) return await presign_get(db, object_key=q.source_object_key) # ════════════════════════════════════════════════════════════════════ # Questions # ════════════════════════════════════════════════════════════════════ async def _next_seq(db: AsyncSession, qid: uuid.UUID) -> int: mx = (await db.execute( select(func.max(AuditorInboundQuestion.seq)).where( AuditorInboundQuestion.questionnaire_id == qid, ) )).scalar() return (mx or 0) + 1 async def add_questions( db: AsyncSession, qid: uuid.UUID, *, items: list[dict[str, Any]], ) -> int: """Bulk-add questions from a list of ``{text, section?}`` (manual/paste).""" q = await get_questionnaire(db, qid) if q.status not in EDITABLE_STATUSES: raise ValidationError("questions can only be added while received/answering/review") seq = await _next_seq(db, qid) added = 0 for it in items: text = (it.get("text") or "").strip() if not text: continue db.add(AuditorInboundQuestion( questionnaire_id=qid, seq=seq, section=(it.get("section") or None), question_text=text, status="pending", )) seq += 1 added += 1 if added and q.status == "received": q.status = "answering" await db.flush() return added async def import_questions_xlsx( db: AsyncSession, qid: uuid.UUID, *, payload: bytes, ) -> dict[str, Any]: """Parse a question column out of an uploaded xlsx and add them.""" from openpyxl import load_workbook from app.core.xlsx_safety import ( MAX_XLSX_COLUMNS, bounded_row, preflight_xlsx, ) if len(payload) > _MAX_IMPORT_BYTES: raise ValidationError("uploaded file exceeds 20 MB limit") preflight_xlsx(payload) wb = None try: wb = load_workbook(io.BytesIO(payload), read_only=True, data_only=True) ws = wb.active if ws.max_column and ws.max_column > MAX_XLSX_COLUMNS: raise ValidationError(f"workbook exceeds {MAX_XLSX_COLUMNS} columns") rows = [] for r in ws.iter_rows(values_only=True, max_col=MAX_XLSX_COLUMNS + 1): rows.append(bounded_row(r)) if len(rows) > _MAX_IMPORT_ROWS: raise ValidationError( f"the workbook exceeds {_MAX_IMPORT_ROWS} rows — split it first", ) except ValidationError: raise except Exception as exc: # noqa: BLE001 raise ValidationError(f"could not read xlsx: {exc}") from exc finally: if wb is not None: wb.close() if not rows: raise ValidationError("the workbook is empty") # Locate a question column + optional section column by header, else # fall back to the first non-empty column. def _norm(v: Any) -> str: return str(v).strip().lower() if v is not None else "" q_col = sec_col = None header_idx = 0 for i, row in enumerate(rows[:5]): for j, cell in enumerate(_norm(c) for c in row): if any(t in cell for t in ("question", "问题", "问题描述", "控制项", "requirement")): q_col = j elif any(t in cell for t in ("section", "分区", "类别", "domain", "category")): sec_col = j if q_col is not None: header_idx = i break data_rows = rows[header_idx + 1:] if q_col is not None else rows if q_col is None: q_col = 0 # first column fallback items: list[dict[str, Any]] = [] for row in data_rows: if q_col >= len(row): continue text = row[q_col] if text is None or not str(text).strip(): continue section = None if sec_col is not None and sec_col < len(row) and row[sec_col] is not None: section = str(row[sec_col]).strip() or None items.append({"text": str(text).strip(), "section": section}) added = await add_questions(db, qid, items=items) return {"added": added, "parsed": len(items)} async def get_question( db: AsyncSession, question_id: uuid.UUID, ) -> AuditorInboundQuestion: q = await db.get(AuditorInboundQuestion, question_id) if q is None: raise NotFoundError("inbound question not found") return q async def set_answer( db: AsyncSession, question_id: uuid.UUID, *, answer_text: str | None, matched_library_entry_id: uuid.UUID | None = None, ) -> AuditorInboundQuestion: q = await get_question(db, question_id) parent = await get_questionnaire(db, q.questionnaire_id) if parent.status not in EDITABLE_STATUSES: raise ValidationError("answers can only be edited while received/answering/review") if answer_text is not None: q.answer_text = answer_text or None if matched_library_entry_id is not None: q.matched_library_entry_id = matched_library_entry_id await lib_svc.increment_usage(db, matched_library_entry_id) if q.answer_text and q.status == "pending": q.status = "drafted" await db.flush() return q async def review_question( db: AsyncSession, question_id: uuid.UUID, *, reviewer_id: uuid.UUID | None, reviewed: bool, ) -> AuditorInboundQuestion: q = await get_question(db, question_id) if reviewed: if not (q.answer_text or "").strip(): raise ValidationError("cannot mark reviewed without an answer") q.status = "reviewed" q.reviewed_by_person_id = reviewer_id q.reviewed_at = _naive_now() else: q.status = "drafted" if (q.answer_text or "").strip() else "pending" q.reviewed_by_person_id = None q.reviewed_at = None await db.flush() return q async def delete_question(db: AsyncSession, question_id: uuid.UUID) -> uuid.UUID: q = await get_question(db, question_id) qid = q.questionnaire_id await db.delete(q) await db.flush() return qid # ════════════════════════════════════════════════════════════════════ # AI-assisted drafting # ════════════════════════════════════════════════════════════════════ async def ai_draft_pending( db: AsyncSession, qid: uuid.UUID, *, only_unanswered: bool = True, ) -> dict[str, Any]: """Match pending questions to the answer library + draft answers (Bailian).""" from app.domains.auditor import inbound_ai parent = await get_questionnaire(db, qid) if parent.status not in EDITABLE_STATUSES: raise ValidationError("AI drafting is only allowed while received/answering/review") q_stmt = select(AuditorInboundQuestion).where( AuditorInboundQuestion.questionnaire_id == qid, ) if only_unanswered: q_stmt = q_stmt.where( (AuditorInboundQuestion.answer_text.is_(None)) | (AuditorInboundQuestion.answer_text == ""), ) questions = list((await db.execute(q_stmt)).scalars().all()) if not questions: return {"drafted": 0, "matched": 0} library = await lib_svc.list_active_for_matching(db) # AI build raises BAILIAN_LLM_NO_KEY (503) if no key — propagate. results = await inbound_ai.match_and_draft( db, questions=[{"id": str(q.id), "text": q.question_text} for q in questions], library=library, ) drafted = matched = 0 for q in questions: r = results.get(str(q.id)) if not r or not r.get("draft"): continue q.answer_text = r["draft"] q.status = "drafted" drafted += 1 if r.get("matched_id"): try: mid = uuid.UUID(r["matched_id"]) except (ValueError, TypeError): mid = None if mid is not None: q.matched_library_entry_id = mid q.match_confidence = r.get("confidence") await lib_svc.increment_usage(db, mid) matched += 1 if drafted and parent.status == "received": parent.status = "answering" await db.flush() return {"drafted": drafted, "matched": matched, "total": len(questions)} async def backfill_question_to_library( db: AsyncSession, question_id: uuid.UUID, *, category: str, actor_id: uuid.UUID | None, existing_entry_id: uuid.UUID | None = None, ) -> dict[str, Any]: """Promote an answered inbound question into the answer library.""" q = await get_question(db, question_id) if not (q.answer_text or "").strip(): raise ValidationError("question has no answer to backfill") entry = await lib_svc.backfill_from_answer( db, question_text=q.question_text, answer_text=q.answer_text, category=category, actor_id=actor_id, existing_entry_id=existing_entry_id, ) # Link the source question to the (possibly new) library entry. q.matched_library_entry_id = entry.id await db.flush() return {"entry_id": str(entry.id)} # ════════════════════════════════════════════════════════════════════ # Export + version archive # ════════════════════════════════════════════════════════════════════ def _build_filled_workbook( q: AuditorInboundQuestionnaire, questions: list[AuditorInboundQuestion], ) -> io.BytesIO: from openpyxl import Workbook from openpyxl.styles import Alignment, Font, PatternFill wb = Workbook() ws = wb.active ws.title = "Responses" header_fill = PatternFill("solid", fgColor="4472C4") header_font = Font(color="FFFFFF", bold=True) wrap = Alignment(wrap_text=True, vertical="top") ws.merge_cells("A1:D1") # to_cell() neutralizes CSV/formula injection: question_text (and # section) originate from an imported CUSTOMER xlsx, so a cell like # ``=HYPERLINK(...)`` would otherwise execute when a colleague opens # the re-exported workbook. Mirrors the platform-wide build_xlsx guard. ws["A1"] = to_cell(f"{q.customer_name} — {q.title}") ws["A1"].font = Font(bold=True, size=14, color="1F3864") cols = [("#", 6), ("分区 / Section", 22), ("问题 / Question", 60), ("回答 / Answer", 70)] for idx, (label, width) in enumerate(cols, 1): c = ws.cell(row=2, column=idx, value=label) c.fill = header_fill c.font = header_font c.alignment = Alignment(horizontal="center", wrap_text=True) ws.column_dimensions[c.column_letter].width = width r = 2 for qn in questions: r += 1 ws.cell(row=r, column=1, value=qn.seq) ws.cell(row=r, column=2, value=to_cell(qn.section or "")).alignment = wrap ws.cell(row=r, column=3, value=to_cell(qn.question_text)).alignment = wrap ws.cell(row=r, column=4, value=to_cell(qn.answer_text or "")).alignment = wrap ws.freeze_panes = "A3" buf = io.BytesIO() wb.save(buf) buf.seek(0) return buf async def _ordered_questions( db: AsyncSession, qid: uuid.UUID, ) -> list[AuditorInboundQuestion]: return list((await db.execute( select(AuditorInboundQuestion).where( AuditorInboundQuestion.questionnaire_id == qid, ).order_by(AuditorInboundQuestion.seq) )).scalars().all()) async def export_and_archive( db: AsyncSession, qid: uuid.UUID, *, actor_id: uuid.UUID | None, note: str | None = None, ) -> tuple[io.BytesIO, str, int]: """Build the filled workbook, archive a copy as a new version, and return ``(buffer, filename, version_no)`` for the caller to stream. Archiving the exact sent bytes is the accountability trail — what we promised which customer, when. """ q = await get_questionnaire(db, qid) questions = await _ordered_questions(db, qid) buf = _build_filled_workbook(q, questions) data = buf.getvalue() version_no = ((await db.execute( select(func.max(AuditorInboundVersion.version_no)).where( AuditorInboundVersion.questionnaire_id == qid, ) )).scalar() or 0) + 1 from app.storage_router import _safe_filename safe_customer = _safe_filename(q.customer_name)[:40] stamp = _naive_now().strftime("%Y%m%d_%H%M%S") filename = f"{safe_customer}_v{version_no}_{stamp}.xlsx" object_key = await put_object_bytes( db, relative_key=f"{_SOURCE_KEY_PREFIX}/{qid}/exports/{filename}", content=data, content_type=( "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet" ), ) db.add(AuditorInboundVersion( questionnaire_id=qid, version_no=version_no, object_key=object_key, filename=filename, note=note or None, exported_by_person_id=actor_id, exported_at=_naive_now(), )) q.exported_at = _naive_now() await db.flush() return io.BytesIO(data), filename, version_no async def presign_version_download( db: AsyncSession, version_id: uuid.UUID, ) -> str: v = await db.get(AuditorInboundVersion, version_id) if v is None: raise NotFoundError("version not found") # The archived key is server-written under this questionnaire's namespace; # re-validate before minting a bypassing GET (defense-in-depth). _validate_source_key(v.object_key) return await presign_get(db, object_key=v.object_key) # ════════════════════════════════════════════════════════════════════ # Views # ════════════════════════════════════════════════════════════════════ def questionnaire_view( q: AuditorInboundQuestionnaire, *, owner: IdentityPerson | None, now: datetime, counts: tuple[int, int, int] | None = None, ) -> dict[str, Any]: due = q.due_date days_to_due = (due - now.date()).days if due else None total, answered, reviewed = counts or (None, None, None) return { "id": str(q.id), "customer_name": q.customer_name, "title": q.title, "due_date": due.isoformat() if due else None, "days_to_due": days_to_due, "is_overdue": ( q.status not in ("completed", "archived", "cancelled") and due is not None and due < now.date() ), "owner": _person_brief(owner) if owner else None, "owner_person_id": str(q.owner_person_id) if q.owner_person_id else None, "status": q.status, "has_source_file": bool(q.source_object_key), "source_filename": q.source_filename, "notes": q.notes, "total_count": total, "answered_count": answered, "reviewed_count": reviewed, "completed_at": _to_naive(q.completed_at).isoformat() if q.completed_at else None, "exported_at": _to_naive(q.exported_at).isoformat() if q.exported_at else None, "created_at": _to_naive(q.created_at).isoformat() if q.created_at else None, "updated_at": _to_naive(q.updated_at).isoformat() if q.updated_at else None, } def question_view(qn: AuditorInboundQuestion) -> dict[str, Any]: return { "id": str(qn.id), "questionnaire_id": str(qn.questionnaire_id), "seq": qn.seq, "section": qn.section, "question_text": qn.question_text, "matched_library_entry_id": ( str(qn.matched_library_entry_id) if qn.matched_library_entry_id else None ), "match_confidence": qn.match_confidence, "answer_text": qn.answer_text, "status": qn.status, "reviewed_at": _to_naive(qn.reviewed_at).isoformat() if qn.reviewed_at else None, } async def _counts(db: AsyncSession, ids: list[uuid.UUID]) -> dict[uuid.UUID, tuple[int, int, int]]: if not ids: return {} total = dict((await db.execute( select(AuditorInboundQuestion.questionnaire_id, func.count()).where( AuditorInboundQuestion.questionnaire_id.in_(ids), ).group_by(AuditorInboundQuestion.questionnaire_id) )).all()) answered = dict((await db.execute( select(AuditorInboundQuestion.questionnaire_id, func.count()).where( AuditorInboundQuestion.questionnaire_id.in_(ids), AuditorInboundQuestion.answer_text.isnot(None), AuditorInboundQuestion.answer_text != "", ).group_by(AuditorInboundQuestion.questionnaire_id) )).all()) reviewed = dict((await db.execute( select(AuditorInboundQuestion.questionnaire_id, func.count()).where( AuditorInboundQuestion.questionnaire_id.in_(ids), AuditorInboundQuestion.status == "reviewed", ).group_by(AuditorInboundQuestion.questionnaire_id) )).all()) return { i: (total.get(i, 0), answered.get(i, 0), reviewed.get(i, 0)) for i in ids } async def list_questionnaires( db: AsyncSession, *, status: str | None = None, owner_person_id: uuid.UUID | None = None, search: str | None = None, page: int = 1, page_size: int = 50, ) -> dict[str, Any]: now = _naive_now() conds = [] if status: conds.append(AuditorInboundQuestionnaire.status == status) if owner_person_id: conds.append(AuditorInboundQuestionnaire.owner_person_id == owner_person_id) if search: like = f"%{search.strip()}%" from sqlalchemy import or_ conds.append(or_( AuditorInboundQuestionnaire.customer_name.ilike(like), AuditorInboundQuestionnaire.title.ilike(like), )) base = select(AuditorInboundQuestionnaire) for c in conds: base = base.where(c) total = (await db.execute( select(func.count()).select_from(base.subquery()) )).scalar_one() rows = (await db.execute( base.order_by(AuditorInboundQuestionnaire.created_at.desc()) .offset((page - 1) * page_size).limit(page_size) )).scalars().all() persons = await _persons_index(db, {r.owner_person_id for r in rows}) counts = await _counts(db, [r.id for r in rows]) return { "items": [ questionnaire_view(r, owner=persons.get(r.owner_person_id), now=now, counts=counts.get(r.id)) for r in rows ], "meta": {"page": page, "page_size": page_size, "total": total}, } async def get_detail(db: AsyncSession, qid: uuid.UUID) -> dict[str, Any]: q = await get_questionnaire(db, qid) questions = await _ordered_questions(db, qid) persons = await _persons_index(db, {q.owner_person_id}) now = _naive_now() total = len(questions) answered = sum(1 for x in questions if (x.answer_text or "").strip()) reviewed = sum(1 for x in questions if x.status == "reviewed") view = questionnaire_view( q, owner=persons.get(q.owner_person_id), now=now, counts=(total, answered, reviewed), ) view["questions"] = [question_view(x) for x in questions] versions = (await db.execute( select(AuditorInboundVersion).where( AuditorInboundVersion.questionnaire_id == qid, ).order_by(AuditorInboundVersion.version_no.desc()) )).scalars().all() view["versions"] = [ { "id": str(v.id), "version_no": v.version_no, "filename": v.filename, "note": v.note, "exported_at": _to_naive(v.exported_at).isoformat() if v.exported_at else None, } for v in versions ] return view async def overview(db: AsyncSession) -> dict[str, Any]: now = _naive_now() by_status = dict((await db.execute( select(AuditorInboundQuestionnaire.status, func.count()) .group_by(AuditorInboundQuestionnaire.status) )).all()) total = sum(by_status.values()) overdue = (await db.execute( select(func.count()).select_from(AuditorInboundQuestionnaire).where( AuditorInboundQuestionnaire.status.notin_(["completed", "archived", "cancelled"]), AuditorInboundQuestionnaire.due_date.isnot(None), AuditorInboundQuestionnaire.due_date < now.date(), ) )).scalar_one() return { "summary": { "total": total, "in_progress": ( (by_status.get("received", 0) or 0) + (by_status.get("answering", 0) or 0) + (by_status.get("review", 0) or 0) ), "completed": by_status.get("completed", 0) or 0, "overdue": overdue, }, "by_status": by_status, } # ════════════════════════════════════════════════════════════════════ # Due reminders (driven by the daily cron) # ════════════════════════════════════════════════════════════════════ # MUST stay ASCENDING by threshold — _stage_for returns the FIRST matching # (tightest) window, so 2 days out yields "3" (not collapsing to "7") and the # "3" / "due" windows stay reachable. A descending order silently collapses # every days_left in [0,7] to "7" and the tighter nudges never fire. (The # vendor twin _REVIEW_STAGES carries the same invariant.) _STAGES: tuple[tuple[int, str], ...] = ((0, "due"), (3, "3"), (7, "7")) _STAGE_ORDER = {"7": 1, "3": 2, "due": 3, "overdue": 4} def _stage_for(days_left: int) -> str | None: """Tightest due-window ``days_left`` has entered (negative = overdue). e.g. 5 days out → "7", 2 → "3", 0 → "due", >7 → None (too early). """ if days_left < 0: return "overdue" for threshold, label in _STAGES: if days_left <= threshold: return label return None async def find_due_reminders( db: AsyncSession, *, ref: date | None = None, ) -> list[tuple[AuditorInboundQuestionnaire, str, int]]: ref = ref or _naive_now().date() horizon = _add_months(ref, 1) rows = (await db.execute( select(AuditorInboundQuestionnaire).where( AuditorInboundQuestionnaire.status.notin_( ["completed", "archived", "cancelled"], ), AuditorInboundQuestionnaire.due_date.isnot(None), AuditorInboundQuestionnaire.due_date <= horizon, ) )).scalars().all() out: list[tuple[AuditorInboundQuestionnaire, str, int]] = [] for q in rows: days_left = (q.due_date - ref).days stage = _stage_for(days_left) if stage is None: continue if _STAGE_ORDER[stage] <= _STAGE_ORDER.get(q.reminder_stage or "", 0): continue out.append((q, stage, days_left)) return out def mark_reminder_sent(q: AuditorInboundQuestionnaire, stage: str) -> None: q.reminder_stage = stage q.reminder_sent_at = _naive_now()