"""Long-idle surplus computers — the list behind the health score.

Why this exists (2026-09-05). The 长期闲置电脑盈余率 metric read 21.9%
(20 points, the worst failure-rate band) and the only way to see WHICH
machines it meant was the health page's evidence list, capped at 50 ids
with no location, no age and no way to act. Production: 617 computers,
**146 long-idle**, of which **136 have never had a single asset
operation in Helix** (they came in with the 2026-04 legacy import and
have sat ever since); 76 in 北京, 34 in 美国; purchase years 2021×59,
2022×29, 2023×31, 2024×11, 2025×15; and **84 have never been counted in
any stocktake**.

The metric's own definition is reproduced here exactly, because a list
that disagrees with the score it explains is worse than no list:

* population = computer-category assets (laptop / desktop / computer);
  peripherals and office-deployed infrastructure have no circulation, so
  scoring them as idle capital both overstates waste and buries the
  devices someone could actually reclaim;
* in stock = ``idle`` / ``available`` / ``held_by_office`` **and** not
  earmarked for an incoming hire;
* the idle clock is the last real custody event in ``asset_operations``
  (claim / return / recycle_to_office / …), falling back to intake.
  ⚠️ **Not** ``updated_at``: answering a stocktake bumps that column, so
  reading it as an idleness clock let a stocktake reset the idle timer
  on 70 idle computers in one day (2026-07-14/15/21);
* **near-term demand is subtracted**: an idle machine that can cover an
  unresolved Day-1 hardware gap is capacity, not waste. The surplus is
  the tail of the idle queue that demand does not claim, oldest first.

Region scoping mirrors the asset write gate: a global-tier admin sees
every device, a regional ``asset_admin`` sees only the locations they
administer, and an asset whose location resolves to nothing belongs to
no region (global tier only) — the same rule as
``person_can_manage_asset``.
"""
from __future__ import annotations

import logging
from dataclasses import dataclass
from datetime import UTC, date, datetime, timedelta
from typing import Any
from uuid import UUID

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

from app.domains.asset.hardware_prep import COMPUTER_CATEGORIES
from app.domains.asset.location_admin_sync import (
    get_location_ids_managed_by,
    resolve_asset_location_ids_bulk,
)
from app.domains.asset.models import (
    ASSET_STATUS_HELD_BY_OFFICE,
    AssetItem,
    AssetOperation,
)
from app.domains.onboarding.models import (
    PROC_OPEN_STATUSES,
    RES_HARDWARE_ALLOCATION,
    RESO_PENDING,
    OnboardingBlocker,
    OnboardingProcess,
)
from app.identity.person_models import IdentityPerson

logger = logging.getLogger(__name__)

# Must stay identical to ``app/intelligence/providers/assets.py``: this
# list explains that metric, and a different threshold here would make
# the page contradict the score that sent the operator to it.
IDLE_DAYS = 90
STOCK_STATUSES = ("idle", "available", ASSET_STATUS_HELD_BY_OFFICE)
# Same +30d horizon the asset provider uses for "near-term demand".
DEMAND_HORIZON_DAYS = 30
# Keep the custody-history lookup off an unbounded IN list.
_ID_CHUNK = 500


@dataclass(frozen=True)
class IdleComputerFilters:
    """Server-side filters. All optional; each narrows independently."""

    location: tuple[str, ...] = ()
    purchase_year: tuple[int, ...] = ()
    # ``True`` = only devices no stocktake has ever covered, ``False`` =
    # only devices that have been counted, ``None`` = both.
    never_inventoried: bool | None = None
    min_idle_days: int | None = None


class IdleComputerService:
    """Read-only view of the long-idle surplus, with the same maths."""

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

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

        filters = filters or IdleComputerFilters()
        now = datetime.now(UTC)
        cutoff = now - timedelta(days=IDLE_DAYS)

        computers = list(
            (
                await self.db.execute(
                    select(AssetItem).where(
                        AssetItem.deleted_at.is_(None),
                        AssetItem.status != "disposed",
                        AssetItem.category.in_(COMPUTER_CATEGORIES),
                    )
                )
            ).scalars()
        )
        if not computers:
            return _empty_payload()

        # Region scope BEFORE any counting: a regional admin's numbers
        # must be about their own estate, or the page would send them
        # after devices they cannot touch.
        global_tier = has_global_asset_scope(role_names)
        if not global_tier:
            managed = await get_location_ids_managed_by(
                self.db, actor_person_id,
            )
            if not managed:
                return _empty_payload()
            location_ids = await resolve_asset_location_ids_bulk(
                self.db, computers,
            )
            computers = [
                row for row in computers
                if location_ids.get(row.id) in managed
            ]
            if not computers:
                return _empty_payload()

        idle_since, moved_ids = await self._idle_since(computers)
        in_stock = [
            row for row in computers
            if row.status in STOCK_STATUSES
            and row.reserved_for_person_id is None
        ]
        long_idle = [
            row for row in in_stock if idle_since[row.id] <= cutoff
        ]
        # Oldest-first: the surplus is the tail of the idle queue that
        # near-term demand does not claim.
        long_idle.sort(key=lambda row: (idle_since[row.id], str(row.id)))

        uncovered_demand = await self._uncovered_demand_count()
        surplus_count = max(0, len(long_idle) - uncovered_demand)
        # ``reserved_for_demand`` = the newest devices, held back for the
        # hires who need one. Naming them keeps the page honest: they are
        # long-idle too, just not surplus.
        surplus_rows = long_idle[:surplus_count]
        held_for_demand = long_idle[surplus_count:]

        rows = [
            self._serialize(
                row, idle_since[row.id], now,
                is_surplus=True, ever_moved=row.id in moved_ids,
            )
            for row in surplus_rows
        ] + [
            self._serialize(
                row, idle_since[row.id], now,
                is_surplus=False, ever_moved=row.id in moved_ids,
            )
            for row in held_for_demand
        ]
        filtered = _apply_filters(rows, filters)

        return {
            "rows": filtered,
            "summary": {
                # The metric as the health page states it, so the two
                # surfaces can be compared directly.
                "computer_total": len(computers),
                "long_idle_total": len(long_idle),
                "surplus_total": surplus_count,
                "held_for_demand": len(held_for_demand),
                "uncovered_demand": uncovered_demand,
                "surplus_rate_pct": (
                    round(surplus_count / len(computers) * 100, 1)
                    if computers
                    else 0.0
                ),
                "never_moved_in_helix": sum(
                    1 for r in rows if r["never_moved_in_helix"]
                ),
                "never_inventoried": sum(
                    1 for r in rows if r["never_inventoried"]
                ),
                "idle_days_threshold": IDLE_DAYS,
                "filtered_count": len(filtered),
                "regional_scope": not global_tier,
            },
            "facets": _build_facets(rows),
        }

    async def _idle_since(
        self, computers: list[AssetItem],
    ) -> tuple[dict[UUID, datetime], set[UUID]]:
        """Last real custody event per asset, falling back to intake.

        ``asset_items.updated_at`` is a row-mutation stamp, not an
        idleness clock. Production proved the difference: on three days
        in July 2026, 134 / 108 / 87 asset rows were bumped by stocktake
        answers while only 3 real custody operations happened — reading
        ``updated_at`` would have reset the idle timer on 70 idle
        computers for answering a survey.
        """
        latest: dict[UUID, datetime] = {}
        ids = [row.id for row in computers]
        for start in range(0, len(ids), _ID_CHUNK):
            chunk = ids[start:start + _ID_CHUNK]
            for asset_id, moved_at in (
                await self.db.execute(
                    select(
                        AssetOperation.asset_id,
                        func.max(AssetOperation.created_at),
                    )
                    .where(AssetOperation.asset_id.in_(chunk))
                    .group_by(AssetOperation.asset_id)
                )
            ).all():
                if moved_at is not None:
                    latest[asset_id] = _utc(moved_at)
        # Second return value: which assets Helix has ever seen move.
        # 136 of production's 146 long-idle computers have NO custody
        # event at all, so "never moved in Helix" is the dominant shape
        # of this list and the page has to be able to say so.
        return (
            {
                row.id: latest.get(row.id) or _utc(row.created_at)
                for row in computers
            },
            set(latest),
        )

    async def _uncovered_demand_count(self) -> int:
        """Near-term hires still missing a computer.

        Read the same way the asset health provider reads it: open
        onboarding processes within the +30d horizon carrying a pending
        hardware blocker, minus the hires who already have a device
        earmarked. Subtracting this from the idle queue is what stops the
        page recommending the reclaim of a machine the next hire needs.
        """
        rows = (
            await self.db.execute(
                select(OnboardingProcess.person_id)
                .join(
                    OnboardingBlocker,
                    OnboardingBlocker.process_id == OnboardingProcess.id,
                )
                .join(
                    IdentityPerson,
                    IdentityPerson.id == OnboardingProcess.person_id,
                )
                .where(
                    IdentityPerson.deleted_at.is_(None),
                    OnboardingProcess.status.in_(PROC_OPEN_STATUSES),
                    OnboardingProcess.hire_date
                    <= date.today() + timedelta(days=DEMAND_HORIZON_DAYS),
                    OnboardingBlocker.resource_type == RES_HARDWARE_ALLOCATION,
                    OnboardingBlocker.resolution == RESO_PENDING,
                )
            )
        ).scalars().all()
        demand_people = set(rows)
        if not demand_people:
            return 0
        reserved_people = set(
            (
                await self.db.execute(
                    select(AssetItem.reserved_for_person_id).where(
                        AssetItem.reserved_for_person_id.in_(demand_people),
                        AssetItem.category.in_(COMPUTER_CATEGORIES),
                        AssetItem.deleted_at.is_(None),
                    )
                )
            ).scalars()
        )
        return len(demand_people - reserved_people)

    @staticmethod
    def _serialize(
        row: AssetItem,
        idle_from: datetime,
        now: datetime,
        *,
        is_surplus: bool,
        ever_moved: bool,
    ) -> dict[str, Any]:
        return {
            "id": str(row.id),
            "asset_tag": row.asset_tag,
            "category": row.category,
            "status": row.status,
            "brand": row.brand,
            "model": row.model,
            "serial_number": row.serial_number,
            "location": row.location,
            "purchase_date": (
                row.purchase_date.isoformat() if row.purchase_date else None
            ),
            "purchase_year": (
                row.purchase_date.year if row.purchase_date else None
            ),
            "idle_since": idle_from.isoformat(),
            "idle_days": max(0, (now - idle_from).days),
            # ``True`` means Helix has never recorded a custody event for
            # this device, so the idle clock started at intake. For the
            # 2026-04 legacy import that is the import day, not the day
            # the machine actually stopped being used — conservative, and
            # worth showing so an operator reads the number correctly.
            "never_moved_in_helix": not ever_moved,
            "last_inventory_status": row.last_inventory_status,
            "last_inventory_at": (
                _utc(row.last_inventory_at).isoformat()
                if row.last_inventory_at
                else None
            ),
            "never_inventoried": row.last_inventory_at is None,
            "device_status": row.device_status,
            "is_surplus": is_surplus,
            "notes": row.notes,
        }


def _utc(value: datetime) -> datetime:
    return value.replace(tzinfo=UTC) if value.tzinfo is None else value


def _empty_payload() -> dict[str, Any]:
    return {
        "rows": [],
        "summary": {
            "computer_total": 0,
            "long_idle_total": 0,
            "surplus_total": 0,
            "held_for_demand": 0,
            "uncovered_demand": 0,
            "surplus_rate_pct": 0.0,
            "never_moved_in_helix": 0,
            "never_inventoried": 0,
            "idle_days_threshold": IDLE_DAYS,
            "filtered_count": 0,
            "regional_scope": False,
        },
        "facets": {"locations": [], "purchase_years": []},
    }


def _apply_filters(
    rows: list[dict[str, Any]], filters: IdleComputerFilters,
) -> list[dict[str, Any]]:
    out = rows
    if filters.location:
        wanted = set(filters.location)
        out = [r for r in out if (r["location"] or "") in wanted]
    if filters.purchase_year:
        years = set(filters.purchase_year)
        out = [r for r in out if r["purchase_year"] in years]
    if filters.never_inventoried is not None:
        out = [
            r for r in out
            if r["never_inventoried"] is filters.never_inventoried
        ]
    if filters.min_idle_days is not None:
        out = [r for r in out if r["idle_days"] >= filters.min_idle_days]
    return out


def _build_facets(rows: list[dict[str, Any]]) -> dict[str, Any]:
    """Filter options derived from the UNFILTERED row set.

    Deliberately built before ``_apply_filters`` runs: deriving options
    from already-filtered results makes a chosen filter the only option
    left and traps the operator (a repo-wide UX rule).
    """
    loc_counts: dict[str, int] = {}
    year_counts: dict[int, int] = {}
    for r in rows:
        loc = r["location"] or ""
        loc_counts[loc] = loc_counts.get(loc, 0) + 1
        if r["purchase_year"] is not None:
            year_counts[r["purchase_year"]] = (
                year_counts.get(r["purchase_year"], 0) + 1
            )
    return {
        "locations": [
            {"value": k, "count": v}
            for k, v in sorted(
                loc_counts.items(), key=lambda kv: (-kv[1], kv[0])
            )
        ],
        "purchase_years": [
            {"value": k, "count": v}
            for k, v in sorted(year_counts.items(), reverse=True)
        ],
    }