STM32F4-Base/.dsh/skills/serial-boot-test/scripts/verify_log.py

#!/usr/bin/env python3
"""verify_log.py - 对抓取的 MCU 调试日志做标记校验（纯标准库，2026-09-20 替代 verify_log.ps1）。

用于 @test 闭环：烧录后从串口助手日志提取的开机日志交给本脚本 grep 标记，
输出结构化判定（格式与旧 PowerShell 版逐字一致，下游文档/习惯不受影响）。

用法:
  & '.venv\\Scripts\\python.exe' .dsh/skills/serial-boot-test/scripts/verify_log.py --log-path <file>
      [--required 'marker1;marker2']
      [--forbidden 'FAIL;ERROR;HardFault;assert;[ERR]']
      [--require-suite-summary 'Storage Tests;GD5F2GQ5UE Tests']
      [--require-selftest-pass] [--selftest-log <selftest.log 本地副本>]
      [--show-tail-lines 40]
      [--max-boot-count 1] [--boot-marker 'reset cause:']
      [--expect-reset-cause 'software;power-on']
      [--require-order 'App init OK>Network Init>net init: OK']
      [--done-marker 'net init: OK'] [--max-boot-ms 8000] [--json]

退出码（与旧版一致）:
  0 = PASS（所有 Required 标记找到，Forbidden 零命中，汇总行要求满足）
  1 = FAIL（缺 Required / 命中 Forbidden / 汇总行要求不满足）
  2 = ERROR（日志文件不存在/不可读）

默认 Required 标记对应 STM32F4-Base 启动序列（见 App/app/app_main.c + App/task/net_task.c），
按测试用例覆盖，例如:
  --required 'App init OK;net init: OK;PHY: 100M FULL'
  --forbidden 'FAIL;ERROR;HardFault;assert;[ERR]'

语义约定（与旧 PowerShell 版对齐）:
  - Required：大小写敏感子串匹配（全文）
  - Forbidden：大小写**不敏感**子串匹配、逐行计数（旧版 -like 语义；
    如 Forbidden 'FAIL' 会命中 "write /log/run.log fail (7)" 这类行）
  - 固件错误级日志前缀是 `[ERR]`（`DBG_ERROR`，形如 `[时间][ERR][TAG] 消息`），
    与 `ERROR` 子串**不匹配**——默认 Forbidden 因此显式包含 `[ERR]`，驱动/模块
    错误行不再漏检（2026-09-21 修复）。测试套件中**预期**出现 `[ERR]` 的用例
    （如 GD5F DMA 守卫负路径、FatFs 非法名校验）须用 --forbidden 覆盖默认值。
  - 读取编码：先按 UTF-8，解码失败回退系统 locale 编码（旧版 Get-Content 行为）
  - 控制台输出重配置为 UTF-8（errors=replace）：任何 Windows 控制台代码页
    都只影响终端显示，不会导致本脚本崩溃
  - 汇总行（2026-10-01 加，T-2）：纯 Required 子串匹配有个盲区——套件启动横幅
    命中即判 PASS，**套件是否跑到汇总行、断言数是否为 0 完全不可见**（套件中途
    挂死/复位、或 Phase 宏全关时会被误判为通过）。`--require-suite-summary`
    要求日志中出现 `=== <名称>: N/M PASSED (failed=0) ===` 且 **N > 0**。
  - 自检记录联合判定（S7，2026-10 实机窗口；判据 2026-10-22 改为"只看本次新增"）：
    `--require-selftest-pass` 对 /log/selftest.log 的本地副本（定长 64B 记录，格式见
    App/log/sys_log.h）判定。自检仅故障写盘（v1 不记通过项）：**文件缺失 / 0 字节 = 无故障
    → PASS**；存在记录 = FAIL（退出码 1），逐条打印时间戳 + 描述。
    ⚠ 该文件是 512 条环形**累计历史**，所以本函数收到的"有记录"必须由调用方先做差集：
    run_boot_test.py 在烧录前取基线、启动后只把"基线之后新增"的记录写成临时副本交给本
    函数（`--selftest-scope this-run`，默认）。**抓取失败/无数据源/无法证明无新增一律由
    调用方判 FAIL**——旧实现把这些都当"无记录 → PASS"，是假绿（详见
    docs/测试说明/自动化测试流程.md 的自检结果判定节）。
  - 启动锚点计数（2026-10-10）：`--max-boot-count N`（run_boot_test.py 默认传 1）统计
    `reset cause:` 锚点出现次数——**0 次** = 采集没从复位开始（标记全中也可能只是上一轮
    的残留日志）；**>N 次** = 板子在采集窗口内自己复位过（IWDG 咬狗 / Fault 后
    `NVIC_SystemReset` / 掉电重启），此前这类运行只要不命中 Forbidden 就判 PASS。
    `--expect-reset-cause` 对第一个锚点做原因白名单（大小写不敏感子串），空 = 只报告不判。
  - Forbidden 命中行（2026-10-10）：命中时逐行打印**行号 + 内容**（最多 6 行，超出报剩余数）。
    旧实现只说 `forbidden pattern: [ERR] (3 line(s))`，人还得自己再 grep 一遍才知道是哪几行
    ——而"是哪几行"恰恰是定位要的东西。
  - 标记顺序（`--require-order 'a>b>c'`）：按**首次出现的行序**校验启动步骤先后。标记全中
    但顺序颠倒 = 某步在另一步之后补做（屏障失序、重试后补跑），旧实现完全看不见。
  - 开机耗时（`--done-marker` + `--max-boot-ms`）：依赖行首 `[+Nms]` 前缀（run_boot_test.py
    2026-10-10 起按 daemon 毫秒时基给每行加，MCU 侧时间戳只有秒级且开机头几行根本没有）。
    量"第一行 → done-marker"的真实耗时，抓"能开机但慢了几秒"（mount 重试 / PHY 重协商）。
    日志里没有该前缀 → 只 WARN 跳过，不假判 FAIL。
  - `--json`：同一套判定以 JSON 输出（供 CI / 上层脚本消费；退出码语义不变）。
"""

import argparse
import json
import locale
import re
import sys
from pathlib import Path

REQUIRED_DEFAULT = "App init OK;Network Init;net init: OK"
FORBIDDEN_DEFAULT = "FAIL;ERROR;HardFault;assert;[ERR]"

# Forbidden 命中时最多打印几条命中行（够定位又不刷屏；剩余条数会报出来）
HIT_LINES_MAX = 6

# 行首 `[+1234ms] `：run_boot_test.py 按 daemon 毫秒时基给每行加的相对偏移
# （MCU 侧时间戳只有秒级，且开机头几行时钟还没起 → 量不了耗时）
_TS_OFF_RE = re.compile(r"^\[\+(\d+)ms\]\s")

# 每轮启动的锚点行（App/app/app_main.c：`reset cause: <原因> (CSR=0x…)`）。
# 一次采集里出现 >1 次 = 板子在采集窗口内自己复位过（IWDG 咬狗 / Fault 后
# NVIC_SystemReset / 掉电重启），此前只有 Forbidden 命中时才被发现，
# "跑完 PASS 但中途重启过"完全不可见（2026-10-10 加）。
BOOT_MARKER_DEFAULT = "reset cause:"

# 套件汇总行：`=== Storage Tests: 3/3 PASSED (failed=0) ===`
#   名称可含空格；N/M 为十进制；failed 允许前导 0。宽松容忍前后空白与
#   日志时间戳前缀（`[2026-10-01 12:00:00.000]`）——故用 search 而非 match。
_SUMMARY_RE = re.compile(
    r"===\s*(?P<name>[^:=]+?)\s*:\s*(?P<passed>\d+)\s*/\s*(?P<total>\d+)\s+"
    r"PASSED\s*\(\s*failed\s*=\s*(?P<failed>\d+)\s*\)\s*==="
)


def _split_list(spec):
    """';' 分隔列表 → 去空白、去空项。"""
    return [p.strip() for p in spec.split(";") if p.strip()]


def _read_lines(path):
    raw = Path(path).read_bytes()
    for enc in ("utf-8", locale.getpreferredencoding(False)):
        try:
            return raw.decode(enc).splitlines()
        except UnicodeDecodeError:
            continue
    return raw.decode("utf-8", errors="replace").splitlines()


def _reset_cause_of(line, marker):
    """从 `reset cause: software (NVIC_SystemReset) (CSR=0x14000000)` 取原因文本。"""
    i = line.find(marker)
    if i < 0:
        return ""
    tail = line[i + len(marker):].strip()
    j = tail.find(" (CSR=")
    return tail[:j] if j > 0 else tail


def verify(log_path, required=REQUIRED_DEFAULT, forbidden=FORBIDDEN_DEFAULT,
           show_tail=40, out=print, suite_summary="",
           selftest_pass=False, selftest_log="",
           boot_marker=BOOT_MARKER_DEFAULT, max_boot_count=0,
           expect_reset_cause="", require_order="", done_marker="",
           max_boot_ms=0, as_json=False):
    """标记校验核心（可被 run_boot_test.py 复用）。返回退出码 0/1/2。

    suite_summary: ';' 分隔的套件名列表。非空时，要求日志中每个套件都出现
                   `=== <名称>: N/M PASSED (failed=0) ===` 且 **N > 0**。
    selftest_pass: True 时对 selftest_log（/log/selftest.log 本地副本）联合判定：
                   文件缺失 / 0 字节 = 无故障（PASS）；存在任何 64B 记录 = FAIL（S7）。
    selftest_log:  自检记录文件本地路径（空串 = 按文件缺失处理）。
    boot_marker / max_boot_count（2026-10-10）：统计启动锚点出现次数。
                   max_boot_count>0 时启用：0 次 = 采集没从复位开始；>N 次 = 板子
                   在采集窗口内自己复位过（IWDG / Fault 后 NVIC_SystemReset）。
    expect_reset_cause: ';' 分隔的原因子串白名单，只对**第一个**锚点生效（大小写不敏感）。
                   空 = 只报告不判定。
    require_order: '>' 分隔的标记序列（2026-10-10），按**首次出现的行序**校验先后；
                   顺序颠倒 = 某步在另一步之后补做（屏障失序 / 重试补跑）。空 = 不查。
    done_marker / max_boot_ms（2026-10-10）：量"第一行 → done_marker"的毫秒耗时
                   （靠行首 `[+Nms]` 前缀）；超过 max_boot_ms 判 FAIL。
                   日志无该前缀 → WARN 跳过，不假判。
    as_json:       True = 不打印文本判定，最后输出一个 JSON 对象（退出码不变）。
    """
    res = {"log_path": str(log_path), "lines": 0, "bytes": 0, "required_missing": [],
           "forbidden_hits": [], "boot": None, "suites_bad": [], "selftest": None,
           "order_bad": [], "timing": None, "tail": [], "verdict": None, "exit": None}
    emit = out

    def finish(code):
        res["exit"] = code
        res["verdict"] = {0: "PASS", 1: "FAIL", 2: "ERROR"}[code]
        if as_json:
            emit(json.dumps(res, ensure_ascii=False, indent=2))
        return code

    if as_json:
        out = lambda *a, **k: None       # 文本判定静默，结果走 JSON

    p = Path(log_path)
    if not p.is_file():
        out("VERIFY_ERROR: log file not found: {}".format(log_path))
        return finish(2)
    try:
        lines = _read_lines(log_path)
    except OSError:
        out("VERIFY_ERROR: cannot read log file: {}".format(log_path))
        return finish(2)

    full = "\n".join(lines)
    req_list = _split_list(required)
    forb_list = _split_list(forbidden)
    summary_list = _split_list(suite_summary)

    out("=== Log verdict: {} ===".format(log_path))
    out("Lines: {}, bytes: {}".format(len(lines), p.stat().st_size))
    res["lines"] = len(lines)
    res["bytes"] = p.stat().st_size

    missing = []
    for m in req_list:
        if m in full:
            out("[OK]   required marker found: {}".format(m))
        else:
            out("[MISS] required marker NOT found: {}".format(m))
            missing.append(m)
    res["required_missing"] = list(missing)

    hits = []
    for f in forb_list:
        fl = f.lower()
        matched = [(i + 1, ln) for i, ln in enumerate(lines) if fl in ln.lower()]
        if matched:
            out("[HIT]  forbidden pattern: {} ({} line(s))".format(f, len(matched)))
            # 命中行必须打出来（2026-10-10）：旧实现只报 "(3 line(s))"，人还得自己
            # 再 grep 一遍才知道是哪几行——而"哪几行"正是定位所需的那一步
            for ln_no, ln in matched[:HIT_LINES_MAX]:
                out("       L{}: {}".format(ln_no, ln.strip()[:200]))
            if len(matched) > HIT_LINES_MAX:
                out("       ... {} more".format(len(matched) - HIT_LINES_MAX))
            hits.append(f)
            res["forbidden_hits"].append({
                "pattern": f, "count": len(matched),
                "lines": [{"no": n, "text": t.strip()[:200]}
                          for n, t in matched[:HIT_LINES_MAX]]})
        else:
            out("[OK]   forbidden pattern absent: {}".format(f))

    # 启动锚点计数（2026-10-10）：抓"跑完但中途复位过"和"采集没从复位开始"
    boot_issue = False
    if max_boot_count > 0:
        banner_lines = [ln for ln in lines if boot_marker in ln]
        n_boot = len(banner_lines)
        causes = [_reset_cause_of(ln, boot_marker) for ln in banner_lines]
        if n_boot == 0:
            out("[FAIL] no boot banner (marker '{}'): capture did not start at a reset".format(
                boot_marker))
            boot_issue = True
        elif n_boot > max_boot_count:
            out("[FAIL] boot banner x{} > allowed {} -> board reset itself during capture".format(
                n_boot, max_boot_count))
            for ln in banner_lines[:6]:
                out("       {}".format(ln.strip()))
            boot_issue = True
        else:
            out("[OK]   boot banner x{} (marker '{}')".format(n_boot, boot_marker))
        if causes:
            out("[INFO] reset cause(s): {}".format("; ".join(causes)))
            exp = _split_list(expect_reset_cause)
            if exp:
                got = causes[0].lower()
                if any(e.lower() in got for e in exp):
                    out("[OK]   reset cause '{}' matches expected {}".format(
                        causes[0], expect_reset_cause))
                else:
                    out("[FAIL] reset cause '{}' not in expected: {}".format(
                        causes[0], expect_reset_cause))
                    boot_issue = True
        res["boot"] = {"marker": boot_marker, "count": n_boot, "allowed": max_boot_count,
                       "causes": causes, "issue": boot_issue}

    # 标记顺序（2026-10-10）：标记全中但**顺序颠倒** = 某步在另一步之后补做
    # （屏障失序 / 失败重试补跑）。旧实现只看子串存在性，这类失序完全看不见。
    order_issue = False
    if require_order:
        seq = [s.strip() for s in require_order.split(">") if s.strip()]
        pos = []
        for m in seq:
            idx = next((i for i, ln in enumerate(lines) if m in ln), -1)
            pos.append((m, idx))
        absent = [m for m, i in pos if i < 0]
        if absent:
            out("[FAIL] require-order: marker(s) not found: {}".format("; ".join(absent)))
            order_issue = True
        else:
            bad = [(pos[k][0], pos[k + 1][0]) for k in range(len(pos) - 1)
                   if pos[k][1] > pos[k + 1][1]]
            if bad:
                out("[FAIL] marker order violated (expected {})".format(require_order))
                for a, b in bad:
                    out("       '{}' first appeared AFTER '{}'".format(a, b))
                for m, i in pos:
                    out("       L{}: {}".format(i + 1, m))
                order_issue = True
            else:
                out("[OK]   marker order ok: {}".format(require_order))
        res["order_bad"] = [m for m, i in pos if i < 0] + [
            "{}>{}".format(a, b) for a, b in
            [(pos[k][0], pos[k + 1][0]) for k in range(len(pos) - 1) if pos[k][1] > pos[k + 1][1]]]

    # 开机耗时（2026-10-10）：行首 [+Nms] = daemon 毫秒时基的相对偏移
    # （MCU 侧时间戳只有秒级，且开机头几行时钟还没起，量不出耗时）。
    timing = None
    timing_issue = False
    offs = []
    for i, ln in enumerate(lines):
        m = _TS_OFF_RE.match(ln)
        if m:
            offs.append((i, int(m.group(1))))
    if offs:
        timing = {"ts_lines": len(offs), "first_ms": offs[0][1], "last_ms": offs[-1][1],
                  "span_ms": offs[-1][1] - offs[0][1]}
        out("[INFO] capture span: +{}ms .. +{}ms ({} timestamped line(s))".format(
            timing["first_ms"], timing["last_ms"], timing["ts_lines"]))
        if done_marker:
            di = next((i for i, ln in enumerate(lines) if done_marker in ln), -1)
            if di < 0:
                out("[WARN] done marker '{}' not found -> boot time not measured".format(
                    done_marker))
            else:
                doff = next((off for i, off in offs if i >= di), None)
                if doff is None:
                    out("[WARN] done marker line has no [+Nms] prefix -> not measured")
                else:
                    boot_ms = doff - offs[0][1]
                    timing["boot_ms"] = boot_ms
                    timing["done_marker"] = done_marker
                    out("[INFO] boot time to '{}': {} ms".format(done_marker, boot_ms))
                    if max_boot_ms > 0 and boot_ms > max_boot_ms:
                        out("[FAIL] boot time {} ms > limit {} ms".format(boot_ms, max_boot_ms))
                        timing_issue = True
    elif done_marker or max_boot_ms > 0:
        out("[WARN] no [+Nms] line prefix in this log -> timing checks skipped "
            "(run_boot_test.py adds it; --raw-log turns it off)")
    res["timing"] = timing

    # 套件汇总行解析（T-2）：抓"没跑到汇总行 / 断言数为 0"的盲区
    bad_summary = []
    if summary_list:
        parsed = {}
        for ln in lines:
            m = _SUMMARY_RE.search(ln)
            if m:
                parsed.setdefault(m.group("name").strip(), []).append(m)
        for name in summary_list:
            cands = parsed.get(name)
            if not cands:
                out("[MISS] suite summary NOT found: '=== {}: N/M PASSED (failed=0) ==='".format(name))
                bad_summary.append(name)
                continue
            ok = False
            for m in cands:
                passed = int(m.group("passed"))
                total = int(m.group("total"))
                failed = int(m.group("failed"))
                if failed != 0:
                    out("[FAIL] suite summary reports failures: {} -> {}/{} PASSED (failed={})".format(
                        name, passed, total, failed))
                    continue
                if total == 0:
                    out("[FAIL] suite summary total is 0 (no assertion ran): '{}'".format(name))
                    continue
                if passed != total:
                    out("[FAIL] suite summary inconsistent: {} -> {}/{} PASSED".format(
                        name, passed, total))
                    continue
                out("[OK]   suite summary ok: {} -> {}/{} PASSED".format(name, passed, total))
                ok = True
            if not ok:
                bad_summary.append(name)
        res["suites_bad"] = list(bad_summary)

    # 自检记录联合判定（S7）：/log/selftest.log 定长 64B 记录（格式见 App/log/sys_log.h），
    # 仅故障写盘：文件缺失 / 0 字节 = 无故障 → PASS；存在任何一条记录 → FAIL
    selftest_fail = False
    if selftest_pass:
        sp = Path(selftest_log) if selftest_log else None
        if sp is None or not sp.is_file() or sp.stat().st_size == 0:
            out("[OK]   selftest: no fault record (file missing or empty)")
            res["selftest"] = {"path": str(sp) if sp else None, "faults": 0, "records": []}
        else:
            raw = sp.read_bytes()
            n = len(raw) // 64
            if len(raw) % 64:
                out("[WARN] selftest log has {} trailing incomplete bytes (ignored)".format(len(raw) % 64))
            recs = []
            for i in range(n):
                rec = raw[i * 64:(i + 1) * 64]
                ts = rec[0:19].decode("ascii", errors="replace")
                desc = rec[31:59].split(b"\0", 1)[0].decode("utf-8", errors="replace")
                recs.append({"no": i, "ts": ts, "desc": desc})
            res["selftest"] = {"path": str(sp), "faults": n, "records": recs}
            if n > 0:
                selftest_fail = True
                out("[FAIL] selftest: {} fault record(s) in {}".format(n, selftest_log))
                for r in recs:
                    out("       #{}: {}  {}".format(r["no"], r["ts"], r["desc"]))
            else:
                out("[OK]   selftest: no fault record (0 complete records)")

    if show_tail > 0 and lines:
        res["tail"] = lines[-show_tail:]
        out("")
        out("=== Last {} line(s) of log ===".format(show_tail))
        for ln in lines[-show_tail:]:
            out(ln)

    out("")
    if (missing or hits or bad_summary or selftest_fail or boot_issue
            or order_issue or timing_issue):
        out("VERDICT: FAIL")
        return finish(1)
    out("VERDICT: PASS")
    return finish(0)


def main(argv=None):
    # 控制台输出固定 UTF-8：Windows 代码页（GBK/UTF-8 终端）均不会让本脚本崩溃
    for stream in (sys.stdout, sys.stderr):
        try:
            stream.reconfigure(encoding="utf-8", errors="replace")
        except Exception:
            pass

    ap = argparse.ArgumentParser(description="MCU debug log marker check (pure stdlib)")
    ap.add_argument("--log-path", required=True, help="boot log file to check")
    ap.add_argument("--required", default=REQUIRED_DEFAULT,
                    help="required markers, ';' separated (case-sensitive)")
    ap.add_argument("--forbidden", default=FORBIDDEN_DEFAULT,
                    help="forbidden patterns, ';' separated (case-insensitive)")
    ap.add_argument("--require-suite-summary", default="",
                    help="suite names that MUST appear as '=== <name>: N/M PASSED (failed=0) ===' "
                         "with N>0, ';' separated (T-2). Empty = off")
    ap.add_argument("--show-tail-lines", type=int, default=40,
                    help="show last N log lines (0 = off)")
    ap.add_argument("--require-selftest-pass", action="store_true",
                    help="judge /log/selftest.log local copy (S7): "
                         "missing/empty = PASS, any 64B record = FAIL")
    ap.add_argument("--selftest-log", default="",
                    help="local copy path of /log/selftest.log (with --require-selftest-pass)")
    ap.add_argument("--boot-marker", default=BOOT_MARKER_DEFAULT,
                    help="per-boot banner substring (default '{}')".format(BOOT_MARKER_DEFAULT))
    ap.add_argument("--max-boot-count", type=int, default=0,
                    help="fail if the boot banner appears 0 times or more than N times "
                         "(0 = check off). run_boot_test.py passes 1 by default")
    ap.add_argument("--expect-reset-cause", default="",
                    help="';'-separated substrings the FIRST reset cause must match "
                         "(case-insensitive). Empty = report only, no verdict")
    ap.add_argument("--require-order", default="",
                    help="'>'-separated markers that must appear in that line order "
                         "(e.g. 'App init OK>Network Init>net init: OK'). Empty = off")
    ap.add_argument("--done-marker", default="",
                    help="marker whose line marks boot completion, for --max-boot-ms")
    ap.add_argument("--max-boot-ms", type=int, default=0,
                    help="fail if time from first line to --done-marker exceeds N ms "
                         "(needs '[+Nms]' line prefix; 0 = off)")
    ap.add_argument("--json", action="store_true",
                    help="emit one JSON object instead of the text verdict (exit code unchanged)")
    args = ap.parse_args(argv)

    return verify(args.log_path, args.required, args.forbidden,
                  args.show_tail_lines, suite_summary=args.require_suite_summary,
                  selftest_pass=args.require_selftest_pass,
                  selftest_log=args.selftest_log,
                  boot_marker=args.boot_marker,
                  max_boot_count=args.max_boot_count,
                  expect_reset_cause=args.expect_reset_cause,
                  require_order=args.require_order,
                  done_marker=args.done_marker,
                  max_boot_ms=args.max_boot_ms,
                  as_json=args.json)


if __name__ == "__main__":
    sys.exit(main())