group_fqcd_jr/alembic/versions/20260909_constraint_fix.py

"""基线约束纠偏：单列唯一键 → docs/00 声明的联合唯一键（幂等收敛）。

修订原因：`tools/generate_baseline_sql.py` 旧实现只按规则列里的"唯一"字样逐字段生成
`UNIQUE KEY`，无法表达文档中的 `唯一键 (a, b)`，导致四张表的联合唯一键被错误拆成两个
单列唯一键。后果是场内日线与净值每产品只能存一行、持仓每客户只能有一个产品。
`docs/00-新数据库基线设计.md` 是不可变基线，本次只修正数据库结构，未修改任何基线文档。

幂等性：本迁移不假设起始状态，而是把每张表**收敛到基线声明的约束**——
- 既有库：存在错误单列唯一键、缺少联合唯一键 → 删旧加新；
- 空库（由已修正的 `baseline_generated.sql` 重建）：联合唯一键已存在 → 自动跳过。

不这样做会出现"空库 upgrade head 失败"：完整重建时 `baseline_schema` 已按修正后的基线
建出联合唯一键，无条件 `DROP INDEX` 会报 1091。

影响面：4 张表、最多删除 8 个错误唯一键、新增 4 个联合唯一键；纠偏时四张表均为 0 行。
证据：`docs/evidence/20260909-before-constraint-fix.sql`、
`docs/evidence/20260909-fingerprint-before.json` 与 `-after.json`、`docs/08`。
"""
from sqlalchemy import text
from sqlalchemy.engine import Connection

from alembic import op

revision = "20260909_constraint_fix"
down_revision = "20260909_api_receipt"
branch_labels = None
depends_on = None

# (表名, 错误唯一键（顺序与列一致）, 基线要求的联合唯一键名, 列顺序)
CORRECTIONS: list[tuple[str, tuple[str, ...], str, tuple[str, ...]]] = [
    (
        "fin_market_price",
        ("uk_fin_market_price_product_id", "uk_fin_market_price_trade_date"),
        "uk_fin_market_price_product_id_trade_date",
        ("product_id", "trade_date"),
    ),
    (
        "fin_nav_history",
        ("uk_fin_nav_history_product_id", "uk_fin_nav_history_nav_date"),
        "uk_fin_nav_history_product_id_nav_date",
        ("product_id", "nav_date"),
    ),
    (
        "fin_holding",
        ("uk_fin_holding_customer_id", "uk_fin_holding_product_id"),
        "uk_fin_holding_customer_id_product_id",
        ("customer_id", "product_id"),
    ),
    (
        "sys_customer_assignment",
        ("uk_sys_customer_assignment_customer_id", "uk_sys_customer_assignment_employee_role"),
        "uk_sys_customer_assignment_customer_id_employee_role",
        ("customer_id", "employee_role"),
    ),
]


def _unique_indexes(bind: Connection, table: str) -> dict[str, tuple[str, ...]]:
    """返回该表所有唯一索引：索引名 → 按序号排列的列元组。"""
    rows = bind.execute(
        text(
            """
            SELECT INDEX_NAME, COLUMN_NAME
            FROM information_schema.STATISTICS
            WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = :table
              AND NON_UNIQUE = 0 AND INDEX_NAME <> 'PRIMARY'
            ORDER BY INDEX_NAME, SEQ_IN_INDEX
            """
        ),
        {"table": table},
    ).all()
    grouped: dict[str, list[str]] = {}
    for index_name, column in rows:
        grouped.setdefault(index_name, []).append(column)
    return {name: tuple(columns) for name, columns in grouped.items()}


def upgrade() -> None:
    bind = op.get_bind()
    for table, wrong_keys, key_name, columns in CORRECTIONS:
        removable = [key for key in wrong_keys if key in _unique_indexes(bind, table)]
        if removable:
            drops = ", ".join(f"DROP INDEX `{key}`" for key in removable)
            op.execute(f"ALTER TABLE `{table}` {drops}")
        present = {cols for cols in _unique_indexes(bind, table).values()}
        if columns not in present:
            rendered = ", ".join(f"`{column}`" for column in columns)
            op.execute(f"ALTER TABLE `{table}` ADD UNIQUE KEY `{key_name}` ({rendered})")


def downgrade() -> None:
    # 回退会恢复已知错误的单列唯一键，仅用于整体回退到纠偏前状态。
    bind = op.get_bind()
    for table, wrong_keys, key_name, columns in CORRECTIONS:
        if key_name in _unique_indexes(bind, table):
            op.execute(f"ALTER TABLE `{table}` DROP INDEX `{key_name}`")
        for key, column in zip(wrong_keys, columns, strict=True):
            present = {cols for cols in _unique_indexes(bind, table).values()}
            if (column,) not in present:
                op.execute(f"ALTER TABLE `{table}` ADD UNIQUE KEY `{key}` (`{column}`)")