cma-management/backend/scripts/kpi_customer_profitability_deriver.py

#!/usr/bin/env python3
"""CPA 客户获利性派生器 —— 客户级收入快照 + 汇总口径 KPI 值（任务 finance-20261003-cpa-... P0-a）

产物
----
1) 明细表 `customer_profitability_snapshot`（每客户每期一行，幂等 upsert；见
   `scripts/migrate_cpa_snapshot_20261004.py`）
2) `kpi_values` 汇总口径（**只存汇总，避免 284 客户 × 期数污染 KPI 宽表**，任务 §2.1）：
   * `C_CUST_REVENUE`       期内客户收入合计（万元）
   * `C_CUST_COUNT`         期内活跃客户数（个）
   * `C_CUST_CONCENTRATION` **补值**：仅对**库内尚无该期行**的期次写入（现网定义 = 前5大客户
                            收入 ÷ 期内合计 ×100；已有 2026-01~09 九期**原样保留、不覆盖**）

fail-closed（本器**不写**以下列/值；理由见 `app/utils/cpa_caliber.py::FAIL_CLOSED_REASONS`）
    `cogs_alloc` / `chan_sub` / `contrib` / `contrib_rate` / `C_CUST_COGS_ALLOC` /
    `C_CUST_CHAN_SUB` / `C_CUST_CONTRIB` / `C_CUST_CONTRIB_RATE` / 客户平均贡献值
    ⇒ 一律不落库（禁 0、禁分摊值），只在快照行 `fail_closed_reason` 留痕。

口径单一事实源
    客户收入聚合 = `app/utils/cpa_caliber.py`（与 `alert_rules._customer_revenue_shares` 同一实现）。

用法
----
    python3 scripts/kpi_customer_profitability_deriver.py --check     # 只读预检 + 对拍锚点（默认）
    python3 scripts/kpi_customer_profitability_deriver.py --apply     # 单事务幂等写
    python3 scripts/kpi_customer_profitability_deriver.py --verify    # 只读写后回查
    python3 scripts/kpi_customer_profitability_deriver.py --anchor    # 只跑对拍锚点断言

选项
    --entity N       账套（默认 1；本单主体范围 = 陕西酣客 e1）
    --db-url URL     显式连接串（离线演练 `sqlite:///<scratch>.db`）
    --since/--until  仅处理该区间（YYYY-MM）
    --batch NAME     source_batch 名（默认 cpa-20261004）

退出码
    0 成功 ｜ 1 fail-closed 前置不满足 / 校验失败 ｜ 2 用法错误
"""
import argparse
import datetime as _dt
import json
import os
import sys
from urllib.parse import quote_plus

HERE = os.path.dirname(os.path.abspath(__file__))
BACKEND = os.path.dirname(HERE)
if BACKEND not in sys.path:
    sys.path.insert(0, BACKEND)

DEFAULT_BATCH = "cpa-20261004"
SNAPSHOT_TABLE = "customer_profitability_snapshot"
FAIL_CLOSED_NOTE = (
    "cogs_alloc/chan_sub/contrib/contrib_rate = P1 fail-closed（未裁定口径/未打标），禁静默 0 或分摊值"
)

ANCHOR_2026_01_09 = {"units": 73, "total_wan": 4883.52, "top1": 25.3835, "top2": 23.4905,
                     "top4_cum": 63.3249, "top10_cum": 81.7675}
ANCHOR_TOL = 0.1


def _engine(args):
    from sqlalchemy import create_engine
    if args.db_url:
        url = args.db_url
    else:
        try:
            from dotenv import load_dotenv
            load_dotenv(os.path.join(BACKEND, ".env"), override=False)
        except Exception:
            pass
        user = os.environ.get("CMA_DB_USER", "cma_api")
        pwd = os.environ.get("CMA_DB_PASS", "")
        host = os.environ.get("CMA_DB_HOST", "127.0.0.1")
        port = os.environ.get("CMA_DB_PORT", "3306")
        name = os.environ.get("CMA_DB_NAME", "cma")
        url = f"mysql+pymysql://{user}:{quote_plus(pwd)}@{host}:{port}/{name}?charset=utf8mb4"
    kw = {"connect_args": {"check_same_thread": False}} if url.startswith("sqlite") else {}
    return create_engine(url, **kw), url


def _dialect(url):
    return "sqlite" if url.startswith("sqlite") else "mysql"


def _table_exists(conn, dialect):
    from sqlalchemy import text
    if dialect == "sqlite":
        return bool(conn.execute(text("SELECT name FROM sqlite_master WHERE type='table' AND name=:t"),
                                 {"t": SNAPSHOT_TABLE}).fetchall())
    return bool(conn.execute(text("SELECT TABLE_NAME FROM information_schema.TABLES "
                                  "WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME=:t"),
                             {"t": SNAPSHOT_TABLE}).fetchall())


def _grants_ok(conn, dialect, table):
    """写前置：应用账号须对 kpi_values 与目标表有 DML。"""
    from sqlalchemy import text
    if dialect == "sqlite":
        return True, "sqlite(离线演练，无授权面)"
    rows = conn.execute(text("SHOW GRANTS FOR CURRENT_USER()")).fetchall()
    joined = "\n".join(r[0] for r in rows)
    has_kv = ("kpi_values" in joined) or ("ON `cma`.*" in joined) or ("ON *.*" in joined)
    has_new = (f"`{table}`" in joined) or ("ON `cma`.*" in joined) or ("ON *.*" in joined)
    return (has_kv and has_new), joined


def _load_caliber():
    from app.utils import cpa_caliber
    return cpa_caliber


def compute(engine, dialect, entity_id):
    """只读计算：{period: {unit_id: amount}} + 汇总。"""
    cal = _load_caliber()
    monthly = cal.fetch_monthly_revenue(engine, entity_id)
    summary = {}
    for p, pairs in monthly.items():
        m = cal.monthly_metrics(pairs)
        m["_total_yuan"] = sum(a for _, a in pairs)
        signature = _rank_signature(pairs) if len(pairs) > 10 else None
        m["_signature"] = signature
        summary[p] = m
    return monthly, summary


def _rank_signature(pairs):
    """前10名 (unit_id, 分) 指纹（防静默漂移）。"""
    return [(u, round(a, 2)) for u, a in pairs[:10]]


def anchor_check(summary):
    """2026-01~09 对拍锚点（与任务验收标准逐项比）。"""
    pairs = []
    for p in sorted(summary):
        if "2026-01" <= p <= "2026-09":
            pairs.append(p)
    # 需要按窗口重算 topN（不能逐月相加），故单独取窗口
    return pairs


def run(engine, dialect, entity_id, batch, apply=False, verify=False, since=None, until=None):
    cal = _load_caliber()
    from sqlalchemy import text
    result = {"entity_id": entity_id, "batch": batch, "apply": apply, "verify": verify}

    with engine.connect() as conn:
        exists = _table_exists(conn, dialect)
        if apply and not exists:
            raise SystemExit("FAIL-CLOSED: 目标表不存在，请先跑 migrate_cpa_snapshot_20261004.py --apply")
        rows_before = None
        if exists:
            rows_before = conn.execute(text(f"SELECT COUNT(*) FROM {SNAPSHOT_TABLE} WHERE entity_id=:e"),
                                       {"e": entity_id}).scalar()
        kpi_ids = {}
        for code in ("C_CUST_REVENUE", "C_CUST_COUNT", "C_CUST_CONCENTRATION"):
            r = conn.execute(text("SELECT id FROM kpi_definitions WHERE entity_id=:e AND kpi_code=:c"),
                             {"e": entity_id, "c": code}).fetchone()
            kpi_ids[code] = r[0] if r else None
        existing_conc_periods = set()
        if kpi_ids["C_CUST_CONCENTRATION"]:
            for (p,) in conn.execute(text("SELECT period FROM kpi_values WHERE entity_id=:e AND kpi_id=:k"),
                                     {"e": entity_id, "k": kpi_ids["C_CUST_CONCENTRATION"]}).fetchall():
                existing_conc_periods.add(p)

    monthly = cal.fetch_monthly_revenue(engine, entity_id)
    periods = sorted(p for p in monthly if (not since or p >= since) and (not until or p <= until))
    result["periods"] = len(periods)
    result["period_range"] = [periods[0], periods[-1]] if periods else None
    result["kpi_ids"] = kpi_ids
    result["existing_concentration_periods"] = sorted(existing_conc_periods)

    if verify:
        return _verify(engine, dialect, entity_id, monthly, periods, kpi_ids, existing_conc_periods, now_ts=None)

    snap_rows = []
    kpi_rows = []
    for p in periods:
        pairs = monthly[p]
        total_yuan = sum(a for _, a in pairs)
        for idx, (uid, amount) in enumerate(pairs, start=1):
            snap_rows.append({
                "entity_id": entity_id, "period": p, "unit_id": uid,
                "revenue": cal.to_wan(amount), "rank_no": idx,
                "calib_status": "revenue_only", "fail_closed_reason": FAIL_CLOSED_NOTE,
            })
        kpi_rows.append(("C_CUST_REVENUE", p, cal.to_wan(total_yuan)))
        kpi_rows.append(("C_CUST_COUNT", p, float(cal.active_customer_count(pairs))))
        if p not in existing_conc_periods:
            kpi_rows.append(("C_CUST_CONCENTRATION", p, cal.concentration_pct(pairs, 5)))
    result["snapshot_rows"] = len(snap_rows)
    result["kpi_rows"] = len(kpi_rows)
    result["kpi_rows_skipped_no_definition"] = [c for c, i in kpi_ids.items() if i is None]

    if not apply:
        result["preview"] = {
            "first_periods": {p: {k: v for k, v in cal.monthly_metrics(monthly[p]).items()} for p in periods[:2]},
            "last_periods": {p: {k: v for k, v in cal.monthly_metrics(monthly[p]).items()} for p in periods[-2:]},
        }
        return result

    # ---- 写路径（单事务） ----
    with engine.connect() as conn:
        ok, grants = _grants_ok(conn, dialect, SNAPSHOT_TABLE)
        result["grants_ok"] = ok
        if not ok:
            raise SystemExit("FAIL-CLOSED: 应用账号缺 DML 授权（kpi_values / 目标表）。原始 SHOW GRANTS:\n" + grants)
        missing = [c for c, i in kpi_ids.items() if i is None]
        if missing:
            raise SystemExit("FAIL-CLOSED: KPI 定义缺失 %s —— 请先执行 scripts/sql/cpa_kpi_defs_20261004.sql" % missing)

    now_ts = _dt.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    inserted_snap = updated_snap = 0
    kpi_inserted = kpi_updated = 0
    with engine.begin() as conn:
        for r in snap_rows:
            row = conn.execute(text(
                f"SELECT id FROM {SNAPSHOT_TABLE} WHERE entity_id=:entity_id AND period=:period AND unit_id=:unit_id"),
                r).fetchone()
            if row:
                conn.execute(text(
                    f"UPDATE {SNAPSHOT_TABLE} SET revenue=:revenue, rank_no=:rank_no, "
                    "calib_status=:calib_status, fail_closed_reason=:fail_closed_reason, updated_at=:ts "
                    "WHERE id=:id"), {**r, "ts": now_ts, "id": row[0]})
                updated_snap += 1
            else:
                conn.execute(text(
                    f"INSERT INTO {SNAPSHOT_TABLE} (entity_id, period, unit_id, revenue, rank_no, calib_status, "
                    "fail_closed_reason, created_at, updated_at) VALUES (:entity_id, :period, :unit_id, :revenue, "
                    ":rank_no, :calib_status, :fail_closed_reason, :ts, :ts)"), {**r, "ts": now_ts})
                inserted_snap += 1
        for code, period, value in kpi_rows:
            kid = kpi_ids[code]
            if value is None:
                continue  # fail-closed：不给值（不写 NULL 新行，避免污染宽表）
            row = conn.execute(text(
                "SELECT id, actual_value, data_status FROM kpi_values "
                "WHERE entity_id=:e AND kpi_id=:k AND period=:p ORDER BY id DESC LIMIT 1"),
                {"e": entity_id, "k": kid, "p": period}).fetchone()
            remark = {"C_CUST_REVENUE": "口径=Σzxt_masterbill.sum_money（期内、unit_id>0、含税净额>0 取舍）",
                      "C_CUST_COUNT": "口径=期内 zxt_masterbill 活跃往来单位数（unit_id 去重）",
                      "C_CUST_CONCENTRATION": "口径=前5大客户收入/期内客户收入合计×100%（补值期次；"
                                              "2026-01~09 现网值原样保留）"}[code]
            if row:
                # 幂等：值不变则零写入；不降级 data_status
                if abs(float(row[1] or 0) - float(value)) < 1e-9 and row[2]:
                    continue
                conn.execute(text("UPDATE kpi_values SET actual_value=:v, source_type='derived', source_batch=:b, "
                                  "remark=:r WHERE id=:id"),
                             {"v": value, "b": batch, "r": remark, "id": row[0]})
                kpi_updated += 1
            else:
                conn.execute(text(
                    "INSERT INTO kpi_values (entity_id, kpi_id, period, actual_value, source_type, source_batch, "
                    "data_status, calculated_at, remark) VALUES (:e, :k, :p, :v, 'derived', :b, 'pending', :ts, :r)"),
                    {"e": entity_id, "k": kid, "p": period, "v": value, "b": batch, "ts": now_ts, "r": remark})
                kpi_inserted += 1
    result.update(snapshot_inserted=inserted_snap, snapshot_updated=updated_snap,
                  kpi_inserted=kpi_inserted, kpi_updated=kpi_updated, ts=now_ts)
    return result


def _verify(engine, dialect, entity_id, monthly, periods, kpi_ids, existing_conc_periods, now_ts=None):
    """只读写后回查：快照行数/值锚点 + kpi_values 三族。"""
    from sqlalchemy import text
    cal = _load_caliber()
    out = {"verify": True, "entity_id": entity_id}
    with engine.connect() as conn:
        out["snapshot_rows"] = conn.execute(text(
            f"SELECT COUNT(*) FROM {SNAPSHOT_TABLE} WHERE entity_id=:e"), {"e": entity_id}).scalar()
        out["snapshot_periods"] = conn.execute(text(
            f"SELECT COUNT(DISTINCT period) FROM {SNAPSHOT_TABLE} WHERE entity_id=:e"), {"e": entity_id}).scalar()
        # 值锚点：2026-01~09 快照合计 == 4,883.52 万
        r = conn.execute(text(
            f"SELECT ROUND(SUM(revenue),2), COUNT(DISTINCT unit_id) FROM {SNAPSHOT_TABLE} "
            "WHERE entity_id=:e AND period>='2026-01' AND period<='2026-09'"), {"e": entity_id}).fetchone()
        out["anchor_2026_01_09_snapshot"] = {"total_wan": float(r[0] or 0), "units": int(r[1] or 0)}
        out["fail_closed_columns_all_null"] = conn.execute(text(
            f"SELECT COUNT(*) FROM {SNAPSHOT_TABLE} WHERE entity_id=:e AND "
            "(cogs_alloc IS NOT NULL OR chan_sub IS NOT NULL OR contrib IS NOT NULL OR contrib_rate IS NOT NULL)"),
            {"e": entity_id}).scalar()
        # kpi_values 回查
        kv = {}
        for code in ("C_CUST_REVENUE", "C_CUST_COUNT", "C_CUST_CONCENTRATION"):
            kid = kpi_ids.get(code)
            if not kid:
                kv[code] = None
                continue
            rows = conn.execute(text("SELECT period, actual_value, data_status FROM kpi_values "
                                     "WHERE entity_id=:e AND kpi_id=:k ORDER BY period"),
                                {"e": entity_id, "k": kid}).fetchall()
            kv[code] = {"n": len(rows), "sample": [[r[0], float(r[1]) if r[1] is not None else None] for r in rows[:3]],
                        "last": [[r[0], float(r[1]) if r[1] is not None else None] for r in rows[-2:]]}
        out["kpi_values"] = kv
        # 期次覆盖：快照期次 == 有数据期次
        out["periods_with_data"] = len(periods)
        out["concentration_untouched_note"] = (
            "现网 2026-01~09 九期在写前已存在 ⇒ 本器不覆盖（见 existing_concentration_periods）")
        out["existing_concentration_periods"] = sorted(existing_conc_periods)
    return out


def anchor_assert(engine, entity_id):
    """2026-01~09 窗口对拍锚点（任务验收标准逐项）。"""
    from sqlalchemy import text
    cal = _load_caliber()
    with engine.connect() as conn:
        rows = conn.execute(text(
            "SELECT unit_id, SUM(sum_money) AS amount FROM zxt_masterbill WHERE entity_id=:e "
            "AND unit_id IS NOT NULL AND unit_id>0 AND (sum_money + sum_tax_money) > 0 "
            "AND bill_date>='2026-01-01' AND bill_date<='2026-09-30' GROUP BY unit_id ORDER BY amount DESC"),
            {"e": entity_id}).fetchall()
    pairs = [(int(r[0]), float(r[1])) for r in rows]
    total = sum(a for _, a in pairs)
    cum = lambda n: round(sum(a for _, a in pairs[:n]) / total * 100, 4) if total else None
    got = {"units": len(pairs), "total_wan": round(total / 10000, 2),
           "top1": round(pairs[0][1] / total * 100, 4) if pairs else None,
           "top2": round(pairs[1][1] / total * 100, 4) if len(pairs) > 1 else None,
           "top4_cum": cum(4), "top10_cum": cum(10)}
    checks = {}
    for k, exp in ANCHOR_2026_01_09.items():
        v = got.get(k)
        checks[k] = {"expected": exp, "got": v, "pass": (v is not None and abs(float(v) - exp) <= ANCHOR_TOL)}
    return {"anchor": got, "checks": checks, "ALL_PASS": all(c["pass"] for c in checks.values())}


def main():
    ap = argparse.ArgumentParser()
    g = ap.add_mutually_exclusive_group()
    g.add_argument("--check", action="store_true")
    g.add_argument("--apply", action="store_true")
    g.add_argument("--verify", action="store_true")
    g.add_argument("--anchor", action="store_true")
    ap.add_argument("--entity", type=int, default=1)
    ap.add_argument("--db-url")
    ap.add_argument("--batch", default=DEFAULT_BATCH)
    ap.add_argument("--since")
    ap.add_argument("--until")
    ap.add_argument("--json", action="store_true")
    args = ap.parse_args()

    engine, url = _engine(args)
    dialect = _dialect(url)

    if args.anchor:
        res = anchor_assert(engine, args.entity)
        print(json.dumps(res, ensure_ascii=False, indent=2))
        return 0 if res["ALL_PASS"] else 1

    res = run(engine, dialect, args.entity, args.batch, apply=args.apply,
              verify=args.verify, since=args.since, until=args.until)
    print(json.dumps(res, ensure_ascii=False, indent=2))
    return 0


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