"""Asset demand, inventory, custody, warranty, and finance facts.""" from __future__ import annotations from collections.abc import Callable, Sequence from datetime import UTC, datetime, timedelta from decimal import Decimal from typing import TypeVar from uuid import UUID from zoneinfo import ZoneInfo from sqlalchemy import func, select from sqlalchemy.exc import SQLAlchemyError from sqlalchemy.ext.asyncio import AsyncSession from app.domains.asset.hardware_prep import COMPUTER_CATEGORIES from app.domains.asset.location_admin_sync import ( resolve_asset_location_ids_bulk, resolve_person_location_ids_bulk, ) from app.domains.asset.models import ( ASSET_STATUS_HELD_BY_LEAVER, PURCHASE_COST_SOURCE_ESTIMATED_MODEL, AssetInventoryCycle, AssetInventoryItem, 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 from app.intelligence.privacy import EvidenceObject from app.intelligence.snapshot_service import ( MetricObservation, ProviderResult, ProviderUnavailable, SnapshotSubject, ) _APP_TZ = ZoneInfo("Asia/Shanghai") _EVIDENCE_LIMIT = 50 # Stock statuses an unassigned device can sit in — the same vocabulary # ``hardware_prep`` reserves from, minus ``held_by_leaver`` (that device is # still physically out with a leaver and is scored by ``leaver_hold_rate``). _STOCK_STATUSES = ("idle", "available", "held_by_office") _IDLE_DAYS = 90 # Keep the custody-history lookup off an unbounded IN list. _ID_CHUNK = 500 _RowT = TypeVar("_RowT") def _utc(value: datetime) -> datetime: return value.replace(tzinfo=UTC) if value.tzinfo is None else value.astimezone(UTC) def _pct(value: int, total: int) -> Decimal: return (Decimal(value) * 100 / Decimal(total)).quantize(Decimal("0.01")) class AssetFactProvider: """Collect a global or explicitly location-scoped asset snapshot. ``managed_location_ids=None`` means global scope; an empty set means a regional actor administering no locations, and intentionally returns no assets. """ dimension = "assets" def __init__( self, *, managed_location_ids: frozenset[UUID] | None = None, now: Callable[[], datetime] | None = None, ) -> None: self._managed_location_ids = managed_location_ids self._now = now or (lambda: datetime.now(UTC)) @staticmethod def _validate_subject(subject: SnapshotSubject) -> None: if subject.subject_type == "platform" and subject.subject_id == "platform": return if subject.subject_type == "domain" and subject.subject_id == "assets": return raise ValueError("Asset facts only support platform or assets domain snapshots") async def collect(self, db: AsyncSession, subject: SnapshotSubject) -> ProviderResult: self._validate_subject(subject) try: return await self._collect(db) except SQLAlchemyError as exc: raise ProviderUnavailable("asset datastore unavailable") from exc async def _collect(self, db: AsyncSession) -> ProviderResult: now = _utc(self._now()) today = now.astimezone(_APP_TZ).date() asset_query = select(AssetItem).where( AssetItem.deleted_at.is_(None), AssetItem.status != "disposed", ) if self._managed_location_ids is not None and not self._managed_location_ids: assets = [] else: # No SQL prefilter: ``asset_items.location`` is a name, not # an id, and the mapping from name to location id runs # through the resolver (normalization map + master lookup). # The old prefilter keyed on ``holder_office_id``, which is # set on ~10% of rows, so it never filtered much anyway — # ``_scope_assets`` below does the real work with the same # rules the write gate uses. assets = list((await db.execute(asset_query)).scalars()) assets = await self._scope_assets(db, assets) asset_ids = {row.id for row in assets} asset_as_of = max((_utc(row.updated_at) for row in assets), default=None) demand_rows = list( ( await db.execute( select( OnboardingProcess.id, OnboardingProcess.person_id, OnboardingProcess.updated_at, ) .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 <= today + timedelta(days=30), OnboardingBlocker.resource_type == RES_HARDWARE_ALLOCATION, OnboardingBlocker.resolution == RESO_PENDING, ) ) ).all() ) demand_rows = await self._scope_people_rows(db, demand_rows) demand_rows = sorted(demand_rows, key=lambda row: str(row.id)) reserved_people = { row.reserved_for_person_id for row in assets if row.reserved_for_person_id is not None and row.category in COMPUTER_CATEGORIES } reserved_demands = [row for row in demand_rows if row.person_id in reserved_people] uncovered_demands = [row for row in demand_rows if row.person_id not in reserved_people] stock = [ row for row in assets if row.category in COMPUTER_CATEGORIES and row.status in _STOCK_STATUSES and row.reserved_for_person_id is None ] demand_covered = min(len(demand_rows), len(reserved_demands) + len(stock)) demand_gap = uncovered_demands[len(stock):] demand_as_of = max((_utc(row.updated_at) for row in demand_rows), default=None) cycle = ( await db.execute( select(AssetInventoryCycle) .where(AssetInventoryCycle.status.in_(("active", "closed"))) .order_by( AssetInventoryCycle.started_at.desc(), AssetInventoryCycle.created_at.desc(), ) .limit(1) ) ).scalar_one_or_none() inventory_items: list[AssetInventoryItem] = [] if cycle is not None and asset_ids: inventory_items = list( ( await db.execute( select(AssetInventoryItem).where( AssetInventoryItem.cycle_id == cycle.id, AssetInventoryItem.asset_id.in_(asset_ids), ) ) ).scalars() ) # "Accuracy" is about the records that were actually checked: an item # nobody has answered yet is missing evidence, not a wrong record. # Counting it as an error made a fresh cycle open at its worst score # and — because closing a cycle freezes the remainder at ``pending`` # (see ``AssetInventoryCycle``) — pinned the metric permanently on # every cycle that ever closed with unanswered items. Response rate # is where that belongs: it is confidence, and the platform already # renders 数据可信度 next to the score. confirmed = [row for row in inventory_items if row.status == "confirmed"] mismatched = [row for row in inventory_items if row.status == "issue_reported"] unanswered = [row for row in inventory_items if row.status == "pending"] responded = len(confirmed) + len(mismatched) inventory_gaps = [*mismatched, *unanswered] inventory_as_of = _utc(cycle.updated_at) if cycle and inventory_items else None idle_cutoff = now - timedelta(days=_IDLE_DAYS) idle_since = await self._idle_since(db, assets) in_stock = [ row for row in assets 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] <= idle_cutoff] # Keep future demand ahead of reclaim recommendations: an idle device # that can cover an unresolved Day-1 gap is capacity, not waste. long_idle_computers = [row for row in long_idle if row.category in COMPUTER_CATEGORIES] idle_surplus = max(0, len(long_idle_computers) - len(uncovered_demands)) # The rate is drawn from the population the surplus is drawn from: # only a computer can be reclaimed and re-issued to the next hire. # Peripherals and office-deployed infrastructure (monitors in a stock # room, meeting-room displays, printers, network gear) have no such # circulation, so scoring them as idle capital both overstates waste # and buries the devices someone could actually reclaim. reclaimable = [row for row in assets if row.category in COMPUTER_CATEGORIES] # Oldest-first: the surplus is the tail of the idle queue that # near-term demand does not claim. surplus_rows = sorted( long_idle_computers, key=lambda row: (idle_since[row.id], str(row.id)), )[:idle_surplus] leaver_held = [row for row in assets if row.status == ASSET_STATUS_HELD_BY_LEAVER] warranty_population = [row for row in assets if row.purchase_date is not None] warranty_complete = [row for row in warranty_population if row.warranty_until is not None] financial_complete = [ row for row in assets if row.purchase_date is not None and row.purchase_cost is not None and row.currency is not None and row.purchase_entity is not None ] # Same population, split by provenance. The metric VALUE stays # complete-vs-total (an estimate is a real, usable amount for # depreciation); this count is what keeps the number readable. financial_estimated = [ row for row in financial_complete if getattr(row, "purchase_cost_source", None) == PURCHASE_COST_SOURCE_ESTIMATED_MODEL ] def refs(kind: str, rows: Sequence[object]) -> tuple[EvidenceObject, ...]: return tuple( EvidenceObject(kind, str(getattr(row, "id"))) for row in rows[:_EVIDENCE_LIMIT] ) return ProviderResult( dimension=self.dimension, metrics=( MetricObservation( "assets.demand_coverage", _pct(demand_covered, len(demand_rows)) if demand_rows else 0, confidence=100 if demand_rows else 0, evidence=refs("onboarding", demand_gap), data_as_of=demand_as_of, ), MetricObservation( "assets.inventory_accuracy", _pct(len(confirmed), responded) if responded else 0, confidence=( _pct(responded, len(inventory_items)) if inventory_items else 0 ), evidence=refs("inventory", inventory_gaps), data_as_of=inventory_as_of, ), MetricObservation( "assets.idle_rate", _pct(idle_surplus, len(reclaimable)) if reclaimable else 0, confidence=100 if reclaimable else 0, evidence=refs("asset", surplus_rows), data_as_of=asset_as_of, ), MetricObservation( "assets.leaver_hold_rate", _pct(len(leaver_held), len(assets)) if assets else 0, confidence=100 if assets else 0, evidence=refs("asset", leaver_held), data_as_of=asset_as_of, ), MetricObservation( "assets.financial_coverage", _pct(len(financial_complete), len(assets)) if assets else 0, confidence=100 if assets else 0, evidence=refs( "asset", [row for row in assets if row not in financial_complete], ), data_as_of=asset_as_of, ), ), aggregate_facts={ "asset_count": len(assets), "demand_count": len(demand_rows), "demand_covered_count": demand_covered, "stock_available_count": len(stock), "inventory_item_count": len(inventory_items), "inventory_confirmed_count": len(confirmed), "inventory_responded_count": responded, "inventory_issue_count": len(mismatched), "long_idle_count": len(long_idle), "reclaimable_count": len(reclaimable), "leaver_held_count": len(leaver_held), # Warranty is a fact, not a score: ``warranty_until`` has no # producer (no intake/edit control writes it, and the legacy # import never carried it), so scoring it only ever asserted # the same unmovable 0%. See DECISIONS.md 2026-08-18. "warranty_population_count": len(warranty_population), "warranty_complete_count": len(warranty_complete), "financial_complete_count": len(financial_complete), # Of the complete rows, how many carry an ESTIMATED # amount (same-model median, migration 604) rather than # a recorded or finance-register figure. Reported so a # rising coverage number cannot quietly mean "we guessed # the difference" — the health page renders it next to # the score. "financial_estimated_count": len(financial_estimated), "optimization": { "idle_surplus_count": idle_surplus, "idle_candidate_count": len(long_idle_computers), "future_demand_gap_count": len(demand_gap), "inventory_unconfirmed_count": len(inventory_gaps), "warranty_gap_count": len(warranty_population) - len(warranty_complete), "financial_gap_count": len(assets) - len(financial_complete), "financial_estimated_count": len(financial_estimated), "optimization_version": "asset-optimization-v1", }, "regional_scope": self._managed_location_ids is not None, }, data_as_of=min( [value for value in (asset_as_of, demand_as_of, inventory_as_of) if value is not None], default=None, ), ) async def _idle_since( self, db: AsyncSession, assets: list[AssetItem], ) -> dict[UUID, datetime]: """Resolve when each asset last actually moved. ``asset_items.updated_at`` is a row-mutation stamp, not an idleness clock, and reading it as one inverts the incentive: on 2026-07-14 / 07-15 / 07-21 production bumped 134 / 108 / 87 asset rows whose ``last_inventory_at`` moved the same day — i.e. *answering a stocktake* reset the idle clock on 70 idle computers while only 3 real asset operations happened across the platform that day. Bulk edits do the same in the other direction. The custody events themselves are the honest clock (``asset_operations``: claim / return / recycle_to_office / release_from_office / …). Rows Helix has never seen move fall back to intake, which is conservative — a legacy device imported in April counts as idle since April, not since whenever it truly stopped being used. """ if not assets: return {} latest: dict[UUID, datetime] = {} asset_ids = [row.id for row in assets] for start in range(0, len(asset_ids), _ID_CHUNK): chunk = asset_ids[start:start + _ID_CHUNK] for asset_id, moved_at in ( await 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) return { row.id: latest.get(row.id) or _utc(row.created_at) for row in assets } async def _scope_assets( self, db: AsyncSession, assets: list[AssetItem], ) -> list[AssetItem]: if self._managed_location_ids is None: return assets if not self._managed_location_ids: return [] location_ids = await resolve_asset_location_ids_bulk(db, assets) return [ row for row in assets if location_ids[row.id] in self._managed_location_ids ] async def _scope_people_rows(self, db: AsyncSession, rows: list[_RowT]) -> list[_RowT]: if self._managed_location_ids is None: return rows if not self._managed_location_ids: return [] person_ids = {getattr(row, "person_id") for row in rows} location_ids = await resolve_person_location_ids_bulk(db, person_ids) return [ row for row in rows if location_ids.get(getattr(row, "person_id")) in self._managed_location_ids ]