"""Scheduled handlers for the unified finance spend ledger.

Two jobs:

* ``finance.spend_rollup`` — daily. (Re)materialises ``finance_spend_facts``
  for the current month and the previous month. The current month is
  provisional (sources still landing); re-running the previous month
  daily absorbs late-arriving cloud bills until it is closed. Closed
  months are skipped (frozen) — a recompute that finds drift logs a
  restatement but does not move the published number.

* ``finance.fx_monthly_refresh`` — weekly. Backfills ``finance_fx_monthly``
  for the trailing window from frankfurter.dev (ECB), freezing the rate
  for fully-elapsed months so a closed period's USD total never moves
  with the live rate. Soft-skips a currency/month it can't fetch (leaves
  any existing row in place); the next rollup falls back to the live
  platform rate and flags the fact provisional.

The handler receives an open session and the framework commits after a
clean return — handlers flush via the ORM but must not commit.
"""
from __future__ import annotations

import functools
import logging
from datetime import UTC, datetime
from typing import Any

from sqlalchemy.ext.asyncio import AsyncSession

from app.scheduled_jobs.registry import register_scheduled_job

logger = logging.getLogger(__name__)

# How many trailing months the FX backfill keeps current. 13 covers a
# full year of YoY comparison plus the current month.
_FX_BACKFILL_MONTHS = 13


def _shift_period(period: str, months_back: int) -> str:
    year, month = (int(x) for x in period.split("-"))
    idx = year * 12 + (month - 1) - months_back
    y, m = divmod(idx, 12)
    return f"{y:04d}-{m + 1:02d}"


def _current_period() -> str:
    now = datetime.now(UTC)
    return f"{now.year:04d}-{now.month:02d}"


async def _reconcile_cost_centre_master(hly: Any) -> dict[str, Any]:
    """Run the HRIS→``finance_cost_centers`` reconcile in its OWN fresh,
    committed session, retrying on transient MySQL/TiDB errors.

    Isolated from the caller's handler session for two reasons:

    * the 2026-06-17 failure was errno 1213 (deadlock) on the autoflushed
      ``INSERT INTO finance_cost_centers`` — a fresh-session re-run is the
      canonical InnoDB "restart transaction" recovery, and the reconcile is
      idempotent (upsert per cost-centre code);
    * a deadlock on the shared handler session would leave it in a
      rollback-pending state, breaking the *unrelated* spend-fact
      materialisation that runs after it on the same session. A dedicated
      session means a reconcile failure can never poison the rollup.
    """
    from app.scheduled_jobs.handlers import (
        _run_in_fresh_session_with_lock_retry,
    )

    return await _run_in_fresh_session_with_lock_retry(
        hly.reconcile_master_from_hris,
        label="finance.cost_center_reconcile",
    )


# ─── Spend ledger rollup ────────────────────────────────────────


@register_scheduled_job(
    key="finance.spend_rollup",
    display_name="财务对账 · 统一花费总账物化",
    description=(
        "每日重算当月与上月的 finance_spend_facts(云/软件/AI/采购四线归一 "
        "USD、AI·云去重、部门+成本中心双轴)。当月为 provisional;上月每日重算"
        "以吸收晚到账单,直到关账冻结。已关账月份跳过(发现漂移记 restatement "
        "但不改已发布数字)。"
    ),
)
async def run_finance_spend_rollup(
    db: AsyncSession, config: dict[str, Any],
) -> dict[str, Any]:
    from app.core.coordinators import finance_huilianyi_coordinator as hly
    from app.core.coordinators import finance_spend_coordinator as coord

    # Rebuild the authoritative cost-centre master from the HRIS truth
    # FIRST (cheap, idempotent) so the coverage / orphan view this rollup
    # feeds is computed against an up-to-date master. 汇联易 returns nothing
    # for this app, so without this the master is ~empty and ~all spend
    # shows as orphan (verified on prod). A failure here must not abort the
    # ledger rollup, so it's isolated and reported, not raised.
    cc_master: dict[str, Any]
    try:
        cc_master = await _reconcile_cost_centre_master(hly)
    except Exception:
        logger.exception("finance.spend_rollup: cost-centre master reconcile failed")
        cc_master = {"error": True}

    from app.scheduled_jobs.handlers import (
        _run_in_fresh_session_with_lock_retry,
    )

    cur = _current_period()
    prev = _shift_period(cur, 1)
    out: list[dict] = []
    for period in (prev, cur):
        try:
            # Materialise EACH period in its OWN fresh, committed session
            # with transient-error retry — NOT on the shared handler session.
            # Two reasons, both proven on prod:
            #
            # * Isolation. A 1213 deadlock / 1205 lock-wait on one period's
            #   bulk DELETE+INSERT is rolled back server-side and leaves a
            #   SHARED session in PendingRollbackError, so the OTHER period's
            #   first statement then fails too — reporting BOTH periods failed
            #   when only one hit contention (2026-06-24 18:16: a transient
            #   error on 2026-05 cascaded 2026-06 into a false failure, the run
            #   self-healed on the next hourly tick). Separate sessions make the
            #   two periods genuinely independent, finally making the old "we
            #   still attempt both periods" promise true.
            # * Retry + durability. ``materialize_period`` is idempotent (same
            #   fact_key set per period), so the canonical InnoDB "restart
            #   transaction" response — re-run on a fresh session — recovers a
            #   transient blip in-run, and a period that succeeds is committed
            #   even if its sibling ultimately fails. (The 2026-06-18 fix added
            #   this retry to the cost-centre reconcile but left the per-period
            #   materialise loop on the un-isolated shared session — this closes
            #   that gap.)
            # ``partial`` rather than a lambda: the period must be bound
            # per iteration (a bare closure would hand every attempt the
            # loop's final value), and binding it eagerly says so without
            # the default-argument trick — which additionally left the
            # callable's type uninferable now that the retry helper
            # passes the work's own return type through.
            result = await _run_in_fresh_session_with_lock_retry(
                functools.partial(coord.materialize_period, period=period),
                label=f"finance.spend_rollup[{period}]",
            )
            out.append(result)
        except Exception:
            # A genuine (non-transient, or retry-exhausted) failure of one
            # period must not abort the other — it already ran in its own
            # session. Record it and re-raise AFTER the loop so the run lands
            # failed (the failure_notifier alarms) without discarding the
            # sibling's now-committed ledger.
            logger.exception("finance.spend_rollup: period %s failed", period)
            out.append({"period": period, "error": True})
    if any(r.get("error") for r in out):
        raise RuntimeError(f"finance.spend_rollup: one or more periods failed: {out}")

    # The ledger is already durable — each period committed in its own
    # session above. Commit the (otherwise idle) handler session to close a
    # clean transaction boundary before the best-effort budget-alert step,
    # so an alert failure rolls back ONLY the alert's writes, never the
    # rollup.
    await db.commit()

    # Budget burn-rate alerting on the freshly-materialised actuals. This
    # transaction ONLY enqueues durable outbox rows and advances their budget
    # threshold claims atomically; the claim-first outbox worker performs the
    # real Feishu calls after commit, with no finance row lock held.
    budget_alerts: dict[str, Any]
    try:
        from app.core import fiscal_year as fy
        from app.domains.finance import budget_service
        # The fiscal year that CONTAINS the current period — not its calendar
        # year. With an April fiscal start, Jan–Mar still belong to the prior
        # FY; ``int(cur[:4])`` would evaluate a future, empty FY and silently
        # stop alerting for a quarter every year.
        start_month = await fy.get_fiscal_year_start_month(db)
        year = fy.fy_of_period(cur, start_month)
        budget_alerts = await budget_service.evaluate_and_alert(
            db, fiscal_year=year, as_of_period=cur,
        )
        await db.commit()
    except Exception:
        await db.rollback()
        logger.exception("finance.spend_rollup: budget alert eval failed")
        budget_alerts = {"error": True}

    return {
        "periods": [r["period"] for r in out],
        "cost_center_master": cc_master,
        "budget_alerts": budget_alerts,
        "results": out,
    }


# ─── Monthly FX backfill ────────────────────────────────────────


@register_scheduled_job(
    key="finance.fx_monthly_refresh",
    display_name="财务对账 · 月度历史汇率回填",
    description=(
        "每周从 frankfurter.dev(ECB)回填 finance_fx_monthly 的滚动 13 个月"
        "窗口,为已结束月份冻结当月汇率(关账后 USD 总额不随实时汇率漂移)。"
        "拉取失败的币种/月份软跳过,下次 rollup 退回实时汇率并标 provisional。"
    ),
)
async def run_finance_fx_monthly_refresh(
    db: AsyncSession, config: dict[str, Any],
) -> dict[str, Any]:
    from app.core.coordinators import finance_spend_coordinator as coord
    from app.domains.finance import fx as finance_fx

    cur = _current_period()
    currencies = await coord.discover_native_currencies(db)
    periods = [_shift_period(cur, n) for n in range(_FX_BACKFILL_MONTHS)]

    fetched = 0
    skipped = 0
    for currency in sorted(currencies):
        for period in periods:
            is_past = period < cur
            row = await finance_fx.get_monthly_row(db, currency, period)
            # Already-frozen past month: leave it (authoritative).
            if row is not None and row.frozen and is_past:
                continue
            rate = await finance_fx.fetch_rate_from_frankfurter(currency, period)
            if rate is None:
                skipped += 1
                continue
            await finance_fx.upsert_monthly_rate(
                db, currency, period, rate,
                source="frankfurter",
                # Freeze only fully-elapsed months; the current month is
                # still moving and stays unfrozen.
                frozen=is_past,
                note="auto-backfill (frankfurter.dev / ECB)",
            )
            fetched += 1
    return {
        "currencies": sorted(currencies),
        "periods": periods,
        "rates_written": fetched,
        "skipped": skipped,
    }


# ─── 汇联易 external-actuals sync (Phase 2) ──────────────────────


@register_scheduled_job(
    key="finance.huilianyi_actuals_sync",
    display_name="财务对账 · 汇联易实际账拉取",
    description=(
        "从汇联易开放 API 增量拉取报销/费用单据(按成本中心/部门分摊)+ 成本中心"
        "主数据,落 finance_external_statements / finance_cost_centers,供内部"
        "vs外部对账与口径补全。**默认禁用**:需先在汇联易开放平台给 Helix app "
        "授予读权限,并跑一次 dry-run 校验字段映射后再启用(见 docs/ops)。"
        "无连接器/无读权限时优雅跳过,不崩。"
    ),
)
async def run_finance_huilianyi_actuals_sync(
    db: AsyncSession, config: dict[str, Any],
) -> dict[str, Any]:
    from app.core.coordinators import finance_huilianyi_coordinator as hly

    cur = _current_period()
    prev = _shift_period(cur, 1)
    # Authoritative HRIS-derived master first (always available), then the
    # 汇联易 enrichment overlay (best-effort — degrades to a no-op when the
    # open API grants no read access).
    hris = await _reconcile_cost_centre_master(hly)
    cc = await hly.sync_cost_centers(db)
    out = []
    for period in (prev, cur):
        out.append(await hly.pull_actuals(db, period))
    return {"cost_centers": cc, "cost_center_master_hris": hris,
            "periods": [r.get("period") for r in out], "results": out}


@register_scheduled_job(
    key="finance.netsuite_gl_actuals_sync",
    display_name="财务对账 · NetSuite GL 实际账拉取",
    description=(
        "通过 SuiteQL 从 NetSuite 增量拉取**汇总** GL(期间×主体×科目×部门粒度,"
        "服务端 GROUP BY,绝不拉原始分录)+ 科目/子公司/会计期间主数据,落 "
        "finance_gl_actuals 等,作为对得上记账本的运营子账本。**默认禁用**:"
        "需先确认 NetSuite 角色的 GL 读权限并跑一次 gl-probe 校验列结构后再启用。"
        "已关账期间拉取一次后冻结,当前+上一期间每日幂等覆盖。无连接器/无权限"
        "优雅跳过,不崩。"
    ),
)
async def run_finance_netsuite_gl_actuals_sync(
    db: AsyncSession, config: dict[str, Any],
) -> dict[str, Any]:
    from app.core.coordinators import finance_netsuite_coordinator as ns

    cur = _current_period()
    prev = _shift_period(cur, 1)
    # Masters first (cheap; GL pull resolves period ids from them), then GL
    # for the previous + current month. Closed periods are pulled once then
    # frozen by the coordinator; the open month overwrites in place.
    masters = await ns.sync_masters(db)
    out = []
    for period in (prev, cur):
        out.append(await ns.pull_gl_actuals(db, period))
    return {"masters": masters,
            "periods": [r.get("period") for r in out], "results": out}


# ─── Month-end book-close automation (book-close PR6) ──────────────
#
# Two crons, both seeded DISABLED (migration 358): the controller enables
# them when ready. Roll-forward only mutates close rows; the reminder is a
# pull-first surface with an opt-in push — and it pushes ONLY to close-task
# owners (the finance team), never to general employees/managers, honouring
# the platform's no-disturb rule.

def _china_period() -> str:
    """Current ``YYYY-MM`` in Asia/Shanghai — the boundary the close team works
    to (a UTC ``_current_period`` would roll a month 8h early/late at the edge)."""
    from zoneinfo import ZoneInfo
    now = datetime.now(ZoneInfo("Asia/Shanghai"))
    return f"{now.year:04d}-{now.month:02d}"


def _china_today():
    from zoneinfo import ZoneInfo
    return datetime.now(ZoneInfo("Asia/Shanghai")).date()


@register_scheduled_job(
    key="finance.close_auto_roll_forward",
    display_name="财务关账 · 月初自动滚动开新月",
    description=(
        "月初自动为当前关账期克隆上一期的关账清单(滚月):状态重置、"
        "「常态」任务起始为进行中、按新期锚点重算截止日。幂等——当前期已有"
        "任务则跳过;上一期为空则不动(初始播种仍人工)。中国时区判月界。"
        "默认禁用——控制器就绪后开启。"
    ),
)
async def run_finance_close_auto_roll_forward(
    db: AsyncSession, config: dict[str, Any],
) -> dict[str, Any]:
    from app.domains.finance import close_task_service

    current = _china_period()
    prev = _shift_period(current, 1)
    existing = await close_task_service.list_tasks(db, current)
    if existing:
        return {"period": current, "rolled": False, "reason": "already has tasks"}
    src = await close_task_service.list_tasks(db, prev)
    if not src:
        return {"period": current, "rolled": False, "reason": "no source period to roll from"}
    result = await close_task_service.roll_forward(
        db, from_period=prev, to_period=current,
    )
    await db.commit()
    return {"period": current, "rolled": True, **result}


@register_scheduled_job(
    key="finance.close_deadline_reminder",
    display_name="财务关账 · 截止日提醒",
    description=(
        "为当前关账期内逾期 / 即将到期(默认 2 天内)且未完成的关账任务,"
        "向其负责人发一条汇总提醒。**纯拉取为主**:站内待办收件箱已常态展示;"
        "本提醒为 opt-in 推送,且**仅推送给关账任务负责人(财务团队),绝不"
        "打扰普通员工/经理**。config: within_days(默认 2)、push_feishu"
        "(默认 false=仅站内信)。中国时区判到期。默认禁用。"
    ),
)
async def run_finance_close_deadline_reminder(
    db: AsyncSession, config: dict[str, Any],
) -> dict[str, Any]:
    from datetime import timedelta

    from app.domains.finance import close_service, close_task_service
    from app.domains.finance.close_models import TASK_DONE_STATUSES
    from app.identity.person_models import IdentityPerson
    from app.notifications.service import (
        LocalizedText,
        NotificationEnvelope,
        NotificationService,
    )

    within_days = int(config.get("within_days", 2))
    push_feishu = bool(config.get("push_feishu", False))
    period = config.get("period") or _china_period()

    run = await close_service.get_run(db, period)
    if run is None or run.status == "closed":
        return {"period": period, "reminded": 0, "reason": "no open run"}

    today = _china_today()
    horizon = today + timedelta(days=within_days)
    tasks = await close_task_service.list_tasks(db, period)

    # owner_person_id -> {overdue, soon, earliest}
    by_owner: dict[str, dict[str, Any]] = {}
    for tk in tasks:
        if tk.status in TASK_DONE_STATUSES or not tk.owner_person_id or tk.deadline is None:
            continue
        if tk.deadline > horizon:
            continue
        bucket = by_owner.setdefault(
            tk.owner_person_id, {"overdue": 0, "soon": 0, "earliest": tk.deadline},
        )
        if tk.deadline < today:
            bucket["overdue"] += 1
        else:
            bucket["soon"] += 1
        bucket["earliest"] = min(bucket["earliest"], tk.deadline)

    if not by_owner:
        return {"period": period, "reminded": 0}

    svc = NotificationService(db)
    reminded = 0
    for owner_id, b in by_owner.items():
        person = await db.get(IdentityPerson, owner_id)
        if person is None or not person.work_email:
            continue
        title = f"月结关账任务提醒 · {run.label}"
        body = (
            f"你有 {b['overdue']} 项逾期、{b['soon']} 项即将到期的关账任务"
            f"(最近截止 {b['earliest'].isoformat()})。"
        )
        channels = ["feishu"] if push_feishu else []
        await svc.push_inbox_and_feishu(
            NotificationEnvelope(
                recipient=person,
                title=title,
                body_text=body,
                channels=channels,
                template_key="finance.close_reminder",
                url="/admin/finance/close",
                # Every placeholder the ``finance.close_reminder``
                # template interpolates. ``title`` / ``body_text`` above
                # stay as the pre-template fallback for the (impossible
                # after this PR, but cheap) no-template path; the
                # template + this context is what actually renders, in
                # the owner's own language.
                context={
                    "period": period,
                    "label": run.label,
                    "overdue": b["overdue"],
                    "soon": b["soon"],
                    "earliest": b["earliest"].isoformat(),
                },
                card_subtitle=LocalizedText(
                    zh="仅供本期关账负责人参考",
                    en="For this period's close owner only",
                    ja="今期の決算担当者向けの参考情報です",
                ),
                related_resource_type="finance.close_run",
                related_resource_id=None,
            ),
            feishu=push_feishu,
        )
        reminded += 1
    await db.commit()
    return {"period": period, "reminded": reminded, "within_days": within_days}