"""Monthly close, reopen approval and locked-period write guards.

A close freezes one calendar month: the snapshot is hashed and stored as an
append-only month report. Ordinary writes that would change that month's
results are rejected until an administrator approves a reopen request.
Bank source files remain immutable evidence and are never rewritten.
"""

from __future__ import annotations

from calendar import monthrange
from datetime import date, datetime, timedelta, timezone
from decimal import Decimal, ROUND_HALF_UP
import hashlib
import json
import re
import sqlite3

from .db import utc_now
from . import calculation, dashboard, settings as settings_mod


YEAR_MONTH_RE = re.compile(r"^\d{4}-(0[1-9]|1[0-2])$")
CLOSE_STATUSES = ("pending", "closing", "closed", "failed", "reopened")
REOPEN_STATUSES = ("pending", "approved", "rejected")
DEFAULT_REOPEN_DAYS = 3
MIN_REASON_LEN = 10
TWOPLACES = Decimal("0.01")

LOCK_TOAST = (
    "{year_month} 已结账锁定，不可直接修改 / 确需更正请前往「结账与期初」"
    "发起重开申请，经审批后修改并留痕。"
)


class PeriodLockedError(ValueError):
    """A write targeted a currently locked month."""

    def __init__(self, year_month: str):
        self.year_month = year_month
        super().__init__(lock_message(year_month))


class PeriodCloseError(ValueError):
    """Close / reopen business-rule failure."""


class PeriodConflictError(ValueError):
    """Idempotency or state conflict."""


def lock_message(year_month: str) -> str:
    return LOCK_TOAST.format(year_month=year_month)


def today_shanghai() -> date:
    return datetime.now(timezone(timedelta(hours=8))).date()


def validate_year_month(value: object) -> str:
    text = str(value or "").strip()
    if not YEAR_MONTH_RE.fullmatch(text):
        raise PeriodCloseError("账期必须是 YYYY-MM。")
    return text


def month_bounds(year_month: str) -> tuple[str, str]:
    year_month = validate_year_month(year_month)
    year, month = (int(part) for part in year_month.split("-"))
    last = monthrange(year, month)[1]
    return f"{year_month}-01", f"{year_month}-{last:02d}"


def add_months(year_month: str, delta: int) -> str:
    year, month = (int(part) for part in validate_year_month(year_month).split("-"))
    month += delta
    while month < 1:
        month += 12
        year -= 1
    while month > 12:
        month -= 12
        year += 1
    return f"{year:04d}-{month:02d}"


def next_month(year_month: str) -> str:
    return add_months(year_month, 1)


def previous_month(year_month: str) -> str:
    return add_months(year_month, -1)


def month_of(iso_date: str | None) -> str | None:
    if not iso_date:
        return None
    text = str(iso_date).strip()
    if len(text) < 7:
        return None
    candidate = text[:7]
    return candidate if YEAR_MONTH_RE.fullmatch(candidate) else None


def _q2(value: Decimal) -> str:
    return str(value.quantize(TWOPLACES, rounding=ROUND_HALF_UP))


def _canonical_json(payload: object) -> str:
    return json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":"))


def _hash_payload(payload: object) -> str:
    return hashlib.sha256(_canonical_json(payload).encode("utf-8")).hexdigest()


def _actor_name(actor: sqlite3.Row | None) -> str:
    if actor is None:
        return "system"
    return str(actor["username"] or actor["display_name"] or "admin")


def _actor_role(actor: sqlite3.Row | None) -> str:
    if actor is None:
        return "system"
    return str(actor["role"] or "admin")


def closing_day(connection: sqlite3.Connection) -> int:
    values = settings_mod.get_settings(connection)
    try:
        day = int(values.get("closing_day") or "5")
    except (TypeError, ValueError):
        day = 5
    return max(1, min(day, 28))


def last_closable_month(connection: sqlite3.Connection, *, today: date | None = None) -> str:
    """The latest month that may be closed given today's date and closing_day.

    Closing day 5 of August means July becomes closable on August 5.
    """
    today = today or today_shanghai()
    day = closing_day(connection)
    if today.day >= day:
        return previous_month(f"{today.year:04d}-{today.month:02d}")
    return add_months(f"{today.year:04d}-{today.month:02d}", -2)


def earliest_month(connection: sqlite3.Connection) -> str | None:
    start = calculation.get_calculation_start_date(connection)
    if start:
        return month_of(start)
    row = connection.execute(
        "SELECT MIN(substr(transaction_at, 1, 7)) AS ym FROM source_rows"
    ).fetchone()
    ym = row["ym"] if row is not None else None
    return ym if ym and YEAR_MONTH_RE.fullmatch(str(ym)) else None


def is_month_locked(connection: sqlite3.Connection, year_month: str) -> bool:
    year_month = validate_year_month(year_month)
    row = connection.execute(
        "SELECT 1 FROM closed_periods WHERE year_month = ?", (year_month,)
    ).fetchone()
    return row is not None


def locked_month_for_date(connection: sqlite3.Connection, iso_date: str | None) -> str | None:
    ym = month_of(iso_date)
    if ym and is_month_locked(connection, ym):
        return ym
    return None


def assert_date_writable(connection: sqlite3.Connection, iso_date: str | None) -> None:
    ym = locked_month_for_date(connection, iso_date)
    if ym:
        raise PeriodLockedError(ym)


def assert_event_writable(connection: sqlite3.Connection, event_id: int) -> None:
    row = connection.execute(
        """
        SELECT d.effective_at
        FROM current_transfer_decisions c
        JOIN transfer_match_decisions d ON d.id = c.decision_id
        WHERE c.event_id = ?
        """,
        (int(event_id),),
    ).fetchone()
    if row is not None:
        assert_date_writable(connection, row["effective_at"])


def assert_ledger_writable(connection: sqlite3.Connection, ledger_event_id: int) -> None:
    row = connection.execute(
        """
        SELECT p.effective_at
        FROM current_ledger_event_revisions cur
        JOIN ledger_event_revisions p ON p.id = cur.revision_id
        WHERE cur.ledger_event_id = ?
        """,
        (int(ledger_event_id),),
    ).fetchone()
    if row is not None:
        assert_date_writable(connection, row["effective_at"])


def split_writable_row_ids(
    connection: sqlite3.Connection, source_row_ids: list[int]
) -> tuple[list[int], list[tuple[int, str]]]:
    """Return (writable_ids, [(row_id, year_month), ...] for locked rows)."""
    if not source_row_ids:
        return [], []
    placeholders = ",".join("?" for _ in source_row_ids)
    rows = connection.execute(
        f"SELECT id, transaction_at FROM source_rows WHERE id IN ({placeholders})",
        [int(item) for item in source_row_ids],
    ).fetchall()
    writable: list[int] = []
    locked: list[tuple[int, str]] = []
    for row in rows:
        ym = month_of(row["transaction_at"])
        if ym and is_month_locked(connection, ym):
            locked.append((int(row["id"]), ym))
        else:
            writable.append(int(row["id"]))
    return writable, locked


def record_late_arrivals(
    connection: sqlite3.Connection,
    items: list[tuple[int, str]],
    actor: sqlite3.Row | None,
) -> int:
    if not items:
        return 0
    now = utc_now()
    inserted = 0
    # One transaction so the rows and their audit trail commit or roll back
    # together; without it the caller's connection.close() silently rolled
    # the late-arrival records back while the API still reported them.
    with connection:
        for row_id, year_month in items:
            existing = connection.execute(
                "SELECT id FROM period_late_arrivals WHERE source_row_id = ?",
                (row_id,),
            ).fetchone()
            if existing is not None:
                continue
            connection.execute(
                """
                INSERT INTO period_late_arrivals (
                    year_month, source_row_id, status, created_at, actor_user_id, actor_username
                ) VALUES (?, ?, 'open', ?, ?, ?)
                """,
                (
                    year_month,
                    row_id,
                    now,
                    actor["id"] if actor is not None else None,
                    _actor_name(actor),
                ),
            )
            inserted += 1
            _append_audit(
                connection,
                actor,
                "late_arrival",
                year_month,
                object_label=f"源行 {row_id}",
                reason="闭期后到达的流水，未改写已结账快照",
                after={"source_row_id": row_id},
            )
    return inserted


def _append_audit(
    connection: sqlite3.Connection,
    actor: sqlite3.Row | None,
    action: str,
    year_month: str | None,
    *,
    object_label: str | None = None,
    reason: str | None = None,
    before: object = None,
    after: object = None,
    report_no: str | None = None,
    related_id: int | None = None,
) -> None:
    connection.execute(
        """
        INSERT INTO period_audit_events (
            created_at, actor_user_id, actor_username, actor_role, action,
            year_month, object_label, reason, before_json, after_json,
            report_no, related_id
        ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
        """,
        (
            utc_now(),
            actor["id"] if actor is not None else None,
            _actor_name(actor),
            _actor_role(actor),
            action,
            year_month,
            object_label,
            reason,
            _canonical_json(before) if before is not None else None,
            _canonical_json(after) if after is not None else None,
            report_no,
            related_id,
        ),
    )


def build_snapshot(connection: sqlite3.Connection, year_month: str) -> dict[str, object]:
    start, end = month_bounds(year_month)
    companies, totals = dashboard.company_summaries(
        connection, from_date=start, cutoff=end
    )
    event_count = connection.execute(
        """
        SELECT COUNT(*) AS n FROM eligible_intercompany_events
        WHERE date(effective_at) >= date(?) AND date(effective_at) <= date(?)
        """,
        (start, end),
    ).fetchone()["n"]
    pending = connection.execute(
        """
        SELECT COUNT(*) AS n
        FROM current_transfer_decisions c
        JOIN transfer_match_decisions d ON d.id = c.decision_id
        JOIN canonical_transfer_events e ON e.id = c.event_id
        WHERE e.lifecycle = 'active'
          AND d.classification IN ('unresolved', 'needs_review')
          AND date(d.effective_at) >= date(?)
          AND date(d.effective_at) <= date(?)
        """,
        (start, end),
    ).fetchone()
    openings = calculation.list_opening_balances(connection)
    confirmed_openings = [
        item for item in openings if item.get("status") == "confirmed"
    ]
    return {
        "year_month": year_month,
        "window": {"start": start, "end": end},
        "companies": companies,
        "totals": totals,
        "event_count": int(event_count),
        "pending_unconfirmed_count": int(pending["n"] if pending else 0),
        "opening_pairs": [
            {
                "id": item["id"],
                "company_id_low": item["company_id_low"],
                "company_id_high": item["company_id_high"],
                "amount": item["amount"],
            }
            for item in confirmed_openings
        ],
        "carry_forward_to": next_month(year_month),
    }


def _precondition_rows(connection: sqlite3.Connection, year_month: str) -> list[dict[str, object]]:
    start, end = month_bounds(year_month)
    companies = connection.execute(
        "SELECT COUNT(*) AS n FROM companies WHERE status != 'disabled'"
    ).fetchone()["n"]
    submitted = connection.execute(
        """
        SELECT COUNT(DISTINCT b.company_id) AS n
        FROM source_rows r
        JOIN sheet_batches s ON s.id = r.sheet_batch_id
        JOIN import_batches b ON b.id = s.import_batch_id
        JOIN sheet_reviews rv ON rv.sheet_batch_id = s.id AND rv.review_status = 'confirmed'
        WHERE date(r.transaction_at) >= date(?) AND date(r.transaction_at) <= date(?)
        """,
        (start, end),
    ).fetchone()["n"]
    attested = connection.execute(
        """
        SELECT COUNT(DISTINCT a.company_id) AS n
        FROM no_business_attestations a
        WHERE a.status = 'approved'
          AND date(a.gap_start) <= date(?)
          AND date(a.gap_end) >= date(?)
        """,
        (end, start),
    ).fetchone()["n"]
    covered = int(submitted) + int(attested)
    open_gaps = connection.execute(
        """
        SELECT COUNT(*) AS n FROM coverage_gaps
        WHERE status = 'open'
          AND date(gap_start) <= date(?)
          AND date(gap_end) >= date(?)
        """,
        (end, start),
    ).fetchone()["n"]
    pending_matches = connection.execute(
        """
        SELECT COUNT(*) AS n
        FROM current_transfer_decisions c
        JOIN transfer_match_decisions d ON d.id = c.decision_id
        JOIN canonical_transfer_events e ON e.id = c.event_id
        WHERE e.lifecycle = 'active'
          AND d.classification IN ('unresolved', 'needs_review')
          AND date(d.effective_at) >= date(?)
          AND date(d.effective_at) <= date(?)
        """,
        (start, end),
    ).fetchone()["n"]
    pending_manuals = connection.execute(
        """
        SELECT COUNT(*) AS n
        FROM manual_records m
        JOIN current_manual_record_decisions c ON c.record_id = m.id
        JOIN manual_record_decisions d ON d.id = c.decision_id
        WHERE d.state = 'pending'
          AND date(m.occurred_at) >= date(?)
          AND date(m.occurred_at) <= date(?)
        """,
        (start, end),
    ).fetchone()["n"]
    pending_accounts = connection.execute(
        "SELECT COUNT(*) AS n FROM bank_accounts WHERE status = 'pending'"
    ).fetchone()["n"]
    pending_attest = connection.execute(
        "SELECT COUNT(*) AS n FROM no_business_attestations WHERE status = 'pending'"
    ).fetchone()["n"]
    review_total = int(pending_matches) + int(pending_manuals) + int(pending_accounts) + int(pending_attest)
    start_date = calculation.get_calculation_start_date(connection)
    opening_ok = True
    opening_sub = "未设置起算日，本月按期间净变动结账"
    if start_date:
        summary = calculation.opening_coverage_summary(connection)
        pair_count = int(summary.get("confirmed_pair_count") or 0)
        opening_ok = pair_count > 0 or int(companies) <= 1
        opening_sub = (
            f"已确认 {pair_count} 对公司对期初"
            if opening_ok
            else "尚未录入已确认的公司对期初"
        )

    submit_ok = int(companies) == 0 or covered >= int(companies) or int(submitted) > 0 and int(open_gaps) == 0
    # Missing submissions only block when some companies exist and none have
    # coverage for the month. A company with no enabled accounts is not required.
    enabled_companies = connection.execute(
        """
        SELECT COUNT(DISTINCT company_id) AS n FROM bank_accounts
        WHERE status = 'active'
        """
    ).fetchone()["n"]
    if int(enabled_companies) == 0:
        submit_ok = True
        submit_sub = "尚无已启用银行账户，不要求流水提交"
    else:
        submit_ok = int(submitted) + int(attested) >= int(enabled_companies)
        submit_sub = f"{int(submitted) + int(attested)} / {int(enabled_companies)} 家已覆盖（含无业务说明）"

    return [
        {
            "key": "submissions",
            "title": "流水提交",
            "ok": submit_ok,
            "blocking": not submit_ok,
            "detail": submit_sub,
        },
        {
            "key": "coverage",
            "title": "账户连续",
            "ok": int(open_gaps) == 0,
            "blocking": int(open_gaps) > 0,
            "detail": "账户覆盖连续" if int(open_gaps) == 0 else f"{int(open_gaps)} 个账户存在断档",
        },
        {
            "key": "reviews",
            "title": "审核完成",
            "ok": review_total == 0,
            "blocking": review_total > 0,
            "detail": "0 项尚未处理" if review_total == 0 else f"{review_total} 项尚未处理",
        },
        {
            "key": "opening",
            "title": "期初锁定",
            "ok": opening_ok,
            "blocking": not opening_ok,
            "detail": opening_sub,
        },
    ]


def evaluate_preconditions(connection: sqlite3.Connection, year_month: str) -> dict[str, object]:
    checks = _precondition_rows(connection, year_month)
    blockers = [item for item in checks if item["blocking"]]
    return {
        "year_month": year_month,
        "checks": checks,
        "ready": len(blockers) == 0,
        "blockers": blockers,
    }


def _current_run(connection: sqlite3.Connection, year_month: str) -> sqlite3.Row | None:
    return connection.execute(
        """
        SELECT * FROM period_close_runs
        WHERE year_month = ?
        ORDER BY version DESC
        LIMIT 1
        """,
        (year_month,),
    ).fetchone()


def _max_version(connection: sqlite3.Connection, year_month: str) -> int:
    row = connection.execute(
        "SELECT MAX(version) AS v FROM period_close_runs WHERE year_month = ?",
        (year_month,),
    ).fetchone()
    return int(row["v"] or 0)


def ensure_pending_tasks(
    connection: sqlite3.Connection, *, today: date | None = None
) -> list[str]:
    """Create pending close tasks for every closable month that has none."""
    last = last_closable_month(connection, today=today)
    first = earliest_month(connection) or last
    created: list[str] = []
    month = first
    with connection:
        while month <= last:
            existing = _current_run(connection, month)
            if existing is None:
                _insert_pending(connection, month, actor=None)
                created.append(month)
            if month == last:
                break
            month = next_month(month)
    return created


def _insert_pending(
    connection: sqlite3.Connection, year_month: str, actor: sqlite3.Row | None
) -> sqlite3.Row:
    version = _max_version(connection, year_month) + 1
    now = utc_now()
    connection.execute(
        """
        INSERT INTO period_close_runs (
            year_month, version, status, created_at, updated_at
        ) VALUES (?, ?, 'pending', ?, ?)
        """,
        (year_month, version, now, now),
    )
    _append_audit(
        connection,
        actor,
        "close_pending",
        year_month,
        object_label=f"{year_month} 待结账任务",
        reason="到达结账日后自动生成待结账任务，待管理员确认锁账",
    )
    row = _current_run(connection, year_month)
    assert row is not None
    return row


def _report_no(year_month: str, version: int) -> str:
    compact = year_month.replace("-", "")
    return f"MR-{compact}-v{version}"


def execute_close(
    connection: sqlite3.Connection,
    year_month: str,
    actor: sqlite3.Row,
    *,
    confirm: bool = False,
) -> dict[str, object]:
    year_month = validate_year_month(year_month)
    if not confirm:
        raise PeriodCloseError("必须勾选确认后才能执行结账。")
    evaluation = evaluate_preconditions(connection, year_month)
    if not evaluation["ready"]:
        raise PeriodCloseError(
            "存在阻断项，暂不能执行月度结账："
            + "；".join(item["detail"] for item in evaluation["blockers"])
        )
    current = _current_run(connection, year_month)
    if current is not None and current["status"] == "closed":
        raise PeriodConflictError(f"{year_month} 已结账，不可重复执行。")
    if current is not None and current["status"] not in ("pending", "failed", "reopened"):
        raise PeriodConflictError(f"{year_month} 当前状态为 {current['status']}，不能结账。")

    snapshot = build_snapshot(connection, year_month)
    digest = _hash_payload(snapshot)
    now = utc_now()
    with connection:
        if current is not None and current["status"] == "pending":
            version = int(current["version"])
            report_no = _report_no(year_month, version)
            connection.execute(
                """
                UPDATE period_close_runs
                SET status = 'closed', snapshot_json = ?, snapshot_hash = ?,
                    report_no = ?, blockers_json = '[]', fail_reason = NULL,
                    closed_at = ?, closed_by = ?, closed_by_username = ?,
                    reopen_window_end = NULL, updated_at = ?
                WHERE id = ?
                """,
                (
                    _canonical_json(snapshot), digest, report_no,
                    now, actor["id"], _actor_name(actor), now, int(current["id"]),
                ),
            )
        else:
            version = _max_version(connection, year_month) + 1
            report_no = _report_no(year_month, version)
            connection.execute(
                """
                INSERT INTO period_close_runs (
                    year_month, version, status, snapshot_json, snapshot_hash,
                    report_no, blockers_json, closed_at, closed_by, closed_by_username,
                    created_at, updated_at
                ) VALUES (?, ?, 'closed', ?, ?, ?, '[]', ?, ?, ?, ?, ?)
                """,
                (
                    year_month, version, _canonical_json(snapshot), digest, report_no,
                    now, actor["id"], _actor_name(actor), now, now,
                ),
            )
        connection.execute(
            """
            INSERT INTO closed_periods (year_month, closed_at, closed_by)
            VALUES (?, ?, ?)
            ON CONFLICT(year_month) DO UPDATE SET
                closed_at = excluded.closed_at,
                closed_by = excluded.closed_by
            """,
            (year_month, now, actor["id"]),
        )
        _append_audit(
            connection,
            actor,
            "close_execute",
            year_month,
            object_label=report_no,
            reason="管理员确认后锁定账期并生成月报",
            after={"snapshot_hash": digest, "report_no": report_no, "totals": snapshot["totals"]},
            report_no=report_no,
        )
    return close_payload(connection, year_month)


def mark_close_failed(
    connection: sqlite3.Connection,
    year_month: str,
    actor: sqlite3.Row,
    reason: str,
) -> dict[str, object]:
    year_month = validate_year_month(year_month)
    reason = str(reason or "").strip() or "结账失败"
    current = _current_run(connection, year_month)
    now = utc_now()
    with connection:
        if current is None:
            version = 1
            connection.execute(
                """
                INSERT INTO period_close_runs (
                    year_month, version, status, fail_reason, created_at, updated_at
                ) VALUES (?, ?, 'failed', ?, ?, ?)
                """,
                (year_month, version, reason, now, now),
            )
        else:
            connection.execute(
                """
                UPDATE period_close_runs
                SET status = 'failed', fail_reason = ?, updated_at = ?
                WHERE id = ?
                """,
                (reason, now, int(current["id"])),
            )
        _append_audit(
            connection,
            actor,
            "close_fail",
            year_month,
            reason=reason,
            after={"fail_reason": reason},
        )
    return close_payload(connection, year_month)


def request_reopen(
    connection: sqlite3.Connection,
    year_month: str,
    actor: sqlite3.Row,
    *,
    reason: str,
    companies_note: str = "",
    window_days: int = DEFAULT_REOPEN_DAYS,
) -> dict[str, object]:
    year_month = validate_year_month(year_month)
    reason = str(reason or "").strip()
    if len(reason) < MIN_REASON_LEN:
        raise PeriodCloseError(f"重开原因不少于 {MIN_REASON_LEN} 个字，将写入审计记录。")
    try:
        days = int(window_days or DEFAULT_REOPEN_DAYS)
    except (TypeError, ValueError):
        days = DEFAULT_REOPEN_DAYS
    days = max(1, min(days, 30))
    current = _current_run(connection, year_month)
    if current is None or current["status"] != "closed":
        raise PeriodCloseError("只有已锁定的账期才能申请重开。")
    pending = connection.execute(
        """
        SELECT id FROM period_reopen_requests
        WHERE year_month = ? AND status = 'pending'
        """,
        (year_month,),
    ).fetchone()
    if pending is not None:
        raise PeriodConflictError("该账期已有待审批的重开申请。")
    now = utc_now()
    with connection:
        cursor = connection.execute(
            """
            INSERT INTO period_reopen_requests (
                period_close_id, year_month, reason, companies_note, window_days,
                status, requester_id, requester_username, requested_at,
                before_json
            ) VALUES (?, ?, ?, ?, ?, 'pending', ?, ?, ?, ?)
            """,
            (
                int(current["id"]),
                year_month,
                reason,
                str(companies_note or "").strip(),
                days,
                actor["id"],
                _actor_name(actor),
                now,
                current["snapshot_json"],
            ),
        )
        request_id = int(cursor.lastrowid)
        _append_audit(
            connection,
            actor,
            "reopen_request",
            year_month,
            object_label=f"RO-{request_id:06d}",
            reason=reason,
            before={"status": "closed", "report_no": current["report_no"]},
            related_id=request_id,
            report_no=current["report_no"],
        )
    return reopen_payload(connection, request_id)


def decide_reopen(
    connection: sqlite3.Connection,
    request_id: int,
    actor: sqlite3.Row,
    *,
    approve: bool,
    comment: str = "",
) -> dict[str, object]:
    row = connection.execute(
        "SELECT * FROM period_reopen_requests WHERE id = ?", (int(request_id),)
    ).fetchone()
    if row is None:
        raise PeriodCloseError("重开申请不存在。")
    if row["status"] != "pending":
        raise PeriodConflictError("该申请已处理。")
    comment = str(comment or "").strip()
    if not approve and len(comment) < 2:
        raise PeriodCloseError("驳回必须填写审批意见。")
    now = utc_now()
    year_month = row["year_month"]
    close_run = connection.execute(
        "SELECT * FROM period_close_runs WHERE id = ?", (row["period_close_id"],)
    ).fetchone()
    with connection:
        if approve:
            window_end = (
                datetime.now(timezone.utc) + timedelta(days=int(row["window_days"]))
            ).date().isoformat()
            connection.execute(
                """
                UPDATE period_reopen_requests
                SET status = 'approved', reviewer_id = ?, reviewer_username = ?,
                    reviewed_at = ?, review_comment = ?
                WHERE id = ?
                """,
                (actor["id"], _actor_name(actor), now, comment, int(request_id)),
            )
            connection.execute(
                """
                UPDATE period_close_runs
                SET status = 'reopened', reopen_window_end = ?, updated_at = ?
                WHERE id = ?
                """,
                (window_end, now, int(row["period_close_id"])),
            )
            connection.execute(
                "DELETE FROM closed_periods WHERE year_month = ?", (year_month,)
            )
            _append_audit(
                connection,
                actor,
                "reopen_approve",
                year_month,
                object_label=f"RO-{int(request_id):06d}",
                reason=comment or row["reason"],
                before={"status": "closed", "report_no": close_run["report_no"] if close_run else None},
                after={"status": "reopened", "window_end": window_end},
                related_id=int(request_id),
                report_no=close_run["report_no"] if close_run else None,
            )
        else:
            connection.execute(
                """
                UPDATE period_reopen_requests
                SET status = 'rejected', reviewer_id = ?, reviewer_username = ?,
                    reviewed_at = ?, review_comment = ?
                WHERE id = ?
                """,
                (actor["id"], _actor_name(actor), now, comment, int(request_id)),
            )
            _append_audit(
                connection,
                actor,
                "reopen_reject",
                year_month,
                object_label=f"RO-{int(request_id):06d}",
                reason=comment,
                related_id=int(request_id),
            )
    return reopen_payload(connection, int(request_id))


def remaining_days(window_end: str | None, *, today: date | None = None) -> int | None:
    if not window_end:
        return None
    today = today or today_shanghai()
    try:
        end = date.fromisoformat(str(window_end)[:10])
    except ValueError:
        return None
    return (end - today).days


def expire_reopen_windows(
    connection: sqlite3.Connection, *, today: date | None = None
) -> list[str]:
    """Auto-restore lock when a reopen window has elapsed with no re-close."""
    today = today or today_shanghai()
    rows = connection.execute(
        """
        SELECT * FROM period_close_runs
        WHERE status = 'reopened' AND reopen_window_end IS NOT NULL
          AND date(reopen_window_end) < date(?)
        """,
        (today.isoformat(),),
    ).fetchall()
    restored: list[str] = []
    for row in rows:
        year_month = row["year_month"]
        now = utc_now()
        with connection:
            connection.execute(
                """
                UPDATE period_close_runs
                SET status = 'closed', reopen_window_end = NULL, updated_at = ?
                WHERE id = ?
                """,
                (now, int(row["id"])),
            )
            connection.execute(
                """
                INSERT INTO closed_periods (year_month, closed_at, closed_by)
                VALUES (?, ?, ?)
                ON CONFLICT(year_month) DO UPDATE SET closed_at = excluded.closed_at
                """,
                (year_month, now, row["closed_by"]),
            )
            _append_audit(
                connection,
                None,
                "reopen_expire",
                year_month,
                object_label=row["report_no"],
                reason="重开窗口到期，自动恢复锁定；原月报继续有效",
                report_no=row["report_no"],
            )
        restored.append(year_month)
    return restored


def close_payload(connection: sqlite3.Connection, year_month: str) -> dict[str, object]:
    year_month = validate_year_month(year_month)
    run = _current_run(connection, year_month)
    evaluation = evaluate_preconditions(connection, year_month)
    snapshot = None
    if run is not None and run["snapshot_json"]:
        snapshot = json.loads(run["snapshot_json"])
    elif evaluation["ready"] or run is None:
        snapshot = build_snapshot(connection, year_month)
    status = run["status"] if run is not None else "pending"
    remaining = remaining_days(run["reopen_window_end"] if run is not None else None)
    pending_reopen = connection.execute(
        """
        SELECT id FROM period_reopen_requests
        WHERE year_month = ? AND status = 'pending'
        ORDER BY id DESC LIMIT 1
        """,
        (year_month,),
    ).fetchone()
    return {
        "year_month": year_month,
        "status": status,
        "ready": evaluation["ready"] and status in ("pending", "failed", "reopened"),
        "checks": evaluation["checks"],
        "blockers": evaluation["blockers"],
        "snapshot": snapshot,
        "snapshot_hash": run["snapshot_hash"] if run is not None else None,
        "report_no": run["report_no"] if run is not None else None,
        "closed_at": run["closed_at"] if run is not None else None,
        "closed_by_username": run["closed_by_username"] if run is not None else None,
        "fail_reason": run["fail_reason"] if run is not None else None,
        "reopen_window_end": run["reopen_window_end"] if run is not None else None,
        "reopen_remaining_days": remaining,
        "pending_reopen_id": int(pending_reopen["id"]) if pending_reopen else None,
        "version": int(run["version"]) if run is not None else 0,
        "locked": is_month_locked(connection, year_month),
    }


def timeline(connection: sqlite3.Connection, *, today: date | None = None) -> list[dict[str, object]]:
    today = today or today_shanghai()
    last = last_closable_month(connection, today=today)
    first = earliest_month(connection) or add_months(last, -5)
    # Always show at least 8 months ending at current calendar month.
    current = f"{today.year:04d}-{today.month:02d}"
    end = current if current >= last else last
    start = add_months(end, -7)
    if first < start:
        start = first
    months: list[dict[str, object]] = []
    month = start
    while month <= end:
        run = _current_run(connection, month)
        status = run["status"] if run is not None else ("open" if month > last else "pending")
        if month == current and status in ("pending", "open"):
            cell = "current"
            label = "进行中"
        elif status == "closed":
            cell = "locked"
            label = "已锁定"
        elif status == "reopened":
            cell = "reopened"
            label = "已重开"
        elif status == "failed":
            cell = "failed"
            label = "结账失败"
        elif month < last and status == "pending":
            cell = "open"
            label = "待结账"
        elif month > last:
            cell = "open"
            label = "归集中"
        else:
            cell = "open"
            label = "待结账"
        months.append(
            {
                "year_month": month,
                "status": status,
                "cell": cell,
                "label": label,
                "locked": is_month_locked(connection, month),
            }
        )
        if month == end:
            break
        month = next_month(month)
    return months


def overview(connection: sqlite3.Connection, *, today: date | None = None) -> dict[str, object]:
    ensure_pending_tasks(connection, today=today)
    expire_reopen_windows(connection, today=today)
    today = today or today_shanghai()
    target = last_closable_month(connection, today=today)
    history = connection.execute(
        """
        SELECT year_month, closed_at, closed_by_username, report_no, status
        FROM period_close_runs
        WHERE status IN ('closed', 'reopened', 'failed')
        ORDER BY closed_at DESC, id DESC
        LIMIT 1
        """
    ).fetchone()
    late = connection.execute(
        "SELECT COUNT(*) AS n FROM period_late_arrivals WHERE status = 'open'"
    ).fetchone()["n"]
    return {
        "today": today.isoformat(),
        "closing_day": closing_day(connection),
        "target_month": target,
        "timeline": timeline(connection, today=today),
        "close": close_payload(connection, target),
        "history": dict(history) if history is not None else None,
        "open_late_arrivals": int(late),
        "lock_toast_template": LOCK_TOAST,
    }


def list_reopen_requests(
    connection: sqlite3.Connection, *, status: str | None = None
) -> list[dict[str, object]]:
    params: list[object] = []
    where = ""
    if status:
        where = "WHERE r.status = ?"
        params.append(status)
    rows = connection.execute(
        f"""
        SELECT r.*, c.report_no, c.snapshot_hash, c.status AS close_status
        FROM period_reopen_requests r
        JOIN period_close_runs c ON c.id = r.period_close_id
        {where}
        ORDER BY r.id DESC
        """,
        params,
    ).fetchall()
    return [reopen_payload(connection, int(row["id"])) for row in rows]


def reopen_payload(connection: sqlite3.Connection, request_id: int) -> dict[str, object]:
    row = connection.execute(
        """
        SELECT r.*, c.report_no, c.snapshot_json, c.snapshot_hash, c.status AS close_status,
               c.closed_at, c.closed_by_username, c.reopen_window_end
        FROM period_reopen_requests r
        JOIN period_close_runs c ON c.id = r.period_close_id
        WHERE r.id = ?
        """,
        (int(request_id),),
    ).fetchone()
    if row is None:
        raise PeriodCloseError("重开申请不存在。")
    before = json.loads(row["before_json"]) if row["before_json"] else None
    after = json.loads(row["after_json"]) if row["after_json"] else None
    if after is None and row["close_status"] == "reopened":
        after = {"status": "reopened", "window_end": row["reopen_window_end"]}
    return {
        "id": int(row["id"]),
        "number": f"RO-{int(row['id']):06d}",
        "year_month": row["year_month"],
        "reason": row["reason"],
        "companies_note": row["companies_note"],
        "window_days": int(row["window_days"]),
        "status": row["status"],
        "requester_username": row["requester_username"],
        "requested_at": row["requested_at"],
        "reviewer_username": row["reviewer_username"],
        "reviewed_at": row["reviewed_at"],
        "review_comment": row["review_comment"],
        "report_no": row["report_no"],
        "close_status": row["close_status"],
        "closed_at": row["closed_at"],
        "closed_by_username": row["closed_by_username"],
        "reopen_window_end": row["reopen_window_end"],
        "reopen_remaining_days": remaining_days(row["reopen_window_end"]),
        "before": before,
        "after": after,
        "diff": _diff_snapshot(before, after if after else {"status": row["close_status"]}),
    }


def _diff_snapshot(before: object, after: object) -> list[dict[str, object]]:
    rows: list[dict[str, object]] = []

    def walk(prefix: str, left: object, right: object) -> None:
        if left == right:
            rows.append({"path": prefix or "状态", "before": left, "after": right, "changed": False})
            return
        if isinstance(left, dict) and isinstance(right, dict):
            keys = sorted(set(left) | set(right))
            for key in keys:
                if key in {"companies", "opening_pairs"}:
                    continue
                walk(f"{prefix}.{key}" if prefix else key, left.get(key), right.get(key))
            return
        rows.append({"path": prefix or "值", "before": left, "after": right, "changed": True})

    if isinstance(before, dict) or isinstance(after, dict):
        walk("", before if isinstance(before, dict) else {}, after if isinstance(after, dict) else {})
    else:
        walk("状态", before, after)
    # Keep a short, UI-friendly subset.
    changed = [item for item in rows if item["changed"]]
    unchanged = [item for item in rows if not item["changed"]]
    return (changed + unchanged)[:12]


def list_audit_events(
    connection: sqlite3.Connection,
    *,
    action: str | None = None,
    year_month: str | None = None,
    company_q: str | None = None,
    since: str | None = None,
    until: str | None = None,
    limit: int = 100,
) -> list[dict[str, object]]:
    clauses: list[str] = []
    params: list[object] = []
    if action:
        clauses.append("action = ?")
        params.append(action)
    if year_month:
        clauses.append("year_month = ?")
        params.append(validate_year_month(year_month))
    if since:
        clauses.append("date(created_at) >= date(?)")
        params.append(since)
    if until:
        clauses.append("date(created_at) <= date(?)")
        params.append(until)
    if company_q:
        clauses.append("(object_label LIKE ? OR reason LIKE ?)")
        like = f"%{company_q}%"
        params.extend([like, like])
    where = f"WHERE {' AND '.join(clauses)}" if clauses else ""
    limit = max(1, min(int(limit), 500))
    rows = connection.execute(
        f"""
        SELECT * FROM period_audit_events
        {where}
        ORDER BY id DESC
        LIMIT ?
        """,
        [*params, limit],
    ).fetchall()
    items = []
    for row in rows:
        items.append(
            {
                "id": int(row["id"]),
                "created_at": row["created_at"],
                "actor_username": row["actor_username"],
                "actor_role": row["actor_role"],
                "action": row["action"],
                "year_month": row["year_month"],
                "object_label": row["object_label"],
                "reason": row["reason"],
                "before": json.loads(row["before_json"]) if row["before_json"] else None,
                "after": json.loads(row["after_json"]) if row["after_json"] else None,
                "report_no": row["report_no"],
            }
        )
    return items


def month_status_label(connection: sqlite3.Connection, year_month: str) -> tuple[str | None, str]:
    run = _current_run(connection, year_month)
    if run is None:
        return None, "—"
    mapping = {
        "closed": ("closed", "已锁定"),
        "reopened": ("reopened", "已重开"),
        "failed": ("failed", "结账失败"),
        "pending": ("pending", "待结账"),
        "closing": ("closing", "处理中"),
    }
    return mapping.get(run["status"], (run["status"], run["status"]))
