"""安全问卷 · 甲方侧供应商安全评估 — service layer.

The engine behind "以甲方身份评估云厂商/供应商的安全态势" (vendor security
assessment / TPRM), one half of the two-sided 安全问卷 feature:

* A **template** is a reusable bilingual questionnaire (seeded from
  ISO/IEC 27001:2022 Annex A + ISO/IEC 27701). It holds **template
  questions** with a stable per-template ``code`` (e.g. ``A.5.1``) that
  survives the export→vendor-fills→import round-trip.
* An **assessment** instantiates one template for one vendor. Creating
  it materialises one **answer** row per template question, snapshotting
  the question so a later template edit never rewrites history (same
  discipline as self-check's ``check_item_snapshot``).
* The blank questionnaire is exported to a bilingual xlsx and sent to
  the vendor OUT OF BAND (Helix never emails an external party). The
  vendor's filled workbook is uploaded back; :func:`import_answers`
  re-binds each row by ``code``, weighted-auto-scores the answers and
  derives a **risk rating**.
* On completion, answers the auditor flagged ``finding`` are promoted to
  ``auditor_findings`` rows (scoped ``finding_type`` per assessment, so
  the reconcile auto-resolves an answer the auditor later un-flags).
* A daily cron reminds the owner before ``next_review_date``.

Datetime discipline: ``DateTime(timezone=True)`` columns round-trip naive
on fresh load (codebase-wide MySQL/TiDB convention), so this module
computes in **naive UTC** throughout.
"""
from __future__ import annotations

import io
import uuid
from datetime import UTC, date, datetime
from typing import Any

from sqlalchemy import func, select
from sqlalchemy import update as sa_update
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.exceptions import NotFoundError, ValidationError
from app.core.storage import presign_get, presign_put, stat_object
from app.domains.auditor import constants as const
from app.domains.auditor.findings_service import FindingSpec, reconcile_findings
from app.domains.auditor.models import (
    AuditorSqTemplate,
    AuditorSqTemplateQuestion,
    AuditorVendorAssessment,
    AuditorVendorAssessmentAnswer,
    AuditorVendorAssessmentAttachment,
)
from app.identity.person_models import IdentityPerson

# ── Enumerations ─────────────────────────────────────────────────────
ASSESSMENT_STATUSES: tuple[str, ...] = (
    "draft", "sent", "answered", "completed", "archived", "cancelled",
)
# Statuses in which the answer set may still be edited / imported.
EDITABLE_STATUSES: frozenset[str] = frozenset({"draft", "sent", "answered"})
TIERS: tuple[str, ...] = ("critical", "high", "medium", "low")
ANSWER_TYPES: tuple[str, ...] = (
    "yes_no", "yes_no_na", "single_choice", "multi_choice", "scale_1_5", "text",
)
REVIEW_FLAGS: tuple[str, ...] = ("ok", "concern", "finding")
RISK_RATINGS: tuple[str, ...] = ("low", "medium", "high", "critical")

# Default re-assessment cadence (months) by data-sensitivity tier.
DEFAULT_REVIEW_MONTHS: dict[str, int] = {
    "critical": 12, "high": 12, "medium": 24, "low": 36,
}
# score_pct → risk rating (higher score = lower risk). Ordered high→low.
_RATING_THRESHOLDS: tuple[tuple[int, str], ...] = (
    (90, "low"), (75, "medium"), (60, "high"), (0, "critical"),
)

# The control key vendor-assessment findings are filed under. Not in the
# compliance catalog (findings tolerate arbitrary control keys — the
# column is explicitly not a FK), so they surface in the findings
# worklist without a catalog entry.
FINDING_CONTROL_KEY = "vendor_security_assessment"

# Evidence: the vendor's certs (pdf/images) + the returned filled
# workbook (xlsx). Scripts excluded; browser uploads direct to S3.
ALLOWED_EVIDENCE_CONTENT_TYPES: frozenset[str] = frozenset({
    "image/jpeg", "image/png", "image/webp", "image/heic", "image/heif",
    "application/pdf",
    "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
    "application/vnd.ms-excel",
})
_EVIDENCE_KEY_PREFIX = "vendor_assessment_evidence"
_MAX_EVIDENCE_BYTES = 50 * 1024 * 1024  # 50 MB — certs / pentest reports.
_MAX_IMPORT_BYTES = 20 * 1024 * 1024  # 20 MB — a filled questionnaire xlsx.
_MAX_IMPORT_ROWS = 20_000  # far above any real questionnaire.


# ════════════════════════════════════════════════════════════════════
# Time helpers
# ════════════════════════════════════════════════════════════════════


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:
    """Naive month arithmetic clamped to the target month's last day."""
    import calendar

    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)


# ════════════════════════════════════════════════════════════════════
# Person enrichment
# ════════════════════════════════════════════════════════════════════


async def _persons_index(
    db: AsyncSession, person_ids: set[uuid.UUID],
) -> dict[uuid.UUID, IdentityPerson]:
    ids = [pid for pid in person_ids if pid]
    if not ids:
        return {}
    rows = (await db.execute(
        select(IdentityPerson).where(IdentityPerson.id.in_(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,
        "employee_no": p.employee_no,
        "work_email": p.work_email,
    }


# ════════════════════════════════════════════════════════════════════
# Scoring
# ════════════════════════════════════════════════════════════════════


def _coerce_scale(value: str | None) -> int | None:
    try:
        n = int(str(value).strip())
    except (TypeError, ValueError):
        return None
    return n if 1 <= n <= 5 else None


def score_answer(ans: AuditorVendorAssessmentAnswer) -> tuple[int | None, int | None]:
    """Compute ``(earned, max)`` points for one answer.

    ``max`` is ``None`` when the question does not participate in scoring
    (``text`` / ``multi_choice``) or was answered N/A — such answers are
    excluded from both numerator and denominator. A blank answer on a
    scorable question scores 0 out of its weight (an undemonstrated
    control is a real gap, not a free pass).
    """
    weight = max(int(ans.weight or 0), 0)
    at = ans.answer_type
    val = (ans.answer_value or "").strip()

    if at in ("text", "multi_choice"):
        return None, None
    if weight == 0:
        return None, None

    if at == "yes_no_na" and val.lower() == "na":
        return None, None

    if at in ("yes_no", "yes_no_na"):
        expected = (ans.expected_answer or "yes").strip().lower()
        earned = weight if val.lower() == expected else 0
        return earned, weight

    if at == "single_choice":
        fraction = 0.0
        for opt in (ans.options or []):
            if str(opt.get("value")) == val:
                try:
                    fraction = float(opt.get("score", 0) or 0)
                except (TypeError, ValueError):
                    fraction = 0.0
                break
        fraction = min(max(fraction, 0.0), 1.0)
        return round(weight * fraction), weight

    if at == "scale_1_5":
        n = _coerce_scale(val)
        if n is None:
            return 0, weight
        return round(weight * (n / 5)), weight

    return None, None


def compute_rating(score_pct: int | None) -> str | None:
    if score_pct is None:
        return None
    for floor, rating in _RATING_THRESHOLDS:
        if score_pct >= floor:
            return rating
    return "critical"


async def _recompute_scores(
    db: AsyncSession, assessment: AuditorVendorAssessment,
) -> None:
    """Re-score every answer + roll up the assessment score/rating.

    Does not touch a manually-overridden rating. Flushes; caller commits.
    """
    answers = (await db.execute(
        select(AuditorVendorAssessmentAnswer).where(
            AuditorVendorAssessmentAnswer.assessment_id == assessment.id,
        )
    )).scalars().all()
    total_earned = 0
    total_max = 0
    for ans in answers:
        earned, mx = score_answer(ans)
        ans.auto_score = earned
        ans.max_score = mx
        ans.is_scorable = mx is not None
        if mx is not None:
            total_earned += earned or 0
            total_max += mx
    if total_max > 0:
        assessment.score_earned = total_earned
        assessment.score_max = total_max
        assessment.score_pct = round(total_earned / total_max * 100)
    else:
        assessment.score_earned = None
        assessment.score_max = None
        assessment.score_pct = None
    if not assessment.rating_overridden:
        assessment.risk_rating = compute_rating(assessment.score_pct)
    await db.flush()


# ════════════════════════════════════════════════════════════════════
# Templates
# ════════════════════════════════════════════════════════════════════


def template_view(
    t: AuditorSqTemplate, *, question_count: int | None = None,
) -> dict[str, Any]:
    return {
        "id": str(t.id),
        "key": t.key,
        "name_zh": t.name_zh,
        "name_en": t.name_en,
        "description_zh": t.description_zh,
        "description_en": t.description_en,
        "kind": t.kind,
        "tier": t.tier,
        "version": t.version,
        "is_system": bool(t.is_system),
        "enabled": bool(t.enabled),
        "sort_order": t.sort_order,
        "question_count": question_count,
        "created_at": _to_naive(t.created_at).isoformat() if t.created_at else None,
        "updated_at": _to_naive(t.updated_at).isoformat() if t.updated_at else None,
    }


def question_view(q: AuditorSqTemplateQuestion) -> dict[str, Any]:
    return {
        "id": str(q.id),
        "template_id": str(q.template_id),
        "code": q.code,
        "section_key": q.section_key,
        "section_zh": q.section_zh,
        "section_en": q.section_en,
        "text_zh": q.text_zh,
        "text_en": q.text_en,
        "guidance_zh": q.guidance_zh,
        "guidance_en": q.guidance_en,
        "answer_type": q.answer_type,
        "options": q.options,
        "weight": q.weight,
        "expected_answer": q.expected_answer,
        "requires_evidence": bool(q.requires_evidence),
        "sort_order": q.sort_order,
    }


async def list_templates(
    db: AsyncSession, *, include_disabled: bool = True,
) -> list[dict[str, Any]]:
    stmt = select(AuditorSqTemplate)
    if not include_disabled:
        stmt = stmt.where(AuditorSqTemplate.enabled.is_(True))
    stmt = stmt.order_by(AuditorSqTemplate.sort_order, AuditorSqTemplate.name_en)
    rows = (await db.execute(stmt)).scalars().all()
    # Batch question counts.
    counts: dict[uuid.UUID, int] = {}
    if rows:
        cnt_rows = (await db.execute(
            select(
                AuditorSqTemplateQuestion.template_id,
                func.count(),
            ).where(
                AuditorSqTemplateQuestion.template_id.in_([r.id for r in rows]),
            ).group_by(AuditorSqTemplateQuestion.template_id)
        )).all()
        counts = {tid: c for tid, c in cnt_rows}
    return [template_view(r, question_count=counts.get(r.id, 0)) for r in rows]


async def get_template(
    db: AsyncSession, template_id: uuid.UUID,
) -> AuditorSqTemplate:
    t = await db.get(AuditorSqTemplate, template_id)
    if t is None:
        raise NotFoundError("questionnaire template not found")
    return t


async def list_template_questions(
    db: AsyncSession, template_id: uuid.UUID,
) -> list[AuditorSqTemplateQuestion]:
    return list((await db.execute(
        select(AuditorSqTemplateQuestion).where(
            AuditorSqTemplateQuestion.template_id == template_id,
        ).order_by(
            AuditorSqTemplateQuestion.sort_order, AuditorSqTemplateQuestion.code,
        )
    )).scalars().all())


async def get_template_detail(
    db: AsyncSession, template_id: uuid.UUID,
) -> dict[str, Any]:
    t = await get_template(db, template_id)
    questions = await list_template_questions(db, template_id)
    view = template_view(t, question_count=len(questions))
    view["questions"] = [question_view(q) for q in questions]
    return view


def _validate_answer_type(answer_type: str) -> None:
    if answer_type not in ANSWER_TYPES:
        raise ValidationError(f"answer_type must be one of {ANSWER_TYPES}")


async def create_template(
    db: AsyncSession, *, payload: dict[str, Any],
) -> AuditorSqTemplate:
    key = (payload.get("key") or "").strip()
    if not key:
        raise ValidationError("key is required")
    existing = (await db.execute(
        select(AuditorSqTemplate).where(AuditorSqTemplate.key == key)
    )).scalar_one_or_none()
    if existing is not None:
        raise ValidationError(f"template key '{key}' already exists")
    t = AuditorSqTemplate(
        key=key,
        name_zh=(payload.get("name_zh") or "").strip(),
        name_en=(payload.get("name_en") or "").strip(),
        description_zh=(payload.get("description_zh") or None),
        description_en=(payload.get("description_en") or None),
        kind=payload.get("kind") or "vendor",
        tier=payload.get("tier") or None,
        version=int(payload.get("version", 1)),
        is_system=False,
        enabled=bool(payload.get("enabled", True)),
        sort_order=int(payload.get("sort_order", 0)),
    )
    db.add(t)
    await db.flush()
    return t


async def update_template(
    db: AsyncSession, template_id: uuid.UUID, *, payload: dict[str, Any],
) -> AuditorSqTemplate:
    t = await get_template(db, template_id)
    for field in ("name_zh", "name_en"):
        if field in payload and payload[field] is not None:
            setattr(t, field, payload[field].strip())
    for field in ("description_zh", "description_en", "tier"):
        if field in payload:
            setattr(t, field, payload[field] or None)
    if "enabled" in payload and payload["enabled"] is not None:
        t.enabled = bool(payload["enabled"])
    if "sort_order" in payload and payload["sort_order"] is not None:
        t.sort_order = int(payload["sort_order"])
    await db.flush()
    return t


async def clone_template(
    db: AsyncSession, template_id: uuid.UUID, *, new_key: str, name_zh: str,
    name_en: str,
) -> AuditorSqTemplate:
    """Clone a template + its questions into a new custom (non-system) one."""
    src = await get_template(db, template_id)
    new_key = new_key.strip()
    if not new_key:
        raise ValidationError("new_key is required")
    dup = (await db.execute(
        select(AuditorSqTemplate).where(AuditorSqTemplate.key == new_key)
    )).scalar_one_or_none()
    if dup is not None:
        raise ValidationError(f"template key '{new_key}' already exists")
    clone = AuditorSqTemplate(
        key=new_key,
        name_zh=name_zh.strip() or src.name_zh,
        name_en=name_en.strip() or src.name_en,
        description_zh=src.description_zh,
        description_en=src.description_en,
        kind=src.kind,
        tier=src.tier,
        version=1,
        is_system=False,
        enabled=True,
        sort_order=src.sort_order,
    )
    db.add(clone)
    await db.flush()
    for q in await list_template_questions(db, template_id):
        db.add(AuditorSqTemplateQuestion(
            template_id=clone.id,
            code=q.code,
            section_key=q.section_key,
            section_zh=q.section_zh,
            section_en=q.section_en,
            text_zh=q.text_zh,
            text_en=q.text_en,
            guidance_zh=q.guidance_zh,
            guidance_en=q.guidance_en,
            answer_type=q.answer_type,
            options=q.options,
            weight=q.weight,
            expected_answer=q.expected_answer,
            requires_evidence=q.requires_evidence,
            sort_order=q.sort_order,
        ))
    await db.flush()
    return clone


async def delete_template(db: AsyncSession, template_id: uuid.UUID) -> None:
    t = await get_template(db, template_id)
    if t.is_system:
        raise ValidationError("system templates cannot be deleted (disable instead)")
    await db.delete(t)  # questions cascade via FK
    await db.flush()


async def add_template_question(
    db: AsyncSession, template_id: uuid.UUID, *, payload: dict[str, Any],
) -> AuditorSqTemplateQuestion:
    t = await get_template(db, template_id)
    if t.is_system:
        raise ValidationError(
            "system templates are read-only — clone before editing questions",
        )
    code = (payload.get("code") or "").strip()
    if not code:
        raise ValidationError("code is required")
    _validate_answer_type(payload["answer_type"])
    dup = (await db.execute(
        select(AuditorSqTemplateQuestion).where(
            AuditorSqTemplateQuestion.template_id == template_id,
            AuditorSqTemplateQuestion.code == code,
        )
    )).scalar_one_or_none()
    if dup is not None:
        raise ValidationError(f"question code '{code}' already exists in template")
    q = AuditorSqTemplateQuestion(
        template_id=template_id,
        code=code,
        section_key=(payload.get("section_key") or "general").strip(),
        section_zh=(payload.get("section_zh") or "").strip(),
        section_en=(payload.get("section_en") or "").strip(),
        text_zh=(payload.get("text_zh") or "").strip(),
        text_en=(payload.get("text_en") or "").strip(),
        guidance_zh=(payload.get("guidance_zh") or None),
        guidance_en=(payload.get("guidance_en") or None),
        answer_type=payload["answer_type"],
        options=payload.get("options"),
        weight=int(payload.get("weight", 1)),
        expected_answer=(payload.get("expected_answer") or None),
        requires_evidence=bool(payload.get("requires_evidence", False)),
        sort_order=int(payload.get("sort_order", 0)),
    )
    db.add(q)
    await db.flush()
    return q


async def update_template_question(
    db: AsyncSession, question_id: uuid.UUID, *, payload: dict[str, Any],
) -> AuditorSqTemplateQuestion:
    q = await db.get(AuditorSqTemplateQuestion, question_id)
    if q is None:
        raise NotFoundError("template question not found")
    t = await get_template(db, q.template_id)
    if t.is_system:
        raise ValidationError("system templates are read-only — clone before editing")
    if "answer_type" in payload and payload["answer_type"] is not None:
        _validate_answer_type(payload["answer_type"])
        q.answer_type = payload["answer_type"]
    for field in ("section_key", "section_zh", "section_en", "text_zh", "text_en"):
        if field in payload and payload[field] is not None:
            setattr(q, field, payload[field].strip())
    for field in ("guidance_zh", "guidance_en", "expected_answer"):
        if field in payload:
            setattr(q, field, payload[field] or None)
    if "options" in payload:
        q.options = payload["options"]
    if "weight" in payload and payload["weight"] is not None:
        q.weight = int(payload["weight"])
    if "requires_evidence" in payload and payload["requires_evidence"] is not None:
        q.requires_evidence = bool(payload["requires_evidence"])
    if "sort_order" in payload and payload["sort_order"] is not None:
        q.sort_order = int(payload["sort_order"])
    await db.flush()
    return q


async def delete_template_question(
    db: AsyncSession, question_id: uuid.UUID,
) -> uuid.UUID:
    q = await db.get(AuditorSqTemplateQuestion, question_id)
    if q is None:
        raise NotFoundError("template question not found")
    t = await get_template(db, q.template_id)
    if t.is_system:
        raise ValidationError("system templates are read-only")
    template_id = q.template_id
    await db.delete(q)
    await db.flush()
    return template_id


# ════════════════════════════════════════════════════════════════════
# Assessment lifecycle
# ════════════════════════════════════════════════════════════════════


async def get_assessment(
    db: AsyncSession, assessment_id: uuid.UUID,
) -> AuditorVendorAssessment:
    a = await db.get(AuditorVendorAssessment, assessment_id)
    if a is None:
        raise NotFoundError("vendor assessment not found")
    return a


async def create_assessment(
    db: AsyncSession, *, payload: dict[str, Any], actor_id: uuid.UUID | None,
) -> AuditorVendorAssessment:
    """Create an assessment from a template + materialise its answers."""
    template = await get_template(db, payload["template_id"])
    questions = await list_template_questions(db, template.id)
    if not questions:
        raise ValidationError("template has no questions")
    tier = payload.get("tier") or "medium"
    if tier not in TIERS:
        raise ValidationError(f"tier must be one of {TIERS}")
    vendor_name = (payload.get("vendor_name") or "").strip()
    if not vendor_name:
        raise ValidationError("vendor_name is required")

    a = AuditorVendorAssessment(
        template_id=template.id,
        template_key_snapshot=template.key,
        template_name_snapshot=template.name_en,
        supplier_id=payload.get("supplier_id"),
        vendor_name=vendor_name,
        vendor_contact_name=(payload.get("vendor_contact_name") or None),
        vendor_contact_email=(payload.get("vendor_contact_email") or None),
        tier=tier,
        status="draft",
        owner_person_id=payload.get("owner_person_id") or actor_id,
        reviewer_person_id=payload.get("reviewer_person_id") or None,
        due_at=payload.get("due_at"),
        review_period_months=(
            int(payload["review_period_months"])
            if payload.get("review_period_months") else DEFAULT_REVIEW_MONTHS[tier]
        ),
        created_by_person_id=actor_id,
    )
    db.add(a)
    await db.flush()
    for q in questions:
        db.add(AuditorVendorAssessmentAnswer(
            assessment_id=a.id,
            question_code=q.code,
            section_key=q.section_key,
            section_zh=q.section_zh,
            section_en=q.section_en,
            text_zh=q.text_zh,
            text_en=q.text_en,
            guidance_zh=q.guidance_zh,
            guidance_en=q.guidance_en,
            answer_type=q.answer_type,
            options=q.options,
            weight=q.weight,
            expected_answer=q.expected_answer,
            requires_evidence=q.requires_evidence,
            sort_order=q.sort_order,
        ))
    await db.flush()
    return a


async def update_assessment_meta(
    db: AsyncSession, assessment_id: uuid.UUID, *, payload: dict[str, Any],
) -> AuditorVendorAssessment:
    a = await get_assessment(db, assessment_id)
    if a.status in ("archived", "cancelled"):
        raise ValidationError("cannot edit an archived/cancelled assessment")
    if "vendor_name" in payload and payload["vendor_name"] is not None:
        name = payload["vendor_name"].strip()
        if not name:
            raise ValidationError("vendor_name cannot be blank")
        a.vendor_name = name
    for field in ("vendor_contact_name", "vendor_contact_email"):
        if field in payload:
            setattr(a, field, payload[field] or None)
    if "supplier_id" in payload:
        a.supplier_id = payload["supplier_id"]
    if "tier" in payload and payload["tier"] is not None:
        if payload["tier"] not in TIERS:
            raise ValidationError(f"tier must be one of {TIERS}")
        a.tier = payload["tier"]
    if "owner_person_id" in payload:
        a.owner_person_id = payload["owner_person_id"]
    if "reviewer_person_id" in payload:
        a.reviewer_person_id = payload["reviewer_person_id"]
    if "due_at" in payload:
        a.due_at = payload["due_at"]
    if "review_period_months" in payload and payload["review_period_months"]:
        a.review_period_months = int(payload["review_period_months"])
    await db.flush()
    return a


async def mark_sent(
    db: AsyncSession, assessment_id: uuid.UUID,
) -> AuditorVendorAssessment:
    """Record that the questionnaire was exported & sent to the vendor."""
    a = await get_assessment(db, assessment_id)
    if a.status not in ("draft", "sent"):
        raise ValidationError("only a draft assessment can be marked as sent")
    a.status = "sent"
    if a.sent_at is None:
        a.sent_at = _naive_now()
    await db.flush()
    return a


async def set_answer(
    db: AsyncSession, answer_id: uuid.UUID, *, payload: dict[str, Any],
) -> AuditorVendorAssessmentAnswer:
    """Manually set one answer (auditor filling on the vendor's behalf)."""
    ans = await db.get(AuditorVendorAssessmentAnswer, answer_id)
    if ans is None:
        raise NotFoundError("assessment answer not found")
    a = await get_assessment(db, ans.assessment_id)
    if a.status not in EDITABLE_STATUSES:
        raise ValidationError(
            "answers can only be edited while draft/sent/answered",
        )
    if "answer_value" in payload:
        val = payload["answer_value"]
        if isinstance(val, (list, tuple)):
            import json
            ans.answer_value = json.dumps(list(val), ensure_ascii=False)
        else:
            ans.answer_value = (str(val).strip() if val is not None else None) or None
    if "answer_comment" in payload:
        ans.answer_comment = payload["answer_comment"] or None
    earned, mx = score_answer(ans)
    ans.auto_score = earned
    ans.max_score = mx
    ans.is_scorable = mx is not None
    await db.flush()
    if a.status == "sent":
        a.status = "answered"
        a.answered_at = _naive_now()
    await _recompute_scores(db, a)
    return ans


async def review_answer(
    db: AsyncSession, answer_id: uuid.UUID, *,
    review_flag: str | None, reviewer_note: str | None,
) -> AuditorVendorAssessmentAnswer:
    ans = await db.get(AuditorVendorAssessmentAnswer, answer_id)
    if ans is None:
        raise NotFoundError("assessment answer not found")
    if review_flag is not None and review_flag not in REVIEW_FLAGS:
        raise ValidationError(f"review_flag must be one of {REVIEW_FLAGS}")
    a = await get_assessment(db, ans.assessment_id)
    if a.status in ("archived", "cancelled"):
        # There is no reconcile path out of these two states —
        # ``complete_assessment`` (the only caller of ``_sync_findings``)
        # refuses them — so a flag edit here would either strand an open
        # finding or fail to raise one, with no way to fix it short of
        # walking the assessment back to draft.
        raise ValidationError(
            "cannot review answers on an archived/cancelled assessment — "
            "set it back to draft first",
        )
    ans.review_flag = review_flag
    if reviewer_note is not None:
        ans.reviewer_note = reviewer_note or None
    await db.flush()
    if a.status == "completed":
        # Findings were already promoted by ``complete_assessment``, so a
        # flag change now has to reconcile immediately. Without this,
        # un-flagging left the finding open on the worklist for ever and
        # ``answer.finding_id`` kept pointing at it (#2165).
        await _sync_findings(db, a)
    return ans


async def complete_assessment(
    db: AsyncSession, assessment_id: uuid.UUID, *, summary: str | None,
    reviewer_person_id: uuid.UUID | None,
) -> AuditorVendorAssessment:
    """Finalize: score, rate, set next review date, promote findings."""
    a = await get_assessment(db, assessment_id)
    if a.status in ("archived", "cancelled"):
        raise ValidationError(
            "cannot complete an archived/cancelled assessment",
        )
    await _recompute_scores(db, a)
    a.status = "completed"
    a.completed_at = _naive_now()
    if summary is not None:
        a.summary = summary or None
    if reviewer_person_id is not None:
        a.reviewer_person_id = reviewer_person_id
    months = a.review_period_months or DEFAULT_REVIEW_MONTHS.get(a.tier, 24)
    a.next_review_date = _add_months(_naive_now().date(), months)
    a.review_reminder_stage = None
    a.review_reminder_sent_at = None
    await db.flush()
    await _sync_findings(db, a)
    return a


async def override_rating(
    db: AsyncSession, assessment_id: uuid.UUID, *, rating: str | None,
) -> AuditorVendorAssessment:
    a = await get_assessment(db, assessment_id)
    if rating is None:
        # Clear the override → fall back to the computed rating.
        a.rating_overridden = False
        a.risk_rating = compute_rating(a.score_pct)
    else:
        if rating not in RISK_RATINGS:
            raise ValidationError(f"rating must be one of {RISK_RATINGS}")
        a.rating_overridden = True
        a.risk_rating = rating
    await db.flush()
    return a


async def reopen_assessment(
    db: AsyncSession, assessment_id: uuid.UUID,
) -> AuditorVendorAssessment:
    a = await get_assessment(db, assessment_id)
    if a.status != "completed":
        raise ValidationError("only a completed assessment can be reopened")
    a.status = "answered"
    a.completed_at = None
    await db.flush()
    return a


async def _close_assessment_findings(
    db: AsyncSession, a: AuditorVendorAssessment,
) -> None:
    """Resolve every open finding this assessment promoted.

    ``reconcile_findings`` with an empty spec list auto-resolves the
    whole ``assessment:{id}`` scope — the same mechanism that closes a
    finding when its answer is un-flagged and the assessment
    re-completed. Also clears the back-reference so the detail view does
    not link answers to a resolved finding.
    """
    await reconcile_findings(
        db,
        control_key=FINDING_CONTROL_KEY,
        finding_type=f"assessment:{a.id}",
        specs=[],
    )
    await db.execute(
        sa_update(AuditorVendorAssessmentAnswer)
        .where(
            AuditorVendorAssessmentAnswer.assessment_id == a.id,
            AuditorVendorAssessmentAnswer.finding_id.is_not(None),
        )
        .values(finding_id=None)
    )
    await db.flush()


async def set_status(
    db: AsyncSession, assessment_id: uuid.UUID, *, status: str,
) -> AuditorVendorAssessment:
    """Archive / cancel / un-archive an assessment.

    Every transition here leaves the assessment in a state that
    ``complete_assessment`` refuses (archived / cancelled) or that means
    "not assessed yet" (draft), so each one has to clean up after the
    completed state it is leaving. It used to just assign ``a.status``:

    * cancel/archive a completed assessment → its promoted findings sat
      on the open worklist for ever, pointing at an assessment nobody
      would ever complete again (the only reconcile entry point).
    * completed → draft kept ``completed_at`` / ``score_pct`` /
      ``risk_rating`` / ``next_review_date``, so the detail page showed
      a draft carrying a completion time and a risk grade — while
      ``reopen_assessment`` right above knew to clear ``completed_at``.
    """
    if status not in ("archived", "cancelled", "draft"):
        raise ValidationError("status must be archived / cancelled / draft")
    a = await get_assessment(db, assessment_id)
    # #2232: close unconditionally. The old predicate — "completed, or
    # carries a completed_at" — missed the REOPEN path:
    # ``reopen_assessment`` sets ``status='answered'`` and clears
    # ``completed_at`` while deliberately leaving the promoted findings
    # OPEN, so a reopen-then-cancel (an auditor deciding to drop the
    # vendor mid-review) computed ``was_assessed=False``, skipped the
    # close, and stranded those findings for ever: ``complete_assessment``
    # and ``review_answer`` both refuse a cancelled assessment, no cron
    # reconciles the ``assessment:{id}`` scope, and even sending it back
    # to draft skipped the close on the same predicate. The only exit was
    # to run the whole assessment again.
    # ``_close_assessment_findings`` is an empty-spec reconcile — it is
    # idempotent and a no-op when the scope holds nothing — so there is
    # nothing to gate on.
    a.status = status
    await _close_assessment_findings(db, a)
    if status == "draft":
        # Same clean-up ``reopen_assessment`` does, plus the grade: a
        # draft has not been assessed, so it carries no verdict.
        a.completed_at = None
        a.score_pct = None
        a.risk_rating = None
        a.rating_overridden = False
        a.next_review_date = None
        a.review_reminder_stage = None
        a.review_reminder_sent_at = None
    await db.flush()
    return a


# ════════════════════════════════════════════════════════════════════
# Findings promotion
# ════════════════════════════════════════════════════════════════════


async def _sync_findings(
    db: AsyncSession, a: AuditorVendorAssessment,
) -> dict[str, int]:
    """Reconcile ``auditor_findings`` from answers flagged ``finding``.

    ``finding_type`` is scoped per assessment so the reconcile's
    auto-resolve only touches THIS assessment's findings — un-flagging an
    answer and re-completing resolves its finding. Does not commit.
    """
    answers = (await db.execute(
        select(AuditorVendorAssessmentAnswer).where(
            AuditorVendorAssessmentAnswer.assessment_id == a.id,
            AuditorVendorAssessmentAnswer.review_flag == "finding",
        )
    )).scalars().all()
    specs: list[FindingSpec] = []
    for ans in answers:
        title = f"[{a.vendor_name}] {ans.question_code}: {ans.text_en[:200]}"
        specs.append(FindingSpec(
            dedup_key=f"{a.id}:{ans.question_code}",
            severity=a.tier if a.tier in const.ALL_EXPOSURE_SEVERITIES else "medium",
            title=title[:512],
            subject_type="vendor",
            subject_ref=str(a.id),
            subject_display=a.vendor_name,
            detail={
                "assessment_id": str(a.id),
                "question_code": ans.question_code,
                "answer_value": ans.answer_value,
                "reviewer_note": ans.reviewer_note,
                "vendor_name": a.vendor_name,
            },
        ))
    counts = await reconcile_findings(
        db,
        control_key=FINDING_CONTROL_KEY,
        finding_type=f"assessment:{a.id}",
        specs=specs,
    )
    # Backfill the finding_id onto each promoted answer for the detail view.
    from app.domains.auditor.models import AuditorFinding
    if specs:
        rows = (await db.execute(
            select(AuditorFinding).where(
                AuditorFinding.control_key == FINDING_CONTROL_KEY,
                AuditorFinding.finding_type == f"assessment:{a.id}",
            )
        )).scalars().all()
        by_code = {r.dedup_key.split(":", 1)[-1]: r.id for r in rows}
        for ans in answers:
            ans.finding_id = by_code.get(ans.question_code)
        await db.flush()
    return counts


# ════════════════════════════════════════════════════════════════════
# Views
# ════════════════════════════════════════════════════════════════════


def assessment_view(
    a: AuditorVendorAssessment, *,
    persons: dict[uuid.UUID, IdentityPerson],
    now: datetime,
    answered_count: int | None = None,
    total_count: int | None = None,
) -> dict[str, Any]:
    due = _to_naive(a.due_at)
    next_review = a.next_review_date
    days_to_review = (
        (next_review - now.date()).days if next_review else None
    )
    return {
        "id": str(a.id),
        "vendor_name": a.vendor_name,
        "vendor_contact_name": a.vendor_contact_name,
        "vendor_contact_email": a.vendor_contact_email,
        "supplier_id": str(a.supplier_id) if a.supplier_id else None,
        "template_id": str(a.template_id) if a.template_id else None,
        "template_key": a.template_key_snapshot,
        "template_name": a.template_name_snapshot,
        "tier": a.tier,
        "status": a.status,
        "score_earned": a.score_earned,
        "score_max": a.score_max,
        "score_pct": a.score_pct,
        "risk_rating": a.risk_rating,
        "rating_overridden": bool(a.rating_overridden),
        "summary": a.summary,
        "owner": _person_brief(persons.get(a.owner_person_id))
        if a.owner_person_id else None,
        "reviewer": _person_brief(persons.get(a.reviewer_person_id))
        if a.reviewer_person_id else None,
        "due_at": due.isoformat() if due else None,
        "is_response_overdue": (
            a.status in ("sent",) and due is not None and now > due
        ),
        "sent_at": _to_naive(a.sent_at).isoformat() if a.sent_at else None,
        "answered_at": (
            _to_naive(a.answered_at).isoformat() if a.answered_at else None
        ),
        "completed_at": (
            _to_naive(a.completed_at).isoformat() if a.completed_at else None
        ),
        "next_review_date": next_review.isoformat() if next_review else None,
        "days_to_review": days_to_review,
        "review_period_months": a.review_period_months,
        "answered_count": answered_count,
        "total_count": total_count,
        "created_at": _to_naive(a.created_at).isoformat() if a.created_at else None,
        "updated_at": _to_naive(a.updated_at).isoformat() if a.updated_at else None,
    }


def answer_view(ans: AuditorVendorAssessmentAnswer) -> dict[str, Any]:
    return {
        "id": str(ans.id),
        "assessment_id": str(ans.assessment_id),
        "question_code": ans.question_code,
        "section_key": ans.section_key,
        "section_zh": ans.section_zh,
        "section_en": ans.section_en,
        "text_zh": ans.text_zh,
        "text_en": ans.text_en,
        "guidance_zh": ans.guidance_zh,
        "guidance_en": ans.guidance_en,
        "answer_type": ans.answer_type,
        "options": ans.options,
        "weight": ans.weight,
        "expected_answer": ans.expected_answer,
        "requires_evidence": bool(ans.requires_evidence),
        "sort_order": ans.sort_order,
        "answer_value": ans.answer_value,
        "answer_comment": ans.answer_comment,
        "auto_score": ans.auto_score,
        "max_score": ans.max_score,
        "is_scorable": bool(ans.is_scorable),
        "review_flag": ans.review_flag,
        "reviewer_note": ans.reviewer_note,
        "finding_id": str(ans.finding_id) if ans.finding_id else None,
        "attachment_count": ans.attachment_count,
    }


async def _answered_counts(
    db: AsyncSession, assessment_ids: list[uuid.UUID],
) -> dict[uuid.UUID, tuple[int, int]]:
    """Return ``{assessment_id: (answered, total)}`` in two grouped queries."""
    if not assessment_ids:
        return {}
    totals = dict((await db.execute(
        select(
            AuditorVendorAssessmentAnswer.assessment_id, func.count(),
        ).where(
            AuditorVendorAssessmentAnswer.assessment_id.in_(assessment_ids),
        ).group_by(AuditorVendorAssessmentAnswer.assessment_id)
    )).all())
    answered = dict((await db.execute(
        select(
            AuditorVendorAssessmentAnswer.assessment_id, func.count(),
        ).where(
            AuditorVendorAssessmentAnswer.assessment_id.in_(assessment_ids),
            AuditorVendorAssessmentAnswer.answer_value.isnot(None),
            AuditorVendorAssessmentAnswer.answer_value != "",
        ).group_by(AuditorVendorAssessmentAnswer.assessment_id)
    )).all())
    return {
        aid: (answered.get(aid, 0), totals.get(aid, 0)) for aid in assessment_ids
    }


async def list_assessments(
    db: AsyncSession, *,
    status: str | None = None,
    tier: str | None = None,
    risk_rating: 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 is not None:
        conds.append(AuditorVendorAssessment.status == status)
    if tier is not None:
        conds.append(AuditorVendorAssessment.tier == tier)
    if risk_rating is not None:
        conds.append(AuditorVendorAssessment.risk_rating == risk_rating)
    if owner_person_id is not None:
        conds.append(AuditorVendorAssessment.owner_person_id == owner_person_id)
    if search:
        conds.append(AuditorVendorAssessment.vendor_name.ilike(f"%{search.strip()}%"))

    base = select(AuditorVendorAssessment)
    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(AuditorVendorAssessment.created_at.desc())
        .offset((page - 1) * page_size).limit(page_size)
    )).scalars().all()

    person_ids: set[uuid.UUID] = set()
    for a in rows:
        if a.owner_person_id:
            person_ids.add(a.owner_person_id)
        if a.reviewer_person_id:
            person_ids.add(a.reviewer_person_id)
    persons = await _persons_index(db, person_ids)
    counts = await _answered_counts(db, [a.id for a in rows])
    items = [
        assessment_view(
            a, persons=persons, now=now,
            answered_count=counts.get(a.id, (0, 0))[0],
            total_count=counts.get(a.id, (0, 0))[1],
        )
        for a in rows
    ]
    return {
        "items": items,
        "meta": {"page": page, "page_size": page_size, "total": total},
    }


async def get_assessment_detail(
    db: AsyncSession, assessment_id: uuid.UUID,
) -> dict[str, Any]:
    a = await get_assessment(db, assessment_id)
    answers = (await db.execute(
        select(AuditorVendorAssessmentAnswer).where(
            AuditorVendorAssessmentAnswer.assessment_id == a.id,
        ).order_by(
            AuditorVendorAssessmentAnswer.sort_order,
            AuditorVendorAssessmentAnswer.question_code,
        )
    )).scalars().all()
    person_ids = {
        pid for pid in (a.owner_person_id, a.reviewer_person_id) if pid
    }
    persons = await _persons_index(db, person_ids)
    now = _naive_now()
    answered = sum(1 for x in answers if (x.answer_value or "").strip())
    view = assessment_view(
        a, persons=persons, now=now,
        answered_count=answered, total_count=len(answers),
    )
    # Group answers by section, preserving encounter order.
    sections: list[dict[str, Any]] = []
    section_index: dict[str, dict[str, Any]] = {}
    for ans in answers:
        sec = section_index.get(ans.section_key)
        if sec is None:
            sec = {
                "section_key": ans.section_key,
                "section_zh": ans.section_zh,
                "section_en": ans.section_en,
                "answers": [],
            }
            section_index[ans.section_key] = sec
            sections.append(sec)
        sec["answers"].append(answer_view(ans))
    view["sections"] = sections
    view["attachments"] = await list_attachments(db, assessment_id)
    view["counts"] = {
        "concern": sum(1 for x in answers if x.review_flag == "concern"),
        "finding": sum(1 for x in answers if x.review_flag == "finding"),
        "scorable": sum(1 for x in answers if x.is_scorable),
    }
    return view


# ════════════════════════════════════════════════════════════════════
# Overview / risk posture
# ════════════════════════════════════════════════════════════════════


async def get_overview(db: AsyncSession) -> dict[str, Any]:
    """Vendor risk posture: counts by status/rating/tier + upcoming reviews."""
    now = _naive_now()
    today = now.date()

    def _count_group(col):
        return select(col, func.count()).select_from(
            AuditorVendorAssessment,
        ).group_by(col)

    by_status = dict((await db.execute(
        _count_group(AuditorVendorAssessment.status)
    )).all())
    by_rating = dict((await db.execute(
        _count_group(AuditorVendorAssessment.risk_rating).where(
            AuditorVendorAssessment.status == "completed",
        )
    )).all())
    by_tier = dict((await db.execute(
        _count_group(AuditorVendorAssessment.tier)
    )).all())

    total = (await db.execute(
        select(func.count()).select_from(AuditorVendorAssessment)
    )).scalar_one()
    # Reviews due within 30 days or overdue (completed assessments only).
    upcoming_review = (await db.execute(
        select(func.count()).select_from(AuditorVendorAssessment).where(
            AuditorVendorAssessment.status == "completed",
            AuditorVendorAssessment.next_review_date.isnot(None),
            AuditorVendorAssessment.next_review_date <= _add_months(today, 1),
        )
    )).scalar_one()
    # Responses overdue (sent, past due_at).
    response_overdue = (await db.execute(
        select(func.count()).select_from(AuditorVendorAssessment).where(
            AuditorVendorAssessment.status == "sent",
            AuditorVendorAssessment.due_at.isnot(None),
            AuditorVendorAssessment.due_at < now,
        )
    )).scalar_one()
    high_risk = (by_rating.get("high", 0) or 0) + (by_rating.get("critical", 0) or 0)

    return {
        "summary": {
            "total": total,
            "in_progress": (
                (by_status.get("draft", 0) or 0)
                + (by_status.get("sent", 0) or 0)
                + (by_status.get("answered", 0) or 0)
            ),
            "completed": by_status.get("completed", 0) or 0,
            "high_risk": high_risk,
            "upcoming_review": upcoming_review,
            "response_overdue": response_overdue,
        },
        "by_status": by_status,
        "by_rating": by_rating,
        "by_tier": by_tier,
    }


# ════════════════════════════════════════════════════════════════════
# Blank questionnaire export (to send to the vendor) + import
# ════════════════════════════════════════════════════════════════════

# The 2-row-header convention the codebase established: row after the
# banner is the machine-key row, so an exported blank round-trips on
# import via the ``code`` column. Bilingual (zh + en) for every label.
_IMPORT_CODE_HEADERS = {"code", "编号", "问题编号", "question code"}
_IMPORT_ANSWER_HEADERS = {"answer", "回答", "回答/answer", "答复", "response"}
_IMPORT_COMMENT_HEADERS = {"comment", "说明", "说明/comment", "备注", "notes"}


def _answer_type_hint(ans_type: str) -> str:
    return {
        "yes_no": "Yes / No",
        "yes_no_na": "Yes / No / N/A",
        "single_choice": "single choice",
        "multi_choice": "multiple choice",
        "scale_1_5": "1-5",
        "text": "free text",
    }.get(ans_type, ans_type)


def build_blank_questionnaire(
    a: AuditorVendorAssessment,
    answers: list[AuditorVendorAssessmentAnswer],
) -> io.BytesIO:
    """Build the bilingual blank questionnaire workbook to send a vendor.

    Layout: a banner, an instruction row, a header row, then one row per
    question. Column A is the machine ``code`` — the round-trip key the
    import re-binds on. The vendor fills the 回答/Answer + 说明/Comment
    columns.
    """
    from openpyxl import Workbook
    from openpyxl.styles import Alignment, Font, PatternFill
    from openpyxl.worksheet.datavalidation import DataValidation

    wb = Workbook()
    ws = wb.active
    ws.title = "Security Questionnaire"

    header_fill = PatternFill("solid", fgColor="4472C4")
    header_font = Font(color="FFFFFF", bold=True, size=11)
    banner_font = Font(bold=True, size=14, color="1F3864")
    section_fill = PatternFill("solid", fgColor="D9E2F3")
    section_font = Font(bold=True, size=11, color="1F3864")
    wrap = Alignment(wrap_text=True, vertical="top")

    columns = [
        ("code", "编号 / Code", 14),
        ("question", "问题 / Question", 60),
        ("type", "类型 / Type", 16),
        ("options", "可选项 / Options", 30),
        ("answer", "回答 / Answer", 20),
        ("comment", "说明与证据 / Comment & Evidence", 40),
    ]

    ws.merge_cells("A1:F1")
    ws["A1"] = f"供应商安全问卷 · Vendor Security Questionnaire — {a.vendor_name}"
    ws["A1"].font = banner_font
    ws.merge_cells("A2:F2")
    ws["A2"] = (
        "请在「回答」列填写,并在「说明与证据」列补充说明或注明所附证据文件名。"
        "请勿修改「编号」列。 / Please fill the Answer column and add context or "
        "reference attached evidence in the Comment column. Do NOT edit the Code column."
    )
    ws["A2"].font = Font(italic=True, size=9, color="555555")
    ws["A2"].alignment = wrap

    header_row = 3
    for idx, (_key, label, width) in enumerate(columns, 1):
        cell = ws.cell(row=header_row, column=idx, value=label)
        cell.fill = header_fill
        cell.font = header_font
        cell.alignment = Alignment(horizontal="center", wrap_text=True)
        ws.column_dimensions[cell.column_letter].width = width

    r = header_row
    last_section: str | None = None
    for ans in answers:
        if ans.section_key != last_section:
            r += 1
            ws.merge_cells(start_row=r, start_column=1, end_row=r, end_column=6)
            sc = ws.cell(
                row=r, column=1,
                value=f"{ans.section_zh} / {ans.section_en}",
            )
            sc.fill = section_fill
            sc.font = section_font
            last_section = ans.section_key
        r += 1
        ws.cell(row=r, column=1, value=ans.question_code)
        ws.cell(row=r, column=2, value=f"{ans.text_zh}\n{ans.text_en}").alignment = wrap
        ws.cell(row=r, column=3, value=_answer_type_hint(ans.answer_type))
        opts = ""
        if ans.options:
            opts = " / ".join(
                str(o.get("label_en") or o.get("value")) for o in ans.options
            )
        elif ans.answer_type in ("yes_no", "yes_no_na"):
            opts = "Yes / No" + (" / N/A" if ans.answer_type == "yes_no_na" else "")
        elif ans.answer_type == "scale_1_5":
            opts = "1 / 2 / 3 / 4 / 5"
        ws.cell(row=r, column=4, value=opts).alignment = wrap
        ws.cell(row=r, column=6).alignment = wrap
        # A dropdown data-validation on the answer cell for closed types.
        allowed: list[str] | None = None
        if ans.answer_type == "yes_no":
            allowed = ["Yes", "No"]
        elif ans.answer_type == "yes_no_na":
            allowed = ["Yes", "No", "N/A"]
        elif ans.answer_type == "scale_1_5":
            allowed = ["1", "2", "3", "4", "5"]
        elif ans.answer_type == "single_choice" and ans.options:
            allowed = [str(o.get("value")) for o in ans.options]
        if allowed:
            dv = DataValidation(
                type="list",
                formula1='"' + ",".join(allowed) + '"',
                allow_blank=True,
            )
            ws.add_data_validation(dv)
            dv.add(ws.cell(row=r, column=5))

    ws.freeze_panes = "A4"
    buf = io.BytesIO()
    wb.save(buf)
    buf.seek(0)
    return buf


def parse_filled_questionnaire(payload: bytes) -> dict[str, dict[str, str]]:
    """Parse a vendor's filled workbook → ``{code: {answer, comment}}``.

    Locates the header row by the presence of a recognised Code column,
    then reads the Answer + Comment columns for every data row keyed by
    the Code cell. Tolerant of the banner/instruction rows above the
    header and of either language's column labels.
    """
    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 i, row in enumerate(
            ws.iter_rows(values_only=True, max_col=MAX_XLSX_COLUMNS + 1),
        ):
            if i >= _MAX_IMPORT_ROWS:
                raise ValidationError(
                    f"workbook has more than {_MAX_IMPORT_ROWS} rows — "
                    "did you upload the exported questionnaire?",
                )
            rows.append(bounded_row(row))
    except ValidationError:
        raise
    except Exception as exc:  # noqa: BLE001 — surface a clean 400
        raise ValidationError(f"could not read xlsx: {exc}") from exc
    finally:
        if wb is not None:
            wb.close()

    def _norm(v: Any) -> str:
        return str(v).strip().lower() if v is not None else ""

    def _match(cell: str, tokens: frozenset[str]) -> bool:
        # Header cells are bilingual ("编号 / Code") so match by token
        # substring rather than exact equality.
        return any(tok in cell for tok in tokens)

    code_col = ans_col = com_col = None
    header_idx = None
    for i, row in enumerate(rows):
        norm = [_norm(c) for c in row]
        for j, cell in enumerate(norm):
            if not cell:
                continue
            if code_col is None and _match(cell, _IMPORT_CODE_HEADERS):
                code_col = j
            elif ans_col is None and _match(cell, _IMPORT_ANSWER_HEADERS):
                ans_col = j
            elif com_col is None and _match(cell, _IMPORT_COMMENT_HEADERS):
                com_col = j
        if code_col is not None and ans_col is not None:
            header_idx = i
            break
        code_col = ans_col = com_col = None
    if header_idx is None or code_col is None or ans_col is None:
        raise ValidationError(
            "could not find the '编号/Code' and '回答/Answer' columns — "
            "did you upload the exported questionnaire?",
        )

    out: dict[str, dict[str, str]] = {}
    for row in rows[header_idx + 1:]:
        if code_col >= len(row):
            continue
        code = row[code_col]
        if code is None or not str(code).strip():
            continue
        code_s = str(code).strip()
        answer = row[ans_col] if ans_col < len(row) else None
        comment = row[com_col] if (com_col is not None and com_col < len(row)) else None
        out[code_s] = {
            "answer": str(answer).strip() if answer is not None else "",
            "comment": str(comment).strip() if comment is not None else "",
        }
    return out


def _normalise_imported_answer(ans_type: str, raw: str) -> str:
    """Map a vendor's free-form cell into the stored canonical value."""
    v = raw.strip()
    low = v.lower()
    if ans_type in ("yes_no", "yes_no_na"):
        if low in ("yes", "y", "true", "是", "有"):
            return "yes"
        if low in ("no", "n", "false", "否", "无"):
            return "no"
        if ans_type == "yes_no_na" and low in ("n/a", "na", "不适用"):
            return "na"
        return v
    if ans_type == "single_choice":
        return v
    if ans_type == "scale_1_5":
        return v
    return v


async def import_answers(
    db: AsyncSession, assessment_id: uuid.UUID, *, payload: bytes,
) -> dict[str, Any]:
    """Apply a vendor's filled workbook to the assessment answers + score."""
    a = await get_assessment(db, assessment_id)
    if a.status not in EDITABLE_STATUSES:
        raise ValidationError(
            "import is only allowed while the assessment is draft/sent/answered",
        )
    parsed = parse_filled_questionnaire(payload)
    answers = (await db.execute(
        select(AuditorVendorAssessmentAnswer).where(
            AuditorVendorAssessmentAnswer.assessment_id == a.id,
        )
    )).scalars().all()
    by_code = {x.question_code: x for x in answers}
    updated = 0
    matched = 0
    unmatched: list[str] = []
    for code, cell in parsed.items():
        ans = by_code.get(code)
        if ans is None:
            unmatched.append(code)
            continue
        # Counted separately from ``updated`` so the diagnostic below can
        # tell "wrong workbook" from "right workbook, nothing filled in"
        # (#2247). A blank export also parses a section header row as an
        # unknown code, so an empty ``unmatched`` is not a usable signal.
        matched += 1
        answer = cell.get("answer", "")
        comment = cell.get("comment", "")
        if answer:
            ans.answer_value = _normalise_imported_answer(ans.answer_type, answer)
        if comment:
            ans.answer_comment = comment
        if answer or comment:
            updated += 1
    if updated == 0:
        # Nothing landed — almost always the wrong workbook (a different
        # template's questionnaire parses fine: the headers are the
        # same, only the codes differ). Flipping to ``answered`` anyway
        # silently cleared the ``response_overdue`` alarm, which only
        # fires while status == 'sent', on an assessment where the
        # vendor had answered exactly nothing (#2167).
        if matched:
            # Sub-case (#2247): the codes DID match — every matched cell
            # was simply blank (the vendor returned the template
            # untouched). Reporting "N unknown code(s)" sent operators
            # hunting a code mismatch that does not exist.
            raise ValidationError(
                "the workbook's question codes match this assessment, but "
                "every matched cell was empty — the vendor appears to have "
                "returned the blank template. The status was left "
                "unchanged; ask them to fill in the answers.",
            )
        raise ValidationError(
            "no answers matched this assessment's question codes — "
            f"{len(unmatched)} unknown code(s)"
            + (f" (e.g. {', '.join(unmatched[:5])})" if unmatched else "")
            + ". The status was left unchanged; check that this is the "
            "workbook exported from THIS assessment.",
        )
    a.status = "answered"
    a.answered_at = _naive_now()
    await db.flush()
    await _recompute_scores(db, a)
    return {
        "updated": updated,
        "matched_codes": len(parsed) - len(unmatched),
        "unmatched_codes": unmatched,
        "score_pct": a.score_pct,
        "risk_rating": a.risk_rating,
    }


# Results export (internal, scored) — flatten to the generic xlsx helper.
EXPORT_COLUMNS: list[tuple[str, str]] = [
    ("question_code", "编号"),
    ("section", "分区"),
    ("question", "问题"),
    ("answer_type", "类型"),
    ("answer_value", "回答"),
    ("answer_comment", "说明"),
    ("auto_score", "得分"),
    ("max_score", "满分"),
    ("review_flag", "复核标记"),
    ("reviewer_note", "复核备注"),
]


async def export_answer_rows(
    db: AsyncSession, assessment_id: uuid.UUID,
) -> tuple[list[dict[str, Any]], int]:
    a = await get_assessment(db, assessment_id)
    answers = (await db.execute(
        select(AuditorVendorAssessmentAnswer).where(
            AuditorVendorAssessmentAnswer.assessment_id == a.id,
        ).order_by(
            AuditorVendorAssessmentAnswer.sort_order,
            AuditorVendorAssessmentAnswer.question_code,
        )
    )).scalars().all()
    rows = [
        {
            "question_code": x.question_code,
            "section": x.section_en,
            "question": x.text_en,
            "answer_type": x.answer_type,
            "answer_value": x.answer_value,
            "answer_comment": x.answer_comment,
            "auto_score": x.auto_score,
            "max_score": x.max_score,
            "review_flag": x.review_flag,
            "reviewer_note": x.reviewer_note,
        }
        for x in answers
    ]
    return rows, len(rows)


# ════════════════════════════════════════════════════════════════════
# Evidence attachments (S3 presign) — mirrors self_check discipline
# ════════════════════════════════════════════════════════════════════


def _validate_evidence_key(object_key: str) -> list[str]:
    segments = object_key.split("/")
    if ".." in segments:
        raise ValidationError("object_key must not contain '..' segments")
    if _EVIDENCE_KEY_PREFIX not in segments:
        raise ValidationError(
            "object_key is not under the vendor-assessment evidence namespace",
        )
    return segments


def _evidence_relative_key(
    *, assessment_id: uuid.UUID, person_id: uuid.UUID | None, filename: str,
) -> str:
    from app.storage_router import _safe_filename

    today = _naive_now().strftime("%Y/%m/%d")
    rand = uuid.uuid4().hex[:12]
    pid = str(person_id) if person_id else "system"
    return (
        f"{_EVIDENCE_KEY_PREFIX}/{today}/{assessment_id}/{pid}/"
        f"{rand}-{_safe_filename(filename)}"
    )


async def presign_evidence_upload(
    db: AsyncSession, *, assessment_id: 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_EVIDENCE_CONTENT_TYPES:
        raise ValidationError(
            f"unsupported content type '{content_type}' — allowed: "
            "images / pdf / xlsx",
        )
    await get_assessment(db, assessment_id)
    relative_key = _evidence_relative_key(
        assessment_id=assessment_id, person_id=person_id, filename=filename,
    )
    from app.core.storage_uploads import create_scoped_upload_grant

    grant = await create_scoped_upload_grant(
        db, person_id=person_id, purpose=_EVIDENCE_KEY_PREFIX,
        relative_key=relative_key, content_type=content_type,
        size_bytes=size_bytes,
        allowed_content_types=ALLOWED_EVIDENCE_CONTENT_TYPES,
        max_bytes=_MAX_EVIDENCE_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 register_attachment(
    db: AsyncSession, *, assessment_id: uuid.UUID, object_key: str, filename: str,
    content_type: str,
    uploaded_by_person_id: uuid.UUID | None, answer_id: uuid.UUID | None = None,
    kind: str = "evidence",
) -> AuditorVendorAssessmentAttachment:
    # No ``size_bytes`` parameter — see the self-check twin (#2247); the
    # authoritative count comes from ``stat_object`` below.
    if content_type not in ALLOWED_EVIDENCE_CONTENT_TYPES:
        raise ValidationError(f"unsupported content type '{content_type}'")
    await get_assessment(db, assessment_id)
    segments = _validate_evidence_key(object_key)
    if str(assessment_id) not in segments:
        raise ValidationError("object_key does not belong to this assessment")
    if answer_id is not None:
        # ``answer_id`` is a soft ref (no FK), so nothing below this line
        # would catch it pointing at another assessment's answer: the row
        # would file under assessment A while bumping B's
        # ``attachment_count``, leaving B's detail page claiming an
        # attachment that its own attachment list cannot show (#2168).
        ans = await db.get(AuditorVendorAssessmentAnswer, answer_id)
        if ans is None or ans.assessment_id != assessment_id:
            raise ValidationError(
                "answer_id does not belong to this assessment",
            )
    if uploaded_by_person_id is not None:
        from app.core.storage_uploads import mark_uploads_referenced

        attachment_id = uuid.uuid4()
        grants = await mark_uploads_referenced(
            db, object_keys=[object_key], person_id=uploaded_by_person_id,
            allowed_purposes=frozenset({_EVIDENCE_KEY_PREFIX}),
            resource_type="auditor_vendor_attachment",
            resource_id=str(attachment_id),
        )
        size_bytes = grants[object_key].declared_size_bytes
        content_type = grants[object_key].content_type
    else:
        attachment_id = uuid.uuid4()
        stat = await stat_object(db, object_key=object_key)
        if stat is None:
            raise ValidationError("uploaded object not found — upload before registering")
        size_bytes = int(stat["size_bytes"])
        content_type = str(stat["content_type"])
        if content_type not in ALLOWED_EVIDENCE_CONTENT_TYPES:
            raise ValidationError(f"unsupported uploaded content type '{content_type}'")
        if size_bytes > _MAX_EVIDENCE_BYTES:
            raise ValidationError("evidence file exceeds 50 MB limit")
    att = AuditorVendorAssessmentAttachment(
        id=attachment_id,
        assessment_id=assessment_id,
        answer_id=answer_id,
        kind=kind if kind in ("evidence", "response_file", "cert") else "evidence",
        object_key=object_key,
        filename=filename[:255],
        content_type=content_type,
        size_bytes=size_bytes,
        uploaded_by_person_id=uploaded_by_person_id,
        uploaded_at=_naive_now(),
    )
    db.add(att)
    await db.flush()
    if answer_id is not None:
        await _refresh_answer_attachment_count(db, answer_id)
    return att


async def _refresh_answer_attachment_count(
    db: AsyncSession, answer_id: uuid.UUID,
) -> None:
    ans = await db.get(AuditorVendorAssessmentAnswer, answer_id)
    if ans is None:
        return
    ans.attachment_count = (await db.execute(
        select(func.count()).select_from(AuditorVendorAssessmentAttachment).where(
            AuditorVendorAssessmentAttachment.answer_id == answer_id,
        )
    )).scalar_one()
    await db.flush()


async def list_attachments(
    db: AsyncSession, assessment_id: uuid.UUID,
) -> list[dict[str, Any]]:
    rows = (await db.execute(
        select(AuditorVendorAssessmentAttachment).where(
            AuditorVendorAssessmentAttachment.assessment_id == assessment_id,
        ).order_by(AuditorVendorAssessmentAttachment.uploaded_at)
    )).scalars().all()
    person_ids = {r.uploaded_by_person_id for r in rows if r.uploaded_by_person_id}
    persons = await _persons_index(db, person_ids)
    return [
        {
            "id": str(r.id),
            "answer_id": str(r.answer_id) if r.answer_id else None,
            "kind": r.kind,
            "filename": r.filename,
            "content_type": r.content_type,
            "size_bytes": r.size_bytes,
            "uploaded_at": (
                _to_naive(r.uploaded_at).isoformat() if r.uploaded_at else None
            ),
            "uploaded_by": _person_brief(persons.get(r.uploaded_by_person_id))
            if r.uploaded_by_person_id else None,
        }
        for r in rows
    ]


async def get_attachment(
    db: AsyncSession, attachment_id: uuid.UUID,
) -> AuditorVendorAssessmentAttachment:
    att = await db.get(AuditorVendorAssessmentAttachment, attachment_id)
    if att is None:
        raise NotFoundError("attachment not found")
    return att


async def presign_attachment_download(
    db: AsyncSession, attachment_id: uuid.UUID,
) -> tuple[str, AuditorVendorAssessmentAttachment]:
    """Return ``(presigned_url, attachment)``.

    The attachment row comes back so the caller can write the audit line
    — minting this URL is a data-egress event (the link bypasses Helix
    auth for 10 minutes and points at raw evidence: pen-test reports,
    certificates), and it has to be as traceable as the xlsx export of
    the same assessment already is.
    """
    att = await get_attachment(db, attachment_id)
    _validate_evidence_key(att.object_key)
    url = await presign_get(db, object_key=att.object_key)
    return url, att


async def delete_attachment(
    db: AsyncSession, attachment_id: uuid.UUID,
) -> uuid.UUID:
    att = await get_attachment(db, attachment_id)
    assessment_id = att.assessment_id
    answer_id = att.answer_id
    await db.delete(att)
    await db.flush()
    if answer_id is not None:
        await _refresh_answer_attachment_count(db, answer_id)
    return assessment_id


# ════════════════════════════════════════════════════════════════════
# Review reminders (driven by the daily cron)
# ════════════════════════════════════════════════════════════════════

# Staged reminder windows (days before next_review_date) → stage label.
# Ordered TIGHTEST-first: ``_reminder_stage_for`` returns the first stage
# whose threshold ``days_left`` has entered, so the nearest bracket must win.
# (Ordered loosest-first, every days_left in [0,30] resolved to "30", and the
# monotonic _STAGE_ORDER guard then made "14"/"due" unreachable — the owner
# got one reminder at ~30 days, then nothing until AFTER the review lapsed.)
_REVIEW_STAGES: tuple[tuple[int, str], ...] = ((0, "due"), (14, "14"), (30, "30"))


def _reminder_stage_for(days_left: int) -> str | None:
    """Which staged reminder applies at ``days_left`` (negative = overdue)."""
    if days_left < 0:
        return "overdue"
    for threshold, label in _REVIEW_STAGES:
        if days_left <= threshold:
            return label
    return None


# Ordering so we only fire a reminder that is a LATER stage than the last.
_STAGE_ORDER = {"30": 1, "14": 2, "due": 3, "overdue": 4}


async def find_due_review_reminders(
    db: AsyncSession, *, ref: date | None = None,
) -> list[tuple[AuditorVendorAssessment, str, int]]:
    """Return ``(assessment, stage, days_left)`` for completed assessments
    whose next review has entered a new reminder stage not yet sent."""
    ref = ref or _naive_now().date()
    horizon = _add_months(ref, 1)
    rows = (await db.execute(
        select(AuditorVendorAssessment).where(
            AuditorVendorAssessment.status == "completed",
            AuditorVendorAssessment.next_review_date.isnot(None),
            AuditorVendorAssessment.next_review_date <= horizon,
        )
    )).scalars().all()
    out: list[tuple[AuditorVendorAssessment, str, int]] = []
    for a in rows:
        days_left = (a.next_review_date - ref).days
        stage = _reminder_stage_for(days_left)
        if stage is None:
            continue
        prev = _STAGE_ORDER.get(a.review_reminder_stage or "", 0)
        if _STAGE_ORDER[stage] <= prev:
            continue  # already reminded at this or a later stage
        out.append((a, stage, days_left))
    return out


def mark_reminder_sent(a: AuditorVendorAssessment, stage: str) -> None:
    a.review_reminder_stage = stage
    a.review_reminder_sent_at = _naive_now()