167 lines
6.3 KiB
Python
167 lines
6.3 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
sensor-verify.py — 确定性传感器（2026-08-12 新增）
|
||
====================================================
|
||
借鉴《AI Agent 的自我进化》：能用代码解决的坚决不用模型。
|
||
|
||
功能（全部确定性检查，0 LLM 成本）：
|
||
1. JSON Schema 验证 — 关键输出文件是否可解析且字段完整
|
||
2. 退出码分级检查 — cron 脚本的退出码语义（0=OK, 1=业务异常, 2=离线常态）
|
||
3. 关键脚本 smoke test — 快速运行确认可执行
|
||
|
||
用法:
|
||
sensor-verify.py → 全量检查，exit 0=健康 1=有异常
|
||
sensor-verify.py json → 只检查 JSON schema
|
||
sensor-verify.py smoke → 只跑 smoke test
|
||
sensor-verify.py --quiet → 静默模式（watchdog 用，只输出异常）
|
||
|
||
退出码: 0 = 全部健康, 1 = 有异常（供 cron 报警）
|
||
"""
|
||
import json, os, sys, subprocess, datetime
|
||
|
||
HOME = os.path.expanduser("~")
|
||
HERMES = HOME + "/.hermes"
|
||
STOCK_BT = HERMES + "/stock_backtest"
|
||
|
||
# ====== 1. JSON Schema 验证 ======
|
||
# 格式: 文件路径 → (必需字段列表, 允许空?)
|
||
JSON_CHECKS = {
|
||
HERMES + "/model-health.json": (["stable", "total_models", "timestamp"], True),
|
||
HERMES + "/skill-health.json": (["summary", "skills"], True),
|
||
HERMES + "/optimization-report.json": (["health_score"], True),
|
||
HERMES + "/learner/state.json": (["total_cycles", "learned_items"], True),
|
||
STOCK_BT + "/fundamental_scan.json": ([], True),
|
||
STOCK_BT + "/sentiment_scan.json": ([], True),
|
||
STOCK_BT + "/macro_score.json": ([], True),
|
||
STOCK_BT + "/industry_scan.json": ([], True),
|
||
}
|
||
|
||
# ====== 2. 退出码语义（已规范化的脚本） ======
|
||
# 脚本名 → 期望退出码分级说明（0=正常, 2=离线常态等）
|
||
EXIT_CODE_CONTRACT = {
|
||
"dual-backup.sh": "0=备份完成/离线静默跳过, 1=真异常",
|
||
"memory-governance.py": "0=正常(即使无候选), 1=API异常",
|
||
"stock_daily_health.py": "0=健康, 1=有异常(数据/cron/账户)",
|
||
"distill-model-watchdog.py": "0=正常, 1=模型异常",
|
||
}
|
||
|
||
# ====== 3. Smoke test 脚本（快速运行验证可执行性） ======
|
||
SMOKE_SCRIPTS = [
|
||
("learner.py status", 15),
|
||
("skill-manager.py dashboard --quiet 2>/dev/null || skill-manager.py scan --quiet 2>/dev/null || true", 20),
|
||
("memory-governance.py", 30),
|
||
]
|
||
|
||
|
||
def check_json_schema():
|
||
issues = []
|
||
for path, (required, allow_empty) in JSON_CHECKS.items():
|
||
if not os.path.exists(path):
|
||
continue # 不存在不报（有些文件按需生成）
|
||
try:
|
||
with open(path, encoding="utf-8") as f:
|
||
data = json.load(f)
|
||
if not isinstance(data, (dict, list)):
|
||
issues.append(f"{os.path.basename(path)}: 不是 JSON 对象/数组")
|
||
continue
|
||
if isinstance(data, dict):
|
||
for field in required:
|
||
if field not in data:
|
||
issues.append(f"{os.path.basename(path)}: 缺少字段 '{field}'")
|
||
if not allow_empty and isinstance(data, list) and len(data) == 0:
|
||
issues.append(f"{os.path.basename(path)}: 空数组（可能有数据问题）")
|
||
except json.JSONDecodeError as e:
|
||
issues.append(f"{os.path.basename(path)}: JSON 解析失败 ({e})")
|
||
except Exception as e:
|
||
issues.append(f"{os.path.basename(path)}: 读取失败 ({e})")
|
||
return issues
|
||
|
||
|
||
def check_exit_codes():
|
||
"""检查 cron 最近输出是否出现异常退出（Status: script failed / error）"""
|
||
issues = []
|
||
cron_out = HERMES + "/cron/output"
|
||
if not os.path.isdir(cron_out):
|
||
return issues
|
||
cutoff = datetime.datetime.now() - datetime.timedelta(days=1)
|
||
for jid in os.listdir(cron_out):
|
||
jdir = os.path.join(cron_out, jid)
|
||
if not os.path.isdir(jdir):
|
||
continue
|
||
files = sorted(os.listdir(jdir))
|
||
if not files:
|
||
continue
|
||
latest = os.path.join(jdir, files[-1])
|
||
try:
|
||
mtime = datetime.datetime.fromtimestamp(os.path.getmtime(latest))
|
||
if mtime < cutoff:
|
||
continue
|
||
with open(latest, encoding="utf-8", errors="ignore") as f:
|
||
content = f.read()
|
||
if "script failed" in content or "Status: **error**" in content or "Traceback" in content:
|
||
# 网络类失败（服务器离线）是常态——过滤已知静默场景
|
||
if "不在局域网" in content or "离线" in content:
|
||
continue
|
||
issues.append(f"cron {jid} 最近输出异常: {os.path.basename(latest)}")
|
||
except Exception:
|
||
continue
|
||
return issues
|
||
|
||
|
||
def run_smoke():
|
||
issues = []
|
||
for cmd, timeout in SMOKE_SCRIPTS:
|
||
try:
|
||
r = subprocess.run(
|
||
f"cd {HERMES}/scripts && python3 {cmd}",
|
||
shell=True, capture_output=True, text=True, timeout=timeout,
|
||
)
|
||
if r.returncode != 0 and "skill-manager" not in cmd:
|
||
issues.append(f"smoke 失败: {cmd} (exit={r.returncode})")
|
||
except subprocess.TimeoutExpired:
|
||
issues.append(f"smoke 超时: {cmd}")
|
||
except Exception as e:
|
||
issues.append(f"smoke 异常: {cmd} ({e})")
|
||
return issues
|
||
|
||
|
||
def main():
|
||
quiet = "--quiet" in sys.argv
|
||
only = None
|
||
for a in sys.argv[1:]:
|
||
if a in ("json", "smoke"):
|
||
only = a
|
||
|
||
all_issues = []
|
||
if only in (None, "json"):
|
||
all_issues += check_json_schema()
|
||
if only in (None, "smoke"):
|
||
all_issues += run_smoke()
|
||
if only is None:
|
||
all_issues += check_exit_codes()
|
||
|
||
if quiet:
|
||
# watchdog 模式：只输出异常
|
||
if all_issues:
|
||
print("⚠️ 确定性传感器发现异常:")
|
||
for i in all_issues:
|
||
print(f" ❌ {i}")
|
||
sys.exit(1)
|
||
sys.exit(0)
|
||
|
||
print(f"🛰️ 确定性传感器 {datetime.datetime.now().strftime('%Y-%m-%d %H:%M')}")
|
||
print("=" * 50)
|
||
print(f"JSON Schema: {len(JSON_CHECKS)} 个文件")
|
||
print(f"Smoke test: {len(SMOKE_SCRIPTS)} 个脚本")
|
||
if not all_issues:
|
||
print("\n✅ 全部健康——确定性检查通过（0 LLM 成本）")
|
||
sys.exit(0)
|
||
print(f"\n⚠️ 发现 {len(all_issues)} 个异常:")
|
||
for i in all_issues:
|
||
print(f" ❌ {i}")
|
||
sys.exit(1)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|