"""Finance unified spend ledger — tables + rollup/FX crons. Creates the finance domain's four standalone tables (no foreign keys — person / integration references are plain String(36) to dodge TiDB's CHAR/VARCHAR + collation FK strictness, same call the AI-spending period rollup made): * finance_fx_monthly — month-frozen USD-per-unit FX rates * finance_spend_facts — the materialised unified ledger * finance_period_close — per-period accounting freeze state * finance_restatements — post-close total-movement audit log …and seeds two scheduled jobs: * finance.spend_rollup (daily) — materialise current + prev month * finance.fx_monthly_refresh (weekly) — backfill/freeze month FX rates Idempotent: every CREATE is guarded on table-absence, every seed INSERT on a NOT EXISTS check, so a re-run converges. """ from __future__ import annotations import uuid import sqlalchemy as sa from alembic import op revision = "294" down_revision = "293" branch_labels = None depends_on = None def _table_exists(bind, table: str) -> bool: row = bind.execute( sa.text( "SELECT 1 FROM information_schema.TABLES " "WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = :t" ), {"t": table}, ).fetchone() return row is not None def upgrade() -> None: bind = op.get_bind() if not _table_exists(bind, "finance_fx_monthly"): op.create_table( "finance_fx_monthly", sa.Column("id", sa.BigInteger(), autoincrement=True, nullable=False), sa.Column("period", sa.String(length=7), nullable=False), sa.Column("currency", sa.String(length=8), nullable=False), sa.Column("usd_per_unit", sa.Numeric(18, 8), nullable=False), sa.Column( "source", sa.String(length=16), nullable=False, server_default="frankfurter", ), sa.Column( "frozen", sa.Boolean(), nullable=False, server_default="0", ), sa.Column("note", sa.String(length=255), nullable=True), sa.Column( "created_at", sa.DateTime(timezone=True), server_default=sa.text("CURRENT_TIMESTAMP"), nullable=False, ), sa.Column( "updated_at", sa.DateTime(timezone=True), server_default=sa.text("CURRENT_TIMESTAMP"), nullable=False, ), sa.PrimaryKeyConstraint("id"), sa.UniqueConstraint( "period", "currency", name="uq_finance_fx_monthly_period_ccy", ), ) if not _table_exists(bind, "finance_spend_facts"): op.create_table( "finance_spend_facts", sa.Column("id", sa.BigInteger(), autoincrement=True, nullable=False), sa.Column("fact_key", sa.String(length=64), nullable=False), sa.Column("period", sa.String(length=7), nullable=False), sa.Column("cost_type", sa.String(length=16), nullable=False), sa.Column( "basis", sa.String(length=16), nullable=False, server_default="consumption", ), sa.Column("sub_category", sa.String(length=255), nullable=True), sa.Column("vendor", sa.String(length=255), nullable=True), sa.Column("department", sa.String(length=255), nullable=True), sa.Column("cost_center", sa.String(length=64), nullable=True), sa.Column("owner_person_id", sa.String(length=36), nullable=True), sa.Column( "allocation_source", sa.String(length=16), nullable=False, server_default="none", ), sa.Column("amount_native", sa.Numeric(18, 6), nullable=False), sa.Column("currency_native", sa.String(length=8), nullable=False), sa.Column("amount_usd", sa.Numeric(18, 6), nullable=False), sa.Column("fx_rate_used", sa.Numeric(18, 8), nullable=True), sa.Column( "fx_basis", sa.String(length=16), nullable=False, server_default="native_usd", ), sa.Column( "is_provisional", sa.Boolean(), nullable=False, server_default="0", ), sa.Column("source_domain", sa.String(length=32), nullable=False), sa.Column("source_ref", sa.String(length=255), nullable=True), sa.Column("rollup_fingerprint", sa.String(length=64), nullable=False), sa.Column( "computed_at", sa.DateTime(timezone=True), server_default=sa.text("CURRENT_TIMESTAMP"), nullable=False, ), sa.PrimaryKeyConstraint("id"), sa.UniqueConstraint("fact_key", name="uq_finance_spend_facts_fact_key"), ) op.create_index( "ix_fin_facts_period_type", "finance_spend_facts", ["period", "cost_type"], ) op.create_index( "ix_fin_facts_period_dept", "finance_spend_facts", ["period", "department"], ) op.create_index( "ix_fin_facts_period_cc", "finance_spend_facts", ["period", "cost_center"], ) op.create_index( "ix_fin_facts_period_vendor", "finance_spend_facts", ["period", "vendor"], ) op.create_index( "ix_fin_facts_period_owner", "finance_spend_facts", ["period", "owner_person_id"], ) if not _table_exists(bind, "finance_period_close"): op.create_table( "finance_period_close", sa.Column("id", sa.BigInteger(), autoincrement=True, nullable=False), sa.Column("period", sa.String(length=7), nullable=False), sa.Column( "status", sa.String(length=16), nullable=False, server_default="open", ), sa.Column("closed_total_usd", sa.Numeric(18, 6), nullable=True), sa.Column("closed_at", sa.DateTime(timezone=True), nullable=True), sa.Column("closed_by_id", sa.String(length=36), nullable=True), sa.Column("reopened_at", sa.DateTime(timezone=True), nullable=True), sa.Column("reopened_by_id", sa.String(length=36), nullable=True), sa.Column("note", sa.Text(), nullable=True), sa.Column( "created_at", sa.DateTime(timezone=True), server_default=sa.text("CURRENT_TIMESTAMP"), nullable=False, ), sa.Column( "updated_at", sa.DateTime(timezone=True), server_default=sa.text("CURRENT_TIMESTAMP"), nullable=False, ), sa.PrimaryKeyConstraint("id"), sa.UniqueConstraint("period", name="uq_finance_period_close_period"), ) if not _table_exists(bind, "finance_restatements"): op.create_table( "finance_restatements", sa.Column("id", sa.BigInteger(), autoincrement=True, nullable=False), sa.Column("period", sa.String(length=7), nullable=False), sa.Column("cost_type", sa.String(length=16), nullable=True), sa.Column("prev_usd", sa.Numeric(18, 6), nullable=False), sa.Column("new_usd", sa.Numeric(18, 6), nullable=False), sa.Column("delta_usd", sa.Numeric(18, 6), nullable=False), sa.Column("reason", sa.Text(), nullable=True), sa.Column( "detected_at", sa.DateTime(timezone=True), server_default=sa.text("CURRENT_TIMESTAMP"), nullable=False, ), sa.PrimaryKeyConstraint("id"), ) op.create_index( "ix_fin_restate_period", "finance_restatements", ["period", "detected_at"], ) # ── Seed scheduled jobs (idempotent) ── if _table_exists(bind, "scheduled_jobs"): _seed_job( bind, name="finance.spend_rollup", display="财务对账 · 统一花费总账物化", desc=( "每日重算当月与上月的 finance_spend_facts(云/软件/AI/采购四线" "归一 USD、AI·云去重、部门+成本中心双轴)。当月 provisional;" "上月每日重算吸收晚到账单,关账后冻结。" ), handler="finance.spend_rollup", interval=86400, ) _seed_job( bind, name="finance.fx_monthly_refresh", display="财务对账 · 月度历史汇率回填", desc=( "每周从 frankfurter.dev(ECB)回填 finance_fx_monthly 滚动 13 " "个月窗口,冻结已结束月份汇率,保证关账后 USD 总额稳定。" ), handler="finance.fx_monthly_refresh", interval=604800, ) def _seed_job(bind, *, name, display, desc, handler, interval) -> None: bind.execute( sa.text( "INSERT INTO scheduled_jobs " "(id, name, display_name, description, handler_key, " "trigger_type, interval_seconds, timezone, enabled, config, " "run_count, failure_count, created_at, updated_at) " "SELECT :id, :name, :display, :desc, :handler, 'interval', " ":interval, 'Asia/Shanghai', 1, :config, 0, 0, NOW(), NOW() " "FROM DUAL WHERE NOT EXISTS " "(SELECT 1 FROM scheduled_jobs WHERE name = :name)" ), { "id": str(uuid.uuid4()), "name": name, "display": display, "desc": desc, "handler": handler, "interval": interval, "config": "{}", }, ) def downgrade() -> None: bind = op.get_bind() if _table_exists(bind, "scheduled_jobs"): bind.execute( sa.text( "DELETE FROM scheduled_jobs WHERE handler_key IN " "('finance.spend_rollup', 'finance.fx_monthly_refresh')" ) ) for table in ( "finance_restatements", "finance_period_close", "finance_spend_facts", "finance_fx_monthly", ): if _table_exists(bind, table): op.drop_table(table)