83 lines
2.6 KiB
Python
83 lines
2.6 KiB
Python
#!/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)")
|