|
|
@@ -0,0 +1,198 @@
|
|
|
+#!/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())
|
Agnes Agent