"""Reconcile the finance fixed-asset register against the asset ledger. Why this is a standing capability and not a one-off script (explicit product decision, 2026-09-05). 资产财务完整度 read 27.6%; migration 604 plus the same-model-median backfill lift it to roughly 50% and no further, because only ~200 of the 692 amount-less rows with a model have any same-model reference (299 monitors / 49 network devices / 36 conference displays have none). The rest can only come from finance, by serial number. And that is not a migration event: the register is re-exported every close, Helix keeps acquiring devices, and the valuable output is not only "amounts written" but the two **disagreement sets** a script would print once into a terminal and lose: * ``conflict`` — Helix already RECORDS a different amount. Two books disagree; that is a finding for a human, not something to overwrite. * ``unmatched`` — a register line matching no Helix asset. **Finance is depreciating a device the IT ledger does not know about.** This is the inverse of the completeness gap and is invisible from every other surface in the platform. Flow: upload → preview (nothing written) → the operator accepts rows → apply. The item rows are what make the apply auditable afterwards: "what did the register say on the day we accepted it" cannot be recovered from the asset row alone. Matching rules, all deliberately refusing to guess: * serial only, normalised through the same ``mdm_serial_key`` every other cross-system serial comparison in this codebase uses (one normaliser, or the two sides silently stop matching); * a serial matching more than one live asset is ``ambiguous`` and never applied — picking one would write a money value onto an arbitrary row; * a line with no serial or an unparseable amount is ``invalid`` and reported, not skipped silently; * an amount whose currency the register omits is ``invalid`` too: an amount without a currency is exactly the 42-row defect this whole series exists to fix, and importing more of them would be absurd. """ from __future__ import annotations import logging from datetime import UTC, date, datetime from decimal import Decimal, InvalidOperation from typing import Any from uuid import UUID from sqlalchemy import case, func, select from sqlalchemy.ext.asyncio import AsyncSession from app.audit.service import AuditService from app.core.exceptions import AppException, NotFoundError from app.core.fixed_currency import SUPPORTED_FIXED_CURRENCIES from app.domains.asset.bulk_import import MAX_BULK_ROWS from app.domains.asset.models import ( PURCHASE_COST_SOURCE_FINANCE_REGISTER, PURCHASE_COST_SOURCE_RECORDED, RECONCILE_DECISION_APPLIED, RECONCILE_DECISION_PENDING, RECONCILE_DECISION_SKIPPED, RECONCILE_MATCH_AMBIGUOUS, RECONCILE_MATCH_CONFLICT, RECONCILE_MATCH_INVALID, RECONCILE_MATCH_MATCHED, RECONCILE_MATCH_UNMATCHED, RECONCILE_RUN_APPLIED, RECONCILE_RUN_DISCARDED, RECONCILE_RUN_PREVIEWING, AssetCostReconcileItem, AssetCostReconcileRun, AssetItem, mdm_serial_key, ) logger = logging.getLogger(__name__) # Upload column vocabulary. Machine keys first, then the Chinese labels a # finance export actually carries — an operator should not have to rename # headers to use this. _COLUMN_ALIASES: dict[str, tuple[str, ...]] = { "serial_number": ( "serial_number", "serial", "序列号", "SN", "sn", "机器序列号", ), "amount": ( "purchase_cost", "amount", "cost", "金额", "原值", "资产原值", "购入金额", "采购金额", ), "currency": ("currency", "币种", "货币", "币别"), "purchase_date": ( "purchase_date", "购入日期", "采购日期", "入账日期", "取得日期", ), "entity": ( "purchase_entity", "entity", "法人实体", "采购主体", "所属公司", "公司", ), "asset_tag": ("asset_tag", "资产编号", "资产标签", "固定资产编号"), } # The per-run cap on how many items we will persist. Deliberately the # SAME number the shared bulk-upload reader enforces (``MAX_BULK_ROWS``, # 2 000) rather than a looser local one: the reader runs first, so a # larger constant here would be unreachable and its error message would # quote a limit no operator can ever hit. A company-wide fixed-asset # register can exceed this — splitting it is the documented answer, and # the message below says so with the real number. MAX_ITEMS_PER_RUN = MAX_BULK_ROWS # Display order for the item list: the two disagreement sets a human must # read first, then the rows that merely need a click, then the ones that # need the FILE fixed rather than a decision. _MATCH_STATUS_RANK: tuple[str, ...] = ( RECONCILE_MATCH_CONFLICT, RECONCILE_MATCH_AMBIGUOUS, RECONCILE_MATCH_UNMATCHED, RECONCILE_MATCH_MATCHED, RECONCILE_MATCH_INVALID, ) _MATCH_STATUS_ORDER = case( {status: rank for rank, status in enumerate(_MATCH_STATUS_RANK)}, value=AssetCostReconcileItem.match_status, else_=len(_MATCH_STATUS_RANK), ) def _pick(row: dict[str, Any], field: str) -> Any: """First non-empty value among a field's accepted header spellings.""" for alias in _COLUMN_ALIASES[field]: if alias in row: value = row[alias] if value not in (None, ""): return value # Case-insensitive second pass — finance exports vary on casing. lowered = { str(k).strip().lower(): v for k, v in row.items() if k is not None } for alias in _COLUMN_ALIASES[field]: value = lowered.get(alias.lower()) if value not in (None, ""): return value return None def _parse_amount(value: Any) -> Decimal | None: if value in (None, ""): return None text = str(value).strip() # Finance exports carry thousands separators, currency symbols and # parenthesised negatives. Strip what we understand; anything left # over makes the row invalid rather than a silent zero. for junk in (",", "¥", "$", "￥", "€", "£", " ", " "): text = text.replace(junk, "") negative = text.startswith("(") and text.endswith(")") if negative: text = text[1:-1] try: amount = Decimal(text) except (InvalidOperation, ValueError): return None if negative: amount = -amount return amount.quantize(Decimal("0.01")) def _parse_date(value: Any) -> date | None: if value in (None, ""): return None if isinstance(value, datetime): return value.date() if isinstance(value, date): return value text = str(value).strip()[:10].replace("/", "-") try: return date.fromisoformat(text) except ValueError: return None def _clip(value: Any, size: int) -> str | None: if value in (None, ""): return None return str(value).strip()[:size] or None class AssetCostReconcileService: """Preview and apply a finance-register reconciliation run.""" def __init__(self, db: AsyncSession) -> None: self.db = db self.audit = AuditService(db) # ───────────────────────── preview ───────────────────────── async def create_run( self, rows: list[dict[str, Any]], *, source_label: str, file_name: str | None, trace_id: str, actor_user_id: UUID, ) -> dict[str, Any]: """Parse, match, persist the preview. Writes NO asset rows.""" if not rows: raise AppException( code="RECONCILE_EMPTY_FILE", message="the uploaded register has no data rows", status_code=400, ) if len(rows) > MAX_ITEMS_PER_RUN: raise AppException( code="RECONCILE_TOO_MANY_ROWS", message=( f"the register has {len(rows)} rows; the per-upload " f"cap is {MAX_ITEMS_PER_RUN}. Split the file and " "upload each part as its own run — the runs are " "independent and their results add up." ), status_code=400, ) parsed: list[dict[str, Any]] = [] for index, row in enumerate(rows, start=1): serial = _clip(_pick(row, "serial_number"), 128) amount = _parse_amount(_pick(row, "amount")) currency = _clip(_pick(row, "currency"), 8) parsed.append( { "row_number": index, "serial_number": serial, "serial_key": mdm_serial_key(serial), "amount": amount, "currency": (currency or "").upper() or None, "purchase_date": _parse_date(_pick(row, "purchase_date")), "entity": _clip(_pick(row, "entity"), 128), "asset_tag_hint": _clip(_pick(row, "asset_tag"), 64), "raw_amount": _pick(row, "amount"), } ) # One query for every serial in the file. Soft-deleted and # disposed assets are excluded from matching: writing a cost onto # a disposed row would move a number nobody reads any more, and a # soft-deleted row is invisible to every read surface. keys = {p["serial_key"] for p in parsed if p["serial_key"]} by_key: dict[str, list[AssetItem]] = {} if keys: assets = ( await self.db.execute( select(AssetItem).where( AssetItem.deleted_at.is_(None), AssetItem.status != "disposed", AssetItem.serial_number.isnot(None), # ``mdm_serial_key`` is exactly strip + lower, so # this predicate is its SQL twin — the Python # re-keying below still decides the match, this # only stops the whole estate being hydrated for # a ten-row file. Kept literally equivalent: if # the normaliser ever gains a rule, this must # gain it too or matches vanish silently. func.lower(func.trim(AssetItem.serial_number)).in_( keys ), ) ) ).scalars().all() for asset in assets: key = mdm_serial_key(asset.serial_number) if key and key in keys: by_key.setdefault(key, []).append(asset) run = AssetCostReconcileRun( source_label=source_label, # Clipped: the column is String(255) and the value is the # browser-supplied filename, so a long one would raise 1406 # under strict mode — a 500 for something entirely cosmetic. file_name=_clip(file_name, 255), uploaded_by=actor_user_id, status=RECONCILE_RUN_PREVIEWING, row_count=len(parsed), ) self.db.add(run) await self.db.flush() counts = {"matched": 0, "conflict": 0, "unmatched": 0} for p in parsed: # ``matched``, not ``asset``: the serial-index loop above # already binds ``asset`` in this scope as a non-optional # ``AssetItem``, and reusing the name here makes the optional # match result unassignable (mypy [assignment]). status, note, matched = self._classify(p, by_key) if status in counts: counts[status] += 1 self.db.add( AssetCostReconcileItem( run_id=run.id, serial_number=p["serial_number"], serial_key=p["serial_key"], asset_tag_hint=p["asset_tag_hint"], register_amount=p["amount"], register_currency=p["currency"], register_purchase_date=p["purchase_date"], register_entity=p["entity"], register_row_number=p["row_number"], asset_id=matched.id if matched is not None else None, current_amount=( matched.purchase_cost if matched is not None else None ), current_currency=( matched.currency if matched is not None else None ), current_cost_source=( matched.purchase_cost_source if matched is not None else None ), match_status=status, match_note=note, ) ) run.matched_count = counts["matched"] run.conflict_count = counts["conflict"] run.unmatched_count = counts["unmatched"] await self.db.flush() await self.audit.log( trace_id=trace_id, actor_type="user", actor_user_id=actor_user_id, action="asset.cost_reconcile_previewed", status="success", resource_type="asset_cost_reconcile_run", resource_id=str(run.id), detail={ "source_label": source_label, "file_name": file_name, "row_count": len(parsed), **{f"{k}_count": v for k, v in counts.items()}, }, ) logger.info( "asset.cost_reconcile: run %s previewed — %d rows " "(matched=%d conflict=%d unmatched=%d)", run.id, len(parsed), counts["matched"], counts["conflict"], counts["unmatched"], ) return await self.get_run(run.id) @staticmethod def _classify( parsed: dict[str, Any], by_key: dict[str, list[AssetItem]], ) -> tuple[str, str | None, AssetItem | None]: """Verdict for one register line. Never guesses.""" if not parsed["serial_key"]: return ( RECONCILE_MATCH_INVALID, "no serial number — cannot be matched to an asset", None, ) if parsed["amount"] is None: raw = parsed["raw_amount"] return ( RECONCILE_MATCH_INVALID, ( "amount is missing" if raw in (None, "") else f"amount could not be parsed: {str(raw)[:40]}" ), None, ) if not parsed["currency"]: # An amount with no currency is precisely the defect this # series exists to fix; importing more would be absurd. return ( RECONCILE_MATCH_INVALID, "amount has no currency — supply a 币种 column", None, ) if parsed["currency"] not in SUPPORTED_FIXED_CURRENCIES: return ( RECONCILE_MATCH_INVALID, f"unsupported currency {parsed['currency']}", None, ) candidates = by_key.get(parsed["serial_key"]) or [] if not candidates: return ( RECONCILE_MATCH_UNMATCHED, ( "no live asset carries this serial — finance may be " "depreciating a device the ledger does not have" ), None, ) if len(candidates) > 1: # Currently unreachable in production: ``asset_items`` carries # ``UNIQUE KEY uk_asset_items_serial_number`` under a # case-insensitive collation, so two LIVE assets cannot share # a serial (verified against prod 2026-09-05: 0 duplicate # live serials). Kept fail-closed anyway — that index ignores # soft deletes, and if it is ever dropped or scoped, silently # picking one candidate would write a money value onto an # arbitrary row. Covered by a classifier-level test rather # than a DB fixture the constraint forbids. tags = ", ".join( sorted(c.asset_tag or str(c.id) for c in candidates) )[:200] return ( RECONCILE_MATCH_AMBIGUOUS, f"serial matches {len(candidates)} assets: {tags}", None, ) asset = candidates[0] # A recorded amount that DISAGREES is a discrepancy between two # books, not something to overwrite silently. An estimate, or no # amount at all, is what the register is here to replace. if ( asset.purchase_cost is not None and asset.purchase_cost_source == PURCHASE_COST_SOURCE_RECORDED and ( asset.purchase_cost != parsed["amount"] or (asset.currency or "") != parsed["currency"] ) ): return ( RECONCILE_MATCH_CONFLICT, ( f"ledger records {asset.purchase_cost} " f"{asset.currency or '?'}, register says " f"{parsed['amount']} {parsed['currency']}" ), asset, ) return (RECONCILE_MATCH_MATCHED, None, asset) # ───────────────────────── reads ───────────────────────── async def list_runs(self, *, limit: int = 50) -> list[dict[str, Any]]: rows = ( await self.db.execute( select(AssetCostReconcileRun) # ``id`` as a tiebreaker: ``uploaded_at`` is a # second-granularity server default, so two uploads in # the same second left the list order undefined — and the # page picks ``items[0]`` as "the newest run". .order_by( AssetCostReconcileRun.uploaded_at.desc(), AssetCostReconcileRun.id.desc(), ) .limit(max(1, min(200, limit))) ) ).scalars().all() return [_serialize_run(r) for r in rows] async def get_run( self, run_id: UUID, *, match_status: str | None = None, limit: int = 500, offset: int = 0, ) -> dict[str, Any]: run = await self.db.get(AssetCostReconcileRun, run_id) if run is None: raise NotFoundError("Reconcile run", str(run_id)) clauses = [AssetCostReconcileItem.run_id == run_id] if match_status: clauses.append(AssetCostReconcileItem.match_status == match_status) total = int( ( await self.db.execute( select(func.count()) .select_from(AssetCostReconcileItem) .where(*clauses) ) ).scalar_one() ) items = ( await self.db.execute( select(AssetCostReconcileItem) .where(*clauses) .order_by( # Worst-first, EXPLICITLY. Ordering by the status # column sorts alphabetically, which happens to put # ``conflict`` first but buries ``unmatched`` last — # and "finance is depreciating a device the ledger # does not have" is one of the two findings this # surface exists to show. _MATCH_STATUS_ORDER, AssetCostReconcileItem.register_row_number, ) .limit(max(1, min(2000, limit))) .offset(max(0, offset)) ) ).scalars().all() # Full breakdown regardless of the filter, so switching tabs # never makes the totals appear to change. breakdown = { status: count for status, count in ( await self.db.execute( select( AssetCostReconcileItem.match_status, func.count(), ) .where(AssetCostReconcileItem.run_id == run_id) .group_by(AssetCostReconcileItem.match_status) ) ).all() } return { "run": _serialize_run(run), "items": [_serialize_item(i) for i in items], "meta": {"total": total, "limit": limit, "offset": offset}, "breakdown": breakdown, } # ───────────────────────── apply ───────────────────────── async def apply( self, run_id: UUID, *, item_ids: list[UUID] | None, include_conflicts: bool, trace_id: str, actor_user_id: UUID, ) -> dict[str, Any]: """Write the accepted register amounts onto their assets. ``item_ids=None`` means "every applicable row", which is the common case after reviewing the preview. Conflicts are excluded unless ``include_conflicts`` is set: overwriting a recorded amount is a deliberate act, so it needs a deliberate flag. Every write re-checks the asset against the snapshot taken at preview time and REFUSES the row if the ledger moved in between. Applying a decision made against stale values is how a reconciliation quietly clobbers someone else's correction. """ run = await self.db.get(AssetCostReconcileRun, run_id) if run is None: raise NotFoundError("Reconcile run", str(run_id)) if run.status == RECONCILE_RUN_DISCARDED: raise AppException( code="RECONCILE_RUN_DISCARDED", message="this run was discarded", status_code=409, ) allowed = {RECONCILE_MATCH_MATCHED} if include_conflicts: allowed.add(RECONCILE_MATCH_CONFLICT) clauses = [ AssetCostReconcileItem.run_id == run_id, AssetCostReconcileItem.decision == RECONCILE_DECISION_PENDING, AssetCostReconcileItem.match_status.in_(allowed), ] if item_ids: clauses.append(AssetCostReconcileItem.id.in_(item_ids)) items = ( await self.db.execute( select(AssetCostReconcileItem).where(*clauses) ) ).scalars().all() applied = 0 stale: list[dict[str, Any]] = [] now = datetime.now(UTC) for item in items: if item.asset_id is None or item.register_amount is None: continue asset = ( await self.db.execute( select(AssetItem) .where( AssetItem.id == item.asset_id, AssetItem.deleted_at.is_(None), ) .with_for_update() .execution_options(populate_existing=True) ) ).scalar_one_or_none() if asset is None: stale.append( {"item_id": str(item.id), "reason": "asset_gone"} ) continue # Snapshot check: the decision was made against these values. if ( asset.purchase_cost != item.current_amount or (asset.currency or None) != item.current_currency or asset.purchase_cost_source != item.current_cost_source ): stale.append( { "item_id": str(item.id), "asset_tag": asset.asset_tag, "reason": "changed_since_preview", } ) continue before = { "purchase_cost": ( str(asset.purchase_cost) if asset.purchase_cost is not None else None ), "currency": asset.currency, "purchase_cost_source": asset.purchase_cost_source, "purchase_date": ( asset.purchase_date.isoformat() if asset.purchase_date else None ), } asset.purchase_cost = item.register_amount asset.currency = item.register_currency asset.purchase_cost_source = ( PURCHASE_COST_SOURCE_FINANCE_REGISTER ) # Only FILL a missing purchase date — never move one the # ledger already has. The register's 入账日期 is an accounting # date and can legitimately differ from the acquisition date # the asset row records; overwriting would silently re-date # the whole depreciation schedule. if asset.purchase_date is None and item.register_purchase_date: asset.purchase_date = item.register_purchase_date if not asset.purchase_entity and item.register_entity: asset.purchase_entity = item.register_entity asset.updated_at = now item.decision = RECONCILE_DECISION_APPLIED item.decided_at = now applied += 1 await self.audit.log( trace_id=trace_id, actor_type="user", actor_user_id=actor_user_id, action="asset.cost_reconcile_applied", status="success", resource_type="asset_item", resource_id=str(asset.id), detail={ "run_id": str(run_id), "item_id": str(item.id), "match_status": item.match_status, "before": before, "after": { "purchase_cost": str(item.register_amount), "currency": item.register_currency, "purchase_cost_source": ( PURCHASE_COST_SOURCE_FINANCE_REGISTER ), }, "source_label": run.source_label, }, ) run.applied_count = (run.applied_count or 0) + applied # Only stamp an apply that actually wrote something. An accept # where every row turned out stale changed no asset, and an # ``applied_at`` on it would read as a completed reconciliation # in the run list. if applied: run.applied_at = now run.applied_by = actor_user_id # The run is only "applied" once nothing actionable is left # pending; otherwise it stays previewing so a partial accept does # not read as a finished reconciliation. remaining = int( ( await self.db.execute( select(func.count()) .select_from(AssetCostReconcileItem) .where( AssetCostReconcileItem.run_id == run_id, AssetCostReconcileItem.decision == RECONCILE_DECISION_PENDING, AssetCostReconcileItem.match_status.in_( ( RECONCILE_MATCH_MATCHED, RECONCILE_MATCH_CONFLICT, ) ), ) ) ).scalar_one() ) if remaining == 0: run.status = RECONCILE_RUN_APPLIED await self.db.flush() await self.audit.log( trace_id=trace_id, actor_type="user", actor_user_id=actor_user_id, action="asset.cost_reconcile_run_applied", status="success", resource_type="asset_cost_reconcile_run", resource_id=str(run_id), detail={ "applied": applied, "stale_skipped": len(stale), "include_conflicts": include_conflicts, "remaining_pending": remaining, }, ) logger.info( "asset.cost_reconcile: run %s applied %d row(s), %d stale", run_id, applied, len(stale), ) return { "applied": applied, # Surfaced, never swallowed: a row that could not be applied # because the ledger moved is something the operator must see. "stale": stale, "remaining_pending": remaining, "run": _serialize_run(run), } async def skip_items( self, run_id: UUID, *, item_ids: list[UUID], trace_id: str, actor_user_id: UUID, ) -> dict[str, Any]: """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. """ if not item_ids: raise AppException( code="RECONCILE_NO_ITEMS", message="no items given", status_code=400, ) items = ( await self.db.execute( select(AssetCostReconcileItem).where( AssetCostReconcileItem.run_id == run_id, AssetCostReconcileItem.id.in_(item_ids), AssetCostReconcileItem.decision == RECONCILE_DECISION_PENDING, ) ) ).scalars().all() now = datetime.now(UTC) for item in items: item.decision = RECONCILE_DECISION_SKIPPED item.decided_at = now await self.db.flush() await self.audit.log( trace_id=trace_id, actor_type="user", actor_user_id=actor_user_id, action="asset.cost_reconcile_skipped", status="success", resource_type="asset_cost_reconcile_run", resource_id=str(run_id), detail={"skipped": len(items)}, ) return {"skipped": len(items)} async def discard_run( self, run_id: UUID, *, trace_id: str, actor_user_id: UUID, ) -> dict[str, Any]: run = await self.db.get(AssetCostReconcileRun, run_id) if run is None: raise NotFoundError("Reconcile run", str(run_id)) if run.status == RECONCILE_RUN_APPLIED: raise AppException( code="RECONCILE_ALREADY_APPLIED", message=( "this run has been applied; discarding it would hide " "the record of what was written" ), status_code=409, ) run.status = RECONCILE_RUN_DISCARDED await self.db.flush() await self.audit.log( trace_id=trace_id, actor_type="user", actor_user_id=actor_user_id, action="asset.cost_reconcile_discarded", status="success", resource_type="asset_cost_reconcile_run", resource_id=str(run_id), ) return _serialize_run(run) def _serialize_run(run: AssetCostReconcileRun) -> dict[str, Any]: return { "id": str(run.id), "source_label": run.source_label, "file_name": run.file_name, "status": run.status, "row_count": run.row_count, "matched_count": run.matched_count, "conflict_count": run.conflict_count, "unmatched_count": run.unmatched_count, "applied_count": run.applied_count, "uploaded_by": str(run.uploaded_by), "uploaded_at": run.uploaded_at.isoformat() if run.uploaded_at else None, "applied_at": run.applied_at.isoformat() if run.applied_at else None, "applied_by": str(run.applied_by) if run.applied_by else None, "notes": run.notes, } def _serialize_item(item: AssetCostReconcileItem) -> dict[str, Any]: return { "id": str(item.id), "run_id": str(item.run_id), "register_row_number": item.register_row_number, "serial_number": item.serial_number, "asset_tag_hint": item.asset_tag_hint, "register_amount": ( str(item.register_amount) if item.register_amount is not None else None ), "register_currency": item.register_currency, "register_purchase_date": ( item.register_purchase_date.isoformat() if item.register_purchase_date else None ), "register_entity": item.register_entity, "asset_id": str(item.asset_id) if item.asset_id else None, "current_amount": ( str(item.current_amount) if item.current_amount is not None else None ), "current_currency": item.current_currency, "current_cost_source": item.current_cost_source, "match_status": item.match_status, "match_note": item.match_note, "decision": item.decision, "decided_at": ( item.decided_at.isoformat() if item.decided_at else None ), } # Re-exported for the router's template endpoint so the download and the # parser can never disagree about accepted headers. UPLOAD_TEMPLATE_COLUMNS: tuple[tuple[str, str], ...] = ( ("serial_number", "序列号"), ("purchase_cost", "金额"), ("currency", "币种"), ("purchase_date", "购入日期"), ("purchase_entity", "法人实体"), ("asset_tag", "资产编号（可选，仅作核对）"), ) __all__ = [ "MAX_ITEMS_PER_RUN", "UPLOAD_TEMPLATE_COLUMNS", "AssetCostReconcileService", ]