|
|
@@ -0,0 +1,378 @@
|
|
|
+#!/usr/bin/env python3
|
|
|
+# -*- coding: utf-8 -*-
|
|
|
+r"""页面归口审计 —— 把"这一页该不该随输入数据变 / 算不算产物"变成机器每天能查的事 (2026-09-17, 用户令 1)。
|
|
|
+
|
|
|
+## 背景(用户的问题)
|
|
|
+
|
|
|
+ http://127.0.0.1:28084/#findings #sim #documents 及它们包含的子页,
|
|
|
+ "应否随着输入数据的变化而变化"、"是否也属于产物"、"如是产物应纳入产物管理"。
|
|
|
+
|
|
|
+答案不是一句话能给的: 三页里既有**受管静态叙述**(方法论/架构), 也有**冻结交付件**(治理清单/报告),
|
|
|
+还有**真·数据派生快照**(把 outputs 里的数据烘进页面) —— 后者才是"属于产物、必须纳入产物管理"的那批。
|
|
|
+本模块把这份判断落成 `configs/portal_pages.yaml` 登记表 + 五条机器可查的规则, 于是:
|
|
|
+
|
|
|
+ · static —— 正文不得引用产物(setting 变了就该改分类);
|
|
|
+ · live —— iframe/链接指向的端口必须是 `configs/serve.json` 里的已知服务;
|
|
|
+ · data-citing —— 引用的产物路径**必须存在**(否则是悬空引用, 挂羊头卖狗肉);
|
|
|
+ · data-derived —— **必须有 source + (source_sha256 或 generated_at)**; 有 sha 的还要跟当前产物
|
|
|
+ 逐字节比对 ⇒ **陈旧检测**(数据换了、页面没换 = 报出来);
|
|
|
+ · frozen-delivery —— 文件名或正文里必须能读到版本号与日期(客户拿到手才知道是哪一版)。
|
|
|
+
|
|
|
+## 为什么要有"陈旧检测"
|
|
|
+
|
|
|
+页面把产物烘进去, 就**脱离**了"产物即时进页面"的机制(§5.1 的指纹热重载只对实时读产物的服务生效)。
|
|
|
+数据重算后, 这种页面**不会自己变**; 没有检测, 它就一直挂着旧数字 —— 这正是用户担心的情形。
|
|
|
+
|
|
|
+## 退出码 (给 rebuild_all / check 用)
|
|
|
+
|
|
|
+ 0 全部一致 · 5 登记缺项/文件缺失/未登记页 · 6 溯源缺失 · 7 页面陈旧(数据已变, 页面没变) · 8 交付件缺版本号 · 9 分类错误
|
|
|
+
|
|
|
+## 用法
|
|
|
+
|
|
|
+ python scripts/pages_audit.py # 表 + 检查(默认)
|
|
|
+ python scripts/pages_audit.py --list # 只打表
|
|
|
+ python scripts/pages_audit.py --check # 只检查(安静模式, 只打结论)
|
|
|
+ python scripts/pages_audit.py --write-doc # 把表写进 docs/系统设计说明.md (标记区间内)
|
|
|
+ python scripts/pages_audit.py --json # 机器可读结果
|
|
|
+"""
|
|
|
+from __future__ import annotations
|
|
|
+
|
|
|
+import argparse
|
|
|
+import fnmatch
|
|
|
+import glob as globmod
|
|
|
+import hashlib
|
|
|
+import html as htmllib
|
|
|
+import json
|
|
|
+import pathlib
|
|
|
+import re
|
|
|
+import sys
|
|
|
+
|
|
|
+ROOT = pathlib.Path(__file__).resolve().parents[1]
|
|
|
+sys.path.insert(0, str(ROOT))
|
|
|
+from src import paths as P # noqa: E402
|
|
|
+
|
|
|
+REG = P.CONFIGS / 'portal_pages.yaml'
|
|
|
+DOC = P.ROOT / 'docs' / '系统设计说明.md'
|
|
|
+DOC_BEGIN, DOC_END = '<!-- PAGES:BEGIN -->', '<!-- PAGES:END -->'
|
|
|
+
|
|
|
+RC_OK, RC_MISS, RC_PROV, RC_STALE, RC_VER, RC_KIND = 0, 5, 6, 7, 8, 9
|
|
|
+KINDS = ('static', 'live', 'data-citing', 'data-derived', 'frozen-delivery')
|
|
|
+
|
|
|
+RE_OUT = re.compile(r'outputs/[A-Za-z0-9_\u4e00-\u9fff\-./]+')
|
|
|
+# 版本号只认"像版本"的: v1.0 / V2.1 / 版本1.2 / 客户版|外发版|正式版|评审版。
|
|
|
+# ★别写成 \d{8} 之类 —— 那会把页面里的机组编号(00019164)、时间戳当成版本号(2026-09-17 实测踩到)。
|
|
|
+RE_VER = re.compile(r'[vV]\d+(?:\.\d+)+|版本\s*[vV]?\d+(?:\.\d+)*|客户版|外发版|正式版|评审版')
|
|
|
+RE_DATE = re.compile(r'20\d\d-\d\d-\d\d|20\d{6}')
|
|
|
+RE_GEN = re.compile(r'生成[^0-9]{0,8}(20\d\d-\d\d-\d\d[ T]?\d?\d?:?\d?\d?)')
|
|
|
+
|
|
|
+
|
|
|
+def load(path=None):
|
|
|
+ import yaml
|
|
|
+ reg_p = pathlib.Path(path) if path else REG
|
|
|
+ if not reg_p.is_file():
|
|
|
+ raise SystemExit(f'缺登记表 {P.rel(reg_p)}')
|
|
|
+ reg = yaml.safe_load(reg_p.read_text(encoding='utf-8'))
|
|
|
+ reg['_file'] = str(reg_p)
|
|
|
+ return reg
|
|
|
+
|
|
|
+
|
|
|
+def sha256(p: pathlib.Path) -> str | None:
|
|
|
+ try:
|
|
|
+ return hashlib.sha256(p.read_bytes()).hexdigest()
|
|
|
+ except OSError:
|
|
|
+ return None
|
|
|
+
|
|
|
+
|
|
|
+def expand(entry) -> list[pathlib.Path]:
|
|
|
+ """一个登记项的 `file` 字段 → 实际文件列表 (支持 * 通配)。"""
|
|
|
+ pat = P.ROOT / entry['file']
|
|
|
+ s = str(pat)
|
|
|
+ if any(c in s for c in '*?['):
|
|
|
+ return sorted(pathlib.Path(x) for x in globmod.glob(s, recursive=True))
|
|
|
+ return [pat] if pat.is_file() else []
|
|
|
+
|
|
|
+
|
|
|
+def texts(files):
|
|
|
+ for f in files:
|
|
|
+ try:
|
|
|
+ yield f, f.read_text(encoding='utf-8', errors='replace')
|
|
|
+ except OSError:
|
|
|
+ continue
|
|
|
+
|
|
|
+
|
|
|
+def read_embedded_json(path: pathlib.Path, var: str):
|
|
|
+ """抓页面里内嵌的 window.<var> = {…} —— 模板里是 HTML 转义过的, 解析前先反转义。"""
|
|
|
+ try:
|
|
|
+ t = path.read_text(encoding='utf-8', errors='replace')
|
|
|
+ except OSError:
|
|
|
+ return None
|
|
|
+ m = re.search(r'window\.' + re.escape(var) + r'\s*=\s*(\{)', t)
|
|
|
+ if not m:
|
|
|
+ return None
|
|
|
+ raw, depth = '', 0
|
|
|
+ for ch in t[m.start(1):]:
|
|
|
+ raw += ch
|
|
|
+ if ch == '{':
|
|
|
+ depth += 1
|
|
|
+ elif ch == '}':
|
|
|
+ depth -= 1
|
|
|
+ if depth == 0:
|
|
|
+ break
|
|
|
+ for cand in (raw, htmllib.unescape(raw)):
|
|
|
+ try:
|
|
|
+ return json.loads(cand)
|
|
|
+ except Exception:
|
|
|
+ continue
|
|
|
+ return None
|
|
|
+
|
|
|
+
|
|
|
+def check_entry(e, results):
|
|
|
+ """按 kind 查一个登记项 → 往 results 里加 (level, 项, 说明, 退出码)。
|
|
|
+
|
|
|
+ level: OK 通过 / X 失败 / ! 要处理 / ? 只能人工看 / i 已知缺口(登记了 known_gap 并写了理由)。
|
|
|
+ ★ known_gap 的存在是**如实**的产物: 交付包里确实没有那件产物(例如振动线六层链的 cleaned/gearbox_life),
|
|
|
+ 页面里却引用了它。我们不改交付件正文(客户手里那一版是冻结的), 但必须把"引用悬空"这件事记在明处,
|
|
|
+ 所以它报 `i` 而不是装作通过, 也不当成新问题反复报警。
|
|
|
+ """
|
|
|
+ eid = e.get('id') or e.get('name') or '?'
|
|
|
+ kind = e.get('kind')
|
|
|
+ gap = e.get('known_gap') or {}
|
|
|
+ if kind not in KINDS:
|
|
|
+ results.append(('X', eid, f'kind 非法: {kind} (允许 {"/".join(KINDS)})', RC_KIND))
|
|
|
+ return
|
|
|
+ files = expand(e)
|
|
|
+ if not files and not e.get('members_in_zip'):
|
|
|
+ results.append(('X', eid, f'登记的 file 不存在或没匹配到: {e.get("file")}', RC_MISS))
|
|
|
+ return
|
|
|
+
|
|
|
+ # ① 引用的产物必须在位 (data-citing / data-derived; 登记了 cites 的也查)
|
|
|
+ for f, t in texts(files):
|
|
|
+ cited = set(e.get('cites') or [])
|
|
|
+ if kind in ('data-derived', 'data-citing'):
|
|
|
+ cited |= set(RE_OUT.findall(t))
|
|
|
+ for c in sorted(cited):
|
|
|
+ if (P.ROOT / c).exists():
|
|
|
+ continue
|
|
|
+ if gap and c in (gap.get('missing') or []):
|
|
|
+ results.append(('i', eid, f'引用悬空(已知): {c} —— {gap.get("why", "见登记表")}', RC_OK))
|
|
|
+ else:
|
|
|
+ results.append(('X', eid, f'{P.rel(f)} 引用的产物不在位: {c} (悬空引用)', RC_MISS))
|
|
|
+
|
|
|
+ # ② 分类规则
|
|
|
+ if kind == 'static':
|
|
|
+ for f, t in texts(files):
|
|
|
+ if RE_OUT.search(t):
|
|
|
+ results.append(('!', eid, f'{P.rel(f)} 正文里出现产物路径 —— 可能该改成 data-citing/data-derived',
|
|
|
+ RC_KIND))
|
|
|
+
|
|
|
+ elif kind == 'live':
|
|
|
+ svc = {}
|
|
|
+ try:
|
|
|
+ svc = json.loads(P.SERVE_JSON.read_text(encoding='utf-8-sig'))
|
|
|
+ except Exception:
|
|
|
+ pass
|
|
|
+ ports = {int(v) for k, v in svc.items() if isinstance(v, int)}
|
|
|
+ tgt = e.get('live_target') or ''
|
|
|
+ m = re.search(r':(\d{2,5})', tgt)
|
|
|
+ if m and ports and int(m.group(1)) not in ports:
|
|
|
+ results.append(('X', eid, f'live_target 端口 {m.group(1)} 不在 configs/serve.json 的服务端口里', RC_MISS))
|
|
|
+
|
|
|
+ elif kind == 'data-derived':
|
|
|
+ src = e.get('source')
|
|
|
+ var = e.get('embedded_json_var')
|
|
|
+ skey, shkey = e.get('source_key'), e.get('source_sha_key')
|
|
|
+ found_sha, found_src, found_gen = None, src, None
|
|
|
+ if var and skey:
|
|
|
+ for f in files:
|
|
|
+ obj = read_embedded_json(f, var)
|
|
|
+ if isinstance(obj, dict):
|
|
|
+ found_src = obj.get(skey) or found_src
|
|
|
+ found_sha = obj.get(shkey) if shkey else None
|
|
|
+ break
|
|
|
+ for f, t in texts(files):
|
|
|
+ if found_gen is None:
|
|
|
+ g = RE_GEN.search(t)
|
|
|
+ if g:
|
|
|
+ found_gen = g.group(1)
|
|
|
+ if not found_src:
|
|
|
+ if gap:
|
|
|
+ results.append(('i', eid, f'溯源缺失(已知): {gap.get("why", "")}', RC_OK))
|
|
|
+ else:
|
|
|
+ results.append(('!', eid, 'data-derived 但没登记 source, 页面里也没有内嵌来源 ⇒ 溯源缺失', RC_PROV))
|
|
|
+ if not (found_sha or found_gen or e.get('generated_at_key')):
|
|
|
+ if not gap:
|
|
|
+ results.append(('!', eid, 'data-derived 但既无 source_sha256 也无生成时间 ⇒ 无法做陈旧检测', RC_PROV))
|
|
|
+ if found_src:
|
|
|
+ sp = P.ROOT / found_src
|
|
|
+ if not sp.exists():
|
|
|
+ if gap and found_src in (gap.get('missing') or []):
|
|
|
+ results.append(('i', eid, f'内嵌来源不在位(已知): {found_src} —— {gap.get("why", "")}', RC_OK))
|
|
|
+ else:
|
|
|
+ results.append(('!', eid, f'内嵌来源 {found_src} 不在位 (页面里烘的是别处/历史数据)', RC_MISS))
|
|
|
+ elif found_sha:
|
|
|
+ cur = sha256(sp)
|
|
|
+ if cur and cur.lower() != str(found_sha).lower():
|
|
|
+ results.append(('X', eid,
|
|
|
+ f'**页面陈旧**: 内嵌快照 source_sha256={str(found_sha)[:16]} 与当前 '
|
|
|
+ f'{found_src} 的 sha256={cur[:16]} 不同 ⇒ 数据变了, 这份页面没跟着重生成',
|
|
|
+ RC_STALE))
|
|
|
+ else:
|
|
|
+ results.append(('OK', eid, f'快照与当前产物一致 ({found_src} sha256 {str(found_sha)[:16]})', RC_OK))
|
|
|
+ else:
|
|
|
+ results.append(('?', eid, f'有来源 {found_src} 但无指纹, 只能按生成时间判断 '
|
|
|
+ f'(记录: {found_gen or "无"})', RC_PROV))
|
|
|
+
|
|
|
+ elif kind == 'frozen-delivery':
|
|
|
+ ok_ver = ok_date = False
|
|
|
+ for f in files:
|
|
|
+ name = f.name
|
|
|
+ ok_ver |= bool(RE_VER.search(name))
|
|
|
+ ok_date |= bool(RE_DATE.search(name))
|
|
|
+ if not (ok_ver and ok_date):
|
|
|
+ t = f.read_text(encoding='utf-8', errors='replace') # 版本号/日期可能在正文深处
|
|
|
+ ok_ver |= bool(RE_VER.search(t))
|
|
|
+ ok_date |= bool(RE_DATE.search(t))
|
|
|
+ if e.get('version'):
|
|
|
+ ok_ver = True
|
|
|
+ if not ok_ver:
|
|
|
+ results.append(('!', eid, '冻结交付件但没有版本号 (文件名/正文都没有 v*/日期) ⇒ 客户拿到手分不清是哪一版',
|
|
|
+ RC_VER))
|
|
|
+ if not ok_date and not e.get('version'):
|
|
|
+ results.append(('?', eid, '冻结交付件没有日期标记', RC_VER))
|
|
|
+
|
|
|
+ # ③ 子项递归
|
|
|
+ for c in e.get('children') or []:
|
|
|
+ check_entry(c, results)
|
|
|
+
|
|
|
+
|
|
|
+def declared_patterns(reg) -> list[str]:
|
|
|
+ """登记表里所有 file 字段 (含通配) —— 覆盖检查用。"""
|
|
|
+ pats = []
|
|
|
+
|
|
|
+ def walk(e):
|
|
|
+ f = e.get('file')
|
|
|
+ if f:
|
|
|
+ pats.append(str(f))
|
|
|
+ for c in e.get('children') or []:
|
|
|
+ walk(c)
|
|
|
+ for e in reg.get('pages') or []:
|
|
|
+ walk(e)
|
|
|
+ return pats
|
|
|
+
|
|
|
+
|
|
|
+def coverage_gaps(reg) -> list[str]:
|
|
|
+ """登记表里没写、但门户里真实存在的内嵌页 —— 漏登记就等于没管。"""
|
|
|
+ pats = declared_patterns(reg)
|
|
|
+ gaps = []
|
|
|
+ tdir = P.RELEASE / 'portal_src' / 'templates'
|
|
|
+ for f in sorted(tdir.glob('*.html')):
|
|
|
+ rel = f.relative_to(P.ROOT).as_posix()
|
|
|
+ if not any(fnmatch.fnmatch(rel, p) or rel == p for p in pats):
|
|
|
+ gaps.append(rel)
|
|
|
+ return gaps
|
|
|
+
|
|
|
+
|
|
|
+def render(reg, results) -> str:
|
|
|
+ lines = ['| 页面/子页 | kind | 随输入数据变? | 依据(为什么这么判) |', '|---|---|---|---|']
|
|
|
+ def walk(e, depth=0):
|
|
|
+ eid = e.get('id') or e.get('name') or '?'
|
|
|
+ title = e.get('title') or ''
|
|
|
+ pre = ' └ ' * depth
|
|
|
+ lines.append(f"| {pre}`{eid}`{' ' + title if title and e.get('id') else ''} | `{e.get('kind')}` | "
|
|
|
+ f"{'**是**' if e.get('changes_with_data') else '否'} | {e.get('evidence', '')} |")
|
|
|
+ for c in e.get('children') or []:
|
|
|
+ walk(c, depth + 1)
|
|
|
+ for e in reg.get('pages') or []:
|
|
|
+ walk(e)
|
|
|
+ return '\n'.join(lines)
|
|
|
+
|
|
|
+
|
|
|
+def audit(registry=None):
|
|
|
+ """→ (rc, results, unregistered)。
|
|
|
+
|
|
|
+ 供 `guanlan.py check` / `scripts/rebuild_all.py` 直接调用 —— 走函数而不是子进程:
|
|
|
+ 子进程的 stdout 是中文, Windows 控制台默认 cp936 会把 UTF-8 输出解成乱码, 解析结论就不可靠了。
|
|
|
+ """
|
|
|
+ reg = load(registry)
|
|
|
+ results: list[tuple[str, str, str, int]] = []
|
|
|
+ for e in reg.get('pages') or []:
|
|
|
+ check_entry(e, results)
|
|
|
+ rc_map = {r[3] for r in results if r[0] not in ('OK', 'i') and r[3]}
|
|
|
+ return (max(rc_map) if rc_map else RC_OK), results, coverage_gaps(reg)
|
|
|
+
|
|
|
+
|
|
|
+def main() -> int:
|
|
|
+ ap = argparse.ArgumentParser()
|
|
|
+ ap.add_argument('--list', action='store_true', help='只打表')
|
|
|
+ ap.add_argument('--check', action='store_true', help='只检查')
|
|
|
+ ap.add_argument('--write-doc', action='store_true', help='把表写进 docs/系统设计说明.md')
|
|
|
+ ap.add_argument('--json', action='store_true', help='机器可读输出')
|
|
|
+ ap.add_argument('--registry', default=None, help='登记表路径 (默认 configs/portal_pages.yaml; 测试用)')
|
|
|
+ a = ap.parse_args()
|
|
|
+ reg = load(a.registry)
|
|
|
+ rc, results, gaps = audit(a.registry)
|
|
|
+
|
|
|
+ if a.json:
|
|
|
+ print(json.dumps(dict(rc=rc,
|
|
|
+ results=[dict(level=r[0], item=r[1], note=r[2], rc=r[3]) for r in results],
|
|
|
+ unregistered=gaps), ensure_ascii=False, indent=1))
|
|
|
+ return rc
|
|
|
+
|
|
|
+ if a.list or not a.check:
|
|
|
+ print('== 门户页面归口 (configs/portal_pages.yaml) ==')
|
|
|
+ for e in reg.get('pages') or []:
|
|
|
+ def walk(x, d=0):
|
|
|
+ eid = x.get('id') or x.get('name')
|
|
|
+ print(f' {" " * d}{eid:34s} {x.get("kind"):16s} '
|
|
|
+ f'{"随数据变" if x.get("changes_with_data") else "不随数据变"}')
|
|
|
+ for c in x.get('children') or []:
|
|
|
+ walk(c, d + 1)
|
|
|
+ walk(e)
|
|
|
+
|
|
|
+ if not a.check or True:
|
|
|
+ lvl = {}
|
|
|
+ for level, item, note, r in results:
|
|
|
+ lvl[level] = lvl.get(level, 0) + 1
|
|
|
+ print(f'\n== 检查: {lvl.get("OK", 0)} 项一致, {lvl.get("i", 0)} 项已知缺口, '
|
|
|
+ f'{sum(v for k, v in lvl.items() if k not in ("OK", "i"))} 项要处理 ==')
|
|
|
+ for level, item, note, r in results:
|
|
|
+ if level not in ('OK', 'i'):
|
|
|
+ print(f' [{level}] {item}: {note}')
|
|
|
+ if lvl.get('i'):
|
|
|
+ print(f' [i] 已知缺口 {lvl["i"]} 条 (引用悬空/溯源缺失, 已在登记表里写明理由, 不重复刷屏):')
|
|
|
+ seen = set()
|
|
|
+ for level, item, note, r in results:
|
|
|
+ if level == 'i' and item not in seen:
|
|
|
+ seen.add(item)
|
|
|
+ print(f' · {item}')
|
|
|
+ if gaps:
|
|
|
+ print(f' [!] 门户里有 {len(gaps)} 个内嵌页/子页**未登记**(漏登记=没管):')
|
|
|
+ for g in gaps[:8]:
|
|
|
+ print(f' · {g}')
|
|
|
+ print(f'结论: {"全部一致" if rc == RC_OK else "见上"}; 退出码 {rc}')
|
|
|
+ if a.write_doc:
|
|
|
+ write_doc(reg)
|
|
|
+ return rc
|
|
|
+ return rc
|
|
|
+
|
|
|
+
|
|
|
+def write_doc(reg) -> None:
|
|
|
+ if not DOC.is_file():
|
|
|
+ print(f' (缺 {P.rel(DOC)}, 跳过写文档)')
|
|
|
+ return
|
|
|
+ t = DOC.read_text(encoding='utf-8')
|
|
|
+ block = (f'{DOC_BEGIN}\n### 10.1 门户页面归口（自动生成，勿手改）\n\n'
|
|
|
+ f'判据与说明见 `configs/portal_pages.yaml` 头部；检查器 `scripts/pages_audit.py`。\n\n'
|
|
|
+ f'{render(reg, None)}\n\n{DOC_END}')
|
|
|
+ if DOC_BEGIN in t and DOC_END in t:
|
|
|
+ t = re.sub(re.escape(DOC_BEGIN) + r'.*?' + re.escape(DOC_END), lambda m: block, t, flags=re.S)
|
|
|
+ else:
|
|
|
+ t = t.rstrip() + '\n\n---\n\n## 10. 页面归口（哪些页面是产物、该不该随数据变）\n\n' + block + '\n'
|
|
|
+ DOC.write_text(t, encoding='utf-8')
|
|
|
+ print(f' 已写入 {P.rel(DOC)} (§10 页面归口)')
|
|
|
+
|
|
|
+
|
|
|
+if __name__ == '__main__':
|
|
|
+ from src import console
|
|
|
+ console.soft()
|
|
|
+ sys.exit(main())
|
zhouyang.xie