280 lines
11 KiB
Python
280 lines
11 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""M1b import 闭包核验器（静态，零执行）。
|
||
|
||
对 modules/ 下所有 .py 做 AST 扫描，解析 `from X import a, b` / `import X` 中
|
||
**可在本仓库定位到源文件**的跨文件符号引用，逐条核对被引符号是否真的存在于
|
||
目标模块的导出面（顶层 def/class/赋值名 + `__all__` + 该模块自身 import 进来的
|
||
再导出名 + `try/except ImportError` 兼容块里的名字）。
|
||
|
||
输出：
|
||
* 逐条断裂清单 `[path, lineno, symbol, kind]`（kind = missing_symbol / missing_module）
|
||
* 按 (kind, symbol) 聚合的汇总
|
||
* 分区统计：本模块（pbl_blueprint）内 vs 跨模块（pbl_common / pbl_agent_runtime / ...）
|
||
|
||
用法：
|
||
python3 tools/m1b_import_closure.py # 全 modules/ 扫描
|
||
python3 tools/m1b_import_closure.py --self # 只扫 pbl_blueprint 包内
|
||
python3 tools/m1b_import_closure.py --json # 额外输出 JSON 明细
|
||
"""
|
||
from __future__ import print_function
|
||
|
||
import ast
|
||
import io
|
||
import json
|
||
import os
|
||
import sys
|
||
from collections import defaultdict
|
||
|
||
HERE = os.path.dirname(os.path.abspath(__file__))
|
||
REPO = os.path.dirname(HERE) # modules/pbl_blueprint
|
||
MODULES = os.path.dirname(REPO) # modules/
|
||
SELF_PKG = os.path.join(REPO, "pbl_blueprint")
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 模块名 -> 源文件路径 索引
|
||
# --------------------------------------------------------------------------
|
||
def build_index():
|
||
"""{dotted.module: filepath}，覆盖 modules/ 下所有包与顶层模块。"""
|
||
idx = {}
|
||
for entry in sorted(os.listdir(MODULES)):
|
||
root = os.path.join(MODULES, entry)
|
||
if not os.path.isdir(root) or entry.startswith("."):
|
||
continue
|
||
# 包目录：modules/{entry}/{entry}/__init__.py
|
||
pkg = os.path.join(root, entry.replace("-", "_"))
|
||
if os.path.isfile(os.path.join(pkg, "__init__.py")):
|
||
base = entry.replace("-", "_")
|
||
idx[base] = os.path.join(pkg, "__init__.py")
|
||
for dirpath, dirnames, filenames in os.walk(pkg):
|
||
dirnames[:] = [d for d in dirnames
|
||
if d not in ("__pycache__", ".git")]
|
||
for fn in filenames:
|
||
if not fn.endswith(".py"):
|
||
continue
|
||
full = os.path.join(dirpath, fn)
|
||
rel = os.path.relpath(full, os.path.dirname(pkg))
|
||
mod = rel[:-3].replace(os.sep, ".")
|
||
if mod.endswith(".__init__"):
|
||
mod = mod[: -len(".__init__")]
|
||
idx[mod] = full
|
||
# 顶层单文件模块：modules/{entry}.py 不存在，跳过
|
||
return idx
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 单文件导出面
|
||
# --------------------------------------------------------------------------
|
||
_EXPORT_CACHE = {}
|
||
|
||
|
||
def _names_from_body(body, out, depth=0):
|
||
"""收集顶层（含 try/if 块内）定义与导入的名字。"""
|
||
for node in body:
|
||
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)):
|
||
out.add(node.name)
|
||
elif isinstance(node, ast.Assign):
|
||
for t in node.targets:
|
||
_collect_targets(t, out)
|
||
elif isinstance(node, ast.AnnAssign):
|
||
_collect_targets(node.target, out)
|
||
elif isinstance(node, ast.AugAssign):
|
||
_collect_targets(node.target, out)
|
||
elif isinstance(node, ast.Import):
|
||
for al in node.names:
|
||
out.add((al.asname or al.name).split(".")[0])
|
||
elif isinstance(node, ast.ImportFrom):
|
||
for al in node.names:
|
||
if al.name == "*":
|
||
out.add("*")
|
||
else:
|
||
out.add(al.asname or al.name)
|
||
elif isinstance(node, (ast.Try, ast.If, ast.With, ast.For, ast.While)):
|
||
sub = getattr(node, "body", []) or []
|
||
for attr in ("orelse", "finalbody"):
|
||
sub += getattr(node, attr, []) or []
|
||
for h in getattr(node, "handlers", []) or []:
|
||
sub += h.body or []
|
||
_names_from_body(sub, out, depth + 1)
|
||
|
||
|
||
def _collect_targets(target, out):
|
||
if isinstance(target, ast.Name):
|
||
out.add(target.id)
|
||
elif isinstance(target, (ast.Tuple, ast.List)):
|
||
for e in target.elts:
|
||
_collect_targets(e, out)
|
||
elif isinstance(target, ast.Starred):
|
||
_collect_targets(target.value, out)
|
||
|
||
|
||
def exports_of(path):
|
||
"""返回 (names:set, has_star:bool, all_list:list|None)。"""
|
||
key = path
|
||
if key in _EXPORT_CACHE:
|
||
return _EXPORT_CACHE[key]
|
||
names, all_list = set(), None
|
||
try:
|
||
with io.open(path, encoding="utf-8") as f:
|
||
src = f.read()
|
||
tree = ast.parse(src, filename=path)
|
||
except Exception as e:
|
||
_EXPORT_CACHE[key] = (set(), False, None, "PARSE_ERROR: %s" % e)
|
||
return _EXPORT_CACHE[key]
|
||
_names_from_body(tree.body, names)
|
||
# __all__ 字面量
|
||
for node in tree.body:
|
||
if isinstance(node, ast.Assign):
|
||
for t in node.targets:
|
||
if isinstance(t, ast.Name) and t.id == "__all__":
|
||
try:
|
||
all_list = [ast.literal_eval(e) for e in node.value.elts]
|
||
names.update(all_list)
|
||
except Exception:
|
||
pass
|
||
parse_err = None
|
||
res = (names, "*" in names, all_list, parse_err)
|
||
_EXPORT_CACHE[key] = res
|
||
return res
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 解析 import 语句
|
||
# --------------------------------------------------------------------------
|
||
def resolve(modname, index, cur_file):
|
||
"""把 dotted 模块名解析成源文件路径；解析不到返回 None（视为外部依赖）。"""
|
||
if modname in index:
|
||
return index[modname]
|
||
# 相对当前包的猜测（from . import x 已由 level 处理）
|
||
return None
|
||
|
||
|
||
def scan_file(path, index, breaks):
|
||
try:
|
||
with io.open(path, encoding="utf-8") as f:
|
||
src = f.read()
|
||
tree = ast.parse(src, filename=path)
|
||
except Exception as e:
|
||
breaks.append((path, 0, "<parse-error:%s>" % e, "parse_error"))
|
||
return
|
||
|
||
pkg_root = None
|
||
for mod, fp in index.items():
|
||
if fp == path:
|
||
pkg_root = mod.rsplit(".", 1)[0] if "." in mod else mod
|
||
break
|
||
|
||
for node in ast.walk(tree):
|
||
if isinstance(node, ast.ImportFrom):
|
||
level = node.level or 0
|
||
mod = node.module or ""
|
||
if level:
|
||
# 相对导入：按当前文件所在包回溯
|
||
base = path
|
||
for _ in range(level):
|
||
base = os.path.dirname(base)
|
||
rel = mod.replace(".", os.sep) if mod else ""
|
||
cand_pkg = os.path.join(base, rel, "__init__.py") if rel else os.path.join(base, "__init__.py")
|
||
cand_mod = os.path.join(base, rel + ".py") if rel else None
|
||
if os.path.isfile(cand_pkg):
|
||
target = cand_pkg
|
||
elif cand_mod and os.path.isfile(cand_mod):
|
||
target = cand_mod
|
||
else:
|
||
target = None
|
||
tname = ("<relative %s%s>" % ("." * level, mod))
|
||
else:
|
||
target = resolve(mod, index, path)
|
||
tname = mod
|
||
if target is None:
|
||
# 外部依赖（标准库/三方/未落库模块）——不计入本仓库闭包断裂，
|
||
# 但若形如 pbl_* / 本仓库已知顶层包前缀，则记 missing_module。
|
||
if mod.startswith("pbl_") or (level and False):
|
||
for al in node.names:
|
||
breaks.append((path, node.lineno,
|
||
"%s.%s" % (tname, al.name),
|
||
"missing_module"))
|
||
continue
|
||
names, has_star, all_list, perr = exports_of(target)
|
||
if perr:
|
||
breaks.append((target, 0, perr, "parse_error"))
|
||
if has_star:
|
||
continue
|
||
for al in node.names:
|
||
if al.name == "*":
|
||
continue
|
||
if al.name not in names:
|
||
breaks.append((path, node.lineno, al.name,
|
||
"missing_symbol:%s" % tname))
|
||
elif isinstance(node, ast.Import):
|
||
for al in node.names:
|
||
if resolve(al.name, index, path) is None and al.name.startswith("pbl_"):
|
||
breaks.append((path, node.lineno, al.name, "missing_module"))
|
||
|
||
|
||
def main():
|
||
argv = sys.argv[1:]
|
||
self_only = "--self" in argv
|
||
as_json = "--json" in argv
|
||
|
||
index = build_index()
|
||
targets = []
|
||
if self_only:
|
||
for dirpath, dirnames, filenames in os.walk(SELF_PKG):
|
||
dirnames[:] = [d for d in dirnames if d != "__pycache__"]
|
||
for fn in filenames:
|
||
if fn.endswith(".py"):
|
||
targets.append(os.path.join(dirpath, fn))
|
||
for extra in ("tests", "tools", "scripts"):
|
||
d = os.path.join(REPO, extra)
|
||
if os.path.isdir(d):
|
||
for dirpath, dirnames, filenames in os.walk(d):
|
||
dirnames[:] = [x for x in dirnames if x != "__pycache__"]
|
||
for fn in filenames:
|
||
if fn.endswith(".py"):
|
||
targets.append(os.path.join(dirpath, fn))
|
||
else:
|
||
for mod, fp in index.items():
|
||
targets.append(fp)
|
||
|
||
breaks = []
|
||
for fp in sorted(set(targets)):
|
||
scan_file(fp, index, breaks)
|
||
|
||
in_self, cross = [], []
|
||
for b in breaks:
|
||
p = b[0]
|
||
if p.startswith(REPO + os.sep) or p == REPO:
|
||
in_self.append(b)
|
||
else:
|
||
cross.append(b)
|
||
|
||
def dump(title, rows):
|
||
print("\n=== %s (%d) ===" % (title, len(rows)))
|
||
agg = defaultdict(list)
|
||
for path, line, sym, kind in rows:
|
||
agg[(kind, sym)].append("%s:%s" % (os.path.relpath(path, MODULES), line))
|
||
for (kind, sym), locs in sorted(agg.items()):
|
||
print(" %-22s %-40s x%-3d %s" % (kind, sym, len(locs), locs[0]))
|
||
|
||
dump("本模块 pbl_blueprint 内断裂", in_self)
|
||
dump("跨模块断裂（他模块职责，冒泡 PM）", cross)
|
||
|
||
if as_json:
|
||
out = os.path.join(REPO, "tools", "m1b_closure_report.json")
|
||
with io.open(out, "w", encoding="utf-8") as f:
|
||
json.dump({
|
||
"self_breaks": [list(b) for b in in_self],
|
||
"cross_breaks": [list(b) for b in cross],
|
||
"self_count": len(in_self),
|
||
"cross_count": len(cross),
|
||
}, f, ensure_ascii=False, indent=2)
|
||
print("\n[json] %s" % out)
|
||
|
||
print("\nTOTAL self=%d cross=%d" % (len(in_self), len(cross)))
|
||
return 0 if not in_self else 2
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|