From 1651bc8d7d9645215bd3524b1312f6b607048ba1 Mon Sep 17 00:00:00 2001 From: DKChan <524912269@qq.com> Date: Sun, 18 Oct 2026 00:54:26 +0800 Subject: [PATCH] =?UTF-8?q?tools(pipeline):=20=E6=96=B0=E5=A2=9E=E6=B5=81?= =?UTF-8?q?=E6=B0=B4=E7=BA=BF=E9=97=B8=E2=80=94=E2=80=9406=20=E6=9D=A1?= =?UTF-8?q?=E7=9B=AE=E7=8A=B6=E6=80=81=E6=9C=BA=EF=BC=88=E6=A0=87=E9=A2=98?= =?UTF-8?q?=E5=85=83=E6=95=B0=E6=8D=AE/=E5=B1=9E=E6=80=A7=E9=BD=90?= =?UTF-8?q?=E5=A4=87/=E7=8A=B6=E6=80=81=E8=AF=8D=E7=99=BD=E5=90=8D?= =?UTF-8?q?=E5=8D=95/=E6=9D=A1=E4=BB=B6=E5=87=AD=E6=8D=AE/=E5=8F=8C?= =?UTF-8?q?=E8=BD=A8/=E7=89=B9=E6=89=B9/=E9=9B=B6=E5=91=BD=E4=B8=AD?= =?UTF-8?q?=EF=BC=89=EF=BC=8Broadmap=20=E8=BF=AD=E4=BB=A3=E6=80=BB?= =?UTF-8?q?=E8=A1=A8=E5=85=AD=E5=88=97=EF=BC=8B=E5=B7=B2=E7=BC=96=E8=BF=AD?= =?UTF-8?q?=E4=BB=A3=E2=86=94=E5=8D=A1=E7=89=87=E5=88=97=E4=BA=A4=E5=8F=89?= =?UTF-8?q?=E6=A0=B8=E4=B8=80=E5=A4=84=EF=BC=8C=E5=8F=98=E5=BC=82=E9=AA=8C?= =?UTF-8?q?=E8=AF=81=2025=20=E5=8F=98=E4=BD=93=E5=85=A8=E7=BA=A2=EF=BC=8B?= =?UTF-8?q?=E5=9F=BA=E5=87=86=E7=BB=BF=EF=BC=9B=E6=9A=82=E4=B8=8D=E6=B3=A8?= =?UTF-8?q?=E5=86=8C=20run=5Fall=5Fgates=EF=BC=88=E7=AC=AC=E4=BA=8C?= =?UTF-8?q?=E6=89=B9=E6=95=B0=E6=8D=AE=E6=95=B4=E6=94=B9=E5=90=8E=E4=B8=8A?= =?UTF-8?q?=E9=97=B8=EF=BC=8C=E5=AF=B9=E7=8E=B0=E8=A1=8C=E6=97=A7=E6=A0=BC?= =?UTF-8?q?=E5=BC=8F=2006/roadmap=20=E7=BA=A2=E5=B1=9E=E9=A2=84=E6=9C=9F?= =?UTF-8?q?=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- tools/pipeline_gate.py | 349 +++++++++++++++++++++++++++++++++++++++++ 1 file changed, 349 insertions(+) create mode 100644 tools/pipeline_gate.py diff --git a/tools/pipeline_gate.py b/tools/pipeline_gate.py new file mode 100644 index 00000000..9ad5379f --- /dev/null +++ b/tools/pipeline_gate.py @@ -0,0 +1,349 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +r"""pipeline_gate.py — 工作流程流水线闸（06 条目状态机＋roadmap 迭代总表＋一次交叉核）。 + +职责：机检 06-待办清单条目（标题元数据/属性齐备/状态词白名单/条件凭据/双轨/特批/零命中） + 与 roadmap 迭代总表（固定六列/状态四形态/卡片与报告存在性/周期≤2 周/单元格白名单/零命中）； + 「已编迭代↔迭代行卡片列」交叉核只在此一处实现。 +约束：每红项一行「文件:行号 判据名 一句白话说明」，exit 0=全绿 / 1=有红 / 2=输入件缺失； + 卡片与报告存在性相对 --repo-root 解析（缺省=本文件上两级）；现行 06 与 roadmap 尚为旧格式， + 对其红属预期（第二批数据整改后才注册上闸）。 +用法：python tools/pipeline_gate.py [--todo PATH] [--roadmap PATH] [--repo-root PATH] +""" +import argparse +import re +import sys +from pathlib import Path + +DEFAULT_TODO = Path("design/gdd/06-待办清单.md") +DEFAULT_ROADMAP = Path("roadmap.md") + +SOURCE_WORDS = ("playnote", "会话提及", "审计发现", "批次移交") +FACE_WORDS = ("文档轨", "美术轨", "双轨", "工程直入", "反哺点") +STATUS_WORDS = ( + "待文档设计", "文档设计中", "文档设计已闭合", + "待美术设计", "美术设计中", "美术设计已闭合", + "待写卡", "已写卡", "已编迭代", "待裁决", "已写入", +) +# 这些状态 = 设计未闭合，条目必须带「**缺口**」属性；其余状态不得带 +GAP_REQUIRED = ("待文档设计", "文档设计中", "待美术设计", "美术设计中") +ROADMAP_HEADER = ("迭代代号", "状态", "需求卡片", "编入时间", "预计周期", "实现报告") +RESERVED_RETIRED = "已退役（git 历史可考）" + +DATE_RE = re.compile(r"\d{4}-\d{2}-\d{2}") +VER_RE = re.compile(r"[vV]\d+(?:\.\d+)*") +RESEARCH_RE = re.compile(r"design/research/[^\s（）()，,]+") +ITER_RE = re.compile(r"迭代\s*(\d+)") +APPROVAL_RE = re.compile(r"〔用户特批([^〕]*)〕") +CRED_RE = re.compile(r"[（(]([^（）()]*)[）)]") +ATTR_LINE_RE = re.compile(r"^\*\*(现状|缺口|归属|状态)\*\*\s*[：:]\s*(.*)$") +HEADING_RE = re.compile(r"^(#{1,4})\s+") +SEP_ROW_RE = re.compile(r"^\|[\s\-|:]+$") +STATUS_COL_RE = re.compile( + r"^(待开工|进行中|已完成\(\d{4}-\d{2}-\d{2}\)|已终止\(\d{4}-\d{2}-\d{2}\))$") +WEEKS_RE = re.compile(r"^≈(\d+(?:\.\d+)?)\s*周$") +ITER_CODE_COL_RE = re.compile(r"^迭代\s*(\d+)") + + +def norm_parens(text): + """全角括号归一成半角，供状态/保留字形态比对。""" + return text.replace("（", "(").replace("）", ")") + + +def display(path_str, root): + """红项输出用的路径：仓库根内的相对 posix 路径，根外原样。""" + p = Path(path_str) + try: + return p.resolve().relative_to(root).as_posix() + except (ValueError, OSError): + return path_str.replace("\\", "/") + + +def parse_todo(text): + """按标题切条目：#### 开行为条目起点，更高层级标题（#/##/###）同为上条终点。""" + entries = [] + cur = None + for lineno, raw in enumerate(text.splitlines(), start=1): + m = HEADING_RE.match(raw) + if m: + if cur is not None: + entries.append(cur) + cur = None + if len(m.group(1)) == 4: + cur = {"line": lineno, "title": raw.lstrip("#").strip(), "body": []} + elif cur is not None: + cur["body"].append((lineno, raw)) + if cur is not None: + entries.append(cur) + return entries + + +def check_title(title, line, issues, where): + """判据 1：标题须含日期、来源=四选一、设计面=五选一。返回设计面词（可能 None）。""" + if not DATE_RE.search(title): + issues.append((where, line, "标题元数据", "标题缺登记日期（YYYY-MM-DD 形态）")) + face = None + m = re.search(r"来源=([^\s，,（()）]+)", title) + if not m: + issues.append((where, line, "标题元数据", "标题缺「来源=」标记")) + elif m.group(1) not in SOURCE_WORDS: + issues.append((where, line, "标题元数据", + "来源词「%s」不在四选一（playnote/会话提及/审计发现/批次移交）" % m.group(1))) + m = re.search(r"设计面=([^\s，,（()）]+)", title) + if not m: + issues.append((where, line, "标题元数据", "标题缺「设计面=」标记")) + elif m.group(1) not in FACE_WORDS: + issues.append((where, line, "标题元数据", + "设计面词「%s」不在五选一（文档轨/美术轨/双轨/工程直入/反哺点）" % m.group(1))) + else: + face = m.group(1) + return face + + +def check_segment(word, creds, line, issues, where, root): + """判据 3 的单段校验。返回 (合法状态词|None, 已写卡路径|None, 已编迭代号|None)。""" + def red(name, msg): + issues.append((where, line, name, msg)) + + im = re.match(r"^已编迭代\s*(\d+)$", word) + base = "已编迭代" if word.startswith("已编迭代") else word + if im is None and base == "已编迭代": + red("状态词非法", "「%s」缺迭代号（须「已编迭代 N」）" % word) + return None, None, None + if base not in STATUS_WORDS: + red("状态词非法", "「%s」不在合法状态词白名单" % word) + return None, None, None + cred_text = "；".join(creds) + if base in ("文档设计已闭合", "已写入"): + if not (VER_RE.search(cred_text) and ("/" in cred_text or ".md" in cred_text)): + red("凭据缺失", "「%s」须带括号落点凭据（文档路径＋版本号形态）" % base) + elif base == "美术设计已闭合": + if not ("art-bible" in cred_text.lower() and VER_RE.search(cred_text)): + red("凭据缺失", "「美术设计已闭合」凭据须含 art-bible 路径＋版本号") + elif base == "已写卡": + rp = None + for c in creds: + m = RESEARCH_RE.search(c) + if m: + rp = m.group(0) + break + if rp is None: + red("凭据缺失", "「已写卡」括号内须为 design/research/ 下的卡片路径") + elif not (root / rp).is_file(): + red("卡片路径不存在", "已写卡路径 %s 相对仓库根不存在" % rp) + return base, rp, None + if im is not None: + return base, None, int(im.group(1)) + return base, None, None + + +def check_status(raw, line, face, issues, where, root): + """判据 3＋双轨/特批/备选。返回 (状态词列表, 已写卡路径列表, 已编迭代号|None)。""" + content = raw.strip() + m = re.match(r"^\*\*(.+)\*\*$", content) + if m: + content = m.group(1).strip() + for am in APPROVAL_RE.finditer(content): + if not DATE_RE.search(am.group(1)): + issues.append((where, line, "特批标记", "〔用户特批〕缺日期（带 YYYY-MM-DD 才放行跳轨）")) + content = APPROVAL_RE.sub("", content) + content = content.replace("（备选）", "").replace("(备选)", "").strip() + segs = [s.strip() for s in content.split("＋") if s.strip()] + if not segs: + issues.append((where, line, "状态词非法", "状态行内容为空")) + return [], [], None + if len(segs) > 2: + issues.append((where, line, "双轨段数", "状态 %d 段（＋连接时恰好两段）" % len(segs))) + if face == "双轨" and len(segs) != 2: + issues.append((where, line, "双轨段数", "设计面=双轨的状态须恰好两段（现 %d 段）" % len(segs))) + words, cards, iter_no = [], [], None + for seg in segs: + creds = CRED_RE.findall(seg) + word = CRED_RE.sub("", seg).strip() + w, rp, it = check_segment(word, creds, line, issues, where, root) + if w: + words.append(w) + if rp: + cards.append(rp) + if it is not None: + iter_no = it + return words, cards, iter_no + + +def check_todo(text, issues, where, root): + """06 全部判据。返回交叉核链（已编迭代条目 → (行号, 迭代号, 卡片路径|None)）。""" + entries = parse_todo(text) + links = [] + if not entries: + issues.append((where, 1, "零命中自检", + "未解析出任何 #### 条目（路径错/文件空/纯垃圾文本＝假绿防线）")) + return links + for e in entries: + line = e["line"] + face = check_title(e["title"], line, issues, where) + attrs, attr_line = {}, {} + for ln, raw in e["body"]: + m = ATTR_LINE_RE.match(raw.strip()) + if m and m.group(1) not in attrs: + attrs[m.group(1)] = m.group(2) + attr_line[m.group(1)] = ln + for req in ("现状", "归属", "状态"): + if req not in attrs: + issues.append((where, line, "属性缺失", "条目缺「**%s**」属性" % req)) + if "状态" not in attrs: + continue + words, cards, iter_no = check_status( + attrs["状态"], attr_line["状态"], face, issues, where, root) + need_gap = any(w in GAP_REQUIRED for w in words) + if need_gap and "缺口" not in attrs: + issues.append((where, line, "缺口属性", + "状态为设计未闭合态（%s），须有「**缺口**」属性" % "、".join(words))) + if not need_gap and words and "缺口" in attrs: + issues.append((where, line, "缺口属性", + "状态为 %s，设计已闭合/后段不得残留「**缺口**」属性" % "、".join(words))) + if iter_no is not None: + links.append((line, iter_no, cards[0] if cards else None)) + return links + + +def row_cells(row): + return [c.strip() for c in row.split("|")[1:-1]] + + +def cell_shape(cell): + """判据 7：单元格白名单外的形态说明；合法返回 None。""" + if cell == "": + return "空白单元格" + if re.fullmatch(r"〔[^〕]*〕", cell): + return None if DATE_RE.search(cell) else "特批标记缺日期" + if "。" in cell or "，" in cell or len(cell) > 60: + return "成段叙述（只许路径/日期/状态词/保留字/—/≈N 周/特批标记）" + return None + + +def check_iter_table(block, issues, where, root): + """判据 1-7 逐行查。返回 {迭代号: (行号, 需求卡片列原文)} 供交叉核。""" + index = {} + hdr_line, hdr = block[0] + cells = row_cells(hdr) + if tuple(cells) != ROADMAP_HEADER: + issues.append((where, hdr_line, "总表表头", + "表头须为「%s」（现：%s）" % ("｜".join(ROADMAP_HEADER), "、".join(cells)))) + data = [(ln, r) for ln, r in block[1:] if not SEP_ROW_RE.match(r)] + if not data: + issues.append((where, hdr_line, "零命中自检", "迭代总表解析出 0 数据行")) + return index + for ln, r in data: + cells = row_cells(r) + if len(cells) != 6: + issues.append((where, ln, "六列数", "本行 %d 列（须恰六列）" % len(cells))) + continue + code, status, cards, date, weeks, report = cells + for ci, cell in enumerate(cells): + bad = cell_shape(cell) + if bad: + issues.append((where, ln, "单元格白名单", + "第 %d 列「%s」：%s" % (ci + 1, cell[:24], bad))) + im = ITER_CODE_COL_RE.match(code) + if not im: + issues.append((where, ln, "迭代代号", "「%s」缺迭代号（须「迭代 N」形态）" % code)) + status_n = norm_parens(status) + if not STATUS_COL_RE.match(status_n): + issues.append((where, ln, "总表状态列", + "「%s」不在四形态（待开工/进行中/已完成（YYYY-MM-DD）/已终止（YYYY-MM-DD））" % status)) + if norm_parens(cards) != norm_parens(RESERVED_RETIRED): + for part in (p.strip() for p in cards.split("、")): + if part and not (root / part).is_file(): + issues.append((where, ln, "卡片路径不存在", + "需求卡片列 %s 相对仓库根不存在" % part)) + if not re.fullmatch(r"\d{4}-\d{2}-\d{2}", date): + issues.append((where, ln, "编入时间", "「%s」须为 YYYY-MM-DD" % date)) + wm = WEEKS_RE.match(weeks) + if not wm or float(wm.group(1)) > 2: + issues.append((where, ln, "预计周期", "「%s」须为「≈N 周」且 N≤2（允许 0.5 步进）" % weeks)) + done = status_n.startswith("已完成(") + report_n = norm_parens(report) + if report_n == "—": + if done: + issues.append((where, ln, "实现报告", "状态已完成但实现报告为「—」（须为存在路径）")) + elif report_n == norm_parens(RESERVED_RETIRED): + if done: + issues.append((where, ln, "实现报告", "状态已完成但实现报告为退役保留字（须为存在路径）")) + elif not (report.startswith("reports/") and (root / report).is_file()): + issues.append((where, ln, "实现报告", + "「%s」须为「—」/退役保留字/reports/ 下存在路径" % report)) + if im: + index.setdefault(int(im.group(1)), (ln, cards)) + return index + + +def check_roadmap(text, issues, where, root): + """定位迭代总表（表头含「迭代代号」的表格）并逐行查。返回交叉核索引。""" + index = {} + lines = text.splitlines() + found = False + i = 0 + while i < len(lines): + if lines[i].lstrip().startswith("|"): + block = [] + while i < len(lines) and lines[i].lstrip().startswith("|"): + block.append((i + 1, lines[i].strip())) + i += 1 + if not found and "迭代代号" in row_cells(block[0][1]): + found = True + index = check_iter_table(block, issues, where, root) + else: + i += 1 + if not found: + issues.append((where, 1, "零命中自检", + "未找到迭代总表（表头须含「迭代代号」的六列表格）")) + return index + + +def cross_check(links, index, issues, where): + """唯一一处交叉核：已编迭代 N ↔ 总表迭代 N 行的需求卡片列。""" + for line, no, card in links: + row = index.get(no) + if row is None: + issues.append((where, line, "交叉核", "迭代总表无迭代 %d 行" % no)) + elif card is None: + issues.append((where, line, "交叉核", + "已编迭代 %d 但条目无可引的已写卡路径（须先已写卡）" % no)) + elif card not in row[1]: + issues.append((where, line, "交叉核", + "迭代 %d 行卡片列不含本条目已写卡路径 %s" % (no, card))) + + +def main(): + ap = argparse.ArgumentParser( + description="工作流程流水线闸：06 条目状态机＋roadmap 迭代总表＋一次交叉核") + ap.add_argument("--todo", help="06-待办清单路径（缺省=<仓库根>/design/gdd/06-待办清单.md）") + ap.add_argument("--roadmap", help="roadmap 路径（缺省=<仓库根>/roadmap.md）") + ap.add_argument("--repo-root", help="仓库根（卡片/报告存在性的相对基准；缺省=本文件上两级）") + args = ap.parse_args() + root = (Path(args.repo_root).resolve() if args.repo_root + else Path(__file__).resolve().parent.parent) + todo_path = Path(args.todo) if args.todo else root / DEFAULT_TODO + roadmap_path = Path(args.roadmap) if args.roadmap else root / DEFAULT_ROADMAP + missing = [p for p in (todo_path, roadmap_path) if not p.is_file()] + if missing: + print("输入件缺失: %s" % "、".join(str(m) for m in missing), file=sys.stderr) + return 2 + issues = [] # (文件, 行号, 判据名, 一句白话说明) + todo_text = todo_path.read_text(encoding="utf-8-sig") + roadmap_text = roadmap_path.read_text(encoding="utf-8-sig") + links = check_todo(todo_text, issues, str(todo_path), root) + index = check_roadmap(roadmap_text, issues, str(roadmap_path), root) + cross_check(links, index, issues, str(todo_path)) + for where, line, name, msg in issues: + print("%s:%d %s %s" % (display(where, root), line, name, msg)) + if issues: + print("流水线闸 FAIL：红项 %d" % len(issues)) + return 1 + print("流水线闸 PASS：06 条目 %d · 总表数据行 %d · 交叉核 %d 链全过" + % (len(parse_todo(todo_text)), len(index), len(links))) + return 0 + + +if __name__ == "__main__": + sys.exit(main())