group_xinghuo_jinrong/scripts/core/rebuild_lots.py

#!/usr/bin/env python3
"""按 `core_holding` 快照重建份额批次（D18 · L-7 · 架构 §2 scripts/core）。

**定位**：`core_share_lot` 是 FIFO 计费的权威源，`core_holding` 是持仓快照，
两者靠交易统一出入口（`trade_gateway` 的 FR-C16 批次维护）保持一致。
当历史数据缺批次（批次机制上线前的存量、或演示库被人工改动）时，用本脚本
**按持仓快照重建**——这是**快照重建、不是交易回滚**（PRD L-7）：不重放流水，
只让 `Σ core_share_lot.remain_qty` 与 `core_holding.qty` 重新对齐。

**⚠️ D18 单一副本（改本文件前必读）**
补建规则**只有一处** —— `service/convert/lot_bootstrap.bootstrap_lots`。
本脚本与 `trade_gateway` 的 D8 兜底补建**同调该函数**。任何情况下都不要在
本文件里另写一份"由 `as_of` 反推 `confirmed_at`"的逻辑：两处各写一遍必然
漂移，同一持仓在两侧算出不同费率档 —— **演示里看不出来、生产里就是错账**。

**权限**：走 `role="admin"`（= `mysql_user`）。`--force` 需要 DELETE 权限，
而业务账号 `xh_core_rw` 刻意不持有（D20 最小权限），故本运维脚本不用业务账号。

用法：
    python scripts/core/rebuild_lots.py                  # 仅补建「无批次」的持仓行（安全、幂等）
    python scripts/core/rebuild_lots.py --customer CUST-4001
    python scripts/core/rebuild_lots.py --force          # 先删后建：全量快照重建
    python scripts/core/rebuild_lots.py --dry-run        # 只报告、不写库

退出码：0 = 正常；1 = 库不可达。
"""

from __future__ import annotations

import argparse
import sys
from pathlib import Path
from typing import Any

from sqlalchemy import text
from sqlalchemy.engine import Engine

ROOT = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(ROOT))

from app.config.settings import settings  # noqa: E402
from app.gateway.gateway_repository import GatewayRepository  # noqa: E402
from app.service.convert.lot_bootstrap import bootstrap_lots  # noqa: E402
from app.utils.db import get_engine  # noqa: E402

_SELECT_HOLDINGS = "SELECT customer_id, product_id, qty, cost_amount, as_of FROM core_holding"

#: 一次取回全部已有批次的 `(客户, 产品)` 分布，避免逐行回查。
_GROUP_EXISTING_LOTS = """
    SELECT customer_id, product_id, COUNT(*) AS n
      FROM core_share_lot
     GROUP BY customer_id, product_id
"""

_DELETE_LOTS = "DELETE FROM core_share_lot WHERE customer_id = :cid AND product_id = :pid"


def rebuild_lots(
    engine: Engine,
    *,
    customer_id: str | None = None,
    force: bool = False,
    dry_run: bool = False,
) -> dict[str, int]:
    """按 `core_holding` 重建批次（CLI 与单测共用入口）。

    - 缺省：**只补建「无批次」的持仓行**（安全、幂等；重复执行零写入）。
    - `--force`：**先删后建**该 `(客户, 产品)` 的全部批次（全量快照重建）。
    - `qty <= 0` 的持仓行**不补建**（D9/P2 归零行保留但不建批次，
      与 `bootstrap_lots` 的语义一致）。

    返回统计，键含义：`holdings` 扫描的持仓行数 · `bootstrapped` 实际补建的
    持仓行数 · `skipped_existing` 因已有批次而跳过 · `skipped_zero` 因份额为 0
    跳过 · `removed` 删除的批次行数（仅 `--force`）· `written` 写入的批次行数。

    写入复用 `GatewayRepository.insert_share_lots` —— `core_share_lot` 的 INSERT
    语句全仓只此一份，本脚本不另写一条（自检第 13 问）。
    """
    stats = {
        "holdings": 0,
        "bootstrapped": 0,
        "skipped_existing": 0,
        "skipped_zero": 0,
        "removed": 0,
        "written": 0,
    }

    sql = _SELECT_HOLDINGS
    params: dict[str, Any] = {}
    if customer_id:
        sql += " WHERE customer_id = :cid"
        params["cid"] = customer_id
    sql += " ORDER BY customer_id, product_id"

    plans: list[tuple[str, str, list[Any]]] = []
    with engine.connect() as conn:
        holdings = [dict(r) for r in conn.execute(text(sql), params).mappings()]
        existing = {
            (r["customer_id"], r["product_id"]): int(r["n"])
            for r in conn.execute(text(_GROUP_EXISTING_LOTS)).mappings()
        }

    for row in holdings:
        stats["holdings"] += 1
        cid, pid = str(row["customer_id"]), str(row["product_id"])
        lots = bootstrap_lots(row)
        if not lots:
            stats["skipped_zero"] += 1
            continue
        has = existing.get((cid, pid), 0)
        if has and not force:
            stats["skipped_existing"] += 1
            continue
        if has:
            stats["removed"] += has
        stats["bootstrapped"] += 1
        plans.append((cid, pid, lots))

    if dry_run:
        return stats

    # `--force`：先删（独立事务）。放在写入之前，且不嵌套写事务 ——
    # sqlite 测试库是 StaticPool 单连接，事务内再开事务会自锁。
    if force:
        with engine.begin() as conn:
            for cid, pid, _ in plans:
                conn.execute(text(_DELETE_LOTS), {"cid": cid, "pid": pid})

    writer = GatewayRepository(engine=engine)
    for _, _, lots in plans:
        writer.insert_share_lots(lots)
        stats["written"] += len(lots)
    return stats


def main() -> None:
    parser = argparse.ArgumentParser(
        description="按 core_holding 快照重建份额批次（补建规则与网关同源，D18）"
    )
    parser.add_argument("--customer", help="只处理该客户（缺省全部）")
    parser.add_argument(
        "--force",
        action="store_true",
        help="先删后建（全量快照重建）；缺省仅补建无批次的持仓行",
    )
    parser.add_argument("--dry-run", action="store_true", help="只报告、不写库")
    args = parser.parse_args()

    engine = get_engine(settings.mysql_core_database, "admin")
    try:
        with engine.connect() as conn:
            conn.execute(text("SELECT 1"))
    except Exception as exc:  # noqa: BLE001 - CLI 兜底，给出可操作的提示
        print(f"MySQL 不可达：{exc}", file=sys.stderr)
        print("请确认本机 MySQL 服务已启动（见 FLOW §0 本机状态）。", file=sys.stderr)
        raise SystemExit(1)

    stats = rebuild_lots(
        engine, customer_id=args.customer, force=args.force, dry_run=args.dry_run
    )
    prefix = "[dry-run] " if args.dry_run else ""
    print(
        f"{prefix}持仓行 {stats['holdings']} · 补建 {stats['bootstrapped']} · "
        f"跳过(已有批次) {stats['skipped_existing']} · 跳过(份额为0) {stats['skipped_zero']} · "
        f"删除批次 {stats['removed']} · 写入批次 {stats['written']}"
    )


if __name__ == "__main__":
    main()