#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""pbl_compiler 接线 / RBAC / CRUD 定义 一致性机械审计（QC 硬门禁自证工具）。

背景：连续多轮 QC 退回的根因都是「写了但没接线 / 接了线但没登记 RBAC /
CRUD 定义格式不合规」——人肉声明不可信，故本脚本用机械比对钉死所有登记面。

比对集合：
  期望端点集合 EXPECTED = CONTRACTS ∪ READONLY_DENY
    ① pbl_compiler/init.py        CONTRACTS 契约名集合（20）
    ② pbl_compiler/__init__.py    __all__ 导出集合（须 ⊇ CONTRACTS）
    ③ wwwroot/api/*.dspy          实际端点文件集合（须 == EXPECTED）
    ④ scripts/load_path.py        PATHS 登记集合（须 == EXPECTED）
    ⑤ scripts/load_path.py        PATHS_READONLY_DENY（CRUD 只读拒绝端点，非契约）
  另加 json/*.json 结构审计（QC #3：改任一 CRUD 文件必须扫 json/ 全部，Pitfall 10）：
    · 根键白名单 {tblname, alias, title, params, _comment}（禁根级 browserfields）
    · browserfields 只允许挂在 params 下
    · alias 不得与模块名同名（会与模块 wwwroot 目录冲突）
    · params.editable 必须齐备 new_data_url / update_data_url / delete_data_url
      且三个 URL 指向的 .dspy 文件真实存在（防 xls2ui 回退默认写入口）
    · browserfields/editexclouded 引用的字段必须存在于 models/{tblname}.json

退出码：0 全通过；1 存在 FAIL。交付前必须实跑并在交付摘要引用输出。
"""
import ast
import json
import os
import re
import sys

HERE = os.path.dirname(os.path.abspath(__file__))
ROOT = os.path.dirname(HERE)            # 模块仓库根
PKG = os.path.join(ROOT, 'pbl_compiler')
WWW = os.path.join(ROOT, 'wwwroot')
API = os.path.join(WWW, 'api')
JSONDIR = os.path.join(ROOT, 'json')
MODELSDIR = os.path.join(ROOT, 'models')

failures = []
notes = []


def fail(msg):
    failures.append(msg)


def ok(msg):
    notes.append('PASS ' + msg)


def parse(path):
    """ast 解析：返回 (tree, 字面量常量, 顶层 def 名集合, 顶层赋值目标名)。"""
    with open(path, 'r', encoding='utf-8') as fh:
        tree = ast.parse(fh.read(), filename=path)
    consts, defs, assigned = {}, set(), []
    for node in tree.body:
        if isinstance(node, ast.Assign):
            for tgt in node.targets:
                if isinstance(tgt, ast.Name):
                    assigned.append(tgt.id)
                    try:
                        consts[tgt.id] = ast.literal_eval(node.value)
                    except Exception:       # noqa: BLE001 - 含函数引用的 dict 非字面量
                        consts[tgt.id] = node.value      # 保留 AST 供专用提取
        elif isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
            defs.add(node.name)
    return tree, consts, defs, assigned


def dict_keys(node):
    """从 dict AST（或已 literal_eval 的 dict）取字符串键集合。

    init.py 的 CONTRACTS 形如 {'name': fn_obj}，值是实现函数引用，
    ast.literal_eval 整体求值会失败——必须只提取键，不能因此判为空。
    """
    if isinstance(node, dict):
        return {k for k in node if isinstance(k, str)}
    if isinstance(node, ast.Dict):
        return {k.value for k in node.keys if isinstance(k, ast.Constant)
                and isinstance(k.value, str)}
    return set()


def literal_or_none(node):
    """AST 节点 → python 值；失败返回 None。"""
    try:
        return ast.literal_eval(node) if isinstance(node, ast.AST) else node
    except Exception:                       # noqa: BLE001
        return None


# ── ① init.py CONTRACTS ────────────────────────────────────────────────────
init_py = os.path.join(PKG, 'init.py')
CONTRACTS = set()
impl_defs = set()
if not os.path.isfile(init_py):
    fail('缺 pbl_compiler/init.py')
else:
    _, consts, _, _ = parse(init_py)
    if 'CONTRACTS' not in consts:
        fail('init.py 无 CONTRACTS 常量（三处同步的事实源丢失）')
    else:
        CONTRACTS = dict_keys(consts['CONTRACTS'])
        if not CONTRACTS:
            fail('init.py CONTRACTS 为空 dict')
        else:
            ok('init.py CONTRACTS = %d 个契约' % len(CONTRACTS))

# 实现层：契约必须在 api.py 或 rules_export_api.py 中有同名 async def
for impl in ('api.py', 'rules_export_api.py'):
    p = os.path.join(PKG, impl)
    if os.path.isfile(p):
        impl_defs |= parse(p)[2]
ghost = sorted(CONTRACTS - impl_defs)
if ghost:
    fail('CONTRACTS 注册了实现层不存在的函数（幽灵注册 → AttributeError）: %s' % ghost)
else:
    ok('%d 个契约在 api.py/rules_export_api.py 均有同名实现' % len(CONTRACTS))

# ── ② __init__.py __all__ ─────────────────────────────────────────────────
pkg_init = os.path.join(PKG, '__init__.py')
if not os.path.isfile(pkg_init):
    fail('缺 pbl_compiler/__init__.py')
else:
    _, consts, _, _ = parse(pkg_init)
    all_names = set(consts.get('__all__') or [])
    missing = sorted(CONTRACTS - all_names)
    if missing:
        fail('__init__.py __all__ 未导出 CONTRACTS 契约（漏 ② → .dspy NameError）: %s' % missing)
    else:
        ok('__init__.py __all__ 覆盖全部 %d 个契约' % len(CONTRACTS))

# ── ⑤ load_path.py：PATHS / PATHS_CONTRACTS / PATHS_READONLY_DENY ──────────
lp_py = os.path.join(HERE, 'load_path.py')
entries, contract_entries, deny_entries = [], [], []
READONLY_DENY = set()
if not os.path.isfile(lp_py):
    fail('缺 scripts/load_path.py')
else:
    _, consts, _, _ = parse(lp_py)
    # 先把各列表常量解析成 python 值，再解析 PATHS = A + B（Name 需回查 consts）。
    # 关键：parse() 对「纯字面量列表」已 literal_eval 成 python list，只有求值失败
    # 的赋值才保留 AST。两种形态都必须收进 resolved —— 否则 PATHS = PATHS_CONTRACTS
    # + PATHS_READONLY_DENY 的 BinOp 回查会拿到空列表，把已登记的 29 条误判成
    # 「PATHS 为空（RBAC 零登记）」，产生假 FAIL（本轮排查出的审计工具自身缺陷）。
    resolved = {}
    for var, raw in consts.items():
        if isinstance(raw, list):               # parse() 已求值的纯字面量列表
            resolved[var] = raw
        elif isinstance(raw, ast.List):         # 含函数引用等求值失败的列表 AST
            try:
                resolved[var] = [ast.literal_eval(e) for e in raw.elts]
            except Exception:                   # noqa: BLE001
                resolved[var] = []
    for var, bucket in (('PATHS', entries),
                        ('PATHS_CONTRACTS', contract_entries),
                        ('PATHS_READONLY_DENY', deny_entries)):
        raw = consts.get(var)
        if raw is None:
            fail('load_path.py 缺 %s 列表' % var)
            continue
        if isinstance(raw, ast.List):
            raw = resolved.get(var, [])
        elif isinstance(raw, ast.Name):
            raw = resolved.get(raw.id, [])
        elif isinstance(raw, ast.BinOp):        # PATHS = PATHS_CONTRACTS + PATHS_READONLY_DENY
            left = resolved.get(getattr(raw.left, 'id', None), []) if isinstance(raw.left, ast.Name) \
                else (literal_or_none(raw.left) or [])
            right = resolved.get(getattr(raw.right, 'id', None), []) if isinstance(raw.right, ast.Name) \
                else (literal_or_none(raw.right) or [])
            raw = list(left) + list(right)
        for item in raw or []:
            if not (isinstance(item, (list, tuple)) and len(item) == 2):
                fail('load_path.py %s 元素必须是 (path, role) 二元组: %r' % (var, item))
                continue
            bucket.append((item[0], item[1]))
    if not entries:
        fail('load_path.py PATHS 为空（RBAC 零登记 → 全部 403）')
    READONLY_DENY = {p.rsplit('/', 1)[-1][:-len('.dspy')]
                     for p, _ in deny_entries if '/api/' in p}
    for p, _role in entries:
        if '%' in p or '*' in p:
            fail('load_path.py 禁用通配符: %s' % p)
    if not any('%' in p or '*' in p for p, _ in entries):
        ok('load_path.py 无通配符（%d 条全显式）' % len(entries))
    dup = len(entries) - len({p for p, _ in entries})
    if dup:
        fail('load_path.py 存在重复登记 %d 条' % dup)
    else:
        ok('load_path.py 登记 %d 条无重复（契约 %d + 只读拒绝 %d）'
           % (len(entries), len(contract_entries), len(deny_entries)))

PATH_NAMES = {p.rsplit('/', 1)[-1][:-len('.dspy')]
              for p, _ in entries if '/api/' in p}
CONTRACT_PATH_NAMES = {p.rsplit('/', 1)[-1][:-len('.dspy')]
                       for p, _ in contract_entries if '/api/' in p}
EXPECTED = CONTRACTS | READONLY_DENY

if CONTRACT_PATH_NAMES != CONTRACTS:
    fail('PATHS_CONTRACTS 与 CONTRACTS 不一致：未登记=%s 幽灵登记=%s'
         % (sorted(CONTRACTS - CONTRACT_PATH_NAMES),
            sorted(CONTRACT_PATH_NAMES - CONTRACTS)))
else:
    ok('PATHS_CONTRACTS 与 CONTRACTS 一一对应（%d 条）' % len(CONTRACTS))
if PATH_NAMES != EXPECTED:
    fail('PATHS 与 CONTRACTS∪READONLY_DENY 不一致：未登记=%s 幽灵登记=%s'
         % (sorted(EXPECTED - PATH_NAMES), sorted(PATH_NAMES - EXPECTED)))
else:
    ok('PATHS 登记 %d 条，与期望端点集合（契约 %d + 只读拒绝 %d）一一对应'
       % (len(PATH_NAMES), len(CONTRACTS), len(READONLY_DENY)))

# ── ③ wwwroot/api/*.dspy ──────────────────────────────────────────────────
dspy_files = sorted(f for f in os.listdir(API) if f.endswith('.dspy')) \
    if os.path.isdir(API) else []
DSPY_NAMES = {f[:-len('.dspy')] for f in dspy_files}
if DSPY_NAMES != EXPECTED:
    fail('wwwroot/api/ 端点与期望集合不一致：缺端点=%s 多余端点=%s'
         % (sorted(EXPECTED - DSPY_NAMES), sorted(DSPY_NAMES - EXPECTED)))
else:
    ok('wwwroot/api/*.dspy = %d 个，与 CONTRACTS∪READONLY_DENY 一一对应'
       % len(DSPY_NAMES))
n_api = len([1 for p, _ in entries if '/api/' in p])
if n_api != len(dspy_files):
    fail('端点数与 RBAC 登记数不等：%d != %d' % (n_api, len(dspy_files)))
else:
    ok('grep 口径自证：PATHS 中 /api/ 条目=%d，wwwroot/api/*.dspy=%d'
       % (n_api, len(dspy_files)))

# ── .dspy 规范审计（dspy-file-implementation-spec）─────────────────────────
bad_import, no_return, env_req = [], [], []
for fname in dspy_files:
    text = open(os.path.join(API, fname), 'r', encoding='utf-8').read()
    for line in text.splitlines():
        s = line.strip()
        if s.startswith('import ') or s.startswith('from '):
            if 'sqlor.filter' not in s:
                bad_import.append('%s: %s' % (fname, s))
    if not re.search(r'(?m)^\s*return\b', text):
        no_return.append(fname)
    if 'ServerEnv()' in text:
        env_req.append(fname)
if bad_import:
    fail('.dspy 存在非法顶层 import（应由 load_*() 导出）: %s' % bad_import)
else:
    ok('%d 个 .dspy 无非法 import（仅允许 sqlor.filter）' % len(dspy_files))
if no_return:
    fail('.dspy 缺显式 return（隐式返回 None → 前端拿到 null）: %s' % no_return)
else:
    ok('%d 个 .dspy 均有显式 return' % len(dspy_files))
if env_req:
    fail('.dspy 不得用 ServerEnv() 取请求态: %s' % env_req)
else:
    ok('.dspy 无 ServerEnv() 取请求态')

# ── 只读拒绝端点必须真返回 Error（防「名义拒绝、实际可写」）─────────────────
deny_hollow = []
for name in sorted(READONLY_DENY):
    p = os.path.join(API, name + '.dspy')
    if not os.path.isfile(p):
        deny_hollow.append('%s: 文件缺失' % name)
        continue
    text = open(p, 'r', encoding='utf-8').read()
    if 'widgettype' not in text or '"Error"' not in text and "'Error'" not in text:
        deny_hollow.append('%s: 未返回 Error widget' % name)
if deny_hollow:
    fail('只读拒绝端点空心化（未显式拒绝写入）: %s' % deny_hollow)
else:
    ok('%d 个只读拒绝端点均显式返回 Error widget（CRUD 写入口已封死）'
       % len(READONLY_DENY))

# ── CRUD 生成目录登记面（防「生成目录未登记 → 403」）───────────────────────
SKIP_DIRS = {'api', 'i18n', 'imgs', 'styles', 'scripts', 'bricks'}
crud_dirs = [d for d in sorted(os.listdir(WWW))
             if os.path.isdir(os.path.join(WWW, d)) and d not in SKIP_DIRS] \
    if os.path.isdir(WWW) else []
lp_paths = {p for p, _ in entries}
for alias in crud_dirs:
    adir = os.path.join(WWW, alias)
    for fn in sorted(os.listdir(adir)):
        if fn == 'index.ui' or re.match(r'^(get_|add_|update_|delete_)', fn):
            expect = '/pbl_compiler/%s/%s' % (alias, fn)
            if expect not in lp_paths:
                fail('CRUD 生成文件未登记 RBAC: %s' % expect)
if crud_dirs:
    ok('发现 CRUD 生成目录 %d 个，已逐文件核对登记' % len(crud_dirs))
else:
    ok('wwwroot/ 无 CRUD 生成子目录（只读列表走手写契约 api/pbl_script_rule_list.dspy）')

# ── json/*.json 结构审计（QC #3 + Pitfall 10：改一个必须全扫）──────────────
ROOT_ALLOWED = {'tblname', 'alias', 'title', 'params', '_comment'}
json_files = sorted(f for f in os.listdir(JSONDIR) if f.endswith('.json')) \
    if os.path.isdir(JSONDIR) else []
if not json_files:
    fail('json/ 目录为空（CRUD 定义缺失）')
root_key_bad, alias_bad, editable_bad, url_missing, field_ghost = [], [], [], [], []
for fn in json_files:
    path = os.path.join(JSONDIR, fn)
    try:
        d = json.load(open(path, 'r', encoding='utf-8'))
    except Exception as exc:                # noqa: BLE001
        fail('json/%s 不是合法 JSON: %s' % (fn, exc))
        continue
    rk = set(d.keys())
    if not rk <= ROOT_ALLOWED:
        root_key_bad.append('%s: 非法根键 %s' % (fn, sorted(rk - ROOT_ALLOWED)))
        continue
    if 'browserfields' in rk:
        root_key_bad.append('%s: 根级 browserfields（规范只允许 params.browserfields）' % fn)
        continue
    alias = d.get('alias')
    if alias == 'pbl_compiler':
        alias_bad.append('%s: alias 与模块同名，生成目录会与模块 wwwroot 冲突' % fn)
    params = d.get('params') or {}
    if 'browserfields' not in params:
        root_key_bad.append('%s: 缺 params.browserfields' % fn)
    editable = params.get('editable')
    if not isinstance(editable, dict):
        editable_bad.append('%s: params.editable 不是对象（%r）' % (fn, type(editable).__name__))
        continue
    need = ('new_data_url', 'update_data_url', 'delete_data_url')
    absent = [k for k in need if k not in editable]
    if absent:
        editable_bad.append('%s: editable 缺 %s' % (fn, absent))
    for k in need:
        v = editable.get(k)
        if v is None:
            continue
        if not isinstance(v, dict) or 'url' not in v:
            editable_bad.append('%s: editable.%s 必须是含 url 的对象' % (fn, k))
            continue
        m = re.search(r"entire_url\('([^']+)'\)", v['url'])
        if not m:
            editable_bad.append('%s: editable.%s.url 未用 entire_url()' % (fn, k))
            continue
        ref = m.group(1)
        base = os.path.basename(ref)
        if base.startswith('/'):
            base = base.lstrip('/')
        if not (os.path.isfile(os.path.join(API, base))
                or os.path.isfile(os.path.join(ROOT, 'wwwroot', base))):
            url_missing.append('%s: editable.%s → %s 无匹配 .dspy' % (fn, k, ref))
    gd = editable.get('get_data_url')
    if isinstance(gd, dict) and gd.get('url'):
        m = re.search(r"entire_url\('([^']+)'\)", gd['url'])
        if m and not os.path.isfile(os.path.join(API, os.path.basename(m.group(1)))):
            url_missing.append('%s: get_data_url → %s 无匹配 .dspy' % (fn, m.group(1)))
    # 字段幽灵检查：browserfields 引用的字段必须在 models/{tblname}.json 存在
    mp = os.path.join(MODELSDIR, (d.get('tblname') or '') + '.json')
    if os.path.isfile(mp):
        cols = {f.get('name') for f in json.load(open(mp, 'r', encoding='utf-8')).get('fields', [])}
        for scope in ('browserfields', 'editexclouded'):
            bf = params.get(scope)
            names = list(bf.keys()) if isinstance(bf, dict) and scope == 'browserfields' \
                else (bf or [])
            for name in names:
                if name not in cols:
                    field_ghost.append('%s: %s 引用 models 中不存在的字段 %s'
                                       % (fn, scope, name))
    else:
        field_ghost.append('%s: 缺 models/%s.json' % (fn, d.get('tblname')))
for lst, msg in ((root_key_bad, 'json/ 根键不合规'),
                 (alias_bad, 'json/ alias 不合规'),
                 (editable_bad, 'json/ editable 段不合规'),
                 (url_missing, 'json/ editable URL 无匹配 .dspy'),
                 (field_ghost, 'json/ 字段与 models 不一致')):
    if lst:
        fail('%s: %s' % (msg, lst))
if not (root_key_bad or alias_bad or editable_bad or url_missing or field_ghost):
    ok('json/*.json 共 %d 个：根键白名单(%s) / alias / editable 三 URL 齐备且指向真实 .dspy '
       '/ 字段与 models 全对齐' % (len(json_files), sorted(ROOT_ALLOWED)))

# ── 输出 ───────────────────────────────────────────────────────────────────
print('=== pbl_compiler 接线/RBAC/CRUD parity 审计 ===')
for line in notes:
    print('  ' + line)
if failures:
    print('--- FAIL (%d) ---' % len(failures))
    for line in failures:
        print('  ' + line)
    print('RESULT: FAIL')
    sys.exit(1)
print('RESULT: PASS (contracts=%d, readonly_deny=%d, dspy=%d, rbac_api_paths=%d, json=%d)'
      % (len(CONTRACTS), len(READONLY_DENY), len(dspy_files), len(PATH_NAMES),
         len(json_files)))
sys.exit(0)