diff --git a/scripts/deploy_sync.sh b/scripts/deploy_sync.sh index 8d597583..bb24cc45 100755 --- a/scripts/deploy_sync.sh +++ b/scripts/deploy_sync.sh @@ -60,3 +60,6 @@ if [ $count -eq 0 ]; then else echo "✅ 同步了 $count 个文件" fi + +# 同步cron prompt注册版本(确保提示词管理系统与jobs.json一致) +python3 /home/hmo/MoFin/scripts/sync_cron_prompts.py 2>/dev/null || true diff --git a/scripts/preflight_verify.py b/scripts/preflight_verify.py index 9d47a324..496009b6 100644 --- a/scripts/preflight_verify.py +++ b/scripts/preflight_verify.py @@ -143,7 +143,51 @@ def check_cron_timestamps(): pass check("cron过期检查", stale_count == 0, f"{stale_count}个cron长时间未正常运行" if stale_count else "") -# ── 6. DB死锁检测 ── +# ── 7. 提示词一致性检查 ── +def check_prompt_sync(): + """检查注册的cron prompt版本与jobs.json实际运行的是否一致""" + try: + reg_path = "/home/hmo/projects/MoFin/data/prompts/registry.json" + cron_path = "/home/hmo/.hermes/profiles/position-analyst/cron/jobs.json" + if not os.path.exists(reg_path) or not os.path.exists(cron_path): + check("提示词一致性", False, "registry或jobs.json不存在") + return + with open(reg_path) as f: + reg = json.load(f) + with open(cron_path) as f: + crons = json.load(f).get("jobs", []) + # 建立cron prompt索引 + cron_prompts = {} + for j in crons: + name = j.get("name", "") + pid = name.replace(" ","-").replace("(","-").replace(")","").replace("(","-").replace(")","") + if j.get("prompt"): + cron_prompts[pid] = {"name": name, "prompt": j["prompt"]} + # 逐个检查注册的版本文件是否与cron一致 + drift = 0 + for p in reg.get("prompts", []): + pid = p["id"] + if pid not in cron_prompts: + continue + cv = p.get("current_version", "v1") + # 找版本文件路径 + content_path = "" + for v in p.get("versions", []): + if v.get("version") == cv: + content_path = v.get("content_path", "") + break + if not content_path or not os.path.exists(content_path): + drift += 1 + continue + with open(content_path) as f: + registered = f.read() + actual = cron_prompts[pid]["prompt"] + if registered != actual: + drift += 1 + check("提示词一致性: 注册版≈运行版", drift == 0, + f"{drift}个prompt不一致" if drift else "") + except Exception as e: + check("提示词一致性", False, str(e)) def check_db_locks(): """检查最近24小时是否有 database is locked 错误""" try: @@ -167,6 +211,7 @@ if __name__ == "__main__": check_smoke() check_cron_timestamps() check_db_locks() + check_prompt_sync() # 汇总 print() diff --git a/scripts/sync_cron_prompts.py b/scripts/sync_cron_prompts.py new file mode 100644 index 00000000..be7a0831 --- /dev/null +++ b/scripts/sync_cron_prompts.py @@ -0,0 +1,82 @@ +#!/usr/bin/env python3 +"""sync_cron_prompts.py — 同步cron prompt到提示词管理系统 + +每次修改cron prompt后运行此脚本,确保注册的版本文件与jobs.json一致。 +可在修改cron prompt后手动调用,或集成到deploy_sync.sh中。 + +用法: python3 sync_cron_prompts.py [--check-only] + --check-only: 只检查不一致,不修改 +""" +import json, sys, os +from datetime import datetime + +REG_PATH = "/home/hmo/projects/MoFin/data/prompts/registry.json" +VERSIONS_DIR = "/home/hmo/projects/MoFin/data/prompts/versions/" +CRON_PATH = "/home/hmo/.hermes/profiles/position-analyst/cron/jobs.json" + +check_only = "--check-only" in sys.argv + +with open(REG_PATH) as f: + reg = json.load(f) +with open(CRON_PATH) as f: + crons = json.load(f).get("jobs", []) + +# Build index of cron prompts +cron_prompts = {} +for j in crons: + name = j.get("name", "") + pid = name.replace(" ","-").replace("(","-").replace(")","").replace("(","-").replace(")","") + if j.get("prompt"): + cron_prompts[pid] = {"name": name, "prompt": j["prompt"]} + +drift_count = 0 +fix_count = 0 + +for p in reg.get("prompts", []): + pid = p["id"] + if pid not in cron_prompts: + continue + + cv = p.get("current_version", "v1") + actual = cron_prompts[pid]["prompt"] + name = cron_prompts[pid]["name"] + + # Find version file path + content_path = "" + for v in p.get("versions", []): + if v.get("version") == cv: + content_path = v.get("content_path", "") + break + + if not content_path: + print(f"⚠️ {name}: 版本文件路径为空") + drift_count += 1 + continue + + if not os.path.exists(content_path): + print(f"⚠️ {name}: 版本文件不存在 {content_path}") + drift_count += 1 + if not check_only: + os.makedirs(os.path.dirname(content_path), exist_ok=True) + with open(content_path, 'w') as f: + f.write(actual) + print(f" → 已创建") + fix_count += 1 + continue + + with open(content_path) as f: + registered = f.read() + + if registered != actual: + print(f"⚠️ {name}: 版本文件与jobs.json不一致") + drift_count += 1 + if not check_only: + with open(content_path, 'w') as f: + f.write(actual) + print(f" → 已同步") + fix_count += 1 + +if drift_count == 0: + print("✅ 全部一致") +else: + print(f"\n⚠️ {drift_count}个不一致, 已修复{fix_count}个" if not check_only else f"\n⚠️ {drift_count}个不一致 (使用--check-only)")