#!/usr/bin/env python3 """deploy_guard.py — 部署一致性守卫（每 15 分钟 cron，全天候） 原则：实际部署的脚本 = MoFin git 最新版代码。 五件事： 1. 代码漂移检测：git 工作区中【被跟踪的代码文件】出现未提交改动 → 自动 git checkout 回滚 + 重链硬链接 + XMPP 告警 （只回滚未提交的工作区改动；git 提交永远安全） 2. 上游自动部署：session-work 分支领先 master 且可快进 → 自动 merge + 重链 + 必要时重启 dashboard + 告警 3. 硬链接一致性：幂等执行 sync_profile_scripts.sh 4. cron 引用完整性：jobs.json 引用的脚本必须存在 5. 状态落盘：gateway/logs/deploy_guard_status.json + deploy_guard.log （有动作/有问题才发 XMPP，沉默即正常） """ import json, os, subprocess, sys, glob from datetime import datetime REPO = "/home/hmo/MoFin" CANONICAL = f"{REPO}/deploy/profile-scripts" PROFILE_DIRS = [ "/home/hmo/.hermes/profiles/position-analyst/scripts", "/home/hmo/.hermes/profiles/default/scripts", ] LOG = f"{REPO}/gateway/logs/deploy_guard.log" STATUS_JSON = f"{REPO}/gateway/logs/deploy_guard_status.json" SYNC_SCRIPT = f"{CANONICAL}/sync_profile_scripts.sh" # 守卫范围：被 git 跟踪的【代码】路径（运行时产物 data/ logs/ static/*.json 不在其列） CODE_PATHS = [ "deploy/profile-scripts", "deploy/bot", "prompt_manager", "scripts", "server.py", "mofin_db.py", "mo_data.py", "strategy_lifecycle.py", "xmpp_logger.py", "mo_config.py", "static/index.html", "static/mofin_health.html", ] actions = [] # 采取了的动作（需告警） problems = [] # 发现的问题（需告警） def log(msg): line = f"[{datetime.now().isoformat(timespec='seconds')}] {msg}" print(line, flush=True) try: os.makedirs(os.path.dirname(LOG), exist_ok=True) with open(LOG, "a", encoding="utf-8") as f: f.write(line + "\n") except Exception: pass def xmpp(msg, level=None): try: sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from alert_helper import notify, INFO notify("部署守卫", msg, level or INFO) except Exception as e: log(f"XMPP发送失败: {e}") def git(*args, check=True): env = dict(os.environ) env["GIT_ALLOW_COMMIT"] = "1" # pre-commit 白名单令牌（知微无此令牌无法提交） r = subprocess.run(["git", "-C", REPO] + list(args), capture_output=True, text=True, timeout=60, env=env) if check and r.returncode != 0: raise RuntimeError(f"git {' '.join(args)}: {r.stderr.strip()[:200]}") return r.stdout.strip() # ── 1. 代码漂移检测与回滚 ── def check_code_drift(): out = git("status", "--porcelain", "--", *CODE_PATHS) if not out: return modified, untracked = [], [] for line in out.splitlines(): if not line.strip(): continue status, path = line[:2], line[2:].strip() # 重命名格式 "R old -> new" 取新路径 if " -> " in path: path = path.split(" -> ", 1)[1] if status == "??": untracked.append(path) else: modified.append(path) if modified: diff_stat = git("diff", "--stat", "--", *modified, check=False) log(f"DRIFT: 检测到 {len(modified)} 个代码文件未入库改动: {modified}") for f in modified: git("checkout", "--", f) subprocess.run(["bash", SYNC_SCRIPT], capture_output=True, timeout=120) actions.append(f"回滚未入库代码改动 {len(modified)} 个: {', '.join(modified[:5])}" + (f" 等(共{len(modified)}个)" if len(modified) > 5 else "")) log(f"DRIFT_REVERTED: {modified}\n{diff_stat[:500]}") if untracked: # 未跟踪的新文件：不删（可能是在途工作），只告警 problems.append(f"权威目录出现未跟踪文件(需人工确认): {', '.join(untracked[:5])}") log(f"UNTRACKED: {untracked}") # ── 2. session-work → master 自动部署 ── def check_upstream_deploy(): try: git("rev-parse", "--verify", "session-work") except Exception: return # 无此分支 ahead = git("rev-list", "--count", "master..session-work", check=False) if not ahead.isdigit() or int(ahead) == 0: return # 仅当可快进（master 是 session-work 祖先）才自动合并 r = subprocess.run(["git", "-C", REPO, "merge-base", "--is-ancestor", "master", "session-work"]) if r.returncode != 0: problems.append(f"session-work 领先 master {ahead} 个提交但存在分叉，需人工合并") log(f"UPSTREAM_DIVERGED: ahead={ahead}") return log(f"UPSTREAM: session-work 领先 {ahead} 个提交，自动合并部署") merge_out = git("merge", "session-work", "--no-edit") sync_out = subprocess.run(["bash", SYNC_SCRIPT], capture_output=True, text=True, timeout=120) # 触及 dashboard 代码则重启 touched = git("diff", "--name-only", f"HEAD~{ahead}", "HEAD", check=False) restarted = "" if any(t.startswith(("server.py", "static/")) for t in touched.splitlines()): rr = subprocess.run(["sudo", "-n", "systemctl", "restart", "mofin-dashboard"], capture_output=True, text=True, timeout=30) restarted = " + dashboard已重启" if rr.returncode == 0 else " (dashboard重启失败,需人工)" actions.append(f"自动部署 session-work→master ({ahead}个提交){restarted}") log(f"UPSTREAM_DEPLOYED: {merge_out[:200]} | sync: {sync_out.stdout.strip()[-100:]}") # ── 部署后自动验证（盲区②：提交级回归检测——知微 stale 提交事件的系统补丁）── try: vr = subprocess.run(["python3", f"{REPO}/scripts/verify_deployment.py"], capture_output=True, text=True, timeout=300) vout = (vr.stdout or "") + (vr.stderr or "") import re as _re m = _re.search(r"汇总:\s*(\d+)\s*项通过,\s*(\d+)\s*项失败", vout) if m and int(m.group(2)) > 0: fm = _re.search(r"失败项:\s*(\[.*?\])", vout) problems.append(f"部署后验证发现 {m.group(2)} 项失败: {fm.group(1) if fm else '详见日志'}" f"（刚合并的 {ahead} 个提交可能引入回归，需人工核查）") log(f"POST_DEPLOY_VERIFY_FAIL: {m.group(2)} failures") elif m: log(f"POST_DEPLOY_VERIFY_OK: {m.group(1)} passed") else: log("POST_DEPLOY_VERIFY: 无法解析验证输出") except Exception as e: log(f"POST_DEPLOY_VERIFY_ERROR: {str(e)[:120]}") # ── 3. 硬链接一致性 ── def check_hardlinks(): r = subprocess.run(["bash", SYNC_SCRIPT], capture_output=True, text=True, timeout=120) tail = (r.stdout or "").strip().splitlines() if tail: log(f"SYNC: {tail[-1]}") if r.returncode != 0: problems.append(f"sync_profile_scripts.sh 执行失败: {(r.stderr or '')[:150]}") # ── 4. cron 引用完整性 ── def check_cron_refs(): missing = [] for pj in glob.glob("/home/hmo/.hermes/profiles/*/cron/jobs.json"): profile = pj.split("/")[-3] try: with open(pj, encoding="utf-8") as f: jobs = json.load(f) jobs = jobs if isinstance(jobs, list) else jobs.get("jobs", []) except Exception: continue for j in jobs: if not j.get("enabled", True): continue script = j.get("script") or "" if not script: continue base = os.path.basename(script) if not any(os.path.exists(os.path.join(d, base)) for d in PROFILE_DIRS + [CANONICAL]): missing.append(f"[{profile}]{j.get('name','?')}→{base}") if missing: problems.append(f"cron引用缺失脚本 {len(missing)} 个: {', '.join(missing[:5])}") log(f"CRON_MISSING: {missing}") def main(): log("── deploy_guard 开始 ──") for fn in (check_code_drift, check_upstream_deploy, check_hardlinks, check_cron_refs): try: fn() except Exception as e: problems.append(f"{fn.__name__} 异常: {str(e)[:150]}") log(f"ERROR in {fn.__name__}: {e}") status = { "ts": datetime.now().isoformat(timespec="seconds"), "ok": not problems, "actions": actions, "problems": problems, } try: with open(STATUS_JSON, "w", encoding="utf-8") as f: json.dump(status, f, ensure_ascii=False, indent=1) except Exception: pass if actions or problems: # 分级：部署后验证失败=需人工核查(ACTION直通)；其余=INFO(限速聚合) info_parts = list(actions) action_parts = [] for p in problems: if "部署后验证发现" in p: action_parts.append(p) else: info_parts.append(p) if info_parts: xmpp("\n".join(f"✅ {a}" if a in actions else f"⚠️ {a}" for a in info_parts)) for p in action_parts: from alert_helper import notify as _notify, ACTION as _ACT _notify("部署验证", f"⚠️ {p}", _ACT) log(f"── 结束: actions={len(actions)} problems={len(problems)} ──") return 0 if not problems else 1 if __name__ == "__main__": sys.exit(main())