冰朔
8544cae938
光湖子系统面板模块包（胖头鱼/守望两系统沉淀）： - panel/server.js 配置驱动审批面板（登录门/签字/写入窗口/部署通道/唤醒导航） - deploy/verify-and-deploy.py 部署校验Agent（地图/因果链/范围/完整性，0或100） - hooks/pre-receive-review.template 推送审核钩子模板 - frontend 两套门面样例（湖绿深色/宠物暖色玻璃动态） - CAUSAL-CHAIN-20260806.hdlp 两系统建设完整因果链
104 lines
6.2 KiB
Python
104 lines
6.2 KiB
Python
#!/usr/bin/env python3
|
||
# verify-and-deploy.py · 光湖子系统部署校验Agent · 可复用模块 v1.0
|
||
# 2026-08-06 · 铸渊 ICE-GL-ZY001 建 · 配置见同目录 deploy-config.json
|
||
# 流程: 取PR → 服务器地图 → 因果链 → 范围白名单 → 运行完整性 → 合并部署 → 同步/重启 → 回执
|
||
# ⊢ 任一步失败 = REJECTED 不部署 ⊢ 进度实时落盘，面板可见
|
||
import json, os, sys, subprocess, datetime, urllib.request, shutil, tempfile
|
||
|
||
CFG = json.load(open(os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "deploy-config.json") if os.path.exists(os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "deploy-config.json")) else os.path.join(os.path.dirname(os.path.abspath(__file__)), "deploy-config.json")))
|
||
PR = sys.argv[1] if len(sys.argv) > 1 else "?"
|
||
RUN_DIR = CFG["runs_dir"]; os.makedirs(RUN_DIR, exist_ok=True)
|
||
RUN = os.path.join(RUN_DIR, "RUN-PR%s-%s.json" % (PR, datetime.datetime.now().strftime("%Y%m%dT%H%M%S")))
|
||
TOKEN = open(CFG["token_file"]).read().strip()
|
||
API = CFG["api"]; REPO = CFG["repo"]; BARE = CFG["bare"]
|
||
|
||
def api(path, method="GET", body=None):
|
||
data = json.dumps(body).encode() if body is not None else None
|
||
req = urllib.request.Request(API + path, data=data, method=method)
|
||
req.add_header("Authorization", "token " + TOKEN)
|
||
req.add_header("X-Forwarded-Proto", "https")
|
||
req.add_header("Content-Type", "application/json")
|
||
with urllib.request.urlopen(req, timeout=25) as r:
|
||
return json.loads(r.read().decode() or "{}")
|
||
|
||
def gitshow(rev, path):
|
||
r = subprocess.run(["git", "-C", BARE, "show", rev + ":" + path], capture_output=True, text=True)
|
||
return r.stdout if r.returncode == 0 else ""
|
||
|
||
state = {"pr": PR, "startedAt": datetime.datetime.now(datetime.timezone.utc).isoformat(), "status": "RUNNING", "steps": []}
|
||
def save():
|
||
tmp = RUN + ".tmp"; open(tmp, "w").write(json.dumps(state, ensure_ascii=False, indent=1)); os.replace(tmp, RUN)
|
||
def step(name, status, detail=""):
|
||
state["steps"].append({"name": name, "status": status, "detail": detail, "ts": datetime.datetime.now(datetime.timezone.utc).isoformat()}); save()
|
||
def finish(status, detail=""):
|
||
state["status"] = status; state["finishedAt"] = datetime.datetime.now(datetime.timezone.utc).isoformat(); state["detail"] = detail
|
||
save(); print(status, detail); sys.exit(0 if status == "DEPLOYED" else 1)
|
||
|
||
save()
|
||
try:
|
||
pr = api("/repos/%s/pulls/%s" % (REPO, PR))
|
||
head = pr["head"]["sha"]
|
||
files = api("/repos/%s/pulls/%s/files?limit=100" % (REPO, PR))
|
||
names = [f["filename"] for f in files]
|
||
step("FETCH_PR", "PASS", "PR#%s · head %s · %d 文件" % (PR, head[:8], len(names)))
|
||
|
||
ok = True; why = []
|
||
for m in CFG.get("map_checks", []):
|
||
p = m["path"]
|
||
if not os.path.isfile(p): ok = False; why.append("缺失 " + p); continue
|
||
if "contains" in m and m["contains"] not in open(p, errors="ignore").read(): ok = False; why.append(p + " 编号不符")
|
||
if ok: step("CHECK_MAP", "PASS", "服务器地图与策略在位且编号吻合")
|
||
else: step("CHECK_MAP", "FAIL", "; ".join(why)); finish("REJECTED", "地图校验失败")
|
||
|
||
causal_ok = False; causal_file = ""
|
||
for fn in names:
|
||
if any(fn.startswith(cp) for cp in CFG.get("causal_paths", [])):
|
||
txt = gitshow(head, fn)
|
||
if ("trigger" in txt and "why" in txt) or ("因果" in txt):
|
||
causal_ok = True; causal_file = fn; break
|
||
if causal_ok: step("CHECK_CAUSAL", "PASS", "因果链齐备 · " + causal_file)
|
||
else: step("CHECK_CAUSAL", "FAIL", "缺少因果链文档（contributions/ 或 docs/ 下需含 trigger+why 或「因果」）"); finish("REJECTED", "因果链审核失败")
|
||
|
||
bad = [n for n in names if not n.startswith(tuple(CFG.get("allowed_prefix", [])))]
|
||
if bad: step("CHECK_SCOPE", "FAIL", "越界文件: " + ", ".join(bad[:5])); finish("REJECTED", "部署通道范围校验失败")
|
||
step("CHECK_SCOPE", "PASS", "全部文件在部署白名单内")
|
||
|
||
alive = True; dead = []
|
||
for u in CFG.get("services", []):
|
||
r = subprocess.run(["systemctl", "is-active", u], capture_output=True, text=True)
|
||
if r.stdout.strip() != "active": alive = False; dead.append(u)
|
||
if alive: step("CHECK_INTEGRITY", "PASS", "现有服务全部在线，部署不破坏运行")
|
||
else: step("CHECK_INTEGRITY", "FAIL", "服务异常: " + ", ".join(dead)); finish("REJECTED", "完整性校验失败")
|
||
|
||
m = api("/repos/%s/pulls/%s/merge" % (REPO, PR), method="POST", body={"Do": "merge", "merge_title_message": "deploy: PR#%s 审批部署（校验Agent全过）" % PR})
|
||
step("DEPLOY_MERGE", "PASS", "已合并入 deploy 分支")
|
||
|
||
synced = []
|
||
for st in CFG.get("sync_trees", []):
|
||
sp, tgt = st["source_prefix"], st["target"]
|
||
hit = [n for n in names if n.startswith(sp)]
|
||
if not hit: continue
|
||
tmpd = tempfile.mkdtemp()
|
||
subprocess.run(["git", "-C", BARE, "archive", head], stdout=open(tmpd + "/a.tar", "wb"))
|
||
subprocess.run(["tar", "-xf", tmpd + "/a.tar", "-C", tmpd])
|
||
for n in hit:
|
||
src = os.path.join(tmpd, n); rel = n[len(sp):]
|
||
dst = os.path.join(tgt, rel)
|
||
if os.path.isfile(src):
|
||
os.makedirs(os.path.dirname(dst), exist_ok=True)
|
||
shutil.copy2(src, dst)
|
||
synced.append(n)
|
||
shutil.rmtree(tmpd, ignore_errors=True)
|
||
if synced:
|
||
step("SYNC_CUSTOM", "PASS", "同步 %d 文件到运行目录" % len(synced))
|
||
for u in CFG.get("restart_services", []):
|
||
subprocess.run(["systemctl", "restart", u])
|
||
step("RESTART", "PASS", "已重启: " + ", ".join(CFG.get("restart_services", [])))
|
||
else:
|
||
step("SYNC_CUSTOM", "PASS", "本次无需同步运行目录")
|
||
|
||
rcpt = {"ts": datetime.datetime.now(datetime.timezone.utc).isoformat(), "type": "DEPLOY_RECEIPT", "pr": PR, "head": head, "run": os.path.basename(RUN), "verdict": "DEPLOYED", "synced": synced}
|
||
open(os.path.join(CFG["receipts_dir"], "DEPLOY-%s.json" % datetime.datetime.now().strftime("%Y%m%dT%H%M%S")), "w").write(json.dumps(rcpt, ensure_ascii=False, indent=1))
|
||
finish("DEPLOYED", "部署完成 · 100 · 回执已生成")
|
||
except Exception as e:
|
||
step("ERROR", "FAIL", str(e)[:200]); finish("REJECTED", "异常: " + str(e)[:120])
|