guanlan-src/app_qualityGate/app_qualityGate_guanlan/audits/page_fingerprint.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
r"""页面/接口指纹 —— 改造前后与换机部署的回归判据 (2026-09-11)。

## 为什么要有

路径收敛、换机器部署这类改动,**危险在静默**: 某处路径漏改 → 页面某块空掉/取不到数, 但不报错。
所以需要一把"改前改后必须一模一样"的尺子。

指纹 = 对一组页面与接口取 `status + 字节数 + sha256`; 另存一份**归一化** sha256:
把安装根路径 (本机绝对路径, 正/反斜杠两种写法) 统一替换成 `<ROOT>` 后再算 —— 这样
"换了机器所以绝对路径前缀不同"不会误报成内容变化, 而"内容真的变了"仍然抓得到。

用法:
    python scripts/page_fingerprint.py --save .fingerprint_before.json    # 取基线 (相对安装根)
    python scripts/page_fingerprint.py --diff .fingerprint_before.json    # 比对 (不一致则退出码 3)
"""
from __future__ import annotations

try:
    from app_common.app_common_guanlan.api import install_root as _install_root
except ImportError:                     # 理论不可达；包结构异常时回退到按位置上跳
    from pathlib import Path as _P
    def _install_root(_f): return _P(_f).resolve().parents[4]

import argparse
import hashlib
import json
import pathlib
import sys
import urllib.error
import urllib.parse
import urllib.request

ROOT = _install_root(__file__)
sys.path.insert(0, str(ROOT))
from src import paths as P          # 配置唯一取用口

# 覆盖六个服务与门户; 页面类要能代表"路径是否解析对"(取数面), 接口类代表"数据是否读得到"
ENDPOINTS = [
    ('gateway.healthz', 'http://127.0.0.1:{gateway}/healthz'),
    ('gateway.version', 'http://127.0.0.1:{gateway}/api/version'),
    ('portal.home', 'http://127.0.0.1:{gateway}/'),
    ('detail.v2', 'http://127.0.0.1:{detail}/v2'),
    ('detail.maint_survey', 'http://127.0.0.1:{detail}/api/maint_survey'),
    ('detail.query.actpower', 'http://127.0.0.1:{detail}/api/query?t=WTG01&ch=grd_wtc_ActPower_mean&win=2026%E5%B9%B4&agg=med'),
    ('detail.turbine_problems', 'http://127.0.0.1:{detail}/api/turbine?t=WTG01&win=2026%E5%B9%B4'),
    ('cms.home', 'http://127.0.0.1:{cms}/?theme=light'),
    ('sim_sys.page', 'http://127.0.0.1:{sim_sys}/' + urllib.parse.quote('0_四系统合页.html')),
    ('viewer.index', 'http://127.0.0.1:{viewer}/index.html'),
]


def ports() -> dict:
    """端口只从 configs/serve.json 读 (相对路径按本文件位置解析, 不依赖 cwd); 缺项用启动器同款默认值。"""
    default = dict(gateway=28084, detail=18033, cms=18020, sim=18791, sim_sys=18792, viewer=64292, ollama=11434)
    p = P.SERVE_JSON                      # 配置唯一取用口
    cfg = json.loads(p.read_text(encoding='utf-8-sig')) if p.exists() else {}
    return {**default, **{k: v for k, v in cfg.items() if isinstance(v, int)}}


# 设计上就会变的字段 (实测: healthz 的 checked/ms 每次不同; version 的 repo_head 随提交变) ——
# 归一化时剔除, 否则每次比对都是噪音; 真实数据/页面内容不在其列。
VOLATILE = {'checked', 'ms', 'repo_head', 'elapsed', 'uptime', 'uptime_s', 'pid', 'generated_at', 'asof'}


def _strip(node):
    if isinstance(node, dict):
        return {k: _strip(v) for k, v in node.items() if k not in VOLATILE}
    if isinstance(node, list):
        return [_strip(v) for v in node]
    return node


def normalize(body: bytes) -> bytes:
    """→ 用于指纹的字节: JSON 先剔易变字段再按键排序; 非 JSON 原样 (页面类实测稳定)。
    另外把安装根 (绝对路径, 两种斜杠写法) 归一成 <ROOT>, 消除"换机器"的前缀差异。"""
    t = body.decode('utf-8', 'replace')
    try:
        j = json.loads(t)
        t = json.dumps(_strip(j), ensure_ascii=False, sort_keys=True)
    except Exception:
        pass
    for form in (str(ROOT), str(ROOT).replace('\\', '/'), str(ROOT).replace('/', '\\')):
        t = t.replace(form, '<ROOT>')
    return t.encode('utf-8')


def fetch(url: str):
    try:
        with urllib.request.urlopen(url, timeout=120) as r:
            return r.status, r.read()
    except urllib.error.HTTPError as e:
        return e.code, e.read()
    except Exception as e:
        return None, f'{type(e).__name__}: {e}'.encode()


def take() -> dict:
    P = ports()
    out = {}
    for name, tpl in ENDPOINTS:
        url = tpl.format(**P)
        status, body = fetch(url)
        out[name] = dict(url=url, status=status, bytes=len(body),
                         sha256=hashlib.sha256(body).hexdigest()[:16],
                         norm_sha256=hashlib.sha256(normalize(body)).hexdigest()[:16],
                         norm_bytes=len(normalize(body)))
    return out


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument('--save', default=None, help='把指纹写到这个 JSON')
    ap.add_argument('--diff', default=None, help='与这个基线 JSON 比对')
    a = ap.parse_args()
    now = take()

    if a.save:
        p = pathlib.Path(a.save)
        p.parent.mkdir(parents=True, exist_ok=True)
        p.write_text(json.dumps(now, ensure_ascii=False, indent=1), encoding='utf-8')
        print(f'已保存 {len(now)} 个端点指纹 → {p}')
        for k, v in now.items():
            print(f'  {k:26s} HTTP {str(v["status"]):>4} {v["bytes"]:>10,} B  {v["sha256"]}  norm {v["norm_sha256"]}')
        return 0

    if a.diff:
        base = json.loads(pathlib.Path(a.diff).read_text(encoding='utf-8'))
        bad = 0
        print(f'{"端点":26s} {"状态":>10s} {"归一化字节":>24s}  norm_sha (基线→现在)')
        for k, v in now.items():
            b = base.get(k)
            if b is None:
                print(f'{k:26s} {"(基线无)":>10s}'); bad += 1; continue
            same = v['norm_sha256'] == b.get('norm_sha256')
            flag = '一致' if same else '★变化'
            if not same:
                bad += 1
            print(f'{k:26s} {str(b["status"])+"→"+str(v["status"]):>10s} '
                  f'{b.get("norm_bytes", b["bytes"]):>11,}→{v["norm_bytes"]:<12,} '
                  f'{b.get("norm_sha256")}→{v["norm_sha256"]}  {flag}')
        print(f'\n结论: {"全部一致 ✔ (页面与取数面零变化)" if not bad else f"{bad} 个端点有变化"}')
        return 0 if not bad else 3

    for k, v in now.items():
        print(f'  {k:26s} HTTP {str(v["status"]):>4} {v["bytes"]:>10,} B  {v["sha256"]}  norm {v["norm_sha256"]}')
    return 0


if __name__ == '__main__':
    # 控制台 GBK 编不出 ✔ 时降级为 '?' (见 src/console.py): 否则"全部一致"这一行 print 会抛
    # UnicodeEncodeError、退出码变 1 —— 闸门报成失败而实际是通过的 (2026-09-11 踩过)。
    from src import console
    console.soft()
    sys.exit(main())