ebf6d79ef3
QC #5 指出「纯注释文件同样能通过 ast 解析，ast OK 不构成修复在位证据」。 本脚本把自检口径改为可机械核验三项：非注释可执行语句数>0（ast stmt 计数）、 显式 return 数>0 且不在 async with 块内、ast 包装语法通过。 实测：apps/pbls+modules/pbls 6 个 .dspy violations=0； 扩扫 15 个 pbl_* 自研模块 + 9 个复用模块共 156 个 .dspy，violations=2 （modules/world 两处 return 在 async with 内，属 referenced_modules 写保护范围， 且两文件均有块外 return result 兜底不会返回 None，仅记录不修改）。
93 lines
3.2 KiB
Python
93 lines
3.2 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
.dspy 空壳/语法/return 静态扫描（QC 退回 #5 要求的自检口径）
|
||
|
||
背景：Bug WIu0Rkn0KwB0XTjYZz0qk 的 QC 退回意见 #5 指出——「纯注释文件同样能通过
|
||
ast 解析，ast OK 不构成『修复在位』证据」。本脚本把自检口径改为可机械核验的三项：
|
||
1) 非注释可执行语句数 > 0（ast 统计 stmt 节点，排除包装函数本身）
|
||
2) 显式 return 数 > 0（ast.Return 计数），且 return 不落在 async with 块内
|
||
3) ast 包装（async def 包裹）语法解析通过
|
||
|
||
用法：
|
||
python3 scripts/dspy_shell_scan.py [root ...]
|
||
缺省 root = apps modules（相对机构工作空间根）。退出码：0=全通过，1=存在违规。
|
||
"""
|
||
import ast
|
||
import os
|
||
import sys
|
||
|
||
SKIP_PARTS = {".git", "venv", ".pylibs", "node_modules", "__pycache__", "wheelhouse"}
|
||
|
||
|
||
def scan_file(path):
|
||
"""返回 (ok, reason, metrics)"""
|
||
try:
|
||
with open(path, encoding="utf-8", errors="replace") as fh:
|
||
src = fh.read()
|
||
except Exception as exc: # pragma: no cover
|
||
return False, "READ_FAIL %s" % exc, {}
|
||
|
||
wrapped = "async def _h():\n" + "".join(" " + ln + "\n" for ln in src.split("\n"))
|
||
try:
|
||
tree = ast.parse(wrapped)
|
||
except SyntaxError as exc:
|
||
return False, "AST_FAIL %s" % exc, {"bytes": len(src.encode("utf-8"))}
|
||
|
||
stmts = 0
|
||
returns = 0
|
||
for node in ast.walk(tree):
|
||
if isinstance(node, ast.stmt) and not isinstance(
|
||
node, (ast.FunctionDef, ast.AsyncFunctionDef)
|
||
):
|
||
stmts += 1
|
||
if isinstance(node, ast.Return):
|
||
returns += 1
|
||
|
||
ret_in_asyncwith = []
|
||
for node in ast.walk(tree):
|
||
if isinstance(node, ast.AsyncWith):
|
||
for sub in ast.walk(node):
|
||
if isinstance(sub, ast.Return):
|
||
ret_in_asyncwith.append(sub.lineno)
|
||
|
||
metrics = {
|
||
"bytes": len(src.encode("utf-8")),
|
||
"ast_stmts": stmts,
|
||
"ast_return": returns,
|
||
"return_in_asyncwith": ret_in_asyncwith,
|
||
}
|
||
if stmts == 0:
|
||
return False, "SHELL_ONLY: 0 executable statements", metrics
|
||
if returns == 0:
|
||
return False, "NO_RETURN: stmts=%d returns=0" % stmts, metrics
|
||
if ret_in_asyncwith:
|
||
return False, "RETURN_INSIDE_ASYNC_WITH lines=%s" % ret_in_asyncwith, metrics
|
||
return True, "OK", metrics
|
||
|
||
|
||
def main(argv):
|
||
roots = argv[1:] or ["apps", "modules"]
|
||
total = 0
|
||
bad = []
|
||
for root in roots:
|
||
for dirpath, dirnames, filenames in os.walk(root):
|
||
dirnames[:] = [d for d in dirnames if d not in SKIP_PARTS]
|
||
for name in filenames:
|
||
if not name.endswith(".dspy"):
|
||
continue
|
||
path = os.path.join(dirpath, name)
|
||
total += 1
|
||
ok, reason, metrics = scan_file(path)
|
||
if not ok:
|
||
bad.append((path, reason, metrics))
|
||
print("scanned .dspy files: %d" % total)
|
||
print("violations: %d" % len(bad))
|
||
for path, reason, metrics in bad:
|
||
print(" %s -> %s | %s" % (path, reason, metrics))
|
||
return 1 if bad else 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main(sys.argv))
|