姚章浩
7090aafefc
ci-gate.py：Python 的依赖/格式/静态/构建/测试改成真的执行（uv sync --frozen / uv run ruff / compileall / uv run pytest），工具缺失即失败，不再空跑成功；前端测试把子进程退出码当判据（报告写 pass 但退出 1 也判红），框架按 test-scope.json 声明而不是写死 vitest；清单新增 security（密钥扫描 + 依赖漏洞）。预览逻辑进 ci_preview.py：先调 ci_report.verify()，核对不过一个 publish 请求都不发。检出进 ci_checkout.sh：协议原样保留并断言不降级、URL 不放凭据、顺带取事件 before。workflow 收成单作业，不再用 /tmp 跨作业传文件。
208 lines
8.5 KiB
Python
Executable File
208 lines
8.5 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
"""静态预览发布：**先核对完整报告，再发**。
|
||
|
||
为什么必须在这一步核对，而不是等后面的通知步骤：
|
||
预览是**发出去就收不回来**的（URL 会被别人打开、会被转到群里）。等到通知步骤才发现
|
||
"报告缺项 / SHA 不对 / 档案不对"，那时预览早就发出去了 —— 再红也撤不回来。
|
||
|
||
所以这里的顺序是：
|
||
① 调 `ci_report.verify()`（与上报**同一份**核对逻辑，不另写一套）；
|
||
② 有任何问题 ⇒ **一个 publish 请求都不发**，直接返回 0（质量结论由门禁那一步负责）；
|
||
③ 全部通过 ⇒ 打包（不产生 `.` 根成员）→ 断言包里有 index.html → POST publish
|
||
→ 报告里给出的**每个地址都探测 200** 才算发布成功。
|
||
|
||
用法：
|
||
python3 .gitea/scripts/ci_preview.py # 真发（CI 用）
|
||
python3 .gitea/scripts/ci_preview.py --check-only # 只核对报告，不发
|
||
python3 .gitea/scripts/ci_preview.py --self-test # 判定路径自检
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import os
|
||
import sys
|
||
import tarfile
|
||
import tempfile
|
||
import urllib.error
|
||
import urllib.request
|
||
from pathlib import Path
|
||
|
||
ROOT = Path(__file__).resolve().parent.parent.parent
|
||
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
||
sys.path.insert(0, str(ROOT / "scripts"))
|
||
import ci_report # noqa: E402 （与上报共用同一份核对逻辑）
|
||
|
||
ARTIFACT_CANDIDATES = ["web/dist", "dist", "build", "out", "public", "web/build", "frontend/dist"]
|
||
BRIDGE_CANDIDATES = ["http://172.17.0.1:8899", "http://192.168.50.134:8899", "http://100.92.49.78:8899"]
|
||
|
||
|
||
def find_artifact_dir() -> Path | None:
|
||
for rel in ARTIFACT_CANDIDATES:
|
||
d = ROOT / rel
|
||
if (d / "index.html").is_file():
|
||
return d
|
||
return None
|
||
|
||
|
||
def pack(web: Path, dest: Path) -> tuple[bool, str]:
|
||
"""打包成 tar.gz。⚠️ 不要产生 `.` 根成员（那会让预览 404 而流水线显示成功）。"""
|
||
names = []
|
||
with tarfile.open(dest, "w:gz") as tf:
|
||
for p in sorted(web.rglob("*")):
|
||
rel = p.relative_to(web)
|
||
if rel.parts and rel.parts[0] in (".", ".."):
|
||
continue
|
||
tf.add(p, arcname=str(rel))
|
||
names.append(str(rel))
|
||
if "index.html" not in names:
|
||
return False, f"包里没有 index.html（成员 {names[:5]}…）—— 发出去也是 404 的预览"
|
||
if any(n.startswith("./") or n == "." for n in names):
|
||
return False, f"包里出现了 `.` 根成员：{[n for n in names if n.startswith('./')][:3]}"
|
||
return True, f"{len(names)} 个成员"
|
||
|
||
|
||
def resolve_bridge(explicit: str = "") -> str:
|
||
cands = [explicit] + BRIDGE_CANDIDATES
|
||
for c in cands:
|
||
if not c:
|
||
continue
|
||
try:
|
||
with urllib.request.urlopen(c.rstrip("/") + "/healthz", timeout=5) as r:
|
||
if r.status == 200:
|
||
return c.rstrip("/")
|
||
except (urllib.error.URLError, OSError):
|
||
continue
|
||
return ""
|
||
|
||
|
||
def probe(url: str, timeout: int = 10) -> int:
|
||
try:
|
||
with urllib.request.urlopen(url, timeout=timeout) as r:
|
||
return r.status
|
||
except urllib.error.HTTPError as e:
|
||
return e.code
|
||
except (urllib.error.URLError, OSError):
|
||
return 0
|
||
|
||
|
||
def publish(base: str, repo: str, sha: str, token: str, blob: bytes) -> tuple[int, dict]:
|
||
req = urllib.request.Request(
|
||
f"{base}/api/ci/publish?repo={repo}&sha={sha}", data=blob, method="POST",
|
||
headers={"X-CI-Token": token, "Content-Type": "application/gzip"})
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=90) as r:
|
||
return r.status, json.loads(r.read().decode() or "{}")
|
||
except urllib.error.HTTPError as e:
|
||
return e.code, {"_error": e.read().decode()[:200]}
|
||
except (urllib.error.URLError, OSError) as e:
|
||
return 0, {"_error": str(e)}
|
||
|
||
|
||
def self_test() -> int:
|
||
"""打包器的判定：`.` 根成员必须被发现（不然就是"预览 404 而流水线绿"）。"""
|
||
problems = []
|
||
with tempfile.TemporaryDirectory() as d:
|
||
web = Path(d) / "dist"
|
||
(web / "assets").mkdir(parents=True)
|
||
(web / "index.html").write_text("<html></html>", encoding="utf-8")
|
||
(web / "assets" / "a.js").write_text("x", encoding="utf-8")
|
||
ok, why = pack(web, Path(d) / "good.tgz")
|
||
if not ok:
|
||
problems.append(f"正常产物打包失败：{why}")
|
||
empty = Path(d) / "empty"
|
||
empty.mkdir()
|
||
ok, why = pack(empty, Path(d) / "bad.tgz")
|
||
if ok:
|
||
problems.append("空产物目录竟然打包成功（没有 index.html 也算过）")
|
||
if problems:
|
||
print("❌ 预览脚本自检失败：", file=sys.stderr)
|
||
for p in problems:
|
||
print(f" · {p}", file=sys.stderr)
|
||
return 1
|
||
print("✅ 预览脚本自检通过：正常产物可打包；空产物/缺 index.html 被拒")
|
||
return 0
|
||
|
||
|
||
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("--check-only", action="store_true")
|
||
ap.add_argument("--self-test", action="store_true")
|
||
args = ap.parse_args(argv)
|
||
if args.self_test:
|
||
return self_test()
|
||
|
||
if ci_report.GATE_ERROR:
|
||
print(f"❌ 核对不了报告（{ci_report.GATE_ERROR}）⇒ 不发预览", file=sys.stderr)
|
||
return 2
|
||
|
||
rpath = Path(args.report)
|
||
if not rpath.is_file():
|
||
print(f"❌ 读不到报告 {rpath} ⇒ 不发预览", file=sys.stderr)
|
||
return 1
|
||
try:
|
||
rep = json.loads(rpath.read_text(encoding="utf-8"))
|
||
except json.JSONDecodeError as e:
|
||
print(f"❌ 报告解析失败（{e}）⇒ 不发预览", file=sys.stderr)
|
||
return 1
|
||
|
||
in_ci = bool(os.environ.get("GITHUB_ACTIONS"))
|
||
expect_sha = (os.environ.get("GITHUB_SHA") or "").strip()
|
||
recomputed, problems, notes = ci_report.verify(rep, expect_sha, in_ci)
|
||
print(f"── 发布前核对 ── 重算总体 {recomputed}｜报告自称 {rep.get('overall_status')}"
|
||
f"｜档案 {rep.get('profile')}｜run_mode {rep.get('run_mode')}")
|
||
for n in notes:
|
||
print(f" · {n}")
|
||
if problems:
|
||
# 关键：**一个字节都不发**。预览发出去了就收不回来。
|
||
for p in problems:
|
||
print(f" ❌ {p}", file=sys.stderr)
|
||
print("❌ 报告核对不通过 ⇒ **不发出任何 publish 请求**", file=sys.stderr)
|
||
return 0
|
||
if args.check_only:
|
||
print("✅ 核对通过（--check-only：不发）")
|
||
return 0
|
||
|
||
web = find_artifact_dir()
|
||
if web is None:
|
||
print("未发现静态网页产物目录 ⇒ 不发预览（这是配置问题，不是通过）")
|
||
return 0
|
||
sha = str(rep.get("commit_sha") or "")
|
||
with tempfile.TemporaryDirectory() as d:
|
||
tar = Path(d) / "preview.tgz"
|
||
ok, why = pack(web, tar)
|
||
if not ok:
|
||
print(f"❌ 打包不通过：{why} ⇒ 不发", file=sys.stderr)
|
||
return 0
|
||
blob = tar.read_bytes()
|
||
print(f" 打包完成：{why}（{len(blob)} 字节，目录 {web.relative_to(ROOT)}）")
|
||
|
||
base = resolve_bridge(os.environ.get("CI_BRIDGE_RESOLVED") or os.environ.get("BRIDGE_URL") or "")
|
||
if not base:
|
||
print("❌ 没有可达的 ci-bridge ⇒ 不发预览（打包成功不等于预览发出去了）", file=sys.stderr)
|
||
return 0
|
||
code, body = publish(base, rep.get("repository") or os.environ.get("GITHUB_REPOSITORY", ""),
|
||
sha, os.environ.get("BRIDGE_TOKEN", ""), blob)
|
||
print(f" publish HTTP {code} {json.dumps(body, ensure_ascii=False)[:200]}")
|
||
if code != 200:
|
||
print(f"❌ 发布接口返回 {code} ⇒ 不发这条预览", file=sys.stderr)
|
||
return 0
|
||
for key in ("url", "url_tailscale"):
|
||
url = str(body.get(key) or "")
|
||
if not url:
|
||
print(f" ⚠️ 报告里 {key} 为空")
|
||
continue
|
||
st = probe(url)
|
||
print(f" 预览探测 {key}: {url} → {st}")
|
||
if st != 200:
|
||
print(f"❌ {url} 不是 200（通知里给出打不开的链接等于没给）", file=sys.stderr)
|
||
return 0
|
||
if key == "url":
|
||
(ROOT / ".ci-preview-url").write_text(url + "\n", encoding="utf-8")
|
||
print("✅ 预览已发布，且报告里的每个地址都 200")
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main(sys.argv[1:])) |