"""Procurement budgets + budget consumptions + asset→PO link column. This migration completes the procurement domain: * ``procurement_budgets`` — department / cost-centre annual budget pool rows. Each row is a (year, department, cost_center, category_id?, currency) tuple holding ``total_amount`` + ``owner_person_id``. Multi-currency: separate row per currency under the same dept (no cross-currency aggregation in Phase 1). * ``procurement_budget_consumptions`` — append-only event log of budget mutations. Each PO write that affects a budget emits one row with ``event ∈ {lock, release, actualize}``. The current remaining balance is computed at read time as ``total - SUM(lock) + SUM( release)`` (open-commitment model — actualize is informational). * ``asset_items.procurement_order_id`` — nullable FK back to ``procurement_orders.id`` so every hardware row can be traced to a PO. Backfill page in PR 3 (admin-only) lets ops attach historic assets to their PO; the column stays nullable forever (legacy / gift / external-transfer assets exist). All FK source columns mirror their targets via the ``_fk_source_col_type`` helper (TiDB strict-FK). Revision ID: 224 Revises: 223 Create Date: 2026-05-30 """ from __future__ import annotations import logging import sqlalchemy as sa from alembic import op revision = "224" down_revision = "223" branch_labels = None depends_on = None _logger = logging.getLogger("alembic.runtime.migration") def _fk_source_col_type( bind, target_table: str, target_column: str, *, fallback_type: str = "varchar", fallback_collation: str | None = "utf8mb4_unicode_ci", length: int = 36, ): """Mirror target FK column's DATA_TYPE + collation (TiDB strict).""" 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) def _table_exists(bind, table: str) -> bool: return bind.execute( sa.text( "SELECT 1 FROM information_schema.TABLES " "WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = :t LIMIT 1" ), {"t": table}, ).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 LIMIT 1" ), {"t": table, "c": column}, ).scalar() is not None def _index_exists(bind, table: str, index: str) -> bool: return bind.execute( sa.text( "SELECT 1 FROM information_schema.STATISTICS " "WHERE TABLE_SCHEMA = DATABASE() " " AND TABLE_NAME = :t AND INDEX_NAME = :n LIMIT 1" ), {"t": table, "n": index}, ).scalar() is not None def upgrade() -> None: bind = op.get_bind() person_id_type = _fk_source_col_type(bind, "identity_persons", "id") # ── procurement_budgets ────────────────────────────────────────── if not _table_exists(bind, "procurement_budgets"): op.create_table( "procurement_budgets", sa.Column("id", sa.String(36), nullable=False), sa.Column("year", sa.Integer(), nullable=False), sa.Column("department", sa.String(128), nullable=False), sa.Column("cost_center", sa.String(64), nullable=True), sa.Column("category_id", sa.String(36), nullable=True), sa.Column( "currency", sa.String(8), nullable=False, server_default=sa.text("'CNY'"), ), sa.Column("total_amount", sa.Numeric(14, 2), nullable=False), sa.Column("owner_person_id", person_id_type, nullable=True), sa.Column("notes", sa.Text(), nullable=True), sa.Column( "is_active", sa.Boolean(), nullable=False, server_default=sa.text("1"), ), 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(), ), sa.PrimaryKeyConstraint("id"), sa.ForeignKeyConstraint(["category_id"], ["procurement_categories.id"]), sa.ForeignKeyConstraint(["owner_person_id"], ["identity_persons.id"]), # A budget row is uniquely keyed by (year, department, # cost_center, category_id, currency). NULL columns make # the unique-constraint loose on MySQL — we accept that; # admin owns dedup at the UI layer. sa.UniqueConstraint( "year", "department", "cost_center", "category_id", "currency", name="uq_procurement_budgets_scope", ), ) op.create_index( "ix_procurement_budgets_year_dept", "procurement_budgets", ["year", "department"], ) op.create_index( "ix_procurement_budgets_owner", "procurement_budgets", ["owner_person_id"], ) # ── procurement_budget_consumptions ────────────────────────────── if not _table_exists(bind, "procurement_budget_consumptions"): op.create_table( "procurement_budget_consumptions", sa.Column("id", sa.String(36), nullable=False), sa.Column("budget_id", sa.String(36), nullable=False), sa.Column("po_id", sa.String(36), nullable=True), sa.Column("event", sa.String(16), nullable=False), sa.Column("amount", sa.Numeric(14, 2), nullable=False), sa.Column("currency", sa.String(8), nullable=False), sa.Column("amount_cny", sa.Numeric(14, 2), nullable=True), sa.Column("fx_rate_to_cny", sa.Numeric(12, 6), nullable=True), sa.Column("notes", sa.Text(), nullable=True), sa.Column("actor_person_id", person_id_type, nullable=True), sa.Column( "occurred_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), ), sa.PrimaryKeyConstraint("id"), sa.ForeignKeyConstraint( ["budget_id"], ["procurement_budgets.id"], ondelete="CASCADE", ), sa.ForeignKeyConstraint( ["po_id"], ["procurement_orders.id"], ondelete="SET NULL", ), sa.ForeignKeyConstraint( ["actor_person_id"], ["identity_persons.id"], ), ) op.create_index( "ix_proc_budget_cons_budget", "procurement_budget_consumptions", ["budget_id"], ) op.create_index( "ix_proc_budget_cons_po", "procurement_budget_consumptions", ["po_id"], ) # ── asset_items.procurement_order_id ───────────────────────────── # The link from a hardware asset back to the PO that procured it. # Nullable forever — legacy / gift / external-transfer assets have # no PO trail. The PR 3 backfill page walks idle / assigned assets # whose procurement_order_id is NULL and lets ops attach them. if not _column_exists(bind, "asset_items", "procurement_order_id"): po_id_type = _fk_source_col_type(bind, "procurement_orders", "id") op.add_column( "asset_items", sa.Column("procurement_order_id", po_id_type, nullable=True), ) # FK + index on the new column. op.create_foreign_key( "fk_asset_items_procurement_order", "asset_items", "procurement_orders", ["procurement_order_id"], ["id"], ondelete="SET NULL", ) op.create_index( "ix_asset_items_procurement_order", "asset_items", ["procurement_order_id"], ) # ── Seed procurement_reminders scheduled job ───────────────────── # Daily 06:30 UTC — contract expiry + budget thresholds + PO # awaiting Helian for too long. Handler lives in # ``app/scheduled_jobs/procurement_handlers.py``. Enabled by # default because it's read-only with bounded fan-out. existing_job = bind.execute( sa.text( "SELECT id FROM scheduled_jobs WHERE name = :n LIMIT 1" ), {"n": "procurement.reminders"}, ).scalar() if existing_job is None: import uuid bind.execute( sa.text( """ INSERT INTO scheduled_jobs (id, name, display_name, description, handler_key, trigger_type, cron_expr, timezone, enabled, config, run_count, failure_count, created_at, updated_at) VALUES (:id, :name, :display, :desc, :handler, 'cron', :cron, :tz, 1, '{}', 0, 0, NOW(), NOW()) """ ), { "id": str(uuid.uuid4()), "name": "procurement.reminders", "display": "采购提醒(合同到期 / 预算阈值 / 汇联易迟批)", "handler": "procurement.reminders", "cron": "30 6 * * *", # 06:30 UTC daily "tz": "UTC", "desc": ( "Daily reminder dispatcher: contract expiry " "(within renew_notice_days), budget thresholds " "(80% / 95%), and POs lingering in submitted " "state without external_po_number. Per-target " "ISO-week dedup; read-only otherwise." ), }, ) bcount = bind.execute( sa.text("SELECT COUNT(*) FROM procurement_budgets") ).scalar() or 0 _logger.info( "procurement migration 224: budgets=%s consumptions table created; " "asset_items.procurement_order_id added; procurement.reminders " "scheduled.", bcount, ) def downgrade() -> None: op.execute(sa.text( "DELETE FROM scheduled_jobs WHERE name = 'procurement.reminders'" )) # Drop the asset link first (FK depends on procurement_orders). bind = op.get_bind() if _index_exists(bind, "asset_items", "ix_asset_items_procurement_order"): op.drop_index("ix_asset_items_procurement_order", "asset_items") try: op.drop_constraint( "fk_asset_items_procurement_order", "asset_items", type_="foreignkey", ) except Exception: # noqa: BLE001 pass # already gone if _column_exists(bind, "asset_items", "procurement_order_id"): op.drop_column("asset_items", "procurement_order_id") op.execute(sa.text("DROP TABLE IF EXISTS procurement_budget_consumptions")) op.execute(sa.text("DROP TABLE IF EXISTS procurement_budgets"))