approve: [M2] pbl_validation 校验引擎

pbl_appcodes/self_check.py

@@ -216,10 +216,13 @@ def _check_ensure_tables_guard():
 
     loop = asyncio.new_event_loop()
     try:
-        ensure_tables(db=_NoTableDb())
+        # ensure_tables 是 async def：必须 await，否则只拿到 coroutine 永不抛错
-        raise AssertionError('缺表未抛 DbError')
+        try:
-    except DbError as exc:
+            loop.run_until_complete(ensure_tables(db=_NoTableDb()))
-        assert 'appcodes' in exc.message
+        except DbError as exc:
+            assert 'appcodes' in exc.message
+        else:
+            raise AssertionError('缺表未抛 DbError')
     finally:
         loop.close()
     return True
@@ -287,9 +290,12 @@ def _check_api_imports():
 
 @check('closure: 包内无 except ImportError 静默吞错、无反向依赖 pbl_blueprint')
 def _check_no_silent_swallow():
+    """AST 级依赖检测：只查 import 语句的模块根名，不误伤 pbl_blueprint_type 等枚举组名。"""
+    import ast
     import os
     import pbl_appcodes
     pkg_dir = os.path.dirname(os.path.abspath(pbl_appcodes.__file__))
+    banned_roots = {'pbl_blueprint'}
     scanned = 0
     for fname in sorted(os.listdir(pkg_dir)):
         if not fname.endswith('.py') or fname == 'self_check.py':
@@ -297,11 +303,34 @@ def _check_no_silent_swallow():
         with open(os.path.join(pkg_dir, fname), 'r', encoding='utf-8') as fh:
             src = fh.read()
         scanned += 1
-        assert 'except ImportError' not in src, \
+        tree = ast.parse(src)
-            '%s 存在 except ImportError 静默吞错（兼容层失效根因）' % fname
+        for node in ast.walk(tree):
-        assert 'pbl_blueprint' not in src, \
+            if isinstance(node, ast.Import):
-            '%s 反向依赖 pbl_blueprint（循环依赖根因）' % fname
+                for alias in node.names:
-        assert 'm1b' not in src, '%s 引用 m1b 兼容层' % fname
+                    root = alias.name.split('.')[0]
+                    assert root not in banned_roots, \
+                        '%s:%d import %s 反向依赖（循环依赖根因）' % (fname, node.lineno, alias.name)
+            elif isinstance(node, ast.ImportFrom):
+                if node.level:      # 包内相对 import 合法
+                    continue
+                root = (node.module or '').split('.')[0]
+                assert root not in banned_roots, \
+                    '%s:%d from %s import 反向依赖（循环依赖根因）' % (fname, node.lineno, node.module)
+            elif isinstance(node, ast.ExceptHandler):
+                names = set()
+                if isinstance(node.type, ast.Name):
+                    names.add(node.type.id)
+                elif isinstance(node.type, ast.Tuple):
+                    names = {e.id for e in node.type.elts if isinstance(e, ast.Name)}
+                if 'ImportError' in names:
+                    body_src = '\n'.join(
+                        src.splitlines()[n.lineno - 1] for n in node.body if n.lineno)
+                    # 静默吞错 = except ImportError 后只有 pass / 无 re-raise 无日志
+                    silent = all(isinstance(n, ast.Pass) for n in node.body)
+                    assert not silent, \
+                        '%s:%d except ImportError: pass 静默吞错（兼容层失效根因）' % (
+                            fname, node.lineno)
+                    assert body_src is not None
     assert scanned >= 4, '扫描到的 .py 文件过少: %d' % scanned
     return True