"""ai_spending_period_rollup — materialised BucketRow set per period window.

Revision ID: 261
Revises: 260
Create Date: 2026-06-06

AI 大盘的每个 endpoint 都从 ``_collect_period_rows`` 派生(KPI / breakdown /
movers / by-bu / attribution / drill),而它每次请求都扫 ~270 万行
``cmdb_cost_facts``。已关闭月份的结果是**不可变**的 —— 本表把闭月的
canonical BucketRow 集物化下来,闭月读直接命中本表,免去重扫。

一行 == 一条 BucketRow(持久化投影,非预聚合总额 —— 大盘的 Python 聚合
仍跑在物化行上,所有维度照常可用)。窗口 = 替换单元(刷新 = 按
``(period_start, period_end_excl)`` 整窗 DELETE 再批量 INSERT)。
``whitelist_fingerprint`` 钉住会改变实算结果的输入(AI service 白名单 +
CNY→USD 汇率 + 口径版本);读时只信指纹匹配的行,不匹配视为 miss 重算。
FX **进**指纹 —— seats 桶用实时汇率把 CNY 单价换 USD(非冻结),改汇率会
变闭月 seat USD,故纳入以保物化表不偏离实算。

无 FK(可重建缓存)+ BigInteger 自增 PK → 免 TiDB CHAR/VARCHAR FK 严格性。
幂等:``_table_exists`` 守卫;若上一轮留下 VARCHAR-id 旧表先 DROP(本表 PK
是 BigInteger,无此风险,守卫从简)。Downgrade 删表。
"""
from __future__ import annotations

import sqlalchemy as sa

from alembic import op

revision = "261"
down_revision = "260"
branch_labels = None
depends_on = None

_TABLE = "ai_spending_period_rollup"


def _table_exists(name: str) -> bool:
    return op.get_bind().execute(
        sa.text(
            "SELECT 1 FROM information_schema.TABLES "
            "WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = :n LIMIT 1"
        ),
        {"n": name},
    ).scalar() is not None


def upgrade() -> None:
    if _table_exists(_TABLE):
        return
    op.create_table(
        _TABLE,
        sa.Column("id", sa.BigInteger(), autoincrement=True, nullable=False),
        sa.Column("period_start", sa.Date(), nullable=False),
        sa.Column("period_end_excl", sa.Date(), nullable=False),
        sa.Column("whitelist_fingerprint", sa.String(length=64), nullable=False),
        sa.Column("bucket", sa.String(length=32), nullable=False),
        sa.Column("fact_date", sa.Date(), nullable=False),
        sa.Column("cost_usd", sa.Numeric(14, 6), nullable=False),
        sa.Column("cost_usd_gross", sa.Numeric(14, 6), nullable=True),
        sa.Column("user_id", sa.String(length=36), nullable=True),
        sa.Column("department", sa.String(length=255), nullable=True),
        sa.Column("auto_owner_source", sa.String(length=32), nullable=True),
        sa.Column("software_id", sa.String(length=36), nullable=True),
        sa.Column("plan_name", sa.String(length=255), nullable=True),
        sa.Column("model_name", sa.String(length=128), nullable=True),
        sa.Column("integration_id", sa.String(length=36), nullable=True),
        sa.Column("scope_id", sa.String(length=36), nullable=True),
        sa.Column("resource_group", sa.String(length=512), nullable=True),
        sa.Column("iam_user", sa.String(length=255), nullable=True),
        sa.Column("service_name", sa.String(length=255), nullable=True),
        sa.Column("aliyun_account_id", sa.String(length=32), nullable=True),
        sa.Column("resource_id", sa.Text(), nullable=True),
        sa.Column("raw_json", sa.JSON(), nullable=True),
        sa.Column(
            "computed_at", sa.DateTime(timezone=True),
            server_default=sa.text("CURRENT_TIMESTAMP"), nullable=False,
        ),
        sa.PrimaryKeyConstraint("id"),
        mysql_charset="utf8mb4",
        mysql_collate="utf8mb4_unicode_ci",
    )
    op.create_index(
        "ix_ai_rollup_window_fp", _TABLE,
        ["period_start", "period_end_excl", "whitelist_fingerprint"],
    )


def downgrade() -> None:
    if _table_exists(_TABLE):
        op.drop_table(_TABLE)