"""Finance fixed-asset register reconciliation — two tables. Why (2026-09-05). 资产财务完整度 sat at 27.6%. Migration 604 + the same-model-median backfill lift that to roughly 50% and no further: only ~200 of the 692 amount-less rows with a model have any same-model reference at all (299 monitors / 49 network devices / 36 conference displays have none). **The remaining gap can only be closed from the finance fixed-asset register**, matched by serial number. Built as a standing capability, not a one-off script (explicit product decision, 2026-09-05): reconciliation is not a migration event. The register is re-exported every close, Helix keeps acquiring devices, and the interesting output is not just "amounts written" — it is the two disagreement sets: * rows where Helix already records a DIFFERENT amount (a real discrepancy someone should look at, not something to silently overwrite); * register rows matching NO Helix asset (finance is depreciating a device the IT ledger does not know about — the inverse of the completeness gap, and invisible from every existing surface). A script would print those once into a terminal. These tables keep them. **Two tables, preview-then-apply.** ``asset_cost_reconcile_runs`` is one upload; ``asset_cost_reconcile_items`` is one register line with its match verdict, the CURRENT Helix values captured at preview time, and a per-row decision. Nothing is written to ``asset_items`` until an explicit apply, and the item rows are what make the apply auditable afterwards — "what did the register say on the day we accepted it" is not recoverable from the asset row alone. No FK from items to ``asset_items``: an unmatched line has no asset, and a matched asset can later be hard-deleted without invalidating the historical record of what was reconciled. Revision ID: 605 Revises: 604 """ from __future__ import annotations import sqlalchemy as sa from alembic import op from app.core.migration_helpers import id_type_for, table_kw_for revision = "605" down_revision = "604" branch_labels = None depends_on = None T_RUNS = "asset_cost_reconcile_runs" T_ITEMS = "asset_cost_reconcile_items" def _has_table(bind, name: str) -> bool: """Existence guard scoped to the CURRENT database. ``DATABASE()`` matters: without it this matches a same-named table in any schema on the instance, which on a shared test server is a false positive that skips the create and leaves the suite querying nothing. """ return bind.execute( sa.text( "SELECT 1 FROM information_schema.tables " "WHERE table_schema = DATABASE() AND table_name = :n" ), {"n": name}, ).scalar() is not None def upgrade() -> None: bind = op.get_bind() if not _has_table(bind, T_RUNS): op.create_table( T_RUNS, sa.Column("id", id_type_for(), primary_key=True), # Operator-supplied label for WHICH register this was # ("2026-08 固定资产台账"). The file name alone is not # enough to tell two uploads apart a month later. sa.Column("source_label", sa.String(255), nullable=False), sa.Column("file_name", sa.String(255), nullable=True), sa.Column("uploaded_by", id_type_for(), nullable=False), sa.Column( "uploaded_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), ), # previewing → applied | discarded. A run stays previewing # until someone applies it, so an abandoned upload is # visibly abandoned rather than looking like it landed. sa.Column( "status", sa.String(16), nullable=False, server_default="previewing", ), sa.Column("row_count", sa.Integer, nullable=False, server_default="0"), sa.Column("matched_count", sa.Integer, nullable=False, server_default="0"), sa.Column("conflict_count", sa.Integer, nullable=False, server_default="0"), sa.Column("unmatched_count", sa.Integer, nullable=False, server_default="0"), sa.Column("applied_count", sa.Integer, nullable=False, server_default="0"), sa.Column("applied_at", sa.DateTime(timezone=True), nullable=True), sa.Column("applied_by", id_type_for(), nullable=True), sa.Column("notes", sa.Text, nullable=True), sa.Column( "updated_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), onupdate=sa.func.now(), ), **table_kw_for(), ) op.create_index( f"ix_{T_RUNS}_status", T_RUNS, ["status", "uploaded_at"], ) if not _has_table(bind, T_ITEMS): op.create_table( T_ITEMS, sa.Column("id", id_type_for(), primary_key=True), sa.Column("run_id", id_type_for(), nullable=False), # ── what the register said ────────────────────────────── sa.Column("serial_number", sa.String(128), nullable=True), # Normalised (upper + stripped) copy used for matching, kept # so a later audit can see WHY a row matched or did not. sa.Column("serial_key", sa.String(128), nullable=True), sa.Column("asset_tag_hint", sa.String(64), nullable=True), sa.Column("register_amount", sa.Numeric(14, 2), nullable=True), sa.Column("register_currency", sa.String(8), nullable=True), sa.Column("register_purchase_date", sa.Date, nullable=True), sa.Column("register_entity", sa.String(128), nullable=True), sa.Column("register_row_number", sa.Integer, nullable=True), # ── what Helix said AT PREVIEW TIME ───────────────────── # Snapshotted on purpose: the apply decision was made against # these values, and re-deriving them at apply time would hide # a concurrent edit instead of surfacing it. sa.Column("asset_id", id_type_for(), nullable=True), sa.Column("current_amount", sa.Numeric(14, 2), nullable=True), sa.Column("current_currency", sa.String(8), nullable=True), sa.Column("current_cost_source", sa.String(24), nullable=True), # matched | conflict | unmatched | ambiguous | invalid sa.Column("match_status", sa.String(16), nullable=False), # Why a row is not simply matched, in operator-readable form. sa.Column("match_note", sa.String(255), nullable=True), # pending | applied | skipped | superseded sa.Column( "decision", sa.String(16), nullable=False, server_default="pending", ), sa.Column("decided_at", sa.DateTime(timezone=True), nullable=True), sa.Column( "created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), ), **table_kw_for(), ) op.create_index( f"ix_{T_ITEMS}_run", T_ITEMS, ["run_id", "match_status"], ) op.create_index(f"ix_{T_ITEMS}_serial", T_ITEMS, ["serial_key"]) op.create_index(f"ix_{T_ITEMS}_asset", T_ITEMS, ["asset_id"]) def downgrade() -> None: bind = op.get_bind() if _has_table(bind, T_ITEMS): op.drop_table(T_ITEMS) if _has_table(bind, T_RUNS): op.drop_table(T_RUNS)