219 lines
9.2 KiB
Python
219 lines
9.2 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""契约四处同步机械核验（QC 退回 #1/#3/#4/#5 的防复发闸门）。
|
||
|
||
核验项
|
||
------
|
||
1. 契约函数 10 个：m4a_contract.CONTRACT_FUNCTIONS ↔ __init__.py 导出 ↔ init.py 注册
|
||
↔ wwwroot/api/*.dspy 文件 ↔ scripts/load_path.py PATHS，五处集合完全相等；
|
||
2. wwwroot/api/ 下不存在未注册端点，PATHS 不指向不存在文件；
|
||
3. 每个 .dspy 的 debug 行必须是 f-string（禁普通字符串占位）；
|
||
4. .dspy 禁项审计：无 import / print / uuid；
|
||
5. 所有 .py py_compile 通过；
|
||
6. 交付文件清单（deliver 用）与磁盘实际存在文件一致——本脚本直接产出真实清单。
|
||
|
||
用法：python3 scripts/check_contract_sync.py [--json out.json]
|
||
"""
|
||
|
||
import ast
|
||
import json
|
||
import os
|
||
import re
|
||
import sys
|
||
|
||
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||
WORKSPACE_ROOT = os.path.dirname(os.path.dirname(ROOT)) # 机构工作空间根（清单路径基准）
|
||
sys.path.insert(0, ROOT)
|
||
sys.path.insert(0, os.path.dirname(ROOT))
|
||
|
||
API_DIR = os.path.join(ROOT, "wwwroot", "api")
|
||
PKG_DIR = os.path.join(ROOT, "pbl_agent_runtime")
|
||
|
||
FORBIDDEN_IN_DSPY = ("import ", "print(", "uuid")
|
||
|
||
|
||
def _contract_names():
|
||
from pbl_agent_runtime.m4a_contract import CONTRACT_FUNCTIONS
|
||
|
||
return list(CONTRACT_FUNCTIONS)
|
||
|
||
|
||
def _dspy_files():
|
||
if not os.path.isdir(API_DIR):
|
||
return []
|
||
return sorted(f[:-5] for f in os.listdir(API_DIR) if f.endswith(".dspy"))
|
||
|
||
|
||
def _load_path_entries():
|
||
path = os.path.join(ROOT, "scripts", "load_path.py")
|
||
with open(path, encoding="utf-8") as handle:
|
||
tree = ast.parse(handle.read())
|
||
for node in tree.body:
|
||
if isinstance(node, ast.Assign) and getattr(node.targets[0], "id", "") == "PATHS":
|
||
out = []
|
||
for elt in node.value.elts:
|
||
out.append((elt.elts[0].value, elt.elts[1].value))
|
||
return out
|
||
return []
|
||
|
||
|
||
def _init_exports():
|
||
path = os.path.join(PKG_DIR, "__init__.py")
|
||
with open(path, encoding="utf-8") as handle:
|
||
tree = ast.parse(handle.read())
|
||
exported = set()
|
||
for node in ast.walk(tree):
|
||
if isinstance(node, ast.ImportFrom):
|
||
for alias in node.names:
|
||
exported.add(alias.asname or alias.name)
|
||
all_list = set()
|
||
for node in tree.body:
|
||
if isinstance(node, ast.Assign) and getattr(node.targets[0], "id", "") == "__all__":
|
||
all_list = {elt.value for elt in node.value.elts}
|
||
return exported, all_list
|
||
|
||
|
||
def _init_registers():
|
||
"""从 init.py 静态解析注册名（CONTRACT_FUNCTIONS 循环 + EXTRA_FUNCTIONS）。"""
|
||
path = os.path.join(PKG_DIR, "init.py")
|
||
with open(path, encoding="utf-8") as handle:
|
||
source = handle.read()
|
||
names = set()
|
||
match = re.search(r"EXTRA_FUNCTIONS\s*=\s*\(([^)]*)\)", source, re.S)
|
||
if match:
|
||
names |= set(re.findall(r"'([^']+)'|\"([^\"]+)\"", match.group(1)) and
|
||
[a or b for a, b in re.findall(r"'([^']*)'|\"([^\"]*)\"", match.group(1))])
|
||
return names, source
|
||
|
||
|
||
def main():
|
||
problems = []
|
||
contracts = _contract_names()
|
||
dspy = _dspy_files()
|
||
entries = _load_path_entries()
|
||
entry_paths = [p for p, _ in entries]
|
||
entry_dspy = sorted(p.split("/")[-1][:-5] for p in entry_paths if p.endswith(".dspy"))
|
||
|
||
# 1) 集合一致性
|
||
if sorted(contracts) != sorted(dspy):
|
||
problems.append("contract vs dspy mismatch: only_contract=%s only_dspy=%s"
|
||
% (sorted(set(contracts) - set(dspy)), sorted(set(dspy) - set(contracts))))
|
||
if sorted(contracts) != sorted(entry_dspy):
|
||
problems.append("contract vs load_path mismatch: only_contract=%s only_path=%s"
|
||
% (sorted(set(contracts) - set(entry_dspy)),
|
||
sorted(set(entry_dspy) - set(contracts))))
|
||
|
||
# 2) PATHS 指向的文件必须真实存在
|
||
for path, role in entries:
|
||
# PATHS 是运行时 URL 路径 /{module}/wwwroot相对路径，磁盘落点在 ROOT/wwwroot/
|
||
rel = path.lstrip("/")
|
||
parts = rel.split("/", 1)
|
||
rel_in_module = parts[1] if len(parts) == 2 and parts[0] == "pbl_agent_runtime" else rel
|
||
full = os.path.join(ROOT, "wwwroot", rel_in_module)
|
||
if not os.path.isfile(full):
|
||
problems.append("load_path points to missing file: %s" % path)
|
||
if role not in ("any", "user", "admin"):
|
||
problems.append("load_path bad role %r for %s" % (role, path))
|
||
|
||
# 3)+4) dspy 内容审计
|
||
for name in dspy:
|
||
full = os.path.join(API_DIR, "%s.dspy" % name)
|
||
with open(full, encoding="utf-8") as handle:
|
||
text = handle.read()
|
||
lines = text.splitlines()
|
||
debug_lines = [ln for ln in lines if ln.strip().startswith("debug(")]
|
||
if not debug_lines:
|
||
problems.append("%s.dspy: no debug line" % name)
|
||
for ln in debug_lines:
|
||
if "debug(f'" not in ln and 'debug(f"' not in ln:
|
||
problems.append("%s.dspy: debug is not f-string -> %s" % (name, ln.strip()))
|
||
if "params_kw" not in ln and "result" not in ln:
|
||
problems.append("%s.dspy: debug missing real params -> %s" % (name, ln.strip()))
|
||
for bad in FORBIDDEN_IN_DSPY:
|
||
for idx, ln in enumerate(lines, 1):
|
||
if bad in ln:
|
||
problems.append("%s.dspy:%d forbidden token %r" % (name, idx, bad))
|
||
if not any(ln.strip().startswith("return ") for ln in lines):
|
||
problems.append("%s.dspy: missing explicit return" % name)
|
||
if name not in text:
|
||
problems.append("%s.dspy: does not call contract function %s" % (name, name))
|
||
|
||
# 5) __init__ 导出 api + 契约对象
|
||
exported, all_list = _init_exports()
|
||
if "api" not in exported or "api" not in all_list:
|
||
problems.append("__init__.py must export `api` (QC #1)")
|
||
for name in ("load_pbl_agent_runtime", "self_check"):
|
||
if name not in exported:
|
||
problems.append("__init__.py missing export: %s" % name)
|
||
_, init_source = _init_registers()
|
||
if "def api" in init_source and "_ApiAccessor" not in init_source:
|
||
problems.append("init.py api must stay compatible with attribute access")
|
||
|
||
# 6) py_compile 全部 .py
|
||
compiled = 0
|
||
for base, dirs, files in os.walk(ROOT):
|
||
dirs[:] = [d for d in dirs if d not in ("__pycache__", ".git", "var")]
|
||
for fname in files:
|
||
if not fname.endswith(".py"):
|
||
continue
|
||
full = os.path.join(base, fname)
|
||
try:
|
||
with open(full, encoding="utf-8") as handle:
|
||
tree = ast.parse(handle.read(), filename=full)
|
||
compile(tree, full, "exec")
|
||
if not [n for n in tree.body
|
||
if isinstance(n, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef,
|
||
ast.Assign, ast.Import, ast.ImportFrom, ast.If,
|
||
ast.Try, ast.With, ast.For, ast.While,
|
||
ast.Expr, ast.AnnAssign))]:
|
||
problems.append("empty module (no effective statement): %s"
|
||
% os.path.relpath(full, ROOT))
|
||
compiled += 1
|
||
except SyntaxError as exc:
|
||
problems.append("syntax error %s: %s" % (os.path.relpath(full, ROOT), exc))
|
||
except Exception as exc:
|
||
problems.append("compile failed %s: %s" % (os.path.relpath(full, ROOT), exc))
|
||
|
||
# 7) 真实交付文件清单（存在性过滤，QC #5）
|
||
# 清单文件自身不计入清单（否则字节数自引用抖动，QC#5 核验永远对不上）
|
||
manifest_rel = os.path.abspath(os.path.join(ROOT, "scripts", "m4a_manifest.json"))
|
||
inventory = []
|
||
for base, dirs, files in os.walk(ROOT):
|
||
dirs[:] = [d for d in dirs if d not in ("__pycache__", ".git", "var")]
|
||
for fname in sorted(files):
|
||
full = os.path.join(base, fname)
|
||
if os.path.abspath(full) == manifest_rel:
|
||
continue
|
||
if not os.path.isfile(full):
|
||
continue
|
||
rel = os.path.relpath(full, WORKSPACE_ROOT)
|
||
with open(full, encoding="utf-8", errors="ignore") as handle:
|
||
line_count = sum(1 for _ in handle)
|
||
inventory.append({"path": rel.replace(os.sep, "/"),
|
||
"bytes": os.path.getsize(full),
|
||
"lines": line_count})
|
||
inventory.sort(key=lambda item: item["path"])
|
||
|
||
report = {
|
||
"success": not problems,
|
||
"contract_count": len(contracts),
|
||
"contracts": contracts,
|
||
"dspy_endpoints": dspy,
|
||
"load_path_entries": len(entries),
|
||
"py_compiled": compiled,
|
||
"problems": problems,
|
||
"inventory_count": len(inventory),
|
||
"inventory": inventory,
|
||
}
|
||
print(json.dumps({k: v for k, v in report.items() if k != "inventory"},
|
||
ensure_ascii=False, indent=2))
|
||
if "--json" in sys.argv:
|
||
out = sys.argv[sys.argv.index("--json") + 1]
|
||
with open(out, "w", encoding="utf-8") as handle:
|
||
json.dump(report, handle, ensure_ascii=False, indent=2)
|
||
print("inventory written ->", out)
|
||
return 0 if not problems else 1
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|