"""资产台账健康度自检 — every root cause behind the asset health score.

Why this exists (2026-09-05). Three metrics on the asset domain's health
page were sitting in their worst band, and an operator could see the
score but not the cause:

| metric | production | band |
|---|---|---|
| Day-1 就绪率 | 22.5% | 30 / 100 |
| 长期闲置电脑盈余率 | 21.9% | 20 / 100 |
| 资产财务完整度 | 27.6% | 30 / 100 |

All three trace to the same thing — **the ledger action lags the physical
reality** — but each had to be re-derived from scratch every time
somebody asked. This page is the standing answer: one read, every gap
category with its own count, its own explanation of what it means, and a
link to the surface that closes it.

Design rules this module exists to hold:

* **A cause with no count is not shown as zero, it is shown as
  unmeasured.** Every category here has a real predicate; nothing is a
  placeholder.
* **Numbers agree with the metric they explain.** ``financial`` reuses
  the exact four-field predicate from
  ``app/intelligence/providers/assets.py``; ``idle`` reuses
  ``IdleComputerService``; ``day1`` reuses the onboarding hardware-gap
  service. No second implementation of any of the three.
* **Estimated amounts are counted separately from recorded ones.**
  After the 2026-09 backfill, "financial completeness 90%" would
  otherwise quietly include same-model medians. The page reports
  ``estimated`` alongside ``complete`` so nobody reads a guess as an
  invoice figure.
"""
from __future__ import annotations

import logging
from typing import Any
from uuid import UUID

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.domains.asset.hardware_prep import COMPUTER_CATEGORIES
from app.domains.asset.idle_computers import IdleComputerService
from app.domains.asset.location_admin_sync import (
    get_location_ids_managed_by,
    resolve_asset_location_ids_bulk,
)
from app.domains.asset.models import (
    PURCHASE_COST_SOURCE_ESTIMATED_MODEL,
    PURCHASE_COST_SOURCE_FINANCE_REGISTER,
    AssetItem,
)

logger = logging.getLogger(__name__)

# The financial-completeness predicate, field for field, as
# ``AssetFactProvider`` applies it. Any divergence here would make this
# page contradict the score it is explaining.
_FINANCIAL_FIELDS = (
    "purchase_date",
    "purchase_cost",
    "currency",
    "purchase_entity",
)


class AssetLedgerHealthService:
    """One read that explains every asset-domain health deduction."""

    def __init__(self, db: AsyncSession) -> None:
        self.db = db

    async def build(
        self, *, actor_person_id: UUID, role_names: list[str],
    ) -> dict[str, Any]:
        from app.core.rbac import has_global_asset_scope

        global_tier = has_global_asset_scope(role_names)
        # 🔴 Every section must be about the SAME estate. ``_idle``
        # region-scopes itself (it reuses ``IdleComputerService``), so
        # ``_financial`` and ``_inventory`` counting platform-wide put
        # three numbers about three different populations on one page
        # and labelled the lot 仅本人管理的位置 — a regional admin read
        # the whole platform's financial gap as their own backlog.
        # ``None`` means "no filter" (global tier); an empty set means
        # "administers nothing", which is not the same thing.
        managed: set[UUID] | None = None
        if not global_tier:
            managed = await get_location_ids_managed_by(
                self.db, actor_person_id,
            )
        financial = await self._financial(managed)
        idle = await self._idle(
            actor_person_id=actor_person_id, role_names=role_names,
        )
        inventory = await self._inventory(managed)

        sections = [financial, idle, inventory]
        # 🔴 The Day-1 section is deliberately GLOBAL-ONLY. It is the
        # one section that CANNOT be region-scoped: the hardware-gap
        # service counts hires, and a hire belongs to an office, not to
        # an asset location — there is no honest way to say "the Day-1
        # gaps in your region". Its closing action is ``require_admin``
        # anyway, so it is not a gap a regional admin could act on.
        # Omitted rather than shown-with-a-caveat: a caveat on one card
        # of four is exactly the kind of thing that gets skimmed past.
        if global_tier:
            sections.insert(2, await self._day1())
        return {
            "sections": sections,
            "totals": {
                # The single number a manager asks for: how many distinct
                # things need doing. Deliberately a SUM of actionable
                # counts, not a score — this page explains scores, it
                # does not mint a new one.
                "actionable": sum(s["actionable_count"] for s in sections),
                "section_count": len(sections),
                # Says which estate every number above is about, so the
                # page can label itself instead of implying platform-wide.
                "regional_scope": not global_tier,
            },
        }

    async def _scoped(
        self, rows: list[AssetItem], managed: set[UUID] | None,
    ) -> list[AssetItem]:
        """Keep only the assets this actor's regions cover.

        Same rule as every other asset surface: an asset whose location
        resolves to nothing belongs to no region and is global-tier only.
        """
        if managed is None:
            return rows
        if not managed or not rows:
            return []
        location_ids = await resolve_asset_location_ids_bulk(self.db, rows)
        return [r for r in rows if location_ids.get(r.id) in managed]

    async def _financial(
        self, managed: set[UUID] | None = None,
    ) -> dict[str, Any]:
        """资产财务完整度 — which of the four fields is actually missing.

        Production's gap is NOT spread across the four fields: on
        2026-09-05, ``purchase_date`` was missing on 0 rows and
        ``purchase_entity`` on 3, while ``purchase_cost`` and
        ``currency`` were each missing on 718. Reporting a single
        "incomplete" count hides that, and it is exactly the fact that
        decides the remediation (42 rows need only a currency; 718 need a
        price source).
        """
        # Full ORM rows rather than a column tuple: the region resolver
        # reads ``location`` / ``assigned_to`` off each asset, and the
        # live estate is ~1k rows, so hydration is not the cost here.
        rows = await self._scoped(
            list(
                (
                    await self.db.execute(
                        select(AssetItem).where(
                            AssetItem.deleted_at.is_(None),
                            AssetItem.status != "disposed",
                        )
                    )
                ).scalars().all()
            ),
            managed,
        )
        total = len(rows)
        missing_by_field: dict[str, int] = {f: 0 for f in _FINANCIAL_FIELDS}
        complete = 0
        estimated = 0
        from_register = 0
        cost_only = 0
        currency_only = 0
        incomplete_by_category: dict[str, int] = {}

        for row in rows:
            values = {
                "purchase_date": row.purchase_date,
                "purchase_cost": row.purchase_cost,
                "currency": row.currency,
                "purchase_entity": row.purchase_entity,
            }
            missing = [f for f, v in values.items() if v is None]
            for f in missing:
                missing_by_field[f] += 1
            if not missing:
                complete += 1
                if row.purchase_cost_source == (
                    PURCHASE_COST_SOURCE_ESTIMATED_MODEL
                ):
                    estimated += 1
                elif row.purchase_cost_source == (
                    PURCHASE_COST_SOURCE_FINANCE_REGISTER
                ):
                    from_register += 1
            else:
                incomplete_by_category[row.category] = (
                    incomplete_by_category.get(row.category, 0) + 1
                )
                # The two cheap-to-fix shapes, called out because their
                # remediation is completely different from "find a price".
                if missing == ["currency"]:
                    cost_only += 1
                elif missing == ["purchase_cost"]:
                    currency_only += 1

        incomplete = total - complete
        return {
            "key": "financial",
            "metric_key": "assets.financial_coverage",
            "actionable_count": incomplete,
            "population": total,
            "complete": complete,
            "coverage_pct": (
                round(complete / total * 100, 1) if total else 0.0
            ),
            "detail": {
                "missing_by_field": missing_by_field,
                # Has an amount, lacks only a currency — fixable from the
                # purchase entity with no new data.
                "amount_without_currency": cost_only,
                # Has a currency, lacks the amount.
                "currency_without_amount": currency_only,
                "incomplete_by_category": dict(
                    sorted(
                        incomplete_by_category.items(),
                        key=lambda kv: -kv[1],
                    )
                ),
                # Of the COMPLETE rows, how many are estimates rather
                # than recorded figures. This is the number that stops
                # "90% complete" from quietly meaning "half guessed".
                "estimated_amounts": estimated,
                "finance_register_amounts": from_register,
            },
            "action_route": "/admin/assets/list",
        }

    async def _idle(
        self, *, actor_person_id: UUID, role_names: list[str],
    ) -> dict[str, Any]:
        """长期闲置电脑盈余率 — reuses the list service, never re-derived."""
        payload = await IdleComputerService(self.db).build(
            actor_person_id=actor_person_id, role_names=role_names,
        )
        s = payload["summary"]
        return {
            "key": "idle",
            "metric_key": "assets.idle_rate",
            "actionable_count": s["surplus_total"],
            "population": s["computer_total"],
            "complete": s["computer_total"] - s["long_idle_total"],
            "coverage_pct": s["surplus_rate_pct"],
            "detail": {
                "long_idle_total": s["long_idle_total"],
                "held_for_demand": s["held_for_demand"],
                "uncovered_demand": s["uncovered_demand"],
                "never_moved_in_helix": s["never_moved_in_helix"],
                "never_inventoried": s["never_inventoried"],
                "idle_days_threshold": s["idle_days_threshold"],
                "regional_scope": s["regional_scope"],
            },
            "action_route": "/admin/assets/list?tab=idle",
        }

    async def _day1(self) -> dict[str, Any]:
        """Day-1 就绪率 — the hires whose computer never reached the ledger.

        Reuses the onboarding hardware-gap service rather than counting
        blockers here: that service already owns the window rules
        (completed processes included, backfilled excluded, leavers out).
        """
        from app.domains.onboarding.hardware_gap_service import (
            OnboardingHardwareGapService,
        )

        payload = await OnboardingHardwareGapService(self.db).list_gaps()
        s = payload["summary"]
        return {
            "key": "day1_hardware",
            "metric_key": "lifecycle.day1_ready",
            "actionable_count": s["total"],
            "population": s["total"],
            "complete": 0,
            "coverage_pct": None,
            "detail": {
                "no_device": s["no_device"],
                "reserved_only": s["reserved_only"],
                "overdue_30d_plus": s["overdue_30d_plus"],
            },
            "action_route": "/admin/lifecycle/onboarding?tab=hardware",
        }

    async def _inventory(
        self, managed: set[UUID] | None = None,
    ) -> dict[str, Any]:
        """Computers no stocktake has ever covered.

        Not one of the three scored metrics, but the reason the other two
        cannot be acted on confidently: disposing of a device nobody has
        confirmed exists, or counting it as idle capital, both rest on a
        record no human has ever checked. 84 of production's 146 long-idle
        computers were in this state.
        """
        # Rows, not two COUNT(*)s: the region filter is resolved in
        # Python (an asset's location comes from the managed-location
        # master, not from a column this query could join), and the
        # computer population is a few hundred rows.
        rows = await self._scoped(
            list(
                (
                    await self.db.execute(
                        select(AssetItem).where(
                            AssetItem.deleted_at.is_(None),
                            AssetItem.status != "disposed",
                            AssetItem.category.in_(COMPUTER_CATEGORIES),
                        )
                    )
                ).scalars().all()
            ),
            managed,
        )
        total = len(rows)
        never = sum(1 for r in rows if r.last_inventory_at is None)
        return {
            "key": "never_inventoried",
            "metric_key": None,
            "actionable_count": never,
            "population": total,
            "complete": total - never,
            "coverage_pct": (
                round((total - never) / total * 100, 1) if total else 0.0
            ),
            "detail": {},
            "action_route": "/admin/assets/inventory",
        }