"""finance-flux domain schema: config, line catalog, NS account map, periods, pulled lines, and the human commentary layer (6 tables) + seed of the config singleton and the BS/PL statement layout. Revision ID: 347 Revises: 346 """ from __future__ import annotations import uuid import sqlalchemy as sa from alembic import op from app.core.migration_helpers import id_type_for, table_kw_for from app.domains.finance_flux import constants as c revision = "347" down_revision = "346" branch_labels = None depends_on = None CONFIG = "finance_flux_config" LINE_DEFS = "finance_flux_line_defs" ACCOUNT_MAP = "finance_flux_account_map" PERIODS = "finance_flux_periods" LINES = "finance_flux_lines" COMMENTARY = "finance_flux_commentary" KPI_DEFS = "finance_flux_kpi_defs" KPI_OVERRIDE = "finance_flux_kpi_override" INSIGHT_CARDS = "finance_flux_insight_cards" FOLLOWUPS = "finance_flux_followups" 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 _has_rows(bind, name: str) -> bool: return bool(bind.execute(sa.text(f"SELECT 1 FROM {name} LIMIT 1")).scalar()) def upgrade() -> None: bind = op.get_bind() pid = id_type_for("identity_persons") tkw = table_kw_for("identity_persons") if not _table_exists(bind, CONFIG): op.create_table( CONFIG, sa.Column("id", pid, nullable=False), sa.Column("amount_threshold", sa.Numeric(20, 2), nullable=False, server_default=str(c.DEFAULT_AMOUNT_THRESHOLD)), sa.Column("pct_threshold", sa.Numeric(7, 2), nullable=False, server_default=str(c.DEFAULT_PCT_THRESHOLD)), sa.Column("currency", sa.String(3), nullable=False, server_default=c.DEFAULT_CURRENCY), sa.Column("consolidation_subsidiary", sa.String(64), nullable=True), sa.Column("accounting_book", sa.String(64), nullable=True), sa.Column("comparison_modes", sa.JSON, nullable=True), sa.Column("closed_periods_only", sa.Boolean, nullable=False, server_default="1"), sa.Column("ai_enabled", sa.Boolean, nullable=False, server_default="1"), sa.Column("ai_prompt_template", sa.Text, nullable=True), sa.Column("updated_by_person_id", pid, 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()), sa.PrimaryKeyConstraint("id"), **tkw, ) if not _table_exists(bind, LINE_DEFS): op.create_table( LINE_DEFS, sa.Column("id", pid, nullable=False), sa.Column("statement", sa.String(2), nullable=False), sa.Column("line_key", sa.String(64), nullable=False), sa.Column("label", sa.String(255), nullable=False), sa.Column("section", sa.String(64), nullable=True), sa.Column("row_type", sa.String(16), nullable=False, server_default=c.ROW_LINE), sa.Column("sort_order", sa.Integer, nullable=False, server_default="0"), sa.Column("owner_person_id", pid, nullable=True), sa.Column("owner_label", sa.String(64), nullable=True), sa.Column("threshold_amount", sa.Numeric(20, 2), nullable=True), sa.Column("threshold_pct", sa.Numeric(7, 2), nullable=True), sa.Column("is_active", sa.Boolean, nullable=False, server_default="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.UniqueConstraint("statement", "line_key", name="uq_finance_flux_line_defs_key"), **tkw, ) op.create_index("ix_finance_flux_line_defs_stmt_sort", LINE_DEFS, ["statement", "sort_order"]) if not _table_exists(bind, ACCOUNT_MAP): op.create_table( ACCOUNT_MAP, sa.Column("id", pid, nullable=False), sa.Column("ns_account_id", sa.String(64), nullable=False), sa.Column("account_number", sa.String(64), nullable=True), sa.Column("account_name", sa.String(255), nullable=True), sa.Column("account_type", sa.String(32), nullable=True), sa.Column("statement", sa.String(2), nullable=False), sa.Column("line_key", sa.String(64), nullable=True), sa.Column("sign", sa.Integer, nullable=False, server_default="1"), sa.Column("is_active", sa.Boolean, nullable=False, server_default="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.UniqueConstraint("ns_account_id", name="uq_finance_flux_account_map_acct"), **tkw, ) op.create_index("ix_finance_flux_account_map_line", ACCOUNT_MAP, ["line_key"]) if not _table_exists(bind, PERIODS): op.create_table( PERIODS, sa.Column("id", pid, nullable=False), sa.Column("period_key", sa.String(7), nullable=False), sa.Column("label", sa.String(32), nullable=False), sa.Column("ns_period_id", sa.String(64), nullable=True), sa.Column("start_date", sa.Date, nullable=True), sa.Column("end_date", sa.Date, nullable=True), sa.Column("is_closed", sa.Boolean, nullable=False, server_default="0"), sa.Column("sync_status", sa.String(16), nullable=False, server_default=c.SYNC_PENDING), sa.Column("last_synced_at", sa.DateTime(timezone=True), nullable=True), sa.Column("last_sync_error", 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()), sa.PrimaryKeyConstraint("id"), sa.UniqueConstraint("period_key", name="uq_finance_flux_periods_key"), **tkw, ) if not _table_exists(bind, LINES): op.create_table( LINES, sa.Column("id", pid, nullable=False), sa.Column("period_id", pid, nullable=False), sa.Column("statement", sa.String(2), nullable=False), sa.Column("line_key", sa.String(64), nullable=False), sa.Column("value", sa.Numeric(20, 2), nullable=False, server_default="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()), sa.PrimaryKeyConstraint("id"), sa.UniqueConstraint("period_id", "statement", "line_key", name="uq_finance_flux_lines_period_line"), sa.ForeignKeyConstraint(["period_id"], [f"{PERIODS}.id"], name="fk_finance_flux_lines_period", ondelete="CASCADE"), **tkw, ) op.create_index("ix_finance_flux_lines_period", LINES, ["period_id"]) if not _table_exists(bind, COMMENTARY): op.create_table( COMMENTARY, sa.Column("id", pid, nullable=False), sa.Column("period_id", pid, nullable=False), sa.Column("statement", sa.String(2), nullable=False), sa.Column("line_key", sa.String(64), nullable=False), sa.Column("commentary", sa.Text, nullable=True), sa.Column("recon_link", sa.Text, nullable=True), sa.Column("ai_narrative", sa.Text, nullable=True), sa.Column("ai_generated_at", sa.DateTime(timezone=True), nullable=True), sa.Column("edited_by_person_id", pid, 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()), sa.PrimaryKeyConstraint("id"), sa.UniqueConstraint("period_id", "statement", "line_key", name="uq_finance_flux_commentary_period_line"), sa.ForeignKeyConstraint(["period_id"], [f"{PERIODS}.id"], name="fk_finance_flux_commentary_period", ondelete="CASCADE"), **tkw, ) op.create_index("ix_finance_flux_commentary_period", COMMENTARY, ["period_id"]) if not _table_exists(bind, KPI_DEFS): op.create_table( KPI_DEFS, sa.Column("id", pid, nullable=False), sa.Column("kpi_key", sa.String(64), nullable=False), sa.Column("label", sa.String(128), nullable=False), sa.Column("statement", sa.String(2), nullable=False), sa.Column("source_line_key", sa.String(64), nullable=False), sa.Column("sort_order", sa.Integer, nullable=False, server_default="0"), sa.Column("higher_is_better", sa.Boolean, nullable=False, server_default="1"), sa.Column("is_active", sa.Boolean, nullable=False, server_default="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.UniqueConstraint("kpi_key", name="uq_finance_flux_kpi_defs_key"), **tkw, ) if not _table_exists(bind, KPI_OVERRIDE): op.create_table( KPI_OVERRIDE, sa.Column("id", pid, nullable=False), sa.Column("period_id", pid, nullable=False), sa.Column("kpi_key", sa.String(64), nullable=False), sa.Column("value", sa.Numeric(20, 2), nullable=False), sa.Column("note", sa.Text, nullable=True), sa.Column("edited_by_person_id", pid, 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()), sa.PrimaryKeyConstraint("id"), sa.UniqueConstraint("period_id", "kpi_key", name="uq_finance_flux_kpi_override_period_kpi"), sa.ForeignKeyConstraint(["period_id"], [f"{PERIODS}.id"], name="fk_finance_flux_kpi_override_period", ondelete="CASCADE"), **tkw, ) op.create_index("ix_finance_flux_kpi_override_period", KPI_OVERRIDE, ["period_id"]) if not _table_exists(bind, INSIGHT_CARDS): op.create_table( INSIGHT_CARDS, sa.Column("id", pid, nullable=False), sa.Column("period_id", pid, nullable=False), sa.Column("card_type", sa.String(16), nullable=False), sa.Column("emoji", sa.String(16), nullable=True), sa.Column("color", sa.String(16), nullable=True), sa.Column("title", sa.String(255), nullable=True), sa.Column("body", sa.Text, nullable=True), sa.Column("sort_order", sa.Integer, nullable=False, server_default="0"), sa.Column("created_by_person_id", pid, 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()), sa.PrimaryKeyConstraint("id"), sa.ForeignKeyConstraint(["period_id"], [f"{PERIODS}.id"], name="fk_finance_flux_insight_cards_period", ondelete="CASCADE"), **tkw, ) op.create_index("ix_finance_flux_insight_cards_period", INSIGHT_CARDS, ["period_id", "card_type"]) if not _table_exists(bind, FOLLOWUPS): op.create_table( FOLLOWUPS, sa.Column("id", pid, nullable=False), sa.Column("period_id", pid, nullable=False), sa.Column("item", sa.Text, nullable=True), sa.Column("owner_person_id", pid, nullable=True), sa.Column("owner_label", sa.String(64), nullable=True), sa.Column("action", sa.Text, nullable=True), sa.Column("priority", sa.String(8), nullable=False, server_default=c.PRIORITY_MEDIUM), sa.Column("sort_order", sa.Integer, nullable=False, server_default="0"), sa.Column("created_by_person_id", pid, 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()), sa.PrimaryKeyConstraint("id"), sa.ForeignKeyConstraint(["period_id"], [f"{PERIODS}.id"], name="fk_finance_flux_followups_period", ondelete="CASCADE"), **tkw, ) op.create_index("ix_finance_flux_followups_period", FOLLOWUPS, ["period_id"]) _seed(bind) def _seed(bind) -> None: # Config singleton — seeded with the user-confirmed NetSuite consolidation # scope (PingCap Ltd. consolidated, Primary Accounting Book, USD). PR2 # resolves these names to NS internal ids at pull time. if _table_exists(bind, CONFIG) and not _has_rows(bind, CONFIG): cfg = sa.table( CONFIG, sa.column("id"), sa.column("comparison_modes", sa.JSON), # type → list JSON-encoded sa.column("consolidation_subsidiary"), sa.column("accounting_book"), ) op.bulk_insert(cfg, [{ "id": str(uuid.uuid4()), "comparison_modes": [c.COMPARE_MOM, c.COMPARE_YOY], "consolidation_subsidiary": "PingCap Ltd. (Consolidated)", "accounting_book": "Primary Accounting Book", }]) # KPI catalog (Summary page cards). if _table_exists(bind, KPI_DEFS) and not _has_rows(bind, KPI_DEFS): ktbl = sa.table( KPI_DEFS, sa.column("id"), sa.column("kpi_key"), sa.column("label"), sa.column("statement"), sa.column("source_line_key"), sa.column("sort_order"), sa.column("higher_is_better"), ) op.bulk_insert(ktbl, [{ "id": str(uuid.uuid4()), "kpi_key": d["k"], "label": d["l"], "statement": d["s"], "source_line_key": d["src"], "sort_order": i * 10, "higher_is_better": d["hib"], } for i, d in enumerate(c.SEED_KPI_DEFS)]) # Statement layout catalog (BS + PL), per-statement sort order. if _table_exists(bind, LINE_DEFS) and not _has_rows(bind, LINE_DEFS): tbl = sa.table( LINE_DEFS, sa.column("id"), sa.column("statement"), sa.column("line_key"), sa.column("label"), sa.column("section"), sa.column("row_type"), sa.column("sort_order"), ) counters: dict[str, int] = {} rows = [] for d in c.SEED_LINE_DEFS: stmt = d["s"] order = counters.get(stmt, 0) counters[stmt] = order + 10 rows.append({ "id": str(uuid.uuid4()), "statement": stmt, "line_key": d["k"], "label": d["l"], "section": d["sec"], "row_type": d["t"], "sort_order": order, }) op.bulk_insert(tbl, rows) def downgrade() -> None: bind = op.get_bind() for name in (FOLLOWUPS, INSIGHT_CARDS, KPI_OVERRIDE, KPI_DEFS, COMMENTARY, LINES, PERIODS, ACCOUNT_MAP, LINE_DEFS, CONFIG): if _table_exists(bind, name): op.drop_table(name)