cfc/docs/scripts/legacy_dimension_zero_call_check.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""旧体系接口连续 N 天零调用巡检（统一维度体系 P2-T3 退役证据判定）。

对应设计文档 2026-11-03-unified-dimension-system-design.md §8.3 退役触发条件
「旧接口连续 14 天零调用 + 双写一致性 0 差异」。本脚本只判定「零调用」这一半证据。

用法（Git Bash / 任意 Python3 环境）：
    python3 legacy_dimension_zero_call_check.py --days 14 \
        --host 127.0.0.1 --port 3306 --db cfc_test --user root --password xxx
    或通过环境变量传入连接参数（无需命令行硬编码生产连接串）：
        DB_HOST / DB_PORT / DB_NAME / DB_USER / DB_PASSWORD

判定：
    PASS  - 近 N 天每天都有 __heartbeat__ 心跳 且 业务路径调用总数=0
           → 连续 N 天零调用，可退役（输出零调用观察窗口起止日期）
    FAIL  - 有心跳但业务调用>0 → 未满足退役条件，列出剩余调用路径与日期
    WARN  - 无心跳（或心跳未覆盖整个观察窗口）→ 采集器未运行 / flag 未开启，
            检查 DIM_REGISTRY_LEGACY_CALL_LOG_ENABLED 是否置 true

依赖：pymysql（pip install pymysql）。采集表 legacy_dimension_call_log 由后端
DatabaseInitializer 迁移454 创建，__heartbeat__ 心跳行证明采集器当天存活，
防止「表空 = 假零调用」误判。
"""

import argparse
import datetime
import os
import sys

HEARTBEAT = "__heartbeat__"


def parse_args():
    env = os.environ
    p = argparse.ArgumentParser(
        description="旧体系接口连续 N 天零调用巡检（P2-T3 退役证据判定）")
    p.add_argument("--days", type=int, default=14,
                   help="观察窗口天数（默认 14）")
    p.add_argument("--host", default=env.get("DB_HOST", "127.0.0.1"),
                   help="数据库主机（默认 env DB_HOST / 127.0.0.1）")
    p.add_argument("--port", type=int, default=int(env.get("DB_PORT", "3306")),
                   help="数据库端口（默认 env DB_PORT / 3306）")
    p.add_argument("--db", default=env.get("DB_NAME", "cfc"),
                   help="数据库名（默认 env DB_NAME / cfc）")
    p.add_argument("--user", default=env.get("DB_USER", "root"),
                   help="数据库用户（默认 env DB_USER / root）")
    p.add_argument("--password", default=env.get("DB_PASSWORD", ""),
                   help="数据库密码（默认 env DB_PASSWORD / 空）")
    return p.parse_args()


def connect(params):
    try:
        import pymysql
    except ImportError:
        sys.exit("缺少 pymysql 依赖，请先执行：pip install pymysql")
    conn = pymysql.connect(
        host=params.host, port=params.port, user=params.user,
        password=params.password, database=params.db, charset="utf8mb4",
        cursorclass=pymysql.cursors.DictCursor, autocommit=True)
    return conn


def load_rows(conn, days):
    """返回近 N 天 (period, api_path, call_count) 列表（period 为 ISO 字符串）。"""
    with conn.cursor() as cur:
        cur.execute(
            "SELECT api_path, period, call_count FROM legacy_dimension_call_log "
            "WHERE period >= DATE_SUB(CURDATE(), INTERVAL %s DAY) "
            "ORDER BY period ASC, api_path ASC",
            (days,))
        rows = cur.fetchall()
    out = []
    for r in rows:
        out.append({
            "period": r["period"].isoformat() if hasattr(r["period"], "isoformat") else str(r["period"]),
            "api_path": r["api_path"],
            "call_count": int(r["call_count"] or 0),
        })
    return out


def window_dates(days, today=None):
    """观察窗口日期（升序，含今天），返回 ISO 字符串列表，长度=days。"""
    if today is None:
        today = datetime.date.today()
    return [(today - datetime.timedelta(days=days - 1 - i)).isoformat()
            for i in range(days)]


def judge(rows, days, today=None):
    """纯函数判定（无 DB），返回 (verdict, detail dict)。

    detail:
      verdict       PASS / FAIL / WARN
      window_dates  [iso...]
      heartbeat_dates  {iso: True}
      business_by_date {iso: {path: count}}
      business_total  int
      missing_heartbeat [iso...]  窗口内无心跳的日期
      zero_start / zero_end  连续零调用起止（PASS 时 = 窗口起止）
      remaining  [(path, iso, count)]  FAIL 时剩余调用
    """
    dates = window_dates(days, today)
    heartbeat_dates = {r["period"] for r in rows if r["api_path"] == HEARTBEAT}
    business_by_date = {}
    business_total = 0
    remaining = []
    for r in rows:
        if r["api_path"] == HEARTBEAT:
            continue
        business_by_date.setdefault(r["period"], {})
        business_by_date[r["period"]][r["api_path"]] = (
            business_by_date[r["period"]].get(r["api_path"], 0) + r["call_count"])
        business_total += r["call_count"]
        remaining.append((r["api_path"], r["period"], r["call_count"]))

    missing_heartbeat = [d for d in dates if d not in heartbeat_dates]
    detail = {
        "verdict": None,
        "window_dates": dates,
        "heartbeat_dates": heartbeat_dates,
        "business_by_date": business_by_date,
        "business_total": business_total,
        "missing_heartbeat": missing_heartbeat,
        "zero_start": dates[0] if dates else None,
        "zero_end": dates[-1] if dates else None,
        "remaining": remaining,
    }

    if missing_heartbeat:
        # 心跳未覆盖整个窗口 → 无法证明「采集器全程活着」，不轻下 PASS
        detail["verdict"] = "WARN"
    elif business_total == 0:
        detail["verdict"] = "PASS"
    else:
        detail["verdict"] = "FAIL"
    return detail["verdict"], detail


def render_matrix(detail):
    """输出 日期×接口路径 调用矩阵（表格文本）。"""
    lines = []
    lines.append("日期         | 心跳 | 业务调用路径（当日次数）")
    lines.append("-" * 72)
    for d in detail["window_dates"][::-1]:  # 最新在上
        hb = "有" if d in detail["heartbeat_dates"] else "无"
        biz = detail["business_by_date"].get(d, {})
        if biz:
            parts = ", ".join("%s×%d" % (p, c) for p, c in sorted(biz.items()))
        else:
            parts = "-"
        lines.append("%s |  %s  | %s" % (d, hb, parts))
    return "\n".join(lines)


def render_verdict(verdict, detail):
    if verdict == "PASS":
        return (
            "判定：PASS —— 连续 %d 天零调用，可退役\n"
            "零调用观察窗口：%s 至 %s（窗口内每天有心跳，业务调用总数=0）"
            % (len(detail["window_dates"]), detail["zero_start"], detail["zero_end"]))
    if verdict == "FAIL":
        rem = "\n".join("  %s  %s  x%d" % (p, d, c)
                        for p, d, c in detail["remaining"])
        return (
            "判定：FAIL —— 仍有业务调用，未满足退役条件\n"
            "剩余调用路径与日期：\n%s" % rem)
    return (
        "判定：WARN —— 采集证据不足，不能判定可退役\n"
        "缺失心跳日期（%d 天）：%s\n"
        "检查项：\n"
        "  1) 采集开关 dimension.registry.legacyCallLog.enabled "
        "（环境变量 DIM_REGISTRY_LEGACY_CALL_LOG_ENABLED）是否置 true；\n"
        "  2) 采集服务 LegacyDimensionCallLogService 每小时心跳是否正常写入；\n"
        "  3) 若表完全为空：采集器未运行 / 采集开关未开启（默认 false 零开销）。"
        % (len(detail["missing_heartbeat"]), ", ".join(detail["missing_heartbeat"])))


def main():
    args = parse_args()
    conn = connect(args)
    try:
        rows = load_rows(conn, args.days)
    finally:
        conn.close()

    verdict, detail = judge(rows, args.days)
    print("=== 旧体系接口调用矩阵（近 %d 天）===" % args.days)
    print(render_matrix(detail))
    print()
    print(render_verdict(verdict, detail))
    return 0


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