"""Asset inventory (盘点) — cycles + items + denormalised columns + jobs. Revision ID: 088 Revises: 087 Create Date: 2026-04-29 Renumbered THREE times mid-deploy because four parallel PRs hit the same numeric slots within an hour: * 085 → 086 (PR #104) — PR #102 landed ``notification_outbox_read_at`` at 085 first. * 086 → 087 (PR #107) — PR #106 then ALSO renumbered ``notification_outbox_read_at`` to 086, collapsing onto the slot we had just taken. * 087 → 088 (this PR) — PR #105 was already in flight from a stale base; its squash-merge re-introduced a now-stale 085_asset_inventory duplicate AND landed its own ``087_bailian_usage_per_account.py``, giving main two 087s. Final shape: ``084 → 086_notification → 087_bailian → 088_asset_inventory``. The duplicate ``085_asset_inventory.py`` (re-introduced by #105's stale base, identical body to this file) is removed in the same PR. Same numbering-race pattern PR #91 used to deconflict 077↔078, and the AI Spending PR used for 078↔079. What this migration does ======================== Adds the schema for the resource-checkup feature laid out in ``DECISIONS.md`` (2026-04-29). One *cycle* row per round; one *item* row per (cycle, asset) snapshot; three denormalised columns on ``asset_items`` so the existing list page can show "最近盘点" without a join. Two scheduled-job rows seeded **disabled** — the user explicitly asked the auto-half-yearly trigger NOT to run on its own until they flip the toggle. Schema is idempotent (safe to re-run on a partial deploy): table / column / index existence is probed first; FKs go through ``_try_swallow`` so a duplicate-FK retry doesn't blow up the deploy. Foreign keys mirror the parent column's *type AND collation* via ``_fk_source_col_type``. TiDB is strict about both — we got bitten on 082→083 deploys when ``identity_persons.id`` came back as ``utf8mb4_bin`` while a child table defaulted to ``utf8mb4_unicode_ci``. """ from __future__ import annotations import logging import uuid import sqlalchemy as sa from alembic import op logger = logging.getLogger(__name__) revision = "088" down_revision = "087" branch_labels = None depends_on = None # Same allow-list used in 079. Anything outside this list is treated as # a structural failure and re-raised — silent swallowing of "FK type # mismatch" is the bug the production-grade rule explicitly forbids. _IDEMPOTENT_RETRY_HINTS = ( "duplicate key name", "duplicate entry", "duplicate foreign key", "errno: 121", "check that column/key exists", ) def _try_swallow(label: str, fn) -> None: try: fn() except Exception as exc: # noqa: BLE001 msg = str(exc).lower() if any(hint in msg for hint in _IDEMPOTENT_RETRY_HINTS): logger.warning( "migration 088: %s skipped (idempotent retry): %s", label, exc, ) return logger.error( "migration 088: %s failed with non-idempotent error: %s", label, exc, ) raise def _table_exists(bind, name: str) -> bool: 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 _column_exists(bind, table: str, column: str) -> bool: return bind.execute( sa.text( "SELECT 1 FROM information_schema.COLUMNS " "WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = :t " " AND COLUMN_NAME = :c" ), {"t": table, "c": column}, ).scalar() is not None def _fk_source_col_type( bind, target_table: str, target_column: str, *, fallback_type: str = "varchar", fallback_collation: str | None = None, length: int = 36, ): row = bind.execute( sa.text( "SELECT DATA_TYPE, COLLATION_NAME FROM " "information_schema.COLUMNS " "WHERE TABLE_SCHEMA = DATABASE() " " AND TABLE_NAME = :t AND COLUMN_NAME = :c" ), {"t": target_table, "c": target_column}, ).fetchone() if row: data_type = (row[0] or fallback_type).lower() collation = row[1] or fallback_collation else: data_type = fallback_type.lower() collation = fallback_collation type_cls = sa.CHAR if data_type == "char" else sa.String if collation: return type_cls(length, collation=collation) return type_cls(length) _AUTO_START_JOB = "asset.inventory_auto_start" _REMINDER_JOB = "asset.inventory_reminder" def upgrade() -> None: bind = op.get_bind() persons_id = _fk_source_col_type(bind, "identity_persons", "id") asset_items_id = _fk_source_col_type(bind, "asset_items", "id") office_id = _fk_source_col_type(bind, "asset_office_addresses", "id") # ── 1. asset_inventory_cycles ───────────────────────────────────── if not _table_exists(bind, "asset_inventory_cycles"): op.create_table( "asset_inventory_cycles", sa.Column("id", sa.CHAR(36), primary_key=True), sa.Column("name", sa.String(128), nullable=False), sa.Column("description", sa.Text, nullable=True), sa.Column( "status", sa.String(16), nullable=False, server_default="draft", ), sa.Column("scope_filter", sa.JSON, nullable=True), sa.Column("started_at", sa.DateTime(timezone=True), nullable=True), sa.Column("due_at", sa.DateTime(timezone=True), nullable=True), sa.Column("closed_at", sa.DateTime(timezone=True), nullable=True), sa.Column( "reminder_sent_at", sa.DateTime(timezone=True), nullable=True, ), sa.Column("created_by", persons_id, nullable=True), sa.Column("closed_by", persons_id, nullable=True), sa.Column( "total_items", sa.Integer, nullable=False, server_default=sa.text("0"), ), sa.Column( "pending_count", sa.Integer, nullable=False, server_default=sa.text("0"), ), sa.Column( "confirmed_count", sa.Integer, nullable=False, server_default=sa.text("0"), ), sa.Column( "issue_count", sa.Integer, nullable=False, server_default=sa.text("0"), ), sa.Column( "created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), ), sa.Column( "updated_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), onupdate=sa.func.now(), ), sa.Index("ix_asset_inv_cycles_status", "status"), sa.Index("ix_asset_inv_cycles_started_at", "started_at"), ) _try_swallow( "fk_asset_inv_cycles_created_by", lambda: op.create_foreign_key( "fk_asset_inv_cycles_created_by", "asset_inventory_cycles", "identity_persons", ["created_by"], ["id"], ondelete="SET NULL", ), ) _try_swallow( "fk_asset_inv_cycles_closed_by", lambda: op.create_foreign_key( "fk_asset_inv_cycles_closed_by", "asset_inventory_cycles", "identity_persons", ["closed_by"], ["id"], ondelete="SET NULL", ), ) # ── 2. asset_inventory_items ────────────────────────────────────── cycle_id_type = _fk_source_col_type(bind, "asset_inventory_cycles", "id") if not _table_exists(bind, "asset_inventory_items"): op.create_table( "asset_inventory_items", sa.Column("id", sa.CHAR(36), primary_key=True), sa.Column("cycle_id", cycle_id_type, nullable=False), sa.Column("asset_id", asset_items_id, nullable=False), sa.Column("holder_kind", sa.String(16), nullable=False), sa.Column("holder_person_id", persons_id, nullable=True), sa.Column("office_id", office_id, nullable=True), sa.Column("asset_tag_snapshot", sa.String(64), nullable=True), sa.Column("asset_name_snapshot", sa.String(128), nullable=True), sa.Column("asset_category_snapshot", sa.String(32), nullable=True), sa.Column("asset_location_snapshot", sa.String(256), nullable=True), sa.Column("asset_status_snapshot", sa.String(32), nullable=True), sa.Column( "status", sa.String(16), nullable=False, server_default="pending", ), sa.Column("response_type", sa.String(32), nullable=True), sa.Column("response_notes", sa.Text, nullable=True), sa.Column("transferred_to_id", persons_id, nullable=True), sa.Column("new_location", sa.String(256), nullable=True), sa.Column( "responded_at", sa.DateTime(timezone=True), nullable=True, ), sa.Column("responded_by", persons_id, nullable=True), sa.Column( "reminded_at", sa.DateTime(timezone=True), nullable=True, ), sa.Column("issue_status", sa.String(16), nullable=True), sa.Column( "issue_resolved_at", sa.DateTime(timezone=True), nullable=True, ), sa.Column("issue_resolved_by", persons_id, nullable=True), sa.Column("issue_resolution_notes", sa.Text, nullable=True), sa.Column( "created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), ), sa.Column( "updated_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), onupdate=sa.func.now(), ), sa.UniqueConstraint( "cycle_id", "asset_id", name="uq_asset_inv_items_cycle_asset", ), sa.Index("ix_asset_inv_items_cycle_status", "cycle_id", "status"), sa.Index("ix_asset_inv_items_holder", "holder_person_id"), ) _try_swallow( "fk_asset_inv_items_cycle", lambda: op.create_foreign_key( "fk_asset_inv_items_cycle", "asset_inventory_items", "asset_inventory_cycles", ["cycle_id"], ["id"], ondelete="CASCADE", ), ) _try_swallow( "fk_asset_inv_items_asset", lambda: op.create_foreign_key( "fk_asset_inv_items_asset", "asset_inventory_items", "asset_items", ["asset_id"], ["id"], ondelete="CASCADE", ), ) _try_swallow( "fk_asset_inv_items_holder", lambda: op.create_foreign_key( "fk_asset_inv_items_holder", "asset_inventory_items", "identity_persons", ["holder_person_id"], ["id"], ondelete="SET NULL", ), ) _try_swallow( "fk_asset_inv_items_office", lambda: op.create_foreign_key( "fk_asset_inv_items_office", "asset_inventory_items", "asset_office_addresses", ["office_id"], ["id"], ondelete="SET NULL", ), ) _try_swallow( "fk_asset_inv_items_transferred_to", lambda: op.create_foreign_key( "fk_asset_inv_items_transferred_to", "asset_inventory_items", "identity_persons", ["transferred_to_id"], ["id"], ondelete="SET NULL", ), ) _try_swallow( "fk_asset_inv_items_responded_by", lambda: op.create_foreign_key( "fk_asset_inv_items_responded_by", "asset_inventory_items", "identity_persons", ["responded_by"], ["id"], ondelete="SET NULL", ), ) _try_swallow( "fk_asset_inv_items_issue_resolved_by", lambda: op.create_foreign_key( "fk_asset_inv_items_issue_resolved_by", "asset_inventory_items", "identity_persons", ["issue_resolved_by"], ["id"], ondelete="SET NULL", ), ) # ── 3. asset_items denormalised columns ─────────────────────────── if not _column_exists(bind, "asset_items", "last_inventory_at"): op.add_column( "asset_items", sa.Column( "last_inventory_at", sa.DateTime(timezone=True), nullable=True, ), ) if not _column_exists(bind, "asset_items", "last_inventory_cycle_id"): op.add_column( "asset_items", sa.Column( "last_inventory_cycle_id", cycle_id_type, nullable=True, ), ) _try_swallow( "fk_asset_items_last_inventory_cycle", lambda: op.create_foreign_key( "fk_asset_items_last_inventory_cycle", "asset_items", "asset_inventory_cycles", ["last_inventory_cycle_id"], ["id"], ondelete="SET NULL", ), ) if not _column_exists(bind, "asset_items", "last_inventory_status"): op.add_column( "asset_items", sa.Column( "last_inventory_status", sa.String(32), nullable=True, ), ) # ── 4. Seed scheduled_jobs rows (both disabled) ─────────────────── # # Auto-start: cron 09:00 CST on Jan 15 + Jul 15 (= 01:00 UTC). Set # ``timezone='UTC'`` so the scheduler interprets ``cron_expr`` in # UTC — matches every other ``cron`` job in the table. The # human-friendly mapping ("09:00 CST") is documented here so a # future maintainer doesn't think the timing was a typo. # # Reminder: hourly interval; the handler short-circuits unless an # active cycle is past ``started_at + 7d`` AND # ``reminder_sent_at IS NULL``. Cheap to run, so an hourly check is # safe — the UPSERT-style guard makes a 1h window indistinguishable # from a 24h window in practice. Disabled by seed; admin enables # alongside the auto-start row when they're ready to commit to the # half-yearly cadence. for job_name, display_name, description, trigger_type, interval_seconds, cron_expr in ( ( _AUTO_START_JOB, "资产盘点 自动触发", ( "每年 1 月 15 日和 7 月 15 日 09:00 (CST) 自动创建一个新的" "资产盘点周期, 范围为全公司活跃资产。默认 disabled — " "由 IT 在确认手动流程跑通后,在 /admin/scheduled-jobs 页面" "手动启用。" ), "cron", None, "0 1 15 1,7 *", ), ( _REMINDER_JOB, "资产盘点 自动提醒", ( "每小时检查活跃盘点周期, 对开始 7 天后仍未响应的接收人" "重新推送一次飞书提醒(每个周期只重推一次)。默认 " "disabled — 周期启用后由 IT 一并打开。" ), "interval", 3600, None, ), ): existing = bind.execute( sa.text( "SELECT id FROM scheduled_jobs WHERE name = :name" ), {"name": job_name}, ).scalar() if existing: continue bind.execute( sa.text( """ INSERT INTO scheduled_jobs (id, name, display_name, description, handler_key, trigger_type, interval_seconds, cron_expr, timezone, enabled, config, run_count, failure_count, created_at, updated_at) VALUES (:id, :name, :display_name, :description, :handler_key, :trigger_type, :interval_seconds, :cron_expr, 'UTC', 0, '{}', 0, 0, NOW(), NOW()) """ ), { "id": str(uuid.uuid4()), "name": job_name, "display_name": display_name, "description": description, "handler_key": job_name, "trigger_type": trigger_type, "interval_seconds": interval_seconds, "cron_expr": cron_expr, }, ) def downgrade() -> None: bind = op.get_bind() # 1. seeded jobs — safe to delete; handler stops being invoked. bind.execute( sa.text( "DELETE FROM scheduled_jobs WHERE name IN (:a, :b)" ), {"a": _AUTO_START_JOB, "b": _REMINDER_JOB}, ) # 2. denormalised columns on asset_items. if _column_exists(bind, "asset_items", "last_inventory_cycle_id"): _try_swallow( "drop_fk_asset_items_last_inventory_cycle", lambda: op.drop_constraint( "fk_asset_items_last_inventory_cycle", "asset_items", type_="foreignkey", ), ) op.drop_column("asset_items", "last_inventory_cycle_id") if _column_exists(bind, "asset_items", "last_inventory_at"): op.drop_column("asset_items", "last_inventory_at") if _column_exists(bind, "asset_items", "last_inventory_status"): op.drop_column("asset_items", "last_inventory_status") # 3. tables (children first). if _table_exists(bind, "asset_inventory_items"): op.drop_table("asset_inventory_items") if _table_exists(bind, "asset_inventory_cycles"): op.drop_table("asset_inventory_cycles")