"""资产台账健康 HTTP endpoints.

Three surfaces that all answer "is the asset ledger telling the truth?",
which is why they share a router (and a mount slot):

* ``GET /assets/idle-computers`` (+ ``/export.xlsx``) — 长期闲置电脑
* ``GET /assets/ledger-health`` — 台账健康度自检
* ``/assets/cost-reconcile*`` — 财务固定资产台账对账 (upload → preview →
  apply, global-admin only)
* ``GET /assets/leaver-held`` — 离职未归还设备对账 (ageing buckets +
  who should chase each one)

Asset-admin read surface (``require_asset_admin``), region-scoped inside
the service the same way the write gate scopes: global tier sees every
device, a regional ``asset_admin`` sees only their own offices.

⚠️ **This router MUST be mounted BEFORE ``asset_router``** — both live
under ``/api/v1/assets`` and ``asset_router``'s ``GET /assets/{asset_id}``
would otherwise swallow these literal path segments as a UUID and answer
422. Same reason ``hardware_prep_router`` is mounted early; see the
comment in ``app/main.py``.

Read-only by design: every action an operator takes from this list —
reserve for an incoming hire, dispose, add to a stocktake — already has
its own audited write endpoint, and duplicating them here would create a
second path with its own region gate to keep in sync.
"""
from __future__ import annotations

from dataclasses import asdict
from datetime import UTC, datetime
from uuid import UUID, uuid4

from fastapi import APIRouter, Body, Depends, File, Query, Request, UploadFile
from fastapi.responses import StreamingResponse
from pydantic import BaseModel, Field
from sqlalchemy.ext.asyncio import AsyncSession

from app.auth.models import AuthUser
from app.core.database import get_db
from app.core.dependencies import require_admin, require_asset_admin
from app.core.exporting import audit_export, build_xlsx
from app.core.response import success_response
from app.domains.asset.bulk_import import (
    read_bulk_rows,
    validate_bulk_upload_contract,
)
from app.domains.asset.cost_reconcile import (
    UPLOAD_TEMPLATE_COLUMNS,
    AssetCostReconcileService,
)
from app.domains.asset.idle_computers import (
    IdleComputerFilters,
    IdleComputerService,
)
from app.domains.asset.leaver_recovery import LeaverAssetRecoveryService
from app.domains.asset.ledger_health import AssetLedgerHealthService

# Prefix is the bare ``/assets`` namespace with LITERAL sub-paths.
# That works only because this router registers before ``asset_router``
# (FastAPI matches in registration order) — the same mechanism
# ``hardware_prep_router`` relies on. Nesting 台账健康度 under
# ``/idle-computers`` just to dodge the collision would have made the URL
# lie about what it returns.
router = APIRouter(
    prefix="/api/v1/assets",
    tags=["assets"],
)

_XLSX_MEDIA = (
    "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"
)

# (machine key, human label) — the machine-key row is what the bulk-update
# template round-trips on, so keys must match the API field names.
_EXPORT_COLUMNS = (
    ("asset_tag", "资产编号"),
    ("category", "类别"),
    ("brand", "品牌"),
    ("model", "型号"),
    ("serial_number", "序列号"),
    ("status", "状态"),
    ("location", "位置"),
    ("idle_days", "闲置天数"),
    ("idle_since", "最后一次流转"),
    ("never_moved_in_helix", "Helix 内从未流转"),
    ("purchase_date", "采购日期"),
    ("purchase_year", "采购年份"),
    ("last_inventory_status", "最近盘点结果"),
    ("last_inventory_at", "最近盘点时间"),
    ("never_inventoried", "从未盘点"),
    ("is_surplus", "属于盈余"),
    ("notes", "备注"),
)


def _filters(
    location: list[str] | None,
    purchase_year: list[int] | None,
    never_inventoried: bool | None,
    min_idle_days: int | None,
) -> IdleComputerFilters:
    return IdleComputerFilters(
        location=tuple(location or ()),
        purchase_year=tuple(purchase_year or ()),
        never_inventoried=never_inventoried,
        min_idle_days=min_idle_days,
    )


@router.get("/idle-computers")
async def list_idle_computers(
    request: Request,
    location: list[str] | None = Query(None),
    purchase_year: list[int] | None = Query(None),
    never_inventoried: bool | None = Query(None),
    min_idle_days: int | None = Query(None, ge=0, le=3650),
    current_user: AuthUser = Depends(require_asset_admin()),
    db: AsyncSession = Depends(get_db),
):
    """Long-idle surplus computers — the list behind the health score.

    Reproduces 长期闲置电脑盈余率 exactly (90-day idle clock read from
    real custody events, computer categories only, near-term Day-1
    demand subtracted) so the page and the score can never disagree.
    Filter options in ``facets`` are built from the UNFILTERED set, so
    choosing one never collapses the rest.
    """
    payload = await IdleComputerService(db).build(
        actor_person_id=current_user.id,
        role_names=list(current_user.role_names or []),
        filters=_filters(
            location, purchase_year, never_inventoried, min_idle_days,
        ),
    )
    return success_response(data=payload, request=request)


@router.get("/idle-computers/export.xlsx")
async def export_idle_computers(
    request: Request,
    location: list[str] | None = Query(None),
    purchase_year: list[int] | None = Query(None),
    never_inventoried: bool | None = Query(None),
    min_idle_days: int | None = Query(None, ge=0, le=3650),
    current_user: AuthUser = Depends(require_asset_admin()),
    db: AsyncSession = Depends(get_db),
):
    """Same rows as the list endpoint, as a workbook.

    Honours the same filters and the same region scope — an export that
    ignored either would hand a regional admin a spreadsheet of devices
    the UI refuses to show them.
    """
    filters = _filters(
        location, purchase_year, never_inventoried, min_idle_days,
    )
    payload = await IdleComputerService(db).build(
        actor_person_id=current_user.id,
        role_names=list(current_user.role_names or []),
        filters=filters,
    )
    # Asset rows carry serials, locations and (via the holder) people —
    # the domain's own ``/assets/export.xlsx`` has audited this egress
    # since it shipped, and a second spreadsheet of the same estate must
    # leave the same trail.
    await audit_export(
        db, request, current_user,
        action="asset.idle_computers.export",
        resource_type="asset",
        row_count=len(payload["rows"]),
        total_matched=payload["summary"]["long_idle_total"],
        # ``dataclasses.asdict`` rather than the dataclass itself:
        # ``audit_export`` takes a Mapping, and a dataclass answers
        # ``.items()`` with an AttributeError at request time.
        filters=asdict(filters),
    )
    stream = build_xlsx(
        columns=_EXPORT_COLUMNS,
        rows=payload["rows"],
        sheet_title="长期闲置电脑",
    )
    stamp = datetime.now(UTC).strftime("%Y%m%d_%H%M%S")
    return StreamingResponse(
        stream,
        media_type=_XLSX_MEDIA,
        headers={
            "Content-Disposition": (
                f"attachment; filename=idle_computers_{stamp}.xlsx"
            ),
        },
    )


@router.get("/ledger-health")
async def get_ledger_health(
    request: Request,
    current_user: AuthUser = Depends(require_asset_admin()),
    db: AsyncSession = Depends(get_db),
):
    """资产台账健康度自检 — every root cause behind the asset health score.

    One read covering financial completeness (broken down by WHICH field
    is missing, because that decides the remediation), long-idle surplus,
    Day-1 hardware gaps, and computers no stocktake has ever covered.
    Each section reuses the service that owns its metric, so this page can
    never disagree with the score it explains.

    """
    payload = await AssetLedgerHealthService(db).build(
        actor_person_id=current_user.id,
        role_names=list(current_user.role_names or []),
    )
    return success_response(data=payload, request=request)


# ─────────────── 财务固定资产台账对账 (2026-09-05) ───────────────
#
# ``require_admin`` (global_admin), NOT ``require_asset_admin``: the
# uploaded file decides which assets get money values written onto them,
# so the region gate cannot be applied row-by-row without parsing the
# file first. Same reasoning as ``POST /assets/bulk-update``, which is
# global-admin-only for exactly this reason.


class ReconcileApplyBody(BaseModel):
    """Which preview rows to write.

    ``item_ids`` omitted means "every applicable row", the normal case
    after reviewing the preview. ``include_conflicts`` is a separate,
    explicit flag: overwriting an amount the ledger already RECORDS is a
    deliberate act and must not ride along with the easy rows.
    """

    # Bounded like its ``skip`` sibling: an unbounded list would build
    # an IN clause from whatever the caller posted, and the biggest
    # legitimate selection is one run's worth of items.
    item_ids: list[UUID] | None = Field(default=None, max_length=2000)
    include_conflicts: bool = False


class ReconcileSkipBody(BaseModel):
    item_ids: list[UUID] = Field(..., min_length=1, max_length=2000)


@router.get("/cost-reconcile/template.xlsx")
async def download_reconcile_template(
    _admin: AuthUser = Depends(require_admin()),
):
    """Upload template for the finance fixed-asset register.

    Columns come from the same constant the parser reads, so the
    download and the accepted headers can never drift apart. The parser
    also accepts the Chinese spellings a real finance export carries —
    an operator should not have to rename headers.
    """
    stream = build_xlsx(
        columns=UPLOAD_TEMPLATE_COLUMNS,
        rows=[
            {
                "serial_number": "C02XK1ABCDEF",
                "purchase_cost": "12999.00",
                "currency": "CNY",
                "purchase_date": "2023-06-01",
                "purchase_entity": "平凯星辰（北京）科技有限公司",
                "asset_tag": "IT-CN-NB-20230601-001",
            }
        ],
        sheet_title="固定资产台账对账模板",
    )
    return StreamingResponse(
        stream,
        media_type=_XLSX_MEDIA,
        headers={
            "Content-Disposition": (
                "attachment; filename=asset_cost_reconcile_template.xlsx"
            ),
        },
    )


@router.get("/cost-reconcile")
async def list_reconcile_runs(
    request: Request,
    limit: int = Query(default=50, ge=1, le=200),
    _admin: AuthUser = Depends(require_admin()),
    db: AsyncSession = Depends(get_db),
):
    """Every register upload, newest first."""
    runs = await AssetCostReconcileService(db).list_runs(limit=limit)
    return success_response(data={"items": runs}, request=request)


@router.post("/cost-reconcile")
async def upload_reconcile_run(
    request: Request,
    file: UploadFile = File(...),
    source_label: str = Query(..., min_length=1, max_length=255),
    admin: AuthUser = Depends(require_admin()),
    db: AsyncSession = Depends(get_db),
):
    """Upload a register, match by serial, return the preview.

    **Writes no asset rows.** The response is the three disagreement
    sets; applying is a separate, explicit call.
    """
    validate_bulk_upload_contract(file)
    rows = await read_bulk_rows(file)
    trace_id = getattr(request.state, "trace_id", str(uuid4()))
    payload = await AssetCostReconcileService(db).create_run(
        rows,
        source_label=source_label,
        file_name=file.filename,
        trace_id=trace_id,
        actor_user_id=admin.id,
    )
    await db.commit()
    return success_response(data=payload, request=request)


@router.get("/cost-reconcile/{run_id}")
async def get_reconcile_run(
    run_id: UUID,
    request: Request,
    match_status: str | None = Query(None, max_length=16),
    limit: int = Query(default=500, ge=1, le=2000),
    offset: int = Query(default=0, ge=0),
    _admin: AuthUser = Depends(require_admin()),
    db: AsyncSession = Depends(get_db),
):
    """One run's rows. ``breakdown`` always covers the WHOLE run, so
    switching the status filter never makes the totals appear to move."""
    payload = await AssetCostReconcileService(db).get_run(
        run_id, match_status=match_status, limit=limit, offset=offset,
    )
    return success_response(data=payload, request=request)


@router.post("/cost-reconcile/{run_id}/apply")
async def apply_reconcile_run(
    run_id: UUID,
    request: Request,
    body: ReconcileApplyBody = Body(default_factory=ReconcileApplyBody),
    admin: AuthUser = Depends(require_admin()),
    db: AsyncSession = Depends(get_db),
):
    """Write the accepted register amounts, tagged ``finance_register``.

    Each row is re-checked against the snapshot taken at preview time and
    REFUSED if the ledger moved in between; those rows come back in
    ``stale`` rather than being silently skipped.
    """
    trace_id = getattr(request.state, "trace_id", str(run_id))
    result = await AssetCostReconcileService(db).apply(
        run_id,
        item_ids=body.item_ids,
        include_conflicts=body.include_conflicts,
        trace_id=trace_id,
        actor_user_id=admin.id,
    )
    await db.commit()
    return success_response(data=result, request=request)


@router.post("/cost-reconcile/{run_id}/skip")
async def skip_reconcile_items(
    run_id: UUID,
    body: ReconcileSkipBody,
    request: Request,
    admin: AuthUser = Depends(require_admin()),
    db: AsyncSession = Depends(get_db),
):
    """Mark rows as deliberately not applied.

    Distinct from leaving them pending: a skipped row is a decision with
    an audit trail, a pending row is unfinished work. Without this the
    queue could never reach a clean state and would stop being read.
    """
    trace_id = getattr(request.state, "trace_id", str(run_id))
    result = await AssetCostReconcileService(db).skip_items(
        run_id,
        item_ids=body.item_ids,
        trace_id=trace_id,
        actor_user_id=admin.id,
    )
    await db.commit()
    return success_response(data=result, request=request)


@router.post("/cost-reconcile/{run_id}/discard")
async def discard_reconcile_run(
    run_id: UUID,
    request: Request,
    admin: AuthUser = Depends(require_admin()),
    db: AsyncSession = Depends(get_db),
):
    """Abandon a preview. Refuses once the run has been applied — that
    record is the only place "what the register said when we accepted
    it" survives."""
    trace_id = getattr(request.state, "trace_id", str(run_id))
    result = await AssetCostReconcileService(db).discard_run(
        run_id, trace_id=trace_id, actor_user_id=admin.id,
    )
    await db.commit()
    return success_response(data=result, request=request)


@router.get("/cost-reconcile/{run_id}/export.xlsx")
async def export_reconcile_run(
    request: Request,
    run_id: UUID,
    match_status: str | None = Query(None, max_length=16),
    admin: AuthUser = Depends(require_admin()),
    db: AsyncSession = Depends(get_db),
):
    """One run's rows as a workbook — the shape finance asks for back."""
    payload = await AssetCostReconcileService(db).get_run(
        run_id, match_status=match_status, limit=2000,
    )
    # Serials plus per-device purchase amounts from both books.
    await audit_export(
        db, request, admin,
        action="asset.cost_reconcile.export",
        resource_type="asset_cost_reconcile_run",
        row_count=len(payload["items"]),
        total_matched=payload["meta"]["total"],
        filters={"run_id": run_id, "match_status": match_status},
    )
    stream = build_xlsx(
        columns=(
            ("register_row_number", "台账行号"),
            ("serial_number", "序列号"),
            ("match_status", "匹配结果"),
            ("match_note", "说明"),
            ("register_amount", "台账金额"),
            ("register_currency", "台账币种"),
            ("current_amount", "Helix 现值"),
            ("current_currency", "Helix 币种"),
            ("current_cost_source", "Helix 金额来源"),
            ("register_purchase_date", "台账购入日期"),
            ("register_entity", "台账法人实体"),
            ("asset_tag_hint", "台账资产编号"),
            ("decision", "处理结果"),
        ),
        rows=payload["items"],
        sheet_title="台账对账明细",
    )
    stamp = datetime.now(UTC).strftime("%Y%m%d_%H%M%S")
    return StreamingResponse(
        stream,
        media_type=_XLSX_MEDIA,
        headers={
            "Content-Disposition": (
                f"attachment; filename=cost_reconcile_{stamp}.xlsx"
            ),
        },
    )


@router.get("/leaver-held")
async def list_leaver_held_assets(
    request: Request,
    bucket: str | None = Query(None, max_length=16),
    current_user: AuthUser = Depends(require_asset_admin()),
    db: AsyncSession = Depends(get_db),
):
    """离职未归还设备 — with a clock and an owner on each row.

    ``held_by_leaver`` had neither: the weekly digest was a count plus
    five sample rows mailed to every asset admin. Rows here carry
    ageing buckets (30 / 60 / 90+ days), the responsible office
    admin(s), and — importantly — WHICH clock produced ``held_since``,
    because a bucket built on a fallback deserves less confidence than
    one built on the HRIS termination date.

    ``bucket`` filters the rows; the summary always covers the whole
    (region-scoped) set so switching buckets never looks like the
    totals moved.
    """
    payload = await LeaverAssetRecoveryService(db).build(
        actor_person_id=current_user.id,
        role_names=list(current_user.role_names or []),
        bucket=bucket,
    )
    return success_response(data=payload, request=request)


@router.get("/leaver-held/export.xlsx")
async def export_leaver_held_assets(
    request: Request,
    bucket: str | None = Query(None, max_length=16),
    current_user: AuthUser = Depends(require_asset_admin()),
    db: AsyncSession = Depends(get_db),
):
    """The same rows as a workbook — the shape a recovery chase needs."""
    payload = await LeaverAssetRecoveryService(db).build(
        actor_person_id=current_user.id,
        role_names=list(current_user.role_names or []),
        bucket=bucket,
    )
    # The most personal export in this router: departed employees by
    # name and employee number, against the device each still holds.
    await audit_export(
        db, request, current_user,
        action="asset.leaver_held.export",
        resource_type="asset",
        row_count=len(payload["rows"]),
        total_matched=payload["summary"]["total"],
        filters={"bucket": bucket},
    )
    stream = build_xlsx(
        columns=(
            ("asset_tag", "资产编号"),
            ("category", "类别"),
            ("brand", "品牌"),
            ("model", "型号"),
            ("serial_number", "序列号"),
            ("location", "位置"),
            ("holder_name", "最后持有人"),
            ("holder_employee_no", "工号"),
            ("held_since", "未归还起始"),
            ("days_held", "未归还天数"),
            ("bucket", "账龄分桶"),
            ("clock_source", "起始日来源"),
            ("purchase_cost", "采购金额"),
            ("currency", "币种"),
            # Provenance rides along with the amount here too — this
            # workbook is what a recovery chase quotes at a leaver.
            ("purchase_cost_source", "金额来源"),
        ),
        rows=payload["rows"],
        sheet_title="离职未归还设备",
    )
    stamp = datetime.now(UTC).strftime("%Y%m%d_%H%M%S")
    return StreamingResponse(
        stream,
        media_type=_XLSX_MEDIA,
        headers={
            "Content-Disposition": (
                f"attachment; filename=leaver_held_{stamp}.xlsx"
            ),
        },
    )