350 lines
16 KiB
Python
350 lines
16 KiB
Python
#!/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())
|