group_fqcd_jr/tools/backfill_product_fees.py

"""把真实费率回填到 `fin_product`（默认 dry-run，只有 `--apply` 才写库）。

## 为什么需要它

`fin_product.management_fee_rate` / `custodian_fee_rate` 在本机 **20 只里 19 只为空**，
客户问"管理费多少"只能转人工。费率可以从东方财富 f10 基金概况页取到
（`EastmoneyFundAdapter.fetch_fees`），本脚本负责把它落到库里。

## 安全设计

1. **默认 dry-run**：只打印将执行的 UPDATE，不写库。要真正写入必须显式给 `--apply`。
2. **默认只补空值**：`NULL` 才写，**已有值一律不动** —— 因为已有值可能是人为设定
   （本机 `510300` 就是：库里 0.5000/0.1000，真实 0.15/0.05）。要覆盖必须再给 `--overwrite`。
3. **`--overwrite` 也只改这一只这一类**：它不会顺手把别的东西一起改掉。
4. 写入走单个事务，任一失败整体回滚。

## 用法

    python tools/backfill_product_fees.py                 # 看将要改什么
    python tools/backfill_product_fees.py --apply         # 只补空值
    python tools/backfill_product_fees.py --apply --overwrite   # 连已有值一起改成真实值
"""

from __future__ import annotations

import argparse
import asyncio
import sys
from datetime import UTC, datetime
from pathlib import Path

from sqlalchemy import text

PROJECT_ROOT = Path(__file__).resolve().parents[1]
if str(PROJECT_ROOT) not in sys.path:
    sys.path.insert(0, str(PROJECT_ROOT))

from app.infrastructure.db import SessionFactory  # noqa: E402
from app.infrastructure.fund_market_adapter import EastmoneyFundAdapter  # noqa: E402

if hasattr(sys.stdout, "reconfigure"):
    sys.stdout.reconfigure(errors="replace")  # type: ignore[union-attr]

#: 与 `MarketQuoteSyncService.sync` 同一批产品。
PRODUCT_QUERY = """
    SELECT id, product_code, product_name, management_fee_rate, custodian_fee_rate
    FROM fin_product
    WHERE fund_manager = '南方基金'
      AND exchange_code IN ('SSE', 'SZSE')
      AND status = '上市'
    ORDER BY product_code
"""

#: 库字段 ↔ 数据源字段。
FEE_FIELDS = (
    ("management_fee_rate", "management_fee_rate"),
    ("custodian_fee_rate", "custodian_fee_rate"),
)


def _same(left: object, right: object) -> bool:
    if left is None or right is None:
        return False
    try:
        return abs(float(str(left)) - float(str(right))) < 1e-9
    except (TypeError, ValueError):
        return str(left) == str(right)


async def main() -> int:
    parser = argparse.ArgumentParser(description="把真实费率回填到 fin_product")
    parser.add_argument("--apply", action="store_true", help="真正写库（默认只看不改）")
    parser.add_argument(
        "--overwrite", action="store_true",
        help="连已有值也改（默认只补 NULL，不动人为设定过的值）",
    )
    args = parser.parse_args()

    async with SessionFactory() as session:
        rows = (await session.execute(text(PRODUCT_QUERY))).mappings().all()

    adapter = EastmoneyFundAdapter()
    fees = await adapter.fetch_fees([str(row["product_code"]) for row in rows])
    print(f"产品 {len(rows)} 只；取到真实费率 {len(fees)} 只\n")

    updates: list[tuple[int, str, str, object]] = []
    conflicts: list[str] = []
    for row in rows:
        code = str(row["product_code"])
        live = fees.get(code)
        if not live:
            continue
        for column, key in FEE_FIELDS:
            live_value = live.get(key)
            if live_value is None:
                continue
            stored = row[column]
            if stored is not None and not _same(stored, live_value):
                conflicts.append(
                    f"{code} {row['product_name']} 的 {column}：库里 {stored} → 真实 {live_value}"
                )
                if not args.overwrite:
                    continue
            if stored is not None and not args.overwrite:
                continue
            if _same(stored, live_value):
                continue
            updates.append((int(row["id"]), code, column, live_value))

    if conflicts:
        print("与库里已有值冲突（默认不动，需要 --overwrite 才覆盖）：")
        for line in conflicts:
            print(f"  * {line}")
        print()

    if not updates:
        print("没有需要回填的字段。")
        return 0

    print(f"将执行 {len(updates)} 条 UPDATE：")
    for _, code, column, value in updates:
        print(f"  · {code}  {column} = {value}")

    if not args.apply:
        print("\n[dry-run] 未写库。加 --apply 执行。")
        return 0

    now = datetime.now(UTC).replace(tzinfo=None)
    async with SessionFactory() as session, session.begin():
        for product_id, _, column, value in updates:
            # 列名来自本模块常量，不是外部输入；值走参数绑定。
            await session.execute(
                text(
                    f"UPDATE fin_product SET {column} = :value, updated_at = :now "
                    "WHERE id = :product_id"
                ),
                {"value": value, "now": now, "product_id": product_id},
            )
    print(f"\n已写入 {len(updates)} 个字段。")
    return 0


if __name__ == "__main__":
    sys.exit(asyncio.run(main()))