tools(pipeline): 新增流水线闸——06 条目状态机（标题元数据/属性齐备/状态词白名单/条件凭据/双轨/特批/零命中）＋roadmap 迭代总表六列＋已编迭代↔卡片列交叉核一处，变异验证 25 变体全红＋基准绿；暂不注册 run_all_gates（第二批数据整改后上闸，对现行旧格式 06/roadmap 红属预期）
This commit is contained in:
DKChan committed
1 parent
9282ce5484
commit
1651bc8d7d
1 file changed
+349
@@ -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())
|
||||
Reference in new issue
Block a user