PRA-only-demo/scripts/audit-catalog-params.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""审计能力参数 v2：catalog 字段 vs 实现读取。

策略：
- 直接读取：handler 函数体内 text(args,["k"]) / args.k / pick(args,[...]) 等模式
- 间接读取：body 中把 args 整体传给函数（xxx(args)），或必填 key 出现在同文件其他位置（兜底）
- A 类（假必填）只在「直接+间接+全文兜底都读不到」时报告，保证高置信
- B 类（隐藏参数）只报直接读取的 key（保守）
"""
import json
import re
import os

ROOT = r"D:\code\PRA-only-demo"
CAP_DIRS = [
    os.path.join(ROOT, "packages", "rpa-core", "src", "capabilities"),
    os.path.join(ROOT, "src", "main", "capabilities"),
]

with open(os.path.join(ROOT, "resources", "catalog.json"), encoding="utf-8") as f:
    catalog = {c["id"]: c for c in json.load(f)["components"]}

src_blobs = []
for d in CAP_DIRS:
    for fn in os.listdir(d):
        if fn.endswith(".ts") and not fn.endswith(".test.ts"):
            with open(os.path.join(d, fn), encoding="utf-8") as f:
                src_blobs.append((fn, f.read()))

def find_register_bodies(src):
    out = []
    for m in re.finditer(r'register\(\s*(\[[^\]]*\]|"[A-Za-z_.0-9]+")\s*,\s*(async)?\s*\(', src):
        ids = re.findall(r'"([A-Za-z_.0-9]+)"', m.group(1))
        i = m.end() - 1
        depth = 0
        start_body = None
        j = i
        while j < len(src):
            ch = src[j]
            if ch == "(":
                depth += 1
            elif ch == ")":
                depth -= 1
                if depth == 0:
                    break
            elif ch == "{" and start_body is None and depth >= 2:
                start_body = j
            j += 1
        if start_body is None:
            body = src[i + 1:j]
        else:
            d2 = 0
            k = start_body
            while k < len(src):
                if src[k] == "{":
                    d2 += 1
                elif src[k] == "}":
                    d2 -= 1
                    if d2 == 0:
                        break
                k += 1
            body = src[start_body:k]
        out.append((ids, body))
    return out

PAT_DIRECT = [
    re.compile(r'\w+\(\s*args\s*,\s*\[([^\]]*)\)'),
    re.compile(r'pick\(\s*args\s*,\s*\[([^\]]*)\)'),
    re.compile(r'\bargs\.([a-zA-Z_][a-zA-Z0-9_]*)'),
    re.compile(r"args\[['\"]([^'\"]+)['\"]\]"),
]
INTERNAL = {"nodeId", "componentId", "timeoutMs", "time_out", "timeout", "evidenceDir"}

info = {}  # cid -> dict(direct=set, indirect=bool)
for fn, src in src_blobs:
    for ids, body in find_register_bodies(src):
        direct = set()
        for pat in PAT_DIRECT:
            for mm in pat.finditer(body):
                g = mm.group(1)
                if g is None:
                    continue
                if "," in g or '"' in g or "'" in g:
                    direct.update(re.findall(r"['\"]([^'\"]+)['\"]", g))
                else:
                    direct.add(g)
        # args 整体传走 → 间接（keys 未知）
        indirect = bool(re.search(r'\(\s*args\s*[,)]|\bargs\s*\)', body)) and not direct
        # selectorOf(args) 这种：args 作为参数出现在调用里
        if re.search(r'\w+\(\s*args\s*\)|\w+\(\s*ctx\s*,\s*node\s*,\s*args', body):
            indirect = True
        for cid in ids:
            rec = info.setdefault(cid, {"direct": set(), "indirect": False, "file": fn})
            rec["direct"] |= direct
            rec["indirect"] = rec["indirect"] or indirect

A, B = [], []
for cid, comp in sorted(catalog.items()):
    rec = info.get(cid)
    if not rec:
        continue
    fields = comp.get("fields") or []
    cat_keys = {f["key"] for f in fields}
    required = {f["key"] for f in fields if f.get("required")}
    direct = rec["direct"]
    if rec["indirect"]:
        # keys 未知：不判 A（无法确定），不判 B
        continue
    a = {k for k in required if k not in direct}
    # 兜底：全目录源码找 key（带引号形式，或 .key 词边界的直接访问）
    if a:
        all_src = "\n".join(s for _, s in src_blobs)
        a = {k for k in a if not (re.search(r'["\']' + re.escape(k) + r'["\']', all_src) or re.search(r'\.' + re.escape(k) + r'\b', all_src))}
    if a:
        A.append((cid, sorted(a)))
    b = {k for k in direct if k not in cat_keys and k not in INTERNAL and len(k) > 2}
    if b:
        B.append((cid, sorted(b)))

print(f"=== A 类：必填但实现全链路都不读（高置信假必填） {len(A)} 个 ===")
for cid, keys in A:
    print(f"  {cid}: {keys}")
print()
print(f"=== B 类：实现直接读取但目录未定义（隐藏参数，保守列表） {len(B)} 个 ===")
for cid, keys in B:
    print(f"  {cid}: {keys}")