"""Finance external-actuals layer (汇联易) — tables + disabled pull cron. Creates: * finance_external_statements — external actual-spend lines (汇联易 expense/reimbursement/invoice/payment), apportioned to cost-centre / department / project, for reconciliation + complete-the-picture. * finance_cost_centers — authoritative cost-centre master mirror. …and seeds the ``finance.huilianyi_actuals_sync`` scheduled job **DISABLED** (``enabled=0``): activation requires granting the read permission in 汇联易 开放平台 + a one-time dry-run field-mapping validation (the live expense API cannot be exercised from CI), then flipping it on. No FKs (TiDB CHAR/VARCHAR FK avoidance, same as the rest of the finance domain). Idempotent. """ from __future__ import annotations import uuid import sqlalchemy as sa from alembic import op revision = "297" down_revision = "296" 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_external_statements"): op.create_table( "finance_external_statements", sa.Column("id", sa.BigInteger(), autoincrement=True, nullable=False), sa.Column("statement_key", sa.String(length=64), nullable=False), sa.Column("source", sa.String(length=32), nullable=False), sa.Column("external_doc_no", sa.String(length=128), nullable=False), sa.Column("line_no", sa.Integer(), nullable=False, server_default="0"), sa.Column("period", sa.String(length=7), nullable=False), sa.Column("expense_category", sa.String(length=128), nullable=True), sa.Column("mapped_cost_type", sa.String(length=16), nullable=True), sa.Column("promote_to_ledger", sa.Boolean(), nullable=False, server_default="0"), sa.Column("cost_center", sa.String(length=64), nullable=True), sa.Column("department", sa.String(length=255), nullable=True), sa.Column("project", sa.String(length=128), nullable=True), sa.Column("vendor", sa.String(length=255), nullable=True), sa.Column("owner_person_id", sa.String(length=36), nullable=True), sa.Column("amount_native", sa.Numeric(18, 6), nullable=False), sa.Column("currency", 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="month_frozen"), sa.Column("invoice_no", sa.String(length=128), nullable=True), sa.Column("invoice_verified", sa.Boolean(), nullable=True), sa.Column("doc_status", sa.String(length=32), nullable=True), sa.Column("payment_status", sa.String(length=32), nullable=True), sa.Column("paid_at", sa.DateTime(timezone=True), nullable=True), sa.Column("raw", sa.JSON(), nullable=True), sa.Column("pulled_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("statement_key", name="uq_finance_ext_stmt_key"), ) op.create_index("ix_fin_ext_period_cc", "finance_external_statements", ["period", "cost_center"]) op.create_index("ix_fin_ext_period_dept", "finance_external_statements", ["period", "department"]) op.create_index("ix_fin_ext_period_vendor", "finance_external_statements", ["period", "vendor"]) op.create_index("ix_fin_ext_source_doc", "finance_external_statements", ["source", "external_doc_no"]) if not _table_exists(bind, "finance_cost_centers"): op.create_table( "finance_cost_centers", sa.Column("id", sa.BigInteger(), autoincrement=True, nullable=False), sa.Column("source", sa.String(length=32), nullable=False), sa.Column("code", sa.String(length=64), nullable=False), sa.Column("name", sa.String(length=255), nullable=True), sa.Column("parent_code", sa.String(length=64), nullable=True), sa.Column("active", sa.Boolean(), nullable=False, server_default="1"), sa.Column("raw", sa.JSON(), nullable=True), sa.Column("synced_at", sa.DateTime(timezone=True), server_default=sa.text("CURRENT_TIMESTAMP"), nullable=False), sa.PrimaryKeyConstraint("id"), sa.UniqueConstraint("source", "code", name="uq_finance_cost_center_src_code"), ) # Seed the pull cron DISABLED (enabled=0). Daily cadence once enabled. if _table_exists(bind, "scheduled_jobs"): 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', 0, :config, 0, 0, NOW(), NOW() " "FROM DUAL WHERE NOT EXISTS " "(SELECT 1 FROM scheduled_jobs WHERE name = :name)" ), { "id": str(uuid.uuid4()), "name": "finance.huilianyi_actuals_sync", "display": "财务对账 · 汇联易实际账拉取", "desc": ( "从汇联易开放 API 增量拉取报销/费用单据 + 成本中心主数据,落" " finance_external_statements / finance_cost_centers。默认" "禁用:需先在汇联易开放平台授予读权限 + dry-run 校验字段映射" "后启用。无连接器/无读权限优雅跳过。" ), "handler": "finance.huilianyi_actuals_sync", "interval": 86400, "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 = 'finance.huilianyi_actuals_sync'" ) ) for table in ("finance_external_statements", "finance_cost_centers"): if _table_exists(bind, table): op.drop_table(table)