hermes
1b0cce283a
templates/ci_report.py 同步项目版（链接基址/提交链接/回退来源 + 通知字段）；templates/gitea-build.yml 上报步骤加非秘密 CI_PUBLIC_BASE_URL；templates/README.md 加'上报脚本的非秘密配置'一节；01 正本 §4.7 增链接基址与'禁止全局 URL 替换'；02 Runbook 增排障条目；04 加 T-9（通知链接指向内网地址）。
522 lines
27 KiB
Python
Executable File
522 lines
27 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
"""核对门禁报告并上报 ci-bridge（bridge 负责落报告页 + 走当前通知出口：微信/钉钉）。
|
||
|
||
**不只看 `overall_status` 一个字段** —— 那是"信一句没有任何依据的结论"。上报前先独立核对：
|
||
|
||
1. 清单一致性：报告里的 `manifest` 必须与实现里的 `CHECK_MANIFEST` 逐项相同；
|
||
2. 完整清单核对：报告必须**恰好一次**覆盖清单里在本档案下必需的每一项 ——
|
||
缺项 / 重复项 / 清单之外的项，都算不通过；
|
||
3. 逐项状态重算总体：只有"必需项全部 success"才算 success，并与报告自称的
|
||
`overall_status` 对照；**两者不一致 ⇒ 汇总不可信**；
|
||
4. 归属：`commit_sha` 必须是**完整 40 位 SHA**，且（CI 上）等于事件 SHA；
|
||
5. 档案：CI 里必须是 `ci` 档案（`local` 档案下 identity 不要求，等于把身份核对关掉）；
|
||
6. 运行模式：`only`（只跑了子集）的报告**不能**当作 CI 证据。
|
||
|
||
退出码只表达一件事：**这次运行的质量能否判定**。
|
||
· 报告读不到 / 核对不过 ⇒ 非 0（无从判定，绝不能报绿）；
|
||
· 报告核对通过、结论是 failure ⇒ 0（作业在门禁那一步已经红了，这里不重复判）。
|
||
通知发不出去也只标注，不改结论。
|
||
|
||
用法：
|
||
CI_BRIDGE_RESOLVED=http://172.17.0.1:8899 python3 .gitea/scripts/ci_report.py
|
||
python3 .gitea/scripts/ci_report.py --report ci-report.json --no-post
|
||
python3 .gitea/scripts/ci_report.py --self-test # 核对逻辑的反向自检
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import os
|
||
import re
|
||
import subprocess
|
||
import sys
|
||
import urllib.error
|
||
import urllib.request
|
||
from pathlib import Path
|
||
|
||
ROOT = Path(__file__).resolve().parent.parent.parent
|
||
|
||
# 清单与汇总逻辑**只应有一份**：直接 import 本仓的门禁实现（scripts/ci_checks.py），
|
||
# 而不是在这里再写一遍核对规则 —— 两份规则迟早会漂移，而漂移的表现是"门禁绿、上报却放行"。
|
||
GATE_PATH = ROOT / "scripts" / "ci_checks.py"
|
||
try:
|
||
sys.path.insert(0, str(ROOT / "scripts"))
|
||
import ci_checks # type: ignore # noqa: E402
|
||
from ci_checks import CHECK_MANIFEST, FULL_SHA, SUCCESS # noqa: E402
|
||
aggregate = ci_checks.aggregate
|
||
GATE_ERROR = ""
|
||
except Exception as _e: # 缺文件 / 语法错 / 导入期异常
|
||
CHECK_MANIFEST, FULL_SHA, SUCCESS = {}, re.compile(r"^$"), "success"
|
||
aggregate = None
|
||
GATE_ERROR = f"{GATE_PATH} 不可用：{_e}"
|
||
|
||
# bridge 的 status 只认 success/failure/cancelled，它的"失败步骤"判定是
|
||
# `status not in (success, skipped)`；把 not_run 归到 failure 是**保守**方向：
|
||
# 未运行绝不能被渲染成"通过"。
|
||
BRIDGE_STATUS = {"success": "success", "skipped": "skipped", "failure": "failure",
|
||
"not_run": "failure", "unknown": "failure"}
|
||
|
||
|
||
def rd(path: str, default: str = "") -> str:
|
||
try:
|
||
return Path(path).read_text(encoding="utf-8").strip()
|
||
except OSError:
|
||
return default
|
||
|
||
|
||
def git(*args: str) -> str:
|
||
try:
|
||
return subprocess.run(["git", *args], cwd=ROOT, capture_output=True,
|
||
text=True).stdout.strip()
|
||
except OSError:
|
||
return ""
|
||
|
||
|
||
def verify(rep: dict, expect_sha: str, in_ci: bool) -> tuple[str, list[str], list[str]]:
|
||
"""独立核对报告。返回 (重算的总体, 问题, 提示)。纯函数，--self-test 直接打它。"""
|
||
problems: list[str] = []
|
||
notes: list[str] = []
|
||
|
||
if rep.get("schema") != "ci-report/1":
|
||
problems.append(f"报告 schema 不是 ci-report/1：{rep.get('schema')!r}")
|
||
|
||
profile = rep.get("profile")
|
||
if profile not in ("local", "ci"):
|
||
problems.append(f"报告档案非法：{profile!r}")
|
||
elif in_ci and profile != "ci":
|
||
problems.append(f"运行环境是 CI，但报告档案是 {profile!r} —— local 档案下 identity 不要求，"
|
||
f"等于把身份核对关掉")
|
||
|
||
if rep.get("run_mode") != "full":
|
||
problems.append(f"报告的 run_mode 是 {rep.get('run_mode')!r} —— 只跑了子集的报告不能当 CI 证据")
|
||
|
||
sha = str(rep.get("commit_sha") or "")
|
||
if not FULL_SHA.match(sha):
|
||
problems.append(f"报告的 commit_sha 不是完整 40 位 SHA：{sha[:20]!r}")
|
||
elif expect_sha and sha != expect_sha:
|
||
problems.append(f"报告的 commit_sha {sha[:12]} != 事件 SHA {expect_sha[:12]} —— 归属对不上")
|
||
event_sha = str(rep.get("event_sha") or "")
|
||
if event_sha and not FULL_SHA.match(event_sha):
|
||
problems.append(f"报告的 event_sha 不是完整 40 位：{event_sha[:20]!r}")
|
||
elif event_sha and sha and event_sha != sha:
|
||
problems.append(f"报告里 event_sha {event_sha[:12]} 与 commit_sha {sha[:12]} 不一致")
|
||
|
||
manifest = rep.get("manifest")
|
||
if not isinstance(manifest, dict) or not manifest:
|
||
problems.append("报告里没有 manifest —— 没有清单就无法核对'必需项齐不齐'")
|
||
else:
|
||
mine = {cid: list(req) for cid, req in CHECK_MANIFEST.items()}
|
||
if {cid: list(req) for cid, req in manifest.items()} != mine:
|
||
diff = sorted(set(mine) ^ set(manifest)) or [
|
||
cid for cid in mine if list(manifest.get(cid) or []) != mine[cid]]
|
||
problems.append(f"报告里的清单与实现不一致（差异：{diff[:6]}）—— 有一次读完的结论不能作数")
|
||
|
||
checks = rep.get("checks")
|
||
if not isinstance(checks, list):
|
||
problems.append("报告的 checks 不是数组")
|
||
checks = []
|
||
|
||
# 用**报告自带的清单**（已确认与实现一致）重算，而不是信 overall_status
|
||
declared = {cid: tuple(req) for cid, req in (manifest or {}).items()} or CHECK_MANIFEST
|
||
recomputed, agg_problems = aggregate(checks, profile if profile in ("local", "ci") else "ci", declared)
|
||
problems += agg_problems
|
||
|
||
claimed = rep.get("overall_status")
|
||
if recomputed != claimed and claimed != "not_run_only":
|
||
problems.append(f"报告自称 overall_status={claimed!r}，按清单逐项重算是 {recomputed!r} —— "
|
||
f"汇总与清单不一致，不能采信")
|
||
elif claimed == "not_run_only":
|
||
problems.append("报告的 overall_status 是 not_run_only（只跑了子集）—— 不能作为 CI 证据")
|
||
|
||
if not problems:
|
||
notes.append(f"核对通过：{len(checks)} 项、档案 {profile}、SHA {sha[:12]}、"
|
||
f"总体 {recomputed}（与报告自称一致）")
|
||
return recomputed, problems, notes
|
||
|
||
|
||
def self_test() -> int:
|
||
if GATE_ERROR:
|
||
# 模板目录里没有 scripts/ci_checks.py（实现随项目走），所以自检跑不了 ——
|
||
# 明说，而不是拿一份空清单去跑出一堆莫名其妙的结论。
|
||
print(f"❌ 自检跑不了：{GATE_ERROR}", file=sys.stderr)
|
||
print(" 本脚本的核对逻辑依赖本仓的 scripts/ci_checks.py；请在项目仓里跑 --self-test。",
|
||
file=sys.stderr)
|
||
return 2
|
||
sha = "a" * 40
|
||
other = "b" * 40
|
||
good_checks = [{"id": cid, "status": SUCCESS, "reason": "", "required_in": list(req),
|
||
"duration_s": 1} for cid, req in CHECK_MANIFEST.items()]
|
||
# 自检里的"挑一项"要**动态取**清单里的第一项：硬编码某个 id（例如 tests-go）在模板仓/新工程里
|
||
# 根本不存在，自检会因为"找不到那一项"而自作聪明地通过或失败。
|
||
first_id = sorted(CHECK_MANIFEST)[0]
|
||
good = {
|
||
"schema": "ci-report/1", "profile": "ci", "run_mode": "full", "commit_sha": sha,
|
||
"event_sha": sha, "manifest": {c: list(r) for c, r in CHECK_MANIFEST.items()},
|
||
"checks": good_checks, "overall_status": SUCCESS,
|
||
}
|
||
cases = [
|
||
("合规报告 ⇒ 通过", good, sha, True, 0),
|
||
("短 SHA ⇒ 不过", {**good, "commit_sha": sha[:12], "event_sha": ""}, sha, True, 1),
|
||
("归错提交 ⇒ 不过", {**good, "commit_sha": other, "event_sha": other}, sha, True, 1),
|
||
# 下面这些会同时命中两条判据（例如"缺项"既触发清单核对，又让报告自称的 success 与重算不符），
|
||
# 所以期望 2 条 —— 两条都要报出来，不能只报一条就被掩盖。
|
||
("缺一项必需检查 ⇒ 不过（缺项 + 自称与重算不符）",
|
||
{**good, "checks": [c for c in good_checks if c["id"] != first_id]}, sha, True, 2),
|
||
("重复一项 ⇒ 不过（重复 + 自称与重算不符）",
|
||
{**good, "checks": good_checks + [good_checks[0]]}, sha, True, 2),
|
||
("清单之外的项 ⇒ 不过（未知项 + 自称与重算不符）",
|
||
{**good, "checks": good_checks + [{"id": "zz", "status": SUCCESS, "reason": ""}]}, sha, True, 2),
|
||
("清单被改过 ⇒ 不过（清单漂移 + 未知项 + 自称不符）",
|
||
{**good, "manifest": {k: list(v) for k, v in CHECK_MANIFEST.items() if k != "format"}}, sha, True, 3),
|
||
("自称成功但某项 failure ⇒ 不过（逐项 + 自称不符）",
|
||
{**good, "overall_status": SUCCESS,
|
||
"checks": [{**c, "status": "failure"} if c["id"] == first_id else c for c in good_checks]},
|
||
sha, True, 2),
|
||
("CI 里用 local 档案 ⇒ 不过", {**good, "profile": "local"}, sha, True, 1),
|
||
("只跑子集的报告 ⇒ 不过（run_mode + 结论不可用）",
|
||
{**good, "overall_status": "not_run_only", "run_mode": "only"}, sha, True, 2),
|
||
("没有 manifest ⇒ 不过", {k: v for k, v in good.items() if k != "manifest"}, sha, True, 1),
|
||
]
|
||
problems = []
|
||
for name, rep, expect, in_ci, want in cases:
|
||
_, got, _ = verify(rep, expect, in_ci)
|
||
if len(got) != want:
|
||
problems.append(f"{name}：期望 {want} 个问题，实际 {len(got)} 个 {got}")
|
||
# 通知链接基址：优先 CI_PUBLIC_BASE_URL，缺/非法都必须**明确回退**（不猜、不静默）
|
||
env_backup = {k: os.environ.get(k) for k in ("CI_PUBLIC_BASE_URL", "GITHUB_SERVER_URL")}
|
||
try:
|
||
os.environ["GITHUB_SERVER_URL"] = "http://internal:3000"
|
||
os.environ.pop("CI_PUBLIC_BASE_URL", None)
|
||
b, src = _link_base()
|
||
if b != "http://internal:3000" or "GITHUB_SERVER_URL" not in src:
|
||
problems.append(f"没配 CI_PUBLIC_BASE_URL 时没回退到 GITHUB_SERVER_URL：{b!r}/{src!r}")
|
||
os.environ["CI_PUBLIC_BASE_URL"] = "https://gitea.zheromiku.top/"
|
||
b2, src2 = _link_base()
|
||
if b2 != "https://gitea.zheromiku.top" or src2 != "CI_PUBLIC_BASE_URL":
|
||
problems.append(f"配了 CI_PUBLIC_BASE_URL 没用它（或没去掉尾斜杠）：{b2!r}/{src2!r}")
|
||
for bad in ("gitea.zheromiku.top", "ftp://gitea.zheromiku.top",
|
||
"https://u:p@gitea.zheromiku.top", "https://a b/"):
|
||
os.environ["CI_PUBLIC_BASE_URL"] = bad
|
||
b3, src3 = _link_base()
|
||
if b3 != "http://internal:3000" or "回退" not in src3:
|
||
problems.append(f"非法 CI_PUBLIC_BASE_URL({bad!r}) 没明确回退：{b3!r}/{src3!r}")
|
||
os.environ["CI_PUBLIC_BASE_URL"] = "https://gitea.zheromiku.top"
|
||
if _commit_url("a/b", "c" * 40, "https://gitea.zheromiku.top") != \
|
||
f"https://gitea.zheromiku.top/a/b/commit/{'c' * 40}":
|
||
problems.append("提交链接拼错")
|
||
for bad_repo, bad_sha in (("onlyone", "c" * 40), ("a/b", "c" * 12), ("a/b", "")):
|
||
if _commit_url(bad_repo, bad_sha, "https://gitea.zheromiku.top"):
|
||
problems.append(f"仓库/短 SHA 不合法却仍拼出链接：{bad_repo!r}/{bad_sha[:12]!r}")
|
||
finally:
|
||
for k2, v2 in env_backup.items():
|
||
if v2 is None:
|
||
os.environ.pop(k2, None)
|
||
else:
|
||
os.environ[k2] = v2
|
||
if problems:
|
||
print("❌ 上报核对自检失败：", file=sys.stderr)
|
||
for p in problems:
|
||
print(f" · {p}", file=sys.stderr)
|
||
return 1
|
||
print(f"✅ 上报核对自检通过：{len(cases)} 个分支（短 SHA / 归错提交 / 缺项 / 重复 / 未知项 / "
|
||
f"清单漂移 / 自称成功 / 档案不符 / 子集报告 / 无清单 全部拦下）")
|
||
return 0
|
||
|
||
|
||
# ── 通知消息规范（ci-notification-message-standard §5）需要的字段 ──────────────
|
||
# 全部**由程序从真实事实算**：job 名来自 workflow 文件、范围来自门禁的同一个解析器、
|
||
# 统计来自权威清单。缺就缺（消息里显示"未取得"），绝不用写死的值凑格式。
|
||
def _job_name() -> str:
|
||
"""真实 job 名：从 workflow 文件里读 `jobs:` 下的第一个 job（不写死 build）。"""
|
||
for p in (ROOT / ".gitea" / "workflows").glob("*.y*ml"):
|
||
try:
|
||
txt = p.read_text(encoding="utf-8")
|
||
except Exception:
|
||
continue
|
||
m = re.search(r"^jobs:\s*$\n(.*)", txt, re.M | re.S)
|
||
if m:
|
||
m2 = re.search(r"^ ([A-Za-z0-9_.-]+):", m.group(1), re.M)
|
||
if m2:
|
||
return m2.group(1)
|
||
return os.environ.get("GITHUB_JOB", "")
|
||
|
||
|
||
def _scope_for_notify():
|
||
"""本次**准确范围**：复用门禁的同一个解析器（不另写一套范围逻辑）。
|
||
|
||
返回 (commits:[{sha,subject}], source:str, err:str)。取不到就给空 —— 不猜。
|
||
"""
|
||
try:
|
||
sys.path.insert(0, str(ROOT / ".gitea" / "scripts"))
|
||
import ci_gate # type: ignore
|
||
commits, source, err = ci_gate.resolve_scope()
|
||
out = []
|
||
for c in (commits or [])[:10]:
|
||
msg = str(c.get("message") or "").strip().splitlines()
|
||
out.append({"sha": str(c.get("sha") or ""), "subject": (msg[0] if msg else "")})
|
||
return out, source, err or ""
|
||
except Exception as e:
|
||
return [], "", f"{e.__class__.__name__}"
|
||
|
||
|
||
def _link_base() -> tuple:
|
||
"""通知里的链接用哪个基址：**CI_PUBLIC_BASE_URL 优先**，没配/非法就退回 GITHUB_SERVER_URL。
|
||
|
||
只接受 `http(s)://host[:port][/path]` 的绝对地址、不接受带凭据（userinfo）的写法。
|
||
返回 (base, source)；source 会写进上报与 CI 日志，方便事后核对"这条链接是谁给的基址"。
|
||
|
||
⚠️ 这是**唯一的**链接基址来源：不做"把消息文本里的 URL 全部替换一遍"那种事 ——
|
||
那样会把提交标题、日志摘要里的 URL 也改掉（那些是事实，不许动）。
|
||
"""
|
||
pub = (os.environ.get("CI_PUBLIC_BASE_URL") or "").strip()
|
||
fallback = (os.environ.get("GITHUB_SERVER_URL") or "").strip().rstrip("/")
|
||
if pub:
|
||
bad = None
|
||
if not re.match(r"^https?://", pub):
|
||
bad = "不是 http(s) 绝对地址"
|
||
elif " " in pub or " " in pub:
|
||
bad = "含空白字符"
|
||
else:
|
||
m = re.match(r"^[a-z]+://([^/]+)", pub, re.I)
|
||
if not m or "@" in m.group(1):
|
||
bad = "主机部分不合法或带凭据（userinfo）"
|
||
if bad:
|
||
return fallback, f"CI_PUBLIC_BASE_URL 非法（{bad}）⇒ 回退 GITHUB_SERVER_URL"
|
||
return pub.rstrip("/"), "CI_PUBLIC_BASE_URL"
|
||
return fallback, "GITHUB_SERVER_URL（未配 CI_PUBLIC_BASE_URL）"
|
||
|
||
|
||
def _repo_ok(repo: str) -> bool:
|
||
"""链接里的仓库必须是真的 owner/repo（两段、都非空），否则宁可不给链接。"""
|
||
parts = [x for x in str(repo or "").split("/")]
|
||
return len(parts) == 2 and all(x.strip() for x in parts)
|
||
|
||
|
||
def _commit_url(repo: str, sha: str, base: str) -> str:
|
||
if not _repo_ok(repo) or not base or not re.match(r"^[0-9a-f]{40}$", str(sha or "")):
|
||
return ""
|
||
return f"{base}/{repo}/commit/{sha}"
|
||
|
||
|
||
def _compare_url(repo: str, base: str) -> str:
|
||
"""变更列表链接：只有拿到真实的 before/after 才给（不然就是编一个打不开的地址）。"""
|
||
ev = (os.environ.get("GITHUB_EVENT_PATH") or "").strip()
|
||
if not base or not repo or not ev or not os.path.exists(ev):
|
||
return ""
|
||
try:
|
||
with open(ev, encoding="utf-8") as fh:
|
||
d = json.load(fh)
|
||
before, after = str(d.get("before") or ""), str(d.get("after") or "")
|
||
if (len(before) == 40 and len(after) == 40 and set(before) != {"0"}
|
||
and set(after) != {"0"} and before != after):
|
||
return f"{base}/{repo}/compare/{before}...{after}"
|
||
except Exception:
|
||
pass
|
||
return ""
|
||
|
||
|
||
def _first_failure(checks: list) -> dict:
|
||
"""首个失败项（优先真失败，其次"未运行/缺项"—— 两者都不等于通过）。"""
|
||
for c in checks or []:
|
||
if str(c.get("status")) == "failure":
|
||
return {"id": c.get("id"), "reason": str(c.get("reason") or "").strip()[:400]}
|
||
for c in checks or []:
|
||
if str(c.get("status")) in ("not_run", "unknown", "skipped"):
|
||
return {"id": c.get("id"),
|
||
"reason": f"未运行/跳过（{c.get('status')}）—— 不算通过"}
|
||
return {}
|
||
|
||
|
||
def _next_step(checks: list) -> str:
|
||
"""可复现入口：指向项目唯一门禁入口与该检查项（真实命令，不是建议口号）。"""
|
||
ff = _first_failure(checks)
|
||
if ff.get("id"):
|
||
return f"本地复现：make verify（或 python3 scripts/ci_checks.py --only {ff['id']}）"
|
||
return "本地复现：make verify（唯一门禁入口）后查看报告"
|
||
|
||
|
||
def _limitations(checks: list, rep: dict) -> list:
|
||
"""范围限制：**只写事实**（跳过了哪些检查、格式债基线还剩多少），不写"全部通过"。"""
|
||
out = []
|
||
skipped = [str(c.get("id")) for c in (checks or []) if str(c.get("status")) == "skipped"]
|
||
if skipped:
|
||
out.append("已跳过的检查（不算通过）：" + "、".join(skipped[:6]))
|
||
try:
|
||
base = json.loads((ROOT / "web" / ".prettier-baseline.json").read_text(encoding="utf-8"))
|
||
n = len(base.get("files") or base if isinstance(base, (list, dict)) else [])
|
||
if isinstance(base, dict):
|
||
n = len(base.get("files") or base.get("entries") or {})
|
||
if n:
|
||
out.append(f"前端格式债基线：{n} 个文件（只减不增，到期前必须清零）")
|
||
except Exception:
|
||
pass
|
||
for x in (rep.get("limitations") or [])[:3]:
|
||
out.append(str(x)[:160])
|
||
return out[:4]
|
||
|
||
|
||
def main(argv: list[str]) -> int:
|
||
ap = argparse.ArgumentParser()
|
||
ap.add_argument("--report", default=os.environ.get("CI_REPORT", str(ROOT / "ci-report.json")))
|
||
ap.add_argument("--no-post", action="store_true")
|
||
ap.add_argument("--verify-only", action="store_true",
|
||
help="只核对报告、不发上报（预览发布前用；退出码=0 通过 / 1 不通过 / 2 核不了）")
|
||
ap.add_argument("--self-test", action="store_true")
|
||
args = ap.parse_args(argv)
|
||
if args.self_test:
|
||
return self_test()
|
||
|
||
if GATE_ERROR:
|
||
print(f"❌ 核对不了报告：{GATE_ERROR}", file=sys.stderr)
|
||
print(" 本脚本需要本仓的权威门禁实现（scripts/ci_checks.py）来做清单核对；"
|
||
"找不到就不能给出任何结论。", file=sys.stderr)
|
||
return 2 if args.verify_only else 1
|
||
|
||
in_ci = bool(os.environ.get("GITHUB_ACTIONS"))
|
||
expect_sha = (os.environ.get("GITHUB_SHA") or "").strip()
|
||
|
||
rpath = Path(args.report)
|
||
if not rpath.is_file():
|
||
print(f"❌ 读不到门禁报告 {rpath} —— 这次运行的质量无从判定，不能报绿", file=sys.stderr)
|
||
return 2 if args.verify_only else 1
|
||
try:
|
||
rep = json.loads(rpath.read_text(encoding="utf-8"))
|
||
except json.JSONDecodeError as e:
|
||
print(f"❌ 门禁报告解析失败（{e}）—— 不能报绿", file=sys.stderr)
|
||
return 2 if args.verify_only else 1
|
||
if not isinstance(rep, dict):
|
||
print("❌ 门禁报告不是 JSON 对象 —— 不能报绿", file=sys.stderr)
|
||
return 2 if args.verify_only else 1
|
||
|
||
if args.verify_only:
|
||
# 预览发布前的这一道：只回答"这份报告能不能采信"，**不做任何网络请求**。
|
||
recomputed, vp, vn = verify(rep, expect_sha, in_ci)
|
||
print(f"── 发布前核对 —— 重算总体 {recomputed}｜报告自称 {rep.get('overall_status')}"
|
||
f"｜档案 {rep.get('profile')}｜run_mode {rep.get('run_mode')}｜"
|
||
f"SHA {(rep.get('commit_sha') or '')[:12]}")
|
||
for n in vn:
|
||
print(f" · {n}")
|
||
for p in vp:
|
||
print(f" ❌ {p}", file=sys.stderr)
|
||
print("✅ 报告可以采信" if not vp else "❌ 报告不可采信", file=sys.stderr if vp else sys.stdout)
|
||
return 1 if vp else 0
|
||
|
||
recomputed, problems, notes = verify(rep, expect_sha, in_ci)
|
||
checks = rep.get("checks") or []
|
||
overall = "failure" if problems else recomputed
|
||
|
||
print(f"── 门禁报告核对 ── 档案 {rep.get('profile')}｜"
|
||
f"提交 {(rep.get('commit_sha') or '')[:12]}｜重算总体 {recomputed}"
|
||
f"{'' if not problems else '（核对不通过）'}")
|
||
for c in checks:
|
||
mark = {"success": "✅", "skipped": "⏭️"}.get(c.get("status"), "❌")
|
||
print(f" {mark} {c.get('id',''):<18} {str(c.get('status')):<8} {str(c.get('reason') or '')[:100]}")
|
||
for n in notes:
|
||
print(f" · {n}")
|
||
for p in problems:
|
||
print(f" ❌ {p}", file=sys.stderr)
|
||
|
||
# 质量**无法判定**（报告缺失/核对不过）时，不能给出任何"成功"结论
|
||
if problems:
|
||
print("❌ 汇总不可信/不完整 ⇒ 不能报绿（作业这次不算通过）", file=sys.stderr)
|
||
|
||
_lb_base, _lb_source = _link_base()
|
||
repo_full = (rep.get("repository") or os.environ.get("GITHUB_REPOSITORY") or "").strip()
|
||
if _lb_source != "CI_PUBLIC_BASE_URL":
|
||
print(f"提示：通知链接基址 = {_lb_base or '（未取得）'}（{_lb_source}）", flush=True)
|
||
else:
|
||
print(f"通知链接基址 = {_lb_base}（来自 CI_PUBLIC_BASE_URL）", flush=True)
|
||
if not _repo_ok(repo_full):
|
||
print(f"⚠️ 仓库名不是 owner/repo（{repo_full[:40]!r}）⇒ 通知里不给 run/提交/变更 链接",
|
||
file=sys.stderr)
|
||
_scope_commits, _scope_source, _scope_err = _scope_for_notify()
|
||
_scope_payload = {"commits": _scope_commits, "source": _scope_source,
|
||
"count": len(_scope_commits)}
|
||
if _scope_err:
|
||
print(f"⚠️ 通知里不含本次范围摘要（范围解析：{_scope_err[:120]}）", file=sys.stderr)
|
||
payload_count = rd(os.environ.get("CI_COMMITS_COUNT_FILE", ""), "") or git(
|
||
"rev-list", "--count", "HEAD~1..HEAD")
|
||
|
||
payload = {
|
||
# —— bridge 既有字段（保持兼容，不改 bridge）——
|
||
"repo": rep.get("repository") or os.environ.get("GITHUB_REPOSITORY", "unknown/unknown"),
|
||
"sha": rep.get("commit_sha") or "",
|
||
"branch": rep.get("branch") or os.environ.get("GITHUB_REF_NAME", ""),
|
||
"actor": os.environ.get("GITHUB_ACTOR", ""),
|
||
"workflow": os.environ.get("GITHUB_WORKFLOW", "build"),
|
||
"status": BRIDGE_STATUS.get(overall, "failure"),
|
||
"commit_subject": git("log", "-1", "--format=%s"),
|
||
"commit_count": payload_count,
|
||
"duration_s": rep.get("duration_s", "?"),
|
||
"preview_url": rd(str(ROOT / ".ci-preview-url")),
|
||
"log": rd(os.environ.get("CI_GATE_LOG", ""))[-3500:],
|
||
"steps": [{"name": c.get("id"), "status": BRIDGE_STATUS.get(c.get("status"), "failure"),
|
||
"duration_s": c.get("duration_s")} for c in checks],
|
||
# —— 结构化字段（正本 §5 的检查报告形状）——
|
||
"schema": rep.get("schema"),
|
||
"policy_revision": rep.get("policy_revision"),
|
||
"overall_status": overall,
|
||
"claimed_overall_status": rep.get("overall_status"),
|
||
"recomputed_overall_status": recomputed,
|
||
"profile": rep.get("profile"),
|
||
"event": rep.get("event"),
|
||
"run_mode": rep.get("run_mode"),
|
||
"run_id": os.environ.get("GITHUB_RUN_ID", ""),
|
||
"run_url": (f"{_lb_base}/{repo_full}/actions/runs/{os.environ.get('GITHUB_RUN_ID', '')}"
|
||
if (os.environ.get("GITHUB_RUN_ID") and _lb_base and _repo_ok(repo_full)) else ""),
|
||
"checks": checks,
|
||
"manifest": rep.get("manifest"),
|
||
"problems": (rep.get("problems") or []) + problems,
|
||
"verification": {"ok": not problems, "problems": problems, "notes": notes,
|
||
"expected_sha": expect_sha, "reported_sha": rep.get("commit_sha")},
|
||
"notification_status": "not_attempted",
|
||
"deployed": False, # 本 workflow 只做 CI：没有部署，报告里也这么写
|
||
|
||
# —— 通知消息规范需要的字段（缺就缺，不许编）——
|
||
"schema_version": 2,
|
||
"job": _job_name(), # 真实 job 名（来自 workflow 文件）
|
||
"run_number": os.environ.get("GITHUB_RUN_NUMBER", ""),
|
||
"run_attempt": os.environ.get("GITHUB_RUN_ATTEMPT", "1"),
|
||
"commits": _scope_payload["commits"], # 准确范围里的提交标题（≤10 条）
|
||
"range_source": _scope_payload["source"], # 范围是怎么算出来的（可复盘）
|
||
"commit_count": (_scope_payload["count"] or payload_count), # 条数由 Git 算
|
||
"compare_url": _compare_url(repo_full, _lb_base),
|
||
"commit_url": _commit_url(repo_full, rep.get("commit_sha") or "", _lb_base),
|
||
"link_base": _lb_base, # 链接基址（可复盘）
|
||
"link_base_source": _lb_source, # 基址从哪来：CI_PUBLIC_BASE_URL / GITHUB_SERVER_URL 回退
|
||
"run_duration_s": "", # 整次运行耗时：本 workflow 没留起点 ⇒ 不编
|
||
"duration_kind": "gate",
|
||
"first_failure": _first_failure(checks),
|
||
"next_step": _next_step(checks),
|
||
"limitations": _limitations(checks, rep),
|
||
}
|
||
|
||
base = (os.environ.get("CI_BRIDGE_RESOLVED") or os.environ.get("BRIDGE_URL") or "").rstrip("/")
|
||
if args.no_post or not base:
|
||
print(f"跳过上报（--no-post={args.no_post}，bridge={base or '未找到'}）")
|
||
return 1 if problems else 0
|
||
|
||
data = json.dumps(payload, ensure_ascii=False).encode("utf-8")
|
||
req = urllib.request.Request(base + "/api/ci/report", data=data, method="POST",
|
||
headers={"Content-Type": "application/json",
|
||
"X-CI-Token": os.environ.get("BRIDGE_TOKEN", "")})
|
||
status = "failed"
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=30) as resp:
|
||
body = resp.read().decode()[:300]
|
||
status = "sent"
|
||
print(f"✅ 已上报 ci-bridge：{body}")
|
||
except (urllib.error.URLError, OSError) as e:
|
||
status = f"failed: {e}"
|
||
# 通知失败**不**改变质量结论：这里只标注，不把运行判红。
|
||
print(f"❌ 上报 ci-bridge 失败（质量结论不变，仅通知失败）：{e}", file=sys.stderr)
|
||
|
||
payload["notification_status"] = status
|
||
rep["notification_status"] = status
|
||
rpath.write_text(json.dumps(rep, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
||
return 1 if problems else 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main(sys.argv[1:]))
|