#!/usr/bin/env python3 # -*- coding: utf-8 -*- r"""pipeline_gate.py — 工作流程流水线闸(待办清单条目状态机+roadmap 迭代总表+需求卡片口径+一次交叉核)。 职责:机检待办清单条目(标题元数据/属性齐备/状态词白名单/条件凭据/状态段累积制在场约束/特批/ 零命中)与 roadmap 迭代总表(六列/状态四形态/卡片与报告存在性/编入时间与预计周期的 历史行「—」豁免/周期≤2 周可特批豁免/单元格白名单/零命中);「已编迭代↔已写卡段↔迭代行 卡片列」交叉核只此一处。2026-10-18 卡片冲突修订批补两条卡片机检:C1 关联文档同步面的 文档义务口径(生效 2026-10-17)、C2 设计事实源声明(生效 2026-10-18)——两条都按文件名 日期只追新规生效日之后的卡,旧卡属 git 历史留档不回头改。 约束:每红项一行「文件:行号 判据名 一句白话说明」,exit 0=全绿 / 1=有红 / 2=输入件缺失; 卡片与报告存在性相对 --repo-root 解析(缺省=本文件上两级);待办清单与 roadmap 均已为新格式 (2026-10-18 数据整改批),本闸已注册进 tools/run_all_gates.sh 文档域(08c 流水线闸)。 用法: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("待办清单.md") DEFAULT_ROADMAP = Path("roadmap.md") SOURCE_WORDS = ("playnote", "会话提及", "审计发现", "批次移交") FACE_WORDS = ("文档轨", "美术轨", "双轨", "工程直入", "反哺点") STATUS_WORDS = ( "待文档设计", "文档设计中", "文档设计已闭合", "待美术设计", "美术设计中", "美术设计已闭合", "待写卡", "已写卡", "已编迭代", "待裁决", "已写入", ) # 这些状态 = 设计未闭合,条目必须带「**缺口**」属性;其余状态不得带 GAP_REQUIRED = ("待文档设计", "文档设计中", "待美术设计", "美术设计中") # 状态段累积制(doc-conventions §12.1):下游段在场时的上游闭合段在场约束 DOWNSTREAM_WORDS = ("待写卡", "已写卡", "已编迭代") PRESENCE_FACES = ("文档轨", "美术轨", "双轨") DOC_DESIGN_WORDS = ("待文档设计", "文档设计中", "文档设计已闭合") ART_DESIGN_WORDS = ("待美术设计", "美术设计中", "美术设计已闭合") ROADMAP_HEADER = ("迭代代号", "状态", "需求卡片", "编入时间", "预计周期", "实现报告") RESERVED_RETIRED = "已退役(git 历史可考)" # ---- 卡片机检(2026-10-18 卡片冲突修订批补)---- CARD_GLOB = "需求卡片-*.md" CARD_DIR = "reports" SYNC_RULE_DATE = (2026, 10, 17) # req-card §6:实现批禁写 design/ 与根规则文件,义务一律列反哺点 SOURCE_RULE_DATE = (2026, 10, 18) # req-card §2:设计事实源必填 SOURCE_MARKS = ("设计事实源", "唯一事实源") # 两种合格写法,与库内既有卡片实际形态对齐 SYNC_HEADING_RE = re.compile(r"^#{2,3}\s*\d*\.?\s*关联文档同步面") BANNED_DOC_MARKS = ("design/", "../gdd/", "../registry/", "../research/", "../qa/", "../engineering/", "../ux/", "../art/", "roadmap.md", "AGENTS.md") DATE_RE = re.compile(r"\d{4}-\d{2}-\d{2}") VER_RE = re.compile(r"[vV]\d+(?:\.\d+)*") RESEARCH_RE = re.compile(r"reports/需求卡片-[^\s()(),,]+") ITER_RE = re.compile(r"迭代\s*(\d+)") APPROVAL_RE = re.compile(r"〔用户特批([^〕]*)〕") CRED_RE = re.compile(r"[((]([^()()]*)[))]") CARD_SEP_RE = re.compile(r"|\n|、") # 迭代行卡片列一格多卡的分隔(§12.2 一格一行一卡) 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, line, issues, where): """判据 3 的单段校验。返回 (合法状态词|None, 已编迭代号|None)。 2026-10-20 状态栏纯化批:状态行只写状态词(§12.1),括号凭据已废止—— 闭合证据改到现状关联文档清单核对(check_todo 内),本函数只认状态词本身。 """ 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 if base not in STATUS_WORDS: red("状态词非法", "「%s」不在合法状态词白名单" % base) return None, None if im is not None: return base, int(im.group(1)) return base, None def check_status(raw, line, face, issues, where): """判据 3+状态段累积制(§12.1)+特批/备选。返回 (状态词列表, 已编迭代号|None)。 累积制:按+切段后段数不限,逐段各自合法;再查两条结构约束—— 下游段(待写卡/已写卡/已编迭代)在场时上游闭合段必须在场(带日期特批可豁免); 双轨设计阶段(无下游段)必须恰两段且一文档一美术。 状态行只写状态词(§12.1):去掉特批标记与(备选)后缀后,残留任何括号即红—— 证据写现状关联文档清单,由 check_todo 到正文核对;此处点名病因后短路, 不再产出碎片「状态词非法」误导红。 """ content = raw.strip() m = re.match(r"^\*\*(.+)\*\*$", content) if m: content = m.group(1).strip() dated_approval = False for am in APPROVAL_RE.finditer(content): if DATE_RE.search(am.group(1)): dated_approval = True else: issues.append((where, line, "特批标记", "〔用户特批〕缺日期(带 YYYY-MM-DD 才放行跳轨)")) content = APPROVAL_RE.sub("", content) content = content.replace("(备选)", "").replace("(备选)", "").strip() if re.search(r"[(())]", content): issues.append((where, line, "状态行含括号内容", "括号里的信息(路径/版本/说明)写进现状关联文档清单,状态行只留状态词(§12.1)")) return [], None segs = [s.strip() for s in content.split("+") if s.strip()] if not segs: issues.append((where, line, "状态词非法", "状态行内容为空")) return [], None words, iter_no = [], None for seg in segs: w, it = check_segment(seg, line, issues, where) if w: words.append(w) if it is not None: iter_no = it downstream = [w for w in words if w in DOWNSTREAM_WORDS] if downstream and face in PRESENCE_FACES and not dated_approval: need = [] if face in ("文档轨", "双轨") and "文档设计已闭合" not in words: need.append("文档设计已闭合") if face in ("美术轨", "双轨") and "美术设计已闭合" not in words: need.append("美术设计已闭合") if need: issues.append((where, line, "在场约束", "下游段(%s)在场但缺上游闭合段 %s(带日期特批可豁免)" % ("、".join(downstream), "、".join(need)))) if face == "双轨" and not downstream: n_doc = sum(1 for w in words if w in DOC_DESIGN_WORDS) n_art = sum(1 for w in words if w in ART_DESIGN_WORDS) if len(segs) != 2 or n_doc != 1 or n_art != 1: issues.append((where, line, "双轨段数", "设计面=双轨设计阶段须恰两段且一文档一美术(现 %d 段:文档轨 %d、美术轨 %d)" % (len(segs), n_doc, n_art))) return words, iter_no def check_todo(text, issues, where, root): """待办清单全部判据。返回交叉核链(已编迭代条目 → (行号, 迭代号, 卡片路径|None))。 2026-10-20 状态栏纯化批:闭合类状态词的证据核对从状态栏括号改到现状关联文档 清单(§12.1「状态行只写状态词」)——已写卡对 reports/ 卡片路径+磁盘存在性, 文档轨闭合/已写入对 design/ 路径+版本号,美术轨闭合对 art-bible+版本号。 """ 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, iter_no = check_status( attrs["状态"], attr_line["状态"], face, issues, where) # 现状属性全文(含清单续行):证据核对只看现状,其他属性不算数 xian_lines, in_xian = [], False for ln, raw in e["body"]: m = ATTR_LINE_RE.match(raw.strip()) if m: in_xian = m.group(1) == "现状" if in_xian: xian_lines.append(m.group(2)) elif in_xian: xian_lines.append(raw) xian = "\n".join(xian_lines) xline = attr_line.get("现状", line) # 证据核对一:已写卡 → 现状清单须有卡片路径,逐条验磁盘存在 cards = [] if "已写卡" in words: cards = RESEARCH_RE.findall(xian) if not cards: issues.append((where, xline, "正文缺卡片路径", "状态段「已写卡」在场,但现状关联文档清单里没有 " "reports/需求卡片-*.md 路径(§12.1:证据写正文)")) else: for rp in cards: if not (root / rp).is_file(): issues.append((where, line, "卡片路径不存在", "现状清单里的卡片路径 %s 相对仓库根不存在" % rp)) # 证据核对二:文档轨闭合/反哺已写入 → 现状清单须有文档路径+版本号 # (待办清单正文里的指针是相对路径,形如 systems/… 或 design/…,只认 .md 收尾+版本号) if "文档设计已闭合" in words or "已写入" in words: if not (re.search(r"[^\s ,,()()]+\.md", xian) and VER_RE.search(xian)): issues.append((where, xline, "正文缺设计定稿指针", "状态段含文档轨闭合词,但现状清单里没有带版本号的文档路径(§12.1:证据写正文)")) # 证据核对三:美术轨闭合 → 现状清单须有 art-bible 版本凭据 if "美术设计已闭合" in words: if not ("art-bible" in xian.lower() and VER_RE.search(xian)): issues.append((where, xline, "正文缺美术凭据", "状态段「美术设计已闭合」在场,但现状清单里没有 " "art-bible 版本凭据(§12.1:证据写正文)")) 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, card_col=False): """判据 7:单元格白名单外的形态说明;合法返回 None。 需求卡片列按多卡分隔符逐段看叙述、总长不设限(§12.2 一格一行一卡是合法形态), 路径存在性由列规则逐卡承担;其余列维持整格叙述检查。 """ if cell == "": return "空白单元格" if re.fullmatch(r"〔[^〕]*〕", cell): return None if DATE_RE.search(cell) else "特批标记缺日期" if card_col: if any(mark in part for part in CARD_SEP_RE.split(cell) for mark in ("。", ",")): return "成段叙述(卡片列一格一行一卡,禁叙述文字)" return None 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, card_col=(ci == 2)) 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) not in (norm_parens(RESERVED_RETIRED), "—"): for part in CARD_SEP_RE.split(cards): part = part.strip() if part and not (root / part).is_file(): issues.append((where, ln, "卡片路径不存在", "需求卡片列 %s 相对仓库根不存在" % part)) if date != "—" and not re.fullmatch(r"\d{4}-\d{2}-\d{2}", date): issues.append((where, ln, "编入时间", "「%s」须为 YYYY-MM-DD(历史迭代不可考填「—」)" % date)) # 预计周期:≈N 周(N≤2);超 2 周须格内带日期特批标记(§12.2/§12.4);历史行「—」跳过 week_marks = APPROVAL_RE.findall(weeks) week_core = APPROVAL_RE.sub("", weeks).strip() wm = None if week_core == "—" else WEEKS_RE.match(week_core) if week_core != "—" and wm is None: issues.append((where, ln, "预计周期", "「%s」须为「≈N 周」形态(N≤2,允许 0.5 步进;历史迭代不可考填「—」)" % weeks)) elif wm is not None and float(wm.group(1)) > 2 and not any(DATE_RE.search(mk) for mk in week_marks): issues.append((where, ln, "预计周期", "「%s」超 2 周且无带日期特批标记〔用户特批 YYYY-MM-DD …〕" % 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): pass # 退役保留字合法(含已完成的历史迭代) 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 card_date(path): """文件名尾日期 → (y, m, d) 元组;取不到返回 None(该卡跳过两条新规机检)。""" found = DATE_RE.findall(path.stem) if not found: return None return tuple(int(x) for x in found[-1].split("-")) def check_cards(root, issues): """C1 文档义务口径(生效 2026-10-17)+C2 设计事实源声明(生效 2026-10-18)。 只检文件名日期达生效日的卡——更早的卡属已实现批次的历史留档,不回头改。 C1 判据:「关联文档同步面」表内,目标格指向 design/ 或根规则文件、而义务格仍写 「随批」者红(2026-10-17 req-card §6 改口径后,这类点只能列 reports/ 反哺点清单)。 C2 判据:全文既无「设计事实源」也无「唯一事实源」者红(两种写法都与库内实态对齐)。 """ dirn = root / CARD_DIR if not dirn.is_dir(): issues.append((str(dirn), 1, "零命中自检", "卡片目录不存在,C1/C2 无卡可检")) return 0 n = 0 for path in sorted(dirn.glob(CARD_GLOB)): when = card_date(path) if when is None: continue n += 1 text = path.read_text(encoding="utf-8-sig") if when >= SOURCE_RULE_DATE and not any(mark in text for mark in SOURCE_MARKS): issues.append((str(path), 1, "设计事实源缺失", "须含「设计事实源」或「唯一事实源」声明句(req-card §2 三选一," "GDD 节+版本号/样张+art-bible 版本/工程直入证据指认)")) if when < SYNC_RULE_DATE: continue in_sec = False for lineno, raw in enumerate(text.splitlines(), start=1): stripped = raw.strip() if SYNC_HEADING_RE.match(stripped): in_sec = True continue if in_sec and stripped.startswith("#"): in_sec = False continue if not in_sec or not stripped.startswith("|") or SEP_ROW_RE.match(stripped): continue cells = row_cells(stripped) if len(cells) < 3 or cells[0] in ("文档", ""): continue if any(mark in cells[0] for mark in BANNED_DOC_MARKS) and "随批" in cells[2]: issues.append((str(path), lineno, "文档义务口径", "目标「%s」属实现批禁写面,义务却写「随批」——改标反哺点并列进 " "reports/ 清单(req-card §6,2026-10-17 起)" % cells[0][:36])) return n def main(): ap = argparse.ArgumentParser( description="工作流程流水线闸:待办清单条目状态机+roadmap 迭代总表+一次交叉核") ap.add_argument("--todo", help="待办清单路径(缺省=<仓库根>/待办清单.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)) n_cards = check_cards(root, issues) 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:待办清单条目 %d · 总表数据行 %d · 交叉核 %d 链 · 卡片口径 %d 张全过" % (len(parse_todo(todo_text)), len(index), len(links), n_cards)) return 0 if __name__ == "__main__": sys.exit(main())