"""离线执行迁移 434 的重键修复（部署窗口保险栓）。

迁移 434 把 ``cmdb_cost_facts`` 里非 ``usage`` 桶的历史行重键到带桶分量
的 ``fact_key``（桶分量改成 ``cost_bucket`` 的纯函数之后,旧行的键需要
重算）。表在生产是 ~840 万行量级,写集是 2026-07-27 之后写入的行加上任何
双口径重摄取 —— 通常几十秒,最坏（全表都是非 usage）20–30 分钟。

**为什么需要这个脚本**:本仓库有过「长回填被部署杀掉」的事故。迁移本身
按批 commit、可中断可续跑,但把它排在部署里就意味着窗口由部署决定。先离线
跑完这个修复,``alembic upgrade head`` 里的 434 就只剩一次纯扫描（每行都
重算出与已存 ``fact_key`` 不匹配 → 跳过,不再改写）。

迁移文件名以数字开头,不能直接 ``import`` —— 这个脚本封装了 importlib
那一层,免得操作者在生产上手搓。它调用的就是迁移**自己**的
``repair_bucket_fact_keys``,不是一份平行实现:平行实现会漂移,而这里漂移
等于错误重键。

用法::

    # 只看要动多少行,不写库
    uv run python scripts/repair_cost_fact_bucket_keys.py --dry-run

    # 真跑（按批 commit;中断了直接再跑一次即可）
    uv run python scripts/repair_cost_fact_bucket_keys.py --apply

跑完看四个计数:

* ``rekeyed``            重键成功的行
* ``scope_recovered``    ``scope_id IS NULL`` 但用云端 external_id 哈希过的
                         行,靠前像搜索找回了哈希输入
* ``deduped``            正确的带桶键行已存在 → 删掉旧的平键行（去重,不是丢数据）
* ``not_reconstructible`` 重算不出前像的行 → **跳过**。``> 0`` 是预期且无害:
                         这些行的 scope 已被硬删,因此不可刷新,但也永不被重新
                         摄取（该 scope 不再是摄取目标),不构成双计。

前后核对 SQL 见 ``docs/ops/cloud-lifecycle-ops.md`` §18 步骤 5。
"""
from __future__ import annotations

import argparse
import importlib.util
import logging
from pathlib import Path
from typing import Any

from sqlalchemy import create_engine

# 生产 SSM 脚本惯例:先 import app.main 完成 mapper 配置。这个修复走原始
# SQL、不碰 ORM,但保持惯例一致,免得下一个人以为可以省。
import app.main  # noqa: F401
from app.core.config import settings
from app.core.database_tls import database_engine_options

logger = logging.getLogger("repair_cost_fact_bucket_keys")

_REVISION_FILE = (
    Path(__file__).resolve().parent.parent
    / "alembic" / "versions" / "434_cost_facts_bucket_fact_key_rekey.py"
)


def _load_revision_module() -> Any:
    """Import the 434 revision module by path.

    Its filename starts with a digit, so it is not a legal module name and
    cannot be imported normally. Loading the real revision (rather than
    copying its logic here) is the point: a parallel implementation would
    drift, and drift here means mis-keying rows.
    """
    if not _REVISION_FILE.exists():
        raise SystemExit(f"revision file not found: {_REVISION_FILE}")
    spec = importlib.util.spec_from_file_location(
        "_alembic_rev_434", _REVISION_FILE,
    )
    if spec is None or spec.loader is None:
        raise SystemExit(f"cannot load revision module: {_REVISION_FILE}")
    mod = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(mod)
    if not hasattr(mod, "repair_bucket_fact_keys"):
        raise SystemExit(
            "revision 434 has no repair_bucket_fact_keys — the migration was "
            "restructured; re-read it before running this script.",
        )
    return mod


def main() -> None:
    ap = argparse.ArgumentParser(
        description="Run migration 434's re-key repair out of band.",
    )
    g = ap.add_mutually_exclusive_group(required=True)
    g.add_argument(
        "--apply", action="store_true",
        help="commit per batch (interruptible; re-run to finish)",
    )
    g.add_argument(
        "--dry-run", action="store_true",
        help="roll everything back at the end — reports the counts a real "
             "run would produce without keeping any write",
    )
    args = ap.parse_args()
    logging.basicConfig(
        level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s",
    )

    mod = _load_revision_module()
    # 同步引擎:``repair_bucket_fact_keys`` 是 alembic 迁移里的函数,拿的是
    # 同步 ``Connection``(``app.core.database`` 只导出异步引擎)。用与
    # alembic/env.py 同一个 ``DATABASE_URL_SYNC``,所以离线跑与部署里跑打的
    # 是同一个库。
    db_url, connect_args = database_engine_options(settings.DATABASE_URL_SYNC)
    engine = create_engine(
        db_url, pool_pre_ping=True, connect_args=connect_args,
    )

    if args.apply:
        # commit=True → the migration's own per-batch commits. An abort
        # leaves the table partially repaired but CONSISTENT; re-running
        # finishes it (already-rewritten rows recompute to a mismatch and
        # are skipped).
        with engine.connect() as conn:
            counts = mod.repair_bucket_fact_keys(conn, commit=True)
    else:
        # ``commit=False`` keeps every write in ONE transaction, which we then
        # roll back. Deliberately ``connect()`` + explicit rollback, NOT
        # ``begin()``: ``engine.begin()``'s context manager commits on exit,
        # so a rollback inside it is self-contradictory — the block would end
        # by committing a transaction we had just discarded.
        #
        # Trade-off: one transaction forfeits the per-batch bound, so a dry run
        # over a large write set can hit the TiDB transaction-size ceiling.
        # That failure means "the real run has work to do", not a data problem
        # — go straight to ``--apply``, which is the bounded path.
        with engine.connect() as conn:
            try:
                counts = mod.repair_bucket_fact_keys(conn, commit=False)
            finally:
                conn.rollback()

    mode = "APPLIED" if args.apply else "DRY-RUN (rolled back)"
    logger.info(
        "%s — rekeyed=%d scope_recovered=%d deduped=%d not_reconstructible=%d",
        mode,
        counts.get("rekeyed", 0),
        counts.get("recovered", 0),
        counts.get("deduped", 0),
        counts.get("skipped", 0),
    )
    if counts.get("skipped", 0):
        logger.info(
            "not_reconstructible > 0 is EXPECTED and benign: those rows' "
            "scopes were hard-deleted, so they are un-refreshable but are "
            "never re-ingested either — no double-count. No action needed.",
        )


if __name__ == "__main__":
    main()