#!/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())