把早已推送到其它远端分支的 lineage 合进 main 时,被带进来的提交不是本次新增, 它们标题不合规属既有事实。闸门先抓全部远端分支头,用 git branch -r --contains 取证:能证明已存在于其它远端分支的只告警放行;证不出来的照旧拦截(fail-closed)。
This commit is contained in:
+76
-15
@@ -3,6 +3,12 @@
|
||||
|
||||
规则与本机 ~/.githooks/commit-msg 完全一致,防止 --no-verify 绕过。
|
||||
放行:Merge / Revert " / fixup! / squash! / amend! 开头的自动生成提交。
|
||||
|
||||
存量历史口径:合并一条早已推送到别的远端分支的lineage(例如把生产分支合进 main)时,
|
||||
被带进来的提交并非本次新增,它们的标题不合规属于既有事实。判定为「已在其它远端分支存在」
|
||||
的提交只告警、不判红;本次真正新造的提交一律照旧严格校验。
|
||||
豁免必须是可证明的:先抓全部远端分支头,用 `git branch -r --contains <sha>` 取证;
|
||||
抓不到 / 证不出 ⇒ 不豁免(宁可红,也不假绿)。
|
||||
"""
|
||||
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:
|
||||
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))
|
||||
else:
|
||||
print(" \u2705 %s" % 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")
|
||||
|
||||
Reference in New Issue
Block a user