diff --git a/.gitea/scripts/ci_gate.py b/.gitea/scripts/ci_gate.py index 6cb8043..9a02731 100644 --- a/.gitea/scripts/ci_gate.py +++ b/.gitea/scripts/ci_gate.py @@ -3,6 +3,12 @@ 规则与本机 ~/.githooks/commit-msg 完全一致,防止 --no-verify 绕过。 放行:Merge / Revert " / fixup! / squash! / amend! 开头的自动生成提交。 + +存量历史口径:合并一条早已推送到别的远端分支的lineage(例如把生产分支合进 main)时, +被带进来的提交并非本次新增,它们的标题不合规属于既有事实。判定为「已在其它远端分支存在」 +的提交只告警、不判红;本次真正新造的提交一律照旧严格校验。 +豁免必须是可证明的:先抓全部远端分支头,用 `git branch -r --contains ` 取证; +抓不到 / 证不出 ⇒ 不豁免(宁可红,也不假绿)。 """ import json import os @@ -14,6 +20,9 @@ TYPES = "feat|fix|docs|style|refactor|perf|test|build|ci|chore|revert" RE = re.compile(r"^(?:" + TYPES + r")(?:\([a-zA-Z0-9._/-]+\))?!?: .{3,}$") ALLOW = re.compile(r'^(?:Merge |Revert "|fixup! |squash! |amend! )') +# 本次推送的分支名,用来把「当前分支自己」排除在豁免证据之外 +PUSH_BRANCH = (os.environ.get("GITHUB_REF_NAME") or "").strip() + HELP = """ \u274c 提交信息不合规,已拦截。 @@ -32,9 +41,13 @@ HELP = """ """ +def sh(args): + return subprocess.run(args, capture_output=True, text=True) + + def collect(): - """优先用 Gitea 事件负载里的本次推送提交,取不到则退回 git log。""" - msgs = [] + """返回 [(sha, message)]:优先 Gitea 事件负载,取不到则退回 git log。""" + items = [] ev = os.environ.get("GITHUB_EVENT_PATH") or "" if ev and os.path.exists(ev): try: @@ -43,14 +56,51 @@ def collect(): for c in (d.get("commits") or []): m = (c.get("message") or "").strip() if m: - msgs.append(m) + items.append(((c.get("id") or "").strip(), m)) except Exception as e: print(" 事件负载解析失败(%s),回退到 git log" % e) - if not msgs: - out = subprocess.run(["git", "log", "-n", "5", "--format=%B%x00"], - capture_output=True, text=True).stdout - msgs = [x.strip() for x in out.split("\x00") if x.strip()] - return msgs + if not items: + out = sh(["git", "log", "-n", "5", "--format=%H%x1f%B%x1e"]).stdout + for rec in out.split("\x1e"): + rec = rec.strip() + if not rec: + continue + sha, _, msg = rec.partition("\x1f") + if msg.strip(): + items.append((sha.strip(), msg.strip())) + return items + + +_fetched = None + + +def ensure_remote_heads(): + """抓全部远端分支头(浅历史足够判断 lineage 末端)。""" + global _fetched + if _fetched is None: + r = sh(["git", "fetch", "-q", "--depth", "200", "origin", "+refs/heads/*:refs/remotes/origin/*"]) + _fetched = (r.returncode == 0, (r.stderr or "").strip()) + if not _fetched[0]: + print(" \u26a0 无法抓取远端分支,存量豁免不可用(将严格校验): %s" % _fetched[1][:200]) + return _fetched[0] + + +def refs_containing(sha): + """除本次推送分支外,还有哪些远端分支包含该提交。""" + if not ensure_remote_heads(): + return [] + r = sh(["git", "branch", "-r", "--contains", sha]) + if r.returncode != 0: + return [] + refs = [] + for line in r.stdout.splitlines(): + ref = line.strip().lstrip("*").strip() + if not ref or "->" in ref: + continue + if PUSH_BRANCH and ref == "origin/" + PUSH_BRANCH: + continue + refs.append(ref) + return refs def subject_of(m): @@ -77,17 +127,28 @@ def judge(subj): def main(): - msgs = collect() + items = collect() bad = [] - print("待校验提交 %d 条:" % len(msgs)) - for m in msgs: - subj = subject_of(m) + excused = [] + print("待校验提交 %d 条:" % len(items)) + for sha, msg in items: + subj = subject_of(msg) why = judge(subj) - if why: - bad.append((subj, why)) - print(" \u274c [%s] %s" % (why, subj)) - else: + if not why: print(" \u2705 %s" % subj) + continue + refs = refs_containing(sha) if sha else [] + if refs: + excused.append((sha, subj, why, refs)) + print(" \u21aa [%s] %s ← 已存在于 %s,按存量历史放行" % (why, subj, ",".join(refs[:3]))) + continue + bad.append((subj, why)) + print(" \u274c [%s] %s" % (why, subj)) + if excused: + print("") + print("存量豁免 %d 条(提交早于本次合并,已在其它远端分支存在):" % len(excused)) + for sha, subj, why, refs in excused: + print(" %s %s [%s]" % (sha[:10], subj, ",".join(refs[:3]))) try: with open("/tmp/ci-gate-status", "w") as fh: fh.write("failure" if bad else "success") @@ -101,4 +162,4 @@ def main(): if __name__ == "__main__": - sys.exit(main()) + sys.exit(main()) \ No newline at end of file