guanlan-src/scripts/pages_audit.py

#!/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'


def dt_from(ts: float | None) -> str:
    """mtime → 'YYYY-mm-dd HH:MM' (报告里给人看的时间)。"""
    if not ts:
        return '?'
    import datetime as _dt
    return _dt.datetime.fromtimestamp(ts).strftime('%Y-%m-%d %H:%M')
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 e.get('generated_at_source') == 'mtime':
                # 页面件自己就是生成物: 用"它自己的 mtime"当生成时间, 与 source 的 mtime 比 ⇒ 可做陈旧代理判据
                src = e.get('source')
                mt = None
                for f in files:
                    try:
                        mt = f.stat().st_mtime
                        break
                    except OSError:
                        continue
                sp = (P.ROOT / src) if isinstance(src, str) else None
                if mt and sp is not None and sp.is_file() and (sp.stat().st_mtime - mt) > 86400:
                    results.append(('X', eid, f'**页面件疑似陈旧**: {P.rel(files[0])} 写于 '
                                              f'{dt_from(mt)}, 而它的源 {src} 更新于 {dt_from(sp.stat().st_mtime)} '
                                              f'⇒ 源换了、这份页面件没重生成', RC_STALE))
                else:
                    results.append(('OK', eid, f'按 mtime 判: 页面件与源 {src or "(未登记)"} 的新旧关系正常 '
                                                f'(页面件 {dt_from(mt) if mt else "?"})', RC_OK))
                if not found_src:
                    pass
            elif 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))
            elif e.get('generated_at_source') != 'mtime':
                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 = '&nbsp;&nbsp;└ ' * 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)
    results += ledger_coverage(reg)
    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 ledger_coverage(reg) -> list:
    """页面侧产物的**台账覆盖**检查 (用户令 2026-09-17: 是产物就要纳入产物管理)。

    凡登记表里指向 `outputs/<场>/…` 的路径 (`file` / `source`), 都必须出现在该场的
    `_provenance.json::files` 里 —— 否则它虽然被页面用了, 却没进产物台账 (来源/类型无从查)。
    """
    out: list[tuple[str, str, str, int]] = []
    prov_p = P.out_root() / '_provenance.json'
    if not prov_p.is_file():
        return out
    try:
        prov = set((json.loads(prov_p.read_text(encoding='utf-8')).get('files') or {}).keys())
    except Exception:
        return out
    prefix = P.out_root().relative_to(P.ROOT).as_posix() + '/'      # outputs/<场>/

    def walk(e):
        eid = e.get('id') or e.get('name') or '?'
        gap = e.get('known_gap') or {}
        gap_missing = set(gap.get('missing') or [])
        for key in ('file', 'source'):
            v = e.get(key)
            if not isinstance(v, str) or not v.startswith(prefix) or any(c in v for c in '*?['):
                continue
            if v in gap_missing or any(v.endswith(g.split('/')[-1]) for g in gap_missing):
                out.append(('i', eid, f'页面侧产物 {v} 不在本包(已知缺口): {gap.get("why", "")[:60]}', RC_OK))
                continue
            rel = v[len(prefix):]
            if rel not in prov:
                out.append(('X', eid, f'页面侧产物 {v} 未进产物台账 (_provenance.json 里没有 {rel}) '
                                      f'⇒ 来源/类型无从查, 不算"纳入产物管理"', RC_PROV))
        for c in e.get('children') or []:
            walk(c)
    for e in reg.get('pages') or []:
        walk(e)
    return out


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())