"""周期性自我检查 — service layer (catalog, records, periods, evidence).

The engine behind the "等保、认证、审计 → 周期性自我检查" workbook:

* A **category** is a recurring check line on its own cadence
  (monthly / quarterly / semiannual / annual).
* A scheduled job (``auditor.self_check_opener_monthly``) opens one
  **record** per enabled category per period — idempotent on
  ``(category, period_year, period_label)``.
* The assigned checker performs the check, attaches evidence
  (screenshots / log exports via S3 presign) and signs off; the records
  are the auditable台账, exportable to xlsx and fed into the GRC control
  ``periodic_self_check`` for continuous monitoring.

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

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

from sqlalchemy import and_, func, or_, select
from sqlalchemy.exc import IntegrityError
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.models import (
    AuditorSelfCheckAttachment,
    AuditorSelfCheckCategory,
    AuditorSelfCheckRecord,
)
from app.identity.person_models import IdentityPerson

# ── Enumerations ─────────────────────────────────────────────────────
PERIODICITIES: tuple[str, ...] = ("monthly", "quarterly", "semiannual", "annual")
RECORD_STATUSES: tuple[str, ...] = ("pending", "completed", "skipped")
OUTCOMES: tuple[str, ...] = ("normal", "issue_found", "not_applicable")

# Evidence files: images (screenshots) + PDF (log exports). Script types
# are excluded; the browser uploads directly to S3 via a conditional presigned PUT.
ALLOWED_EVIDENCE_CONTENT_TYPES: frozenset[str] = frozenset({
    "image/jpeg",
    "image/png",
    "image/webp",
    "image/heic",
    "image/heif",
    "application/pdf",
})
_EVIDENCE_KEY_PREFIX = "self_check_evidence"
_MAX_EVIDENCE_BYTES = 25 * 1024 * 1024  # 25 MB — generous for a screenshot/log


# ════════════════════════════════════════════════════════════════════
# 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 _month_end(year: int, month: int) -> date:
    last = calendar.monthrange(year, month)[1]
    return date(year, month, last)


def _due_at(period_end: date) -> datetime:
    """Naive-UTC deadline = last instant of the period's final day."""
    return datetime(period_end.year, period_end.month, period_end.day, 23, 59, 59)


def compute_current_period(
    periodicity: str, ref: date,
) -> tuple[int, str, date, date]:
    """Return ``(year, label, start, end)`` for the period containing ``ref``."""
    y = ref.year
    if periodicity == "monthly":
        m = ref.month
        return y, f"M{m:02d}", date(y, m, 1), _month_end(y, m)
    if periodicity == "quarterly":
        q = (ref.month - 1) // 3 + 1
        start_m = (q - 1) * 3 + 1
        return y, f"Q{q}", date(y, start_m, 1), _month_end(y, start_m + 2)
    if periodicity == "semiannual":
        if ref.month <= 6:
            return y, "H1", date(y, 1, 1), date(y, 6, 30)
        return y, "H2", date(y, 7, 1), date(y, 12, 31)
    if periodicity == "annual":
        return y, "FY", date(y, 1, 1), date(y, 12, 31)
    raise ValidationError(f"unknown periodicity: {periodicity}")


def parse_period_bounds(
    periodicity: str, year: int, label: str,
) -> tuple[date, date]:
    """Reverse of :func:`compute_current_period` for an explicit period.

    Used by the manual "open a past/specific period" admin path so a
    back-dated台账 row carries correct start/end/due. Raises on a label
    that does not match the category's cadence.
    """
    label = label.strip().upper()
    if periodicity == "annual":
        if label not in ("FY", "ANNUAL"):
            raise ValidationError("annual period label must be 'FY'")
        return date(year, 1, 1), date(year, 12, 31)
    if periodicity == "semiannual":
        if label == "H1":
            return date(year, 1, 1), date(year, 6, 30)
        if label == "H2":
            return date(year, 7, 1), date(year, 12, 31)
        raise ValidationError("semiannual period label must be H1/H2")
    if periodicity == "quarterly":
        if label not in ("Q1", "Q2", "Q3", "Q4"):
            raise ValidationError("quarterly period label must be Q1..Q4")
        q = int(label[1])
        start_m = (q - 1) * 3 + 1
        return date(year, start_m, 1), _month_end(year, start_m + 2)
    if periodicity == "monthly":
        if not (label.startswith("M") and label[1:].isdigit()):
            raise ValidationError("monthly period label must be M01..M12")
        m = int(label[1:])
        if not 1 <= m <= 12:
            raise ValidationError("monthly period label must be M01..M12")
        return date(year, m, 1), _month_end(year, m)
    raise ValidationError(f"unknown periodicity: {periodicity}")


# ════════════════════════════════════════════════════════════════════
# 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,
        "employment_status": p.employment_status,
    }


# ════════════════════════════════════════════════════════════════════
# Category catalog
# ════════════════════════════════════════════════════════════════════


def category_view(c: AuditorSelfCheckCategory) -> dict[str, Any]:
    return {
        "id": str(c.id),
        "key": c.key,
        "name": c.name,
        "group_label": c.group_label,
        "periodicity": c.periodicity,
        "check_item": c.check_item,
        "guidance": c.guidance,
        "default_checker_person_id": (
            str(c.default_checker_person_id)
            if c.default_checker_person_id else None
        ),
        "requires_attachment": bool(c.requires_attachment),
        "enabled": bool(c.enabled),
        "sort_order": c.sort_order,
        "created_at": _to_naive(c.created_at).isoformat() if c.created_at else None,
        "updated_at": _to_naive(c.updated_at).isoformat() if c.updated_at else None,
    }


async def list_categories(
    db: AsyncSession, *, include_disabled: bool = True,
) -> list[dict[str, Any]]:
    stmt = select(AuditorSelfCheckCategory)
    if not include_disabled:
        stmt = stmt.where(AuditorSelfCheckCategory.enabled.is_(True))
    stmt = stmt.order_by(
        AuditorSelfCheckCategory.sort_order, AuditorSelfCheckCategory.name,
    )
    rows = (await db.execute(stmt)).scalars().all()
    out: list[dict[str, Any]] = []
    person_ids = {
        r.default_checker_person_id for r in rows if r.default_checker_person_id
    }
    persons = await _persons_index(db, person_ids)
    for r in rows:
        view = category_view(r)
        view["default_checker"] = _person_brief(
            persons.get(r.default_checker_person_id),
        ) if r.default_checker_person_id else None
        out.append(view)
    return out


async def get_category(
    db: AsyncSession, category_id: uuid.UUID,
) -> AuditorSelfCheckCategory:
    cat = await db.get(AuditorSelfCheckCategory, category_id)
    if cat is None:
        raise NotFoundError("self-check category not found")
    return cat


async def create_category(
    db: AsyncSession, *, payload: dict[str, Any],
) -> AuditorSelfCheckCategory:
    key = (payload.get("key") or "").strip()
    if not key:
        raise ValidationError("key is required")
    if payload["periodicity"] not in PERIODICITIES:
        raise ValidationError(f"periodicity must be one of {PERIODICITIES}")
    existing = (await db.execute(
        select(AuditorSelfCheckCategory).where(
            AuditorSelfCheckCategory.key == key,
        )
    )).scalar_one_or_none()
    if existing is not None:
        raise ValidationError(f"category key '{key}' already exists")
    cat = AuditorSelfCheckCategory(
        key=key,
        name=payload["name"].strip(),
        group_label=(payload.get("group_label") or None),
        periodicity=payload["periodicity"],
        check_item=payload["check_item"].strip(),
        guidance=(payload.get("guidance") or None),
        default_checker_person_id=payload.get("default_checker_person_id"),
        requires_attachment=bool(payload.get("requires_attachment", True)),
        enabled=bool(payload.get("enabled", True)),
        sort_order=int(payload.get("sort_order", 0)),
    )
    db.add(cat)
    await db.flush()
    return cat


async def update_category(
    db: AsyncSession, category_id: uuid.UUID, *, payload: dict[str, Any],
) -> AuditorSelfCheckCategory:
    cat = await get_category(db, category_id)
    if "name" in payload and payload["name"] is not None:
        cat.name = payload["name"].strip()
    if "group_label" in payload:
        cat.group_label = payload["group_label"] or None
    if "periodicity" in payload and payload["periodicity"] is not None:
        if payload["periodicity"] not in PERIODICITIES:
            raise ValidationError(f"periodicity must be one of {PERIODICITIES}")
        cat.periodicity = payload["periodicity"]
    if "check_item" in payload and payload["check_item"] is not None:
        cat.check_item = payload["check_item"].strip()
    if "guidance" in payload:
        cat.guidance = payload["guidance"] or None
    if "default_checker_person_id" in payload:
        cat.default_checker_person_id = payload["default_checker_person_id"]
    if "requires_attachment" in payload and payload["requires_attachment"] is not None:
        cat.requires_attachment = bool(payload["requires_attachment"])
    if "enabled" in payload and payload["enabled"] is not None:
        cat.enabled = bool(payload["enabled"])
    if "sort_order" in payload and payload["sort_order"] is not None:
        cat.sort_order = int(payload["sort_order"])
    await db.flush()
    return cat


# ════════════════════════════════════════════════════════════════════
# Records — open / lifecycle
# ════════════════════════════════════════════════════════════════════


async def _open_record_for_period(
    db: AsyncSession,
    cat: AuditorSelfCheckCategory,
    *,
    year: int,
    label: str,
    start: date,
    end: date,
) -> tuple[AuditorSelfCheckRecord | None, bool]:
    """Idempotently create a pending record for ``(cat, year, label)``.

    Returns ``(record, created)``. ``created=False`` when the record
    already existed (idempotent re-run).
    """
    existing = (await db.execute(
        select(AuditorSelfCheckRecord).where(
            AuditorSelfCheckRecord.category_id == cat.id,
            AuditorSelfCheckRecord.period_year == year,
            AuditorSelfCheckRecord.period_label == label,
        )
    )).scalar_one_or_none()
    if existing is not None:
        return existing, False
    rec = AuditorSelfCheckRecord(
        category_id=cat.id,
        period_year=year,
        period_label=label,
        period_start=start,
        period_end=end,
        due_at=_due_at(end),
        status="pending",
        check_item_snapshot=cat.check_item,
        assigned_checker_person_id=cat.default_checker_person_id,
    )
    db.add(rec)
    await db.flush()
    return rec, True


async def open_due_records(
    db: AsyncSession, *, ref: date | None = None,
) -> dict[str, Any]:
    """Open the current-period record for every enabled category.

    Idempotent: a category whose current period already has a record is
    skipped. Safe to run monthly — a quarter boundary is also a month
    boundary, so a single monthly cadence catches every periodicity.
    """
    ref = ref or _naive_now().date()
    cats = (await db.execute(
        select(AuditorSelfCheckCategory).where(
            AuditorSelfCheckCategory.enabled.is_(True),
        )
    )).scalars().all()
    opened: list[str] = []
    skipped = 0
    raced = 0
    for cat in cats:
        year, label, start, end = compute_current_period(cat.periodicity, ref)
        # Per-category SAVEPOINT: a unique-key race (concurrent /open-due
        # or duplicate tick) on ONE category must not roll back the
        # categories already opened in this same transaction. Without the
        # nested begin, a single IntegrityError poisons the whole flush
        # and discards the month's work, reporting a misleading 0.
        try:
            async with db.begin_nested():
                _rec, created = await _open_record_for_period(
                    db, cat, year=year, label=label, start=start, end=end,
                )
            if created:
                opened.append(f"{cat.key}:{year}-{label}")
            else:
                skipped += 1
        except IntegrityError:
            raced += 1
    return {
        "opened": opened,
        "opened_count": len(opened),
        "skipped": skipped,
        "raced": raced,
    }


async def create_record_manual(
    db: AsyncSession, *, category_id: uuid.UUID, year: int, label: str,
) -> AuditorSelfCheckRecord:
    """Open a specific (possibly back-dated) period for a category."""
    cat = await get_category(db, category_id)
    start, end = parse_period_bounds(cat.periodicity, year, label)
    rec, created = await _open_record_for_period(
        db, cat, year=year, label=label.strip().upper(), start=start, end=end,
    )
    if not created:
        raise ValidationError(
            f"record for {cat.key} {year}-{label} already exists",
        )
    return rec


async def get_record(
    db: AsyncSession, record_id: uuid.UUID,
) -> AuditorSelfCheckRecord:
    rec = await db.get(AuditorSelfCheckRecord, record_id)
    if rec is None:
        raise NotFoundError("self-check record not found")
    return rec


async def complete_record(
    db: AsyncSession,
    record_id: uuid.UUID,
    *,
    checker_person_id: uuid.UUID | None,
    outcome: str,
    remark: str | None,
) -> AuditorSelfCheckRecord:
    rec = await get_record(db, record_id)
    if outcome not in OUTCOMES:
        raise ValidationError(f"outcome must be one of {OUTCOMES}")
    if rec.status != "pending":
        # An already signed-off (completed/skipped) record must not be
        # silently overwritten — that would erase the prior checker's
        # identity + outcome with no audit of the change. Require an
        # explicit reopen first (which resets the row back to pending).
        raise ValidationError(
            "record is not pending — reopen it before completing again",
        )
    cat = await db.get(AuditorSelfCheckCategory, rec.category_id)
    if (
        cat is not None
        and cat.requires_attachment
        and outcome != "not_applicable"
        and rec.attachment_count == 0
    ):
        raise ValidationError(
            "this category requires at least one evidence attachment "
            "before it can be completed",
        )
    rec.status = "completed"
    rec.outcome = outcome
    rec.remark = remark
    rec.checker_person_id = checker_person_id
    rec.completed_at = _naive_now()
    await db.flush()
    return rec


async def skip_record(
    db: AsyncSession,
    record_id: uuid.UUID,
    *,
    checker_person_id: uuid.UUID | None,
    remark: str | None,
) -> AuditorSelfCheckRecord:
    rec = await get_record(db, record_id)
    if rec.status != "pending":
        raise ValidationError(
            "record is not pending — reopen it before skipping again",
        )
    rec.status = "skipped"
    rec.outcome = "not_applicable"
    rec.remark = remark
    rec.checker_person_id = checker_person_id
    rec.completed_at = _naive_now()
    await db.flush()
    return rec


async def reopen_record(
    db: AsyncSession, record_id: uuid.UUID,
) -> AuditorSelfCheckRecord:
    rec = await get_record(db, record_id)
    rec.status = "pending"
    rec.outcome = None
    rec.completed_at = None
    rec.checker_person_id = None
    await db.flush()
    return rec


async def update_remark(
    db: AsyncSession, record_id: uuid.UUID, *, remark: str | None,
) -> AuditorSelfCheckRecord:
    rec = await get_record(db, record_id)
    rec.remark = remark
    await db.flush()
    return rec


# ════════════════════════════════════════════════════════════════════
# Records — query + export
# ════════════════════════════════════════════════════════════════════


def record_view(
    rec: AuditorSelfCheckRecord,
    *,
    category: AuditorSelfCheckCategory | None,
    persons: dict[uuid.UUID, IdentityPerson],
    now: datetime,
) -> dict[str, Any]:
    due = _to_naive(rec.due_at)
    is_overdue = rec.status == "pending" and due is not None and now > due
    return {
        "id": str(rec.id),
        "category_id": str(rec.category_id),
        "category_name": category.name if category else None,
        "category_key": category.key if category else None,
        "group_label": category.group_label if category else None,
        "periodicity": category.periodicity if category else None,
        "period_year": rec.period_year,
        "period_label": rec.period_label,
        "period_start": rec.period_start.isoformat() if rec.period_start else None,
        "period_end": rec.period_end.isoformat() if rec.period_end else None,
        "due_at": due.isoformat() if due else None,
        "status": rec.status,
        "is_overdue": is_overdue,
        "is_checked": rec.status == "completed",
        "check_item": rec.check_item_snapshot,
        "outcome": rec.outcome,
        "remark": rec.remark,
        "attachment_count": rec.attachment_count,
        "assigned_checker": _person_brief(
            persons.get(rec.assigned_checker_person_id),
        ) if rec.assigned_checker_person_id else None,
        "checker": _person_brief(
            persons.get(rec.checker_person_id),
        ) if rec.checker_person_id else None,
        "completed_at": (
            _to_naive(rec.completed_at).isoformat() if rec.completed_at else None
        ),
        "created_at": _to_naive(rec.created_at).isoformat() if rec.created_at else None,
    }


async def get_record_detail(
    db: AsyncSession, record_id: uuid.UUID,
) -> dict[str, Any]:
    """Full single-record view + its attachments."""
    rec = await get_record(db, record_id)
    cat = await db.get(AuditorSelfCheckCategory, rec.category_id)
    person_ids = {
        pid for pid in (rec.assigned_checker_person_id, rec.checker_person_id)
        if pid
    }
    persons = await _persons_index(db, person_ids)
    view = record_view(rec, category=cat, persons=persons, now=_naive_now())
    view["attachments"] = await list_attachments(db, record_id)
    return view


async def list_records(
    db: AsyncSession,
    *,
    category_id: uuid.UUID | None = None,
    status: str | None = None,
    period_year: int | None = None,
    overdue_only: bool = False,
    page: int = 1,
    page_size: int = 50,
) -> dict[str, Any]:
    now = _naive_now()
    conds = []
    if category_id is not None:
        conds.append(AuditorSelfCheckRecord.category_id == category_id)
    if status is not None:
        conds.append(AuditorSelfCheckRecord.status == status)
    if period_year is not None:
        conds.append(AuditorSelfCheckRecord.period_year == period_year)
    if overdue_only:
        conds.append(AuditorSelfCheckRecord.status == "pending")
        conds.append(AuditorSelfCheckRecord.due_at < now)

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

    cat_ids = {r.category_id for r in rows}
    cats = {
        c.id: c for c in (await db.execute(
            select(AuditorSelfCheckCategory).where(
                AuditorSelfCheckCategory.id.in_(cat_ids),
            )
        )).scalars().all()
    } if cat_ids else {}
    person_ids: set[uuid.UUID] = set()
    for r in rows:
        if r.assigned_checker_person_id:
            person_ids.add(r.assigned_checker_person_id)
        if r.checker_person_id:
            person_ids.add(r.checker_person_id)
    persons = await _persons_index(db, person_ids)

    items = [
        record_view(r, category=cats.get(r.category_id), persons=persons, now=now)
        for r in rows
    ]
    return {
        "items": items,
        "meta": {"page": page, "page_size": page_size, "total": total},
    }


async def export_rows(
    db: AsyncSession,
    *,
    category_id: uuid.UUID | None,
    status: str | None,
    period_year: int | None,
) -> tuple[list[dict[str, Any]], int]:
    """Flatten records to export dicts (no pagination — capped by caller)."""
    from app.core.exporting import EXPORT_ROW_LIMIT

    result = await list_records(
        db,
        category_id=category_id,
        status=status,
        period_year=period_year,
        page=1,
        page_size=EXPORT_ROW_LIMIT,
    )
    status_label = {"pending": "待检查", "completed": "已检查", "skipped": "不适用"}
    outcome_label = {
        "normal": "正常", "issue_found": "发现异常", "not_applicable": "不适用",
    }
    rows: list[dict[str, Any]] = []
    for it in result["items"]:
        checker = it.get("checker") or it.get("assigned_checker") or {}
        rows.append({
            "category_name": it["category_name"],
            "group_label": it["group_label"],
            "period_year": it["period_year"],
            "period_label": it["period_label"],
            "is_checked": "是" if it["is_checked"] else "否",
            "check_item": it["check_item"],
            "outcome": outcome_label.get(it["outcome"], "") if it["outcome"] else "",
            "status": status_label.get(it["status"], it["status"]),
            "attachment_count": it["attachment_count"],
            "checker_name": checker.get("display_name") if checker else None,
            "completed_at": it["completed_at"],
            "remark": it["remark"],
        })
    return rows, result["meta"]["total"]


EXPORT_COLUMNS: list[tuple[str, str]] = [
    ("category_name", "检查类目"),
    ("group_label", "分组"),
    ("period_year", "年份"),
    ("period_label", "周期"),
    ("is_checked", "是否检查"),
    ("check_item", "检查项"),
    ("outcome", "检查结果"),
    ("status", "状态"),
    ("attachment_count", "附件数"),
    ("checker_name", "检查人"),
    ("completed_at", "完成时间"),
    ("remark", "备注"),
]


# ════════════════════════════════════════════════════════════════════
# Overview + control evaluation
# ════════════════════════════════════════════════════════════════════


async def get_overview(db: AsyncSession) -> dict[str, Any]:
    """Per-category current-period status + global counters."""
    now = _naive_now()
    ref = now.date()
    cats = (await db.execute(
        select(AuditorSelfCheckCategory).where(
            AuditorSelfCheckCategory.enabled.is_(True),
        ).order_by(
            AuditorSelfCheckCategory.sort_order, AuditorSelfCheckCategory.name,
        )
    )).scalars().all()

    # Overdue across all periods (the real backlog signal).
    overdue_total = (await db.execute(
        select(func.count()).select_from(AuditorSelfCheckRecord).where(
            AuditorSelfCheckRecord.status == "pending",
            AuditorSelfCheckRecord.due_at < now,
        )
    )).scalar_one()

    person_ids = {
        c.default_checker_person_id for c in cats if c.default_checker_person_id
    }
    persons = await _persons_index(db, person_ids)

    # One query for every (category, current period) instead of one per
    # category (#2188-9). This overview is reused by the HTTP route, the
    # MCP tool and the security-governance overview, so the N+1 was
    # multiplied three ways on pages people open constantly.
    periods_by_cat = {
        cat.id: compute_current_period(cat.periodicity, ref) for cat in cats
    }
    recs_by_key: dict[tuple[Any, int, str], AuditorSelfCheckRecord] = {}
    if periods_by_cat:
        wanted = (await db.execute(
            select(AuditorSelfCheckRecord).where(
                AuditorSelfCheckRecord.category_id.in_(list(periods_by_cat)),
                or_(*[
                    and_(
                        AuditorSelfCheckRecord.category_id == cid,
                        AuditorSelfCheckRecord.period_year == year,
                        AuditorSelfCheckRecord.period_label == label,
                    )
                    for cid, (year, label, _s, _e) in periods_by_cat.items()
                ]),
            )
        )).scalars().all()
        recs_by_key = {
            (r.category_id, r.period_year, r.period_label): r for r in wanted
        }

    categories: list[dict[str, Any]] = []
    current_done = 0
    for cat in cats:
        year, label, _start, _end = periods_by_cat[cat.id]
        rec = recs_by_key.get((cat.id, year, label))
        cur_status = rec.status if rec is not None else "not_opened"
        if cur_status in ("completed", "skipped"):
            current_done += 1
        categories.append({
            "id": str(cat.id),
            "key": cat.key,
            "name": cat.name,
            "group_label": cat.group_label,
            "periodicity": cat.periodicity,
            "current_period": f"{year}-{label}",
            "current_status": cur_status,
            "current_record_id": str(rec.id) if rec is not None else None,
            "default_checker": _person_brief(
                persons.get(cat.default_checker_person_id),
            ) if cat.default_checker_person_id else None,
        })

    return {
        "summary": {
            "category_count": len(cats),
            "current_period_done": current_done,
            "current_period_pending": len(cats) - current_done,
            "overdue_total": overdue_total,
        },
        "categories": categories,
    }


async def evaluate_self_check_control(db: AsyncSession) -> dict[str, Any]:
    """Compute the ``periodic_self_check`` GRC control status.

    Pass when nothing is overdue; fail with the overdue count otherwise.
    Score = on-time completion rate of all due records (those past due_at).
    Returns the shape ``recompute_control_status`` expects.
    """
    now = _naive_now()
    enabled_count = (await db.execute(
        select(func.count()).select_from(AuditorSelfCheckCategory).where(
            AuditorSelfCheckCategory.enabled.is_(True),
        )
    )).scalar_one()
    if enabled_count == 0:
        return {
            "status": "not_configured",
            "score": None,
            "summary": "未配置任何周期自检类目",
            "detail": {},
        }

    # Due records = those whose deadline has passed.
    due_total = (await db.execute(
        select(func.count()).select_from(AuditorSelfCheckRecord).where(
            AuditorSelfCheckRecord.due_at < now,
        )
    )).scalar_one()
    overdue = (await db.execute(
        select(func.count()).select_from(AuditorSelfCheckRecord).where(
            AuditorSelfCheckRecord.status == "pending",
            AuditorSelfCheckRecord.due_at < now,
        )
    )).scalar_one()
    completed_due = due_total - overdue
    score = round(completed_due / due_total * 100) if due_total else 100
    status = "pass" if overdue == 0 else "fail"
    summary = (
        "周期自检全部按时完成" if overdue == 0
        else f"{overdue} 项周期自检逾期未完成"
    )
    return {
        "status": status,
        "score": score,
        "summary": summary,
        "detail": {
            "enabled_categories": enabled_count,
            "due_total": due_total,
            "completed_due": completed_due,
            "overdue": overdue,
        },
    }


# ════════════════════════════════════════════════════════════════════
# Evidence attachments (S3 presign)
# ════════════════════════════════════════════════════════════════════


def _validate_evidence_key(object_key: str) -> list[str]:
    """Assert ``object_key`` lives in the self-check evidence namespace.

    Returns the path segments (for the caller's per-record check). Raises
    :class:`ValidationError` when the key has no ``self_check_evidence``
    segment or contains a ``..`` traversal segment. Segment-exact (not
    substring) so a crafted key can't smuggle a pointer to an unrelated
    object elsewhere in the bucket.
    """
    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 self-check evidence namespace",
        )
    return segments


def _evidence_relative_key(
    *, record_id: uuid.UUID, person_id: uuid.UUID | None, filename: str,
) -> str:
    from app.storage_router import _safe_filename  # narrow reuse of sanitiser

    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}/{record_id}/{pid}/"
        f"{rand}-{_safe_filename(filename)}"
    )


async def presign_evidence_upload(
    db: AsyncSession,
    *,
    record_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 (jpeg/png/webp/heic) or pdf",
        )
    await get_record(db, record_id)  # 404 if the record is gone
    relative_key = _evidence_relative_key(
        record_id=record_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,
    *,
    record_id: uuid.UUID,
    object_key: str,
    filename: str,
    content_type: str,
    uploaded_by_person_id: uuid.UUID | None,
) -> AuditorSelfCheckAttachment:
    # NOTE (#2247): there is deliberately NO ``size_bytes`` parameter. The
    # size is HEADed off the object below and the client's assertion was
    # unconditionally overwritten — keeping the argument only invited a
    # future caller to believe the value it passes means something.
    if content_type not in ALLOWED_EVIDENCE_CONTENT_TYPES:
        raise ValidationError(f"unsupported content type '{content_type}'")
    rec = await get_record(db, record_id)
    # Defence-in-depth: the object key must live under THIS record's
    # evidence prefix — never trust a client-supplied key that points
    # elsewhere in the bucket. Validate on EXACT path segments (not a
    # substring `in`, which a crafted key like
    # ``self_check_evidence/<rid>/../other/secret.pdf`` would slip past):
    # ``self_check_evidence`` must be a literal segment, the record id
    # must be a literal segment AFTER it, and no ``..`` traversal segment
    # may appear.
    _segments = _validate_evidence_key(object_key)
    if str(record_id) not in _segments:
        raise ValidationError("object_key does not belong to this record")
    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_self_check_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"])  # type: ignore[call-overload]
        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 25 MB limit")
    att = AuditorSelfCheckAttachment(
        id=attachment_id,
        record_id=record_id,
        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()
    rec.attachment_count = await _count_attachments(db, record_id)
    await db.flush()
    return att


async def _count_attachments(db: AsyncSession, record_id: uuid.UUID) -> int:
    return (await db.execute(
        select(func.count()).select_from(AuditorSelfCheckAttachment).where(
            AuditorSelfCheckAttachment.record_id == record_id,
        )
    )).scalar_one()


async def list_attachments(
    db: AsyncSession, record_id: uuid.UUID,
) -> list[dict[str, Any]]:
    await get_record(db, record_id)
    rows = (await db.execute(
        select(AuditorSelfCheckAttachment).where(
            AuditorSelfCheckAttachment.record_id == record_id,
        ).order_by(AuditorSelfCheckAttachment.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),
            "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,
) -> AuditorSelfCheckAttachment:
    att = await db.get(AuditorSelfCheckAttachment, 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, AuditorSelfCheckAttachment]:
    """Return ``(presigned_url, attachment)``.

    The 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 it must be as traceable as the xlsx export of the same
    record already is.
    """
    att = await get_attachment(db, attachment_id)
    # Belt-and-suspenders: presign_get bypasses Helix auth once minted,
    # so never presign a stored key that isn't inside the evidence
    # namespace (guards against a poisoned row from any future write path).
    _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)
    record_id = att.record_id
    await db.delete(att)
    await db.flush()
    rec = await db.get(AuditorSelfCheckRecord, record_id)
    if rec is not None:
        rec.attachment_count = await _count_attachments(db, record_id)
        await db.flush()
    return record_id