"""Asset domain scheduled handlers.

Currently one cron:

* ``asset.asset_admin_role_reconcile`` (daily 03:15 Asia/Shanghai) —
  full reconciliation of the ``asset_admin`` role membership against
  the ``asset_location_admins`` join table. Acts as a backstop: the
  primary path is in-process sync via
  :func:`app.domains.asset.location_admin_sync.replace_location_admins`
  which fires on every location-admin mutation. The cron exists in
  case a code path mutates the join table outside the canonical
  service, or a previous in-process sync failed silently. Re-runs
  are cheap (~hundreds of rows max).
"""
from __future__ import annotations

import logging
import uuid
from typing import Any

from sqlalchemy.ext.asyncio import AsyncSession

from app.scheduled_jobs.registry import register_scheduled_job

logger = logging.getLogger(__name__)

# Asia/Shanghai is the zone the asset crons are scheduled in (migration
# 283). Resolving "today" in this zone — not UTC — keeps the monthly
# book-value snapshot anchored to the calendar the trigger fires on.
_ASSET_CRON_TZ = "Asia/Shanghai"


def _local_today():
    """Today's calendar date in the asset crons' scheduling zone."""
    from datetime import datetime
    from zoneinfo import ZoneInfo

    return datetime.now(ZoneInfo(_ASSET_CRON_TZ)).date()


@register_scheduled_job(
    key="asset.asset_admin_role_reconcile",
    display_name="资产管理员角色对账",
    description=(
        "每日 03:15(Asia/Shanghai) 对账 asset_admin 角色:"
        "确保 auth_user_roles 与 asset_location_admins 完全一致。"
        "正常路径在位置管理员增删时实时同步,本 cron 仅作兜底。"
    ),
)
async def run_asset_admin_role_reconcile(
    db: AsyncSession, config: dict[str, Any],
) -> dict[str, Any]:
    from app.domains.asset.location_admin_sync import (
        reconcile_asset_admin_role,
    )

    summary = await reconcile_asset_admin_role(db)
    await db.commit()

    # Per ``feedback_handler_status_must_raise``: a handler that
    # returns a soft-failure dict lands ``runs.status='ok'`` while
    # the failure_notifier never fires. If the ``asset_admin`` row
    # is missing from ``auth_roles`` (someone deleted it, fresh
    # bootstrap skipped the seed, ...) the reconcile is a no-op
    # forever and persons who SHOULD lose the role keep it. Raise
    # so the next-revolution-of-the-wheel-loop fires the dedup'd
    # in-app admin push.
    if summary.get("role_missing"):
        raise RuntimeError(
            "asset.asset_admin_role_reconcile: asset_admin row missing "
            "from auth_roles — reconcile is a no-op until the role is "
            "re-seeded. Restore via the canonical role bootstrap."
        )

    logger.info(
        "asset.asset_admin_role_reconcile: granted=%s revoked=%s expected=%s",
        summary.get("granted"),
        summary.get("revoked"),
        summary.get("total_expected"),
    )
    return summary


@register_scheduled_job(
    key="asset.book_value_snapshot",
    display_name="硬件资产账面价值月度物化",
    description=(
        "每月 1 日 02:00(Asia/Shanghai)物化上一月末活跃硬件资产的账面"
        "价值(按品类、CNY 归一)写入 asset_book_value_snapshots,支撑折旧"
        "趋势图。当前月账面价值仍实时计算;本 cron 只固化历史月,幂等。"
    ),
)
async def run_book_value_snapshot(
    db: AsyncSession, config: dict[str, Any],
) -> dict[str, Any]:
    """Materialise the just-closed month's fleet net book value.

    ``config`` may carry ``period`` (YYYY-MM) to backfill a specific
    month; otherwise the previous calendar month relative to now is
    snapshotted. Idempotent upsert on (period, asset_category).
    """
    from app.domains.asset import financial_service

    period = config.get("period") if isinstance(config, dict) else None
    if not period:
        # The cron fires at 02:00 on the 1st in ``Asia/Shanghai`` (see
        # migration 283). Anchoring ``_prev_period`` on the UTC date is
        # an off-by-one: at 02:00 CST on day 1, UTC is still 18:00 on the
        # LAST day of the previous month, so ``now(UTC).date()`` lands in
        # that previous month and ``_prev_period`` returns the month
        # *before* the just-closed one — permanently skipping one month
        # of the depreciation trend. Resolve "today" in the trigger's
        # zone so day-1 local maps to the just-closed month.
        period = financial_service._prev_period(_local_today())

    summary = await financial_service.snapshot_book_values(db, period)
    # #1798: if USD assets were skipped purely because the CNY/USD rate was
    # unavailable, the just-built snapshot under-reports the whole-fleet CNY
    # book value for the month. Per ``feedback_handler_status_must_raise``,
    # RAISE *before* committing so the incomplete snapshot is NOT persisted
    # (rolled back), the run records 'failed', and the failure-notifier pages
    # IT to restore FX + re-run (idempotent upsert). A partial commit would
    # leave a permanently-underreported month until someone noticed. Note
    # this fires ONLY on the USD-FX outage — genuinely-unsupported
    # currencies (EUR/JPY/…) still snapshot normally and don't fail the cron.
    if summary.get("skipped_usd_no_fx"):
        raise RuntimeError(
            "asset.book_value_snapshot: CNY/USD FX rate unavailable — "
            f"{summary['skipped_usd_no_fx']} USD asset(s) skipped for "
            f"period={summary.get('period')}, so the CNY book value would "
            "be under-reported. Refusing to persist a partial snapshot; "
            "restore the FX rate and re-run (idempotent)."
        )
    await db.commit()
    logger.info(
        "asset.book_value_snapshot: period=%s categories=%s skipped_fx=%s",
        summary.get("period"),
        summary.get("categories_written"),
        summary.get("skipped_fx"),
    )
    return summary


@register_scheduled_job(
    key="asset.stale_reservation_sweep",
    display_name="备机预留僵尸清理",
    description=(
        "每日 04:10(Asia/Shanghai)释放「预留给某个新人、但那个新人已经"
        "没有在办入职单」的备机预留(撤销 offer / 入职单已完成或已取消)。"
        "以前这个判断只在有人打开「待备机」tab 时算一次,平台里没有任何"
        "定时任务看过预留 —— 一个被撤销的 offer 会把一台笔记本无限期地"
        "扣在库外。2026-09-05 之后预留还会满足 Day-1 硬件维度,所以僵尸"
        "预留同时意味着「设备不可用」且「被算成某个人的 Day-1 就绪」。"
        "释放走与人工释放同一条凭证路径(reserve_release),带 "
        "detail.via='stale_reservation_sweep';锁下重新断言,不会抢掉"
        "刚建立的预留。"
    ),
)
async def run_stale_reservation_sweep(
    db: AsyncSession, config: dict[str, Any],
) -> dict[str, Any]:
    from app.domains.asset.hardware_prep import HardwarePrepService

    trace_id = f"system:asset:stale_reservation_sweep:{uuid.uuid4()}"
    summary = await HardwarePrepService(db).sweep_stale_reservations(
        trace_id=trace_id,
    )
    await db.commit()
    logger.info(
        "asset.stale_reservation_sweep: candidates=%s released=%s",
        summary.get("candidates"), summary.get("released"),
    )
    return summary