"""Backfill ``asset_items.holder_kind`` + ``holder_office_id`` for existing rows. Revision ID: 188 Revises: 187 Create Date: 2026-05-19 Renumbered from 185 → 188 alongside the office-held DDL pair (186/187) because PR #403 landed migration 183 first. The "183" references in the rule descriptions below now refer to migration 186 — the body is identical to the pre-renumber 185. Why this exists --------------- Migration 186 (was 183) added ``holder_kind`` (default ``'personal'``) and ``holder_office_id`` (default NULL) — so the DDL is complete, but every existing row is in the safe-but-uninformative state ``personal`` / ``NULL``. This migration walks the table and reclassifies rows that should actually be ``office`` (or ``held_by_leaver``, for rows that slipped past the T+0 handler): Rule 1 — Terminal status (``disposed`` / ``retired`` / ``scrapped`` / ``sold`` / ``lost`` / ``held_by_leaver``): SKIP. The row's lifecycle is over; relabelling its holder_kind would lie. Rule 2 — T+0 force_transfer victims. Detected by ``asset_history.action='force_transfer'`` joined to a leaver ``from_user``. These rows were wrongly attributed to whichever IT admin caught the toss. Reset to ``held_by_office`` and route to the *leaver's* home office (so the city makes sense for inventory routing — not the recipient's city). Rule 3 — ``status IN ('idle','available','in_stock')`` with a ``location`` that resolves to an office: promote to ``holder_kind='office'`` + ``status='held_by_office'``. The inventory routing was already implicitly doing this via the location-string match; now the model is explicit. Rule 4 — ``status IN ('idle','available')`` with no office match: leave as personal / idle. Truly idle stock. Rule 5 — ``status='claimed'`` or ``'assigned'`` with ``assigned_to`` pointing at an *inactive* (``left``/``leaver_pending``) person: promote to ``held_by_leaver`` (the canonical frozen state). These are missed by the T+0 handler — usually because HRIS marked them leaver after the handler ran. Rule 6 — Anything else (``claimed`` / ``borrowed`` / etc. with active holder): leave at the new default ``holder_kind='personal'``. Location → office resolution ---------------------------- Free-text ``asset_items.location`` strings actually seen in prod: Exact city match: 北京, 上海, 杭州, 广州, 成都, 深圳 Country-only: 美国 → Sunnyvale, 日本 → Tokyo, 新加坡 → Singapore, 马来西亚 → Kuala Lumpur Ambiguous: 国内 (could be any China office) — NOT resolved Non-office: IDC机房, remote, NULL — NOT resolved For leaver work_location_name → office resolution we additionally strip trailing ``市`` (e.g. ``北京市`` → ``北京``). Audit trail ----------- Every row touched produces one ``asset_history`` row + one ``asset_operations`` row + one ``audit_logs`` row. The ``audit_logs`` entry uses ``actor_type='system'`` / ``action='asset.holder_kind_backfill'`` so future ops can grep the exact rows changed. Idempotency ----------- The backfill scans only rows that look "untouched": ``holder_kind='personal' AND holder_office_id IS NULL`` — re-running after partial completion is safe. Already-relabelled rows are skipped. Memory pressure --------------- At ~1100 candidate rows × ~3 INSERT/row = ~3300 row writes total, this migration is well under TiDB's 1 GiB memtable cap and runs cleanly in a single transaction (the cap that bit migration 173 was on a 10M+-row FK backfill with 16 KiB JSON payloads per row — a different scale). Earlier drafts of this migration used explicit batch ``COMMIT`` / ``BEGIN`` to chunk the work; we removed both (a) because Alembic owns the outer transaction and raw ``COMMIT`` corrupts its bookkeeping, and (b) because OFFSET-based pagination silently skips rows once the WHERE filter shrinks (UPDATEd rows fall out of the predicate), losing exactly the rows the migration is supposed to fix. The replacement uses keyset pagination (``WHERE ... AND id > :last_id``) so each iteration sees a strictly-monotone window with no skip risk — even though everything runs in one transaction, the keyset cursor keeps memory bounded per fetch. Downgrade is a no-op: we never want to un-relabel a held_by_office / held_by_leaver row by going back through the migration chain. """ from __future__ import annotations import json import uuid from datetime import UTC, datetime import sqlalchemy as sa from alembic import op revision = "188" down_revision = "187" branch_labels = None depends_on = None BATCH_SIZE = 500 TERMINAL_STATUSES = { "disposed", "retired", "scrapped", "sold", "lost", "held_by_leaver", } IDLE_STATUSES = {"idle", "available", "in_stock"} CLAIMED_STATUSES = {"claimed", "assigned"} def _normalise_location(s: str | None) -> str | None: if not s: return None s = s.strip() if not s: return None # Drop trailing 市 / 省 / 县 (HRIS work_location_name carries these, # asset.location occasionally too). for suffix in ("市", "省", "县"): if s.endswith(suffix): s = s[:-1] break return s def _build_office_lookup(bind) -> dict[str, str]: """Map normalised location string → office_id. Handles direct city match, country_or_region single-office shortcut, and a small manual alias table for free-text quirks seen in prod (``美国`` is country_or_region NOT city for the US office; etc.). """ rows = bind.execute(sa.text( "SELECT id, city, country_or_region FROM asset_office_addresses " "WHERE is_active = 1" )).fetchall() # Group offices by country to detect single-office countries (for # country-name fallback resolution). by_country: dict[str, list[tuple[str, str]]] = {} out: dict[str, str] = {} for office_id, city, country in rows: office_id_str = str(office_id) if city: out[city.strip()] = office_id_str if country: by_country.setdefault(country.strip(), []).append( (office_id_str, (city or "").strip()) ) # Country-name → office (only when country has exactly ONE office). for country, offices in by_country.items(): if len(offices) == 1 and country not in out: out[country] = offices[0][0] # Manual aliases for prod free-text quirks. Keep tight — every entry # is a real value seen in ``SELECT DISTINCT location FROM asset_items`` # on prod 2026-05-19. Don't add speculative aliases. aliases = { # English country / region aliases. "USA": out.get("美国"), "US": out.get("美国"), "United States": out.get("美国"), "Japan": out.get("日本"), "Malaysia": out.get("马来西亚"), # City-name variants. "Singapore": out.get("新加坡") or out.get("Singapore"), "Sunnyvale": out.get("美国") or out.get("Sunnyvale"), "Kuala Lumpur": out.get("马来西亚") or out.get("Kuala Lumpur"), "Tokyo": out.get("日本") or out.get("Tokyo"), } for k, v in aliases.items(): if v and k not in out: out[k] = v return out def _resolve_office( location: str | None, lookup: dict[str, str] ) -> str | None: norm = _normalise_location(location) if not norm: return None return lookup.get(norm) def _new_operation_number(now: datetime) -> str: suffix = uuid.uuid4().hex[:4].upper() return f"OP-{now.strftime('%Y%m%d%H%M%S%f')}-{suffix}" def _new_uuid() -> str: return str(uuid.uuid4()) def _write_history( bind, *, asset_id: str, action: str, from_user: str | None, to_user: str | None, operator_id: str | None, notes: str, now: datetime, ) -> None: bind.execute(sa.text( "INSERT INTO asset_history " "(id, asset_id, action, from_user, to_user, operator_id, notes, created_at) " "VALUES (:id, :aid, :action, :from_u, :to_u, :op, :notes, :ts)" ), { "id": _new_uuid(), "aid": asset_id, "action": action, "from_u": from_user, "to_u": to_user, "op": operator_id, "notes": notes, "ts": now, }) def _write_operation( bind, *, asset_id: str, operation_type: str, from_user: str | None, operator_id: str | None, notes: str, detail: dict, now: datetime, ) -> None: # Hotfix (2026-05-20): the production ``asset_operations.operator_id`` # column is NOT NULL + a real FK to ``identity_persons.id`` — the # original PR #406 backfill passed ``operator_id=None`` for every # one of the 6 backfill rules (system-initiated, no human actor) # and the migration crashed on the first attempted INSERT with # ``(1048, "Column 'operator_id' cannot be null")``, blocking the # whole chain at revision 187 in production. # # The companion ``asset_history`` row that ``_write_history`` # produces IS the canonical audit trail for asset ownership # changes (see the AssetHistory model docstring); ``asset_operations`` # mirrors it for the operations console listing but is not the # source of truth. Skipping the asset_operations write here keeps # the backfill's audit semantics intact (asset_history captures # every transition with ``operator_id IS NULL`` allowed for system # writes) while letting the migration progress. Operators viewing # the asset history page will see the transition; the operations # console will simply not show a "migration backfill" row, which # is the right semantic — these aren't admin-initiated operations, # they're a data fix-up. if operator_id is None: return bind.execute(sa.text( "INSERT INTO asset_operations " "(id, operation_number, asset_id, operation_type, status, " " from_user_id, to_user_id, operator_id, notes, detail, " " created_at, confirmed_at, confirmed_by) " "VALUES (:id, :num, :aid, :otype, 'confirmed', " " :from_u, NULL, :op, :notes, :detail, " " :ts, :ts, :op)" ), { "id": _new_uuid(), "num": _new_operation_number(now), "aid": asset_id, "otype": operation_type, "from_u": from_user, "op": operator_id, "notes": notes, "detail": json.dumps(detail, ensure_ascii=False), "ts": now, }) def _write_audit( bind, *, action: str, resource_id: str, payload: dict, now: datetime, ) -> None: """Write to ``audit_logs`` if the table exists with the expected shape. The audit_logs schema has evolved a few times — we guard against column drift by looking up the columns at runtime and only sending the fields that exist. Hotfix (2026-05-20): the original shape produced only ``{id, action, resource_type, resource_id, actor_type, payload, created_at}``. Production's ``audit_logs`` table requires ``trace_id`` and ``status`` as NOT NULL (no default), and stores the structured body under ``detail`` rather than ``payload``. The runtime column-drift filter silently dropped both NOT NULL fields AND the body column, so every audit INSERT crashed with ``(1364, "Field 'trace_id' doesn't have a default value")`` after the first asset_history row succeeded. Migration 188 dead- lettered the chain at revision 187 in production. Fix: pre-populate every NOT NULL column the live AuditService writes with sensible system-write defaults — ``trace_id`` gets a sentinel that names this exact backfill so forensics can search for it, ``status`` is ``'success'`` (the backfill writes are after-the-fact records of completed in-row state flips), and the structured body lands under BOTH ``payload`` and ``detail`` so the runtime filter keeps whichever column the live schema has. """ cols = {row[0] for row in bind.execute(sa.text( "SELECT COLUMN_NAME FROM information_schema.COLUMNS " "WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'audit_logs'" )).fetchall()} if not cols: return # Hotfix #3 (2026-05-20): the previous shape used a sentinel # string ``"migration:188:backfill_asset_holder_kind"`` (39 chars) # but prod's ``audit_logs.trace_id`` is ``CHAR(36)`` (UUID-shaped). # Insert failed with ``(1406, "Data too long for column 'trace_id'")``. # Use a fresh UUID per row + stash the human-readable sentinel # inside the JSON body so forensics can find every backfill row # via ``WHERE JSON_EXTRACT(detail, '$.migration') = '188_...'`` # regardless of the trace_id randomization. body = dict(payload) body.setdefault("migration", "188_backfill_asset_holder_kind") body_json = json.dumps(body, ensure_ascii=False) fields: dict[str, object] = { "id": _new_uuid(), "trace_id": _new_uuid(), "action": action, "resource_type": "asset_item", "resource_id": resource_id, "actor_type": "system", "status": "success", # Send the body under both legacy + current names; the column # filter below drops whichever the live schema doesn't have. "payload": body_json, "detail": body_json, "created_at": now, } fields = {k: v for k, v in fields.items() if k in cols} placeholders = ", ".join(f":{k}" for k in fields) columns = ", ".join(fields) bind.execute( sa.text(f"INSERT INTO audit_logs ({columns}) VALUES ({placeholders})"), fields, ) def upgrade() -> None: bind = op.get_bind() office_lookup = _build_office_lookup(bind) # Pull active (non-terminal) candidate rows in batches. counts = {"rule2_force_transfer": 0, "rule3_idle_office": 0, "rule5_leaver": 0, "rule_skipped": 0} # Keyset pagination — we cannot use OFFSET here because the WHERE # filter shrinks every iteration (UPDATEd rows leave # `holder_kind='personal' AND holder_office_id IS NULL`) so OFFSET # silently skips unprocessed rows that shifted into positions we # already advanced past. Tracking ``last_id`` keeps the cursor # strictly monotone and immune to the filter shrinking under us. last_id: str | None = None while True: params: dict[str, object] = { "terminal": list(TERMINAL_STATUSES), "limit": BATCH_SIZE, } if last_id is None: cursor_clause = "" else: cursor_clause = " AND id > :last_id " params["last_id"] = last_id # Filter keeps the migration idempotent on re-run (already- # promoted rows have holder_kind='office' or status='held_by_*' # and are skipped). Combined with keyset pagination this is safe # under WHERE-filter shrinkage during the same run: rows that # flip out of the filter mid-loop simply don't reappear in a # later fetch, but ``id > last_id`` guarantees we never revisit # them either (their id is already <= last_id). rows = bind.execute(sa.text( "SELECT id, status, assigned_to, location, holder_kind, " " holder_office_id " "FROM asset_items " "WHERE deleted_at IS NULL " " AND status NOT IN :terminal " " AND holder_kind = 'personal' " " AND holder_office_id IS NULL " f"{cursor_clause}" "ORDER BY id LIMIT :limit" ).bindparams( sa.bindparam("terminal", expanding=True), ), params).fetchall() if not rows: break # Advance the cursor to the largest id we just fetched, regardless # of whether each row was rule-2/3/5 promoted or rule-4/6 skipped. # This guarantees forward progress even when every row in the # batch is a Rule 4/6 no-op (otherwise we'd loop forever on the # same window). last_id = str(rows[-1][0]) for asset_id, status, assigned_to, location, _hk, _hoid in rows: asset_id_str = str(asset_id) assigned_str = str(assigned_to) if assigned_to else None now = datetime.now(UTC) # Rule 2: T+0 force_transfer victim? force_row = bind.execute(sa.text( "SELECT h.from_user, p.work_location_name " "FROM asset_history h " "JOIN identity_persons p ON p.id = h.from_user " "WHERE h.asset_id = :aid " " AND h.action = 'force_transfer' " " AND p.employment_status IN ('left', 'leaver_pending') " "ORDER BY h.created_at DESC LIMIT 1" ), {"aid": asset_id_str}).fetchone() if force_row: leaver_id, leaver_loc = force_row office_id = _resolve_office(leaver_loc, office_lookup) if office_id: bind.execute(sa.text( "UPDATE asset_items " "SET holder_kind = 'office', " " holder_office_id = :oid, " " status = 'held_by_office', " " assigned_to = NULL, " " assigned_at = NULL, " " updated_at = :ts " "WHERE id = :aid" ), {"oid": office_id, "ts": now, "aid": asset_id_str}) notes = ( f"Backfill (migration 185): 撤销 T+0 force_transfer, " f"回收到 leaver ({leaver_id}) 的归属办公室 ({office_id})。" ) _write_history( bind, asset_id=asset_id_str, action="recycle_to_office", from_user=assigned_str, to_user=None, operator_id=None, notes=notes, now=now, ) _write_operation( bind, asset_id=asset_id_str, operation_type="recycle_to_office", from_user=assigned_str, operator_id=None, notes=notes, detail={ "source": "migration_185_backfill", "rule": "rule2_force_transfer", "leaver_person_id": str(leaver_id), "office_id": office_id, "prior_status": status, "prior_assigned_to": assigned_str, }, now=now, ) _write_audit( bind, action="asset.holder_kind_backfill", resource_id=asset_id_str, payload={ "rule": "rule2_force_transfer", "prior_status": status, "prior_assigned_to": assigned_str, "new_office_id": office_id, }, now=now, ) counts["rule2_force_transfer"] += 1 continue # Office can't be resolved — fall through to rule 5 if # still assigned to a leaver, otherwise skip. # Rule 5: claimed/assigned to a left person (T+0 handler # missed it; HRIS marked leaver after the cron ran). if status in CLAIMED_STATUSES and assigned_str: leaver_check = bind.execute(sa.text( "SELECT employment_status FROM identity_persons " "WHERE id = :pid" ), {"pid": assigned_str}).fetchone() if leaver_check and leaver_check[0] in ( "left", "leaver_pending" ): bind.execute(sa.text( "UPDATE asset_items " "SET status = 'held_by_leaver', updated_at = :ts " "WHERE id = :aid" ), {"ts": now, "aid": asset_id_str}) notes = ( f"Backfill (migration 185): 持有人 ({assigned_str}) " f"已离职但 T+0 handler 未触发 — 标记 held_by_leaver。" ) _write_history( bind, asset_id=asset_id_str, action="held_by_leaver", from_user=assigned_str, to_user=None, operator_id=None, notes=notes, now=now, ) _write_operation( bind, asset_id=asset_id_str, operation_type="mark_held_by_leaver", from_user=assigned_str, operator_id=None, notes=notes, detail={ "source": "migration_185_backfill", "rule": "rule5_missed_leaver", "prior_status": status, }, now=now, ) _write_audit( bind, action="asset.holder_kind_backfill", resource_id=asset_id_str, payload={ "rule": "rule5_missed_leaver", "prior_status": status, }, now=now, ) counts["rule5_leaver"] += 1 continue # Rule 3: idle row with office-matching location. if status in IDLE_STATUSES and not assigned_str: office_id = _resolve_office(location, office_lookup) if office_id: bind.execute(sa.text( "UPDATE asset_items " "SET holder_kind = 'office', " " holder_office_id = :oid, " " status = 'held_by_office', " " updated_at = :ts " "WHERE id = :aid" ), {"oid": office_id, "ts": now, "aid": asset_id_str}) notes = ( f"Backfill (migration 185): idle + location='{location}' " f"→ held_by_office, office={office_id}。" ) _write_history( bind, asset_id=asset_id_str, action="recycle_to_office", from_user=None, to_user=None, operator_id=None, notes=notes, now=now, ) _write_operation( bind, asset_id=asset_id_str, operation_type="recycle_to_office", from_user=None, operator_id=None, notes=notes, detail={ "source": "migration_185_backfill", "rule": "rule3_idle_office", "prior_status": status, "prior_location": location, "office_id": office_id, }, now=now, ) _write_audit( bind, action="asset.holder_kind_backfill", resource_id=asset_id_str, payload={ "rule": "rule3_idle_office", "prior_status": status, "new_office_id": office_id, }, now=now, ) counts["rule3_idle_office"] += 1 continue # Rule 4 / 6: leave as personal. No update needed (defaults # already match). counts["rule_skipped"] += 1 # No batch COMMIT/BEGIN — see module docstring "Memory pressure". # Alembic owns the outer transaction; raw COMMIT here would # corrupt its bookkeeping and the dataset is small enough to # complete in one tx without memtable pressure. Keyset cursor # already advanced above. print( f"[185] backfill complete: " f"rule2 force_transfer rollback = {counts['rule2_force_transfer']}, " f"rule3 idle→office = {counts['rule3_idle_office']}, " f"rule5 missed leaver = {counts['rule5_leaver']}, " f"skipped = {counts['rule_skipped']}" ) def downgrade() -> None: # Intentional no-op. The backfill is data-correcting; reverting it # would re-introduce the bug ("everything stacked on Meng"). pass