test: 强化 async 阻塞门禁真实性 (#6398)

scripts/architecture/async_blocking.py

@@ -6,7 +6,8 @@ import argparse
 import ast
 import json
 from collections import Counter
-from collections.abc import Iterator
+from collections.abc import Iterator, Sequence
+from dataclasses import dataclass
 from pathlib import Path
 
 PROJECT_ROOT = Path(__file__).resolve().parents[2]
@@ -33,159 +34,545 @@ SCAN_ROOTS = (
     "app/main.py",
     "app/scheduler.py",
 )
-BLOCKING_EXACT = {
-    "open",
-    "time.sleep",
-    "subprocess.call",
-    "subprocess.check_call",
-    "subprocess.check_output",
-    "subprocess.Popen",
-    "subprocess.run",
-    "os.listdir",
-    "os.scandir",
-    "os.walk",
-    "requests.delete",
-    "requests.get",
-    "requests.head",
-    "requests.patch",
-    "requests.post",
-    "requests.put",
-    "requests.request",
+
+_SYNC_HTTP_METHODS = {
+    "delete_res",
+    "get",
+    "get_json",
+    "get_res",
+    "get_stream",
+    "post",
+    "post_json",
+    "post_res",
+    "put",
+    "put_res",
+    "request",
+    "response_manager",
 }
-BLOCKING_ATTRIBUTES = {
+_REQUESTS_METHODS = {
+    "delete",
+    "get",
+    "head",
+    "patch",
+    "post",
+    "put",
+    "request",
+}
+_PATH_IO_METHODS = {
+    "exists",
     "glob",
+    "is_dir",
+    "is_file",
     "iterdir",
+    "mkdir",
     "open",
     "read_bytes",
     "read_text",
+    "rename",
     "rglob",
+    "stat",
+    "unlink",
     "write_bytes",
     "write_text",
 }
+_SHUTIL_METHODS = {
+    "copy",
+    "copyfile",
+    "copytree",
+    "make_archive",
+    "move",
+    "rmtree",
+    "unpack_archive",
+    "which",
+}
+_SUBPROCESS_METHODS = {
+    "Popen",
+    "call",
+    "check_call",
+    "check_output",
+    "run",
+}
+_OS_IO_METHODS = {"listdir", "scandir", "walk"}
+_SYSTEM_CONFIG_MEMORY_READS = {
+    "app.db.oper.SystemConfigOper.all",
+    "app.db.oper.SystemConfigOper.get",
+    "app.db.oper.systemconfig.SystemConfigOper.all",
+    "app.db.oper.systemconfig.SystemConfigOper.get",
+}
 
 
-def _call_name(node: ast.Call) -> str:
-    """把简单名称和属性调用还原为点分文本。"""
-    parts = []
-    target: ast.expr = node.func
-    while isinstance(target, ast.Attribute):
-        parts.append(target.attr)
-        target = target.value
-    if isinstance(target, ast.Name):
-        parts.append(target.id)
-    return ".".join(reversed(parts))
+@dataclass(frozen=True)
+class _Binding:
+    """描述由明确 import 或局部赋值建立的符号来源。"""
+
+    family: str
+    qualified_name: str
+    kind: str = "module"
 
 
-def _root_name(expression: ast.expr) -> str | None:
-    """返回属性调用最左侧的变量名。"""
-    while isinstance(expression, ast.Attribute):
-        expression = expression.value
-    return expression.id if isinstance(expression, ast.Name) else None
+_UNKNOWN_BINDING = _Binding("unknown", "", "unknown")
+_BLOCKING_FAMILIES = {
+    "os",
+    "requests",
+    "shutil",
+    "subprocess",
+    "sync_http",
+    "sync_oper",
+    "sync_path",
+    "time",
+}
 
 
-class _AsyncPathCollector(ast.NodeVisitor):
-    """用局部数据流识别 anyio AsyncPath 变量及其派生值。"""
+def _binding_for_qualified(qualified_name: str) -> _Binding:
+    """按稳定模块路径识别门禁关心的符号族。"""
+    if qualified_name == "app.adapters.network.http.RequestUtils":
+        return _Binding("sync_http", qualified_name, "class")
+    if qualified_name == "app.adapters.network.http.AsyncRequestUtils":
+        return _Binding("async_http", qualified_name, "class")
+    if qualified_name in {"requests.Session", "requests.sessions.Session"}:
+        return _Binding("sync_http", qualified_name, "class")
+    if qualified_name == "pathlib.Path":
+        return _Binding("sync_path", qualified_name, "class")
+    if qualified_name == "anyio.Path":
+        return _Binding("async_path", qualified_name, "class")
+    if qualified_name.startswith("app.db.oper.") and qualified_name.endswith("Oper"):
+        return _Binding("sync_oper", qualified_name, "class")
+    if qualified_name.startswith("requests."):
+        return _Binding("requests", qualified_name, "callable")
+    if qualified_name.startswith("shutil."):
+        return _Binding("shutil", qualified_name, "callable")
+    if qualified_name.startswith("subprocess."):
+        return _Binding("subprocess", qualified_name, "callable")
+    if qualified_name.startswith("os."):
+        return _Binding("os", qualified_name, "callable")
+    if qualified_name.startswith("time."):
+        return _Binding("time", qualified_name, "callable")
+    return _Binding("module", qualified_name)
 
-    def __init__(self, function: ast.AsyncFunctionDef) -> None:
-        """从参数注解初始化 AsyncPath 变量集合。"""
-        self.paths = {
-            argument.arg
-            for argument in (*function.args.posonlyargs, *function.args.args)
-            if argument.annotation and "AsyncPath" in ast.unparse(argument.annotation)
-        }
-        self.path_collections: set[str] = set()
+
+def _resolve_binding(
+    expression: ast.expr,
+    bindings: dict[str, _Binding],
+) -> _Binding | None:
+    """解析明确 import、构造、属性访问和简单路径派生。"""
+    if isinstance(expression, ast.Name):
+        return bindings.get(expression.id)
+    if isinstance(expression, ast.Attribute):
+        base = _resolve_binding(expression.value, bindings)
+        if not base or base.kind in {"collection", "unknown"}:
+            return None
+        qualified_name = f"{base.qualified_name}.{expression.attr}"
+        if base.kind == "module":
+            return _binding_for_qualified(qualified_name)
+        return _Binding(base.family, qualified_name, "callable")
+    if isinstance(expression, ast.Call):
+        target = _resolve_binding(expression.func, bindings)
+        if target and target.kind == "class":
+            return _Binding(target.family, target.qualified_name, "instance")
+        return None
+    if isinstance(expression, ast.BinOp) and isinstance(expression.op, ast.Div):
+        left = _resolve_binding(expression.left, bindings)
+        if left and left.family in {"async_path", "sync_path"}:
+            return left
+    if isinstance(expression, ast.IfExp):
+        return _merge_binding_options(
+            "",
+            (
+                _resolve_binding(expression.body, bindings),
+                _resolve_binding(expression.orelse, bindings),
+            ),
+        )
+    return None
+
+
+def _merge_binding_options(
+    name: str,
+    options: Sequence[_Binding | None],
+) -> _Binding | None:
+    """控制流合流时保留任一分支可能进入的同步阻塞类型。"""
+    if options and all(option == options[0] for option in options):
+        return options[0]
+    blocking = sorted(
+        (
+            option
+            for option in options
+            if option and option.family in _BLOCKING_FAMILIES
+        ),
+        key=lambda item: (item.family, item.qualified_name, item.kind),
+    )
+    if blocking:
+        return blocking[0]
+    if name == "open" and any(option is None for option in options):
+        return None
+    return _UNKNOWN_BINDING
+
+
+def _merge_binding_states(
+    states: Sequence[dict[str, _Binding]],
+) -> dict[str, _Binding]:
+    """合并互斥控制流的局部符号表。"""
+    names = set().union(*(state.keys() for state in states))
+    merged: dict[str, _Binding] = {}
+    for name in names:
+        binding = _merge_binding_options(name, tuple(state.get(name) for state in states))
+        if binding:
+            merged[name] = binding
+    return merged
+
+
+def _load_oper_methods(root: Path) -> dict[str, set[str]]:
+    """从 Oper 源码建立同步方法索引，避免按方法名猜测数据库调用。"""
+    methods: dict[str, set[str]] = {}
+    oper_root = root / "app/db/oper"
+    if not oper_root.is_dir():
+        return methods
+    for path in sorted(oper_root.glob("*.py")):
+        tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
+        module = ".".join(path.relative_to(root).with_suffix("").parts)
+        for node in tree.body:
+            if not isinstance(node, ast.ClassDef) or not node.name.endswith("Oper"):
+                continue
+            qualified_name = f"{module}.{node.name}"
+            sync_methods = {
+                child.name
+                for child in node.body
+                if isinstance(child, ast.FunctionDef)
+                and not child.name.startswith("__")
+            }
+            methods[qualified_name] = sync_methods
+            methods[f"app.db.oper.{node.name}"] = sync_methods
+    return methods
+
+
+class _ImportCollector(ast.NodeVisitor):
+    """收集模块级 import，作为每个 async 函数的初始符号表。"""
+
+    def __init__(self) -> None:
+        self.bindings: dict[str, _Binding] = {}
+
+    def visit_Import(self, node: ast.Import) -> None:
+        for alias in node.names:
+            local_name = alias.asname or alias.name.split(".", 1)[0]
+            qualified_name = alias.name if alias.asname else local_name
+            self.bindings[local_name] = _binding_for_qualified(qualified_name)
+
+    def visit_ImportFrom(self, node: ast.ImportFrom) -> None:
+        if node.level or not node.module:
+            return
+        for alias in node.names:
+            if alias.name == "*":
+                continue
+            local_name = alias.asname or alias.name
+            qualified_name = f"{node.module}.{alias.name}"
+            self.bindings[local_name] = _binding_for_qualified(qualified_name)
+
+    def _bind_target(self, target: ast.expr, binding: _Binding | None) -> None:
+        if isinstance(target, ast.Name):
+            self.bindings[target.id] = binding or _UNKNOWN_BINDING
+        elif isinstance(target, (ast.Tuple, ast.List)):
+            for element in target.elts:
+                self._bind_target(element, None)
 
     def visit_Assign(self, node: ast.Assign) -> None:
-        """识别 AsyncPath 构造和已知路径的 `/` 派生赋值。"""
-        is_async_path = self._is_async_path_expression(node.value)
-        if is_async_path:
-            for target in node.targets:
-                if isinstance(target, ast.Name):
-                    self.paths.add(target.id)
-        self.generic_visit(node)
-
-    def _is_async_path_expression(self, expression: ast.expr) -> bool:
-        """识别 AsyncPath 构造及任意层级的 `/` 路径派生表达式。"""
-        if isinstance(expression, ast.Call):
-            return _call_name(expression).endswith("AsyncPath")
-        if isinstance(expression, ast.Name):
-            return expression.id in self.paths
-        if isinstance(expression, ast.BinOp):
-            return self._is_async_path_expression(expression.left)
-        if isinstance(expression, ast.IfExp):
-            return (
-                self._is_async_path_expression(expression.body)
-                and self._is_async_path_expression(expression.orelse)
-            )
-        return False
+        binding = _resolve_binding(node.value, self.bindings)
+        for target in node.targets:
+            self._bind_target(target, binding)
 
     def visit_AnnAssign(self, node: ast.AnnAssign) -> None:
-        """识别 `list[AsyncPath]` 等路径集合。"""
-        if isinstance(node.target, ast.Name):
-            annotation = ast.unparse(node.annotation)
-            if "AsyncPath" in annotation:
-                if "list" in annotation or "List" in annotation:
-                    self.path_collections.add(node.target.id)
-                else:
-                    self.paths.add(node.target.id)
-        self.generic_visit(node)
-
-    def visit_AsyncFor(self, node: ast.AsyncFor) -> None:
-        """AsyncPath.iterdir 产出的元素仍是 AsyncPath。"""
-        if isinstance(node.target, ast.Name) and isinstance(node.iter, ast.Call):
-            receiver = _root_name(node.iter.func)
-            if receiver in self.paths:
-                self.paths.add(node.target.id)
-        self.generic_visit(node)
-
-    def visit_For(self, node: ast.For) -> None:
-        """从 `list[AsyncPath]` 迭代得到的元素仍是 AsyncPath。"""
-        if (
-            isinstance(node.target, ast.Name)
-            and isinstance(node.iter, ast.Name)
-            and node.iter.id in self.path_collections
-        ):
-            self.paths.add(node.target.id)
-        self.generic_visit(node)
+        binding = _resolve_binding(node.value, self.bindings) if node.value else None
+        self._bind_target(node.target, binding)
 
     def visit_FunctionDef(self, node: ast.FunctionDef) -> None:
-        """不分析嵌套同步函数。"""
+        return
 
     def visit_AsyncFunctionDef(self, node: ast.AsyncFunctionDef) -> None:
-        """不分析嵌套异步函数。"""
+        return
+
+    def visit_ClassDef(self, node: ast.ClassDef) -> None:
+        return
 
 
 class _AsyncCallVisitor(ast.NodeVisitor):
-    """只收集一个 async 函数本体中的阻塞调用，不进入嵌套函数定义。"""
+    """用局部符号传播识别一个 async 函数直接执行的阻塞调用。"""
 
-    def __init__(self, async_paths: set[str]) -> None:
-        """初始化违规计数器和异步文件对象白名单。"""
+    def __init__(
+        self,
+        function: ast.FunctionDef | ast.AsyncFunctionDef,
+        module_bindings: dict[str, _Binding],
+        oper_methods: dict[str, set[str]],
+        *,
+        record_calls: bool,
+    ) -> None:
         self.calls: Counter[str] = Counter()
-        self._async_paths = async_paths
+        self.nested_functions: list[
+            tuple[ast.FunctionDef | ast.AsyncFunctionDef, dict[str, _Binding]]
+        ] = []
+        self._bindings = dict(module_bindings)
+        self._oper_methods = oper_methods
+        self._record_calls = record_calls
+        self._bind_arguments(function)
+
+    def _bind_arguments(
+        self,
+        function: ast.FunctionDef | ast.AsyncFunctionDef,
+    ) -> None:
+        arguments = (
+            *function.args.posonlyargs,
+            *function.args.args,
+            *function.args.kwonlyargs,
+        )
+        for argument in arguments:
+            binding = self._annotation_binding(argument.annotation)
+            if binding:
+                self._bindings[argument.arg] = binding
+            else:
+                self._bindings[argument.arg] = _UNKNOWN_BINDING
+
+    def _annotation_binding(self, annotation: ast.expr | None) -> _Binding | None:
+        if annotation is None:
+            return None
+        if isinstance(annotation, ast.Subscript):
+            container = ast.unparse(annotation.value).rsplit(".", 1)[-1]
+            elements = (
+                annotation.slice.elts
+                if isinstance(annotation.slice, ast.Tuple)
+                else (annotation.slice,)
+            )
+            for element in elements:
+                binding = self._annotation_binding(element)
+                if not binding:
+                    continue
+                if container in {"list", "List", "Sequence", "set", "tuple"}:
+                    return _Binding(
+                        binding.family,
+                        binding.qualified_name,
+                        "collection",
+                    )
+                if container in {"Annotated", "Optional", "Union"}:
+                    return binding
+            return None
+        if isinstance(annotation, ast.BinOp) and isinstance(annotation.op, ast.BitOr):
+            return self._annotation_binding(annotation.left) or self._annotation_binding(
+                annotation.right
+            )
+        binding = _resolve_binding(annotation, self._bindings)
+        if binding and binding.kind == "class":
+            return _Binding(binding.family, binding.qualified_name, "instance")
+        return None
+
+    def _resolve(self, expression: ast.expr) -> _Binding | None:
+        return _resolve_binding(expression, self._bindings)
+
+    @staticmethod
+    def _call_label(binding: _Binding) -> str:
+        parts = binding.qualified_name.split(".")
+        if binding.family == "sync_oper" and len(parts) >= 2:
+            return ".".join(parts[-2:])
+        if binding.family in {"async_path", "sync_path"} and parts:
+            return f"Path.{parts[-1]}"
+        if binding.family == "sync_http" and len(parts) >= 2:
+            return ".".join(parts[-2:])
+        if binding.family in {"os", "requests", "shutil", "subprocess", "time"}:
+            return ".".join(parts[-2:])
+        return binding.qualified_name
+
+    def _record_call(self, node: ast.Call, binding: _Binding | None) -> None:
+        if not self._record_calls:
+            return
+        if not binding:
+            if isinstance(node.func, ast.Name) and node.func.id == "open":
+                self.calls["open"] += 1
+            return
+        method = binding.qualified_name.rsplit(".", 1)[-1]
+        blocked = False
+        if binding.family == "sync_http":
+            blocked = binding.kind == "callable" and method in _SYNC_HTTP_METHODS
+        elif binding.family == "requests":
+            blocked = method in _REQUESTS_METHODS
+        elif binding.family == "sync_path":
+            blocked = binding.kind == "callable" and method in _PATH_IO_METHODS
+        elif binding.family == "shutil":
+            blocked = method in _SHUTIL_METHODS
+        elif binding.family == "subprocess":
+            blocked = method in _SUBPROCESS_METHODS
+        elif binding.family == "os":
+            blocked = method in _OS_IO_METHODS
+        elif binding.family == "time":
+            blocked = method == "sleep"
+        elif binding.family == "sync_oper" and binding.kind == "callable":
+            class_name, method_name = binding.qualified_name.rsplit(".", 1)
+            blocked = (
+                method_name in self._oper_methods.get(class_name, set())
+                and binding.qualified_name not in _SYSTEM_CONFIG_MEMORY_READS
+            )
+        if blocked:
+            self.calls[self._call_label(binding)] += 1
+
+    def _bind_target(self, target: ast.expr, binding: _Binding | None) -> None:
+        if isinstance(target, ast.Name):
+            self._bindings[target.id] = binding or _UNKNOWN_BINDING
+        elif isinstance(target, (ast.Tuple, ast.List)):
+            for element in target.elts:
+                self._bind_target(element, None)
+
+    def visit_Import(self, node: ast.Import) -> None:
+        collector = _ImportCollector()
+        collector.visit_Import(node)
+        self._bindings.update(collector.bindings)
+
+    def visit_ImportFrom(self, node: ast.ImportFrom) -> None:
+        collector = _ImportCollector()
+        collector.visit_ImportFrom(node)
+        self._bindings.update(collector.bindings)
+
+    def visit_Assign(self, node: ast.Assign) -> None:
+        self.visit(node.value)
+        binding = self._resolve(node.value)
+        for target in node.targets:
+            self._bind_target(target, binding)
+
+    def visit_AnnAssign(self, node: ast.AnnAssign) -> None:
+        if node.value:
+            self.visit(node.value)
+        binding = self._resolve(node.value) if node.value else None
+        binding = binding or self._annotation_binding(node.annotation)
+        self._bind_target(node.target, binding)
+
+    def visit_AugAssign(self, node: ast.AugAssign) -> None:
+        self.visit(node.value)
+        self._bind_target(node.target, None)
+
+    def _visit_branch(
+        self,
+        statements: Sequence[ast.stmt],
+        initial: dict[str, _Binding],
+    ) -> dict[str, _Binding]:
+        saved = self._bindings
+        self._bindings = dict(initial)
+        for statement in statements:
+            self.visit(statement)
+        result = self._bindings
+        self._bindings = saved
+        return result
+
+    def visit_If(self, node: ast.If) -> None:
+        self.visit(node.test)
+        initial = dict(self._bindings)
+        body = self._visit_branch(node.body, initial)
+        alternate = self._visit_branch(node.orelse, initial) if node.orelse else initial
+        self._bindings = _merge_binding_states((body, alternate))
+
+    def visit_For(self, node: ast.For | ast.AsyncFor) -> None:
+        self.visit(node.iter)
+        initial = dict(self._bindings)
+        collection = self._resolve(node.iter)
+        element = (
+            _Binding(collection.family, collection.qualified_name, "instance")
+            if collection and collection.kind == "collection"
+            else None
+        )
+        self._bind_target(node.target, element)
+        for statement in node.body:
+            self.visit(statement)
+        iterated = dict(self._bindings)
+        completed = self._visit_branch(node.orelse, iterated)
+        self._bindings = _merge_binding_states((initial, completed))
+
+    def visit_AsyncFor(self, node: ast.AsyncFor) -> None:
+        self.visit_For(node)
+
+    def _visit_comprehension(
+        self,
+        generators: Sequence[ast.comprehension],
+        outputs: Sequence[ast.expr],
+    ) -> None:
+        saved = self._bindings
+        self._bindings = dict(saved)
+        for generator in generators:
+            self.visit(generator.iter)
+            collection = self._resolve(generator.iter)
+            element = (
+                _Binding(collection.family, collection.qualified_name, "instance")
+                if collection and collection.kind == "collection"
+                else None
+            )
+            self._bind_target(generator.target, element)
+            for condition in generator.ifs:
+                self.visit(condition)
+        for output in outputs:
+            self.visit(output)
+        self._bindings = saved
+
+    def visit_ListComp(self, node: ast.ListComp) -> None:
+        self._visit_comprehension(node.generators, (node.elt,))
+
+    def visit_SetComp(self, node: ast.SetComp) -> None:
+        self._visit_comprehension(node.generators, (node.elt,))
+
+    def visit_GeneratorExp(self, node: ast.GeneratorExp) -> None:
+        self._visit_comprehension(node.generators, (node.elt,))
+
+    def visit_DictComp(self, node: ast.DictComp) -> None:
+        self._visit_comprehension(node.generators, (node.key, node.value))
+
+    def visit_NamedExpr(self, node: ast.NamedExpr) -> None:
+        self.visit(node.value)
+        self._bind_target(node.target, self._resolve(node.value))
 
     def visit_Call(self, node: ast.Call) -> None:
-        """记录命中阻塞名单的调用并继续遍历参数表达式。"""
-        name = _call_name(node)
-        attribute = name.rsplit(".", 1)[-1]
-        receiver = _root_name(node.func)
-        async_safe = name.startswith("aiofiles.") or receiver in self._async_paths
-        if not async_safe and (
-            name in BLOCKING_EXACT or attribute in BLOCKING_ATTRIBUTES
-        ):
-            self.calls[name or attribute] += 1
+        binding = self._resolve(node.func)
+        self._record_call(node, binding)
+        if isinstance(node.func, ast.Lambda):
+            self._visit_lambda_defaults(node.func)
+            for argument in node.args:
+                self.visit(argument)
+            for keyword in node.keywords:
+                self.visit(keyword.value)
+            self.visit(node.func.body)
+            return
         self.generic_visit(node)
 
+    def _visit_definition_expressions(
+        self,
+        node: ast.FunctionDef | ast.AsyncFunctionDef,
+    ) -> None:
+        for decorator in node.decorator_list:
+            self.visit(decorator)
+        for default in node.args.defaults:
+            self.visit(default)
+        for keyword_default in node.args.kw_defaults:
+            if keyword_default:
+                self.visit(keyword_default)
+
+    def _visit_nested_function(
+        self,
+        node: ast.FunctionDef | ast.AsyncFunctionDef,
+    ) -> None:
+        self._visit_definition_expressions(node)
+        self.nested_functions.append((node, dict(self._bindings)))
+        self._bindings[node.name] = _UNKNOWN_BINDING
+
     def visit_FunctionDef(self, node: ast.FunctionDef) -> None:
-        """嵌套同步函数不属于外层 async 的直接执行体。"""
+        self._visit_nested_function(node)
 
     def visit_AsyncFunctionDef(self, node: ast.AsyncFunctionDef) -> None:
-        """嵌套异步函数由模块级收集器单独治理。"""
+        self._visit_nested_function(node)
+
+    def _visit_lambda_defaults(self, node: ast.Lambda) -> None:
+        for default in node.args.defaults:
+            self.visit(default)
+        for keyword_default in node.args.kw_defaults:
+            if keyword_default:
+                self.visit(keyword_default)
+
+    def visit_Lambda(self, node: ast.Lambda) -> None:
+        self._visit_lambda_defaults(node)
 
 
-def _async_functions(
+def _root_functions(
     tree: ast.Module,
 ) -> Iterator[tuple[str, ast.AsyncFunctionDef]]:
-    """产出模块顶层及类直接拥有的 async 函数限定名。"""
+    """产出模块顶层和类直接拥有的 async 入口。"""
     for node in tree.body:
         if isinstance(node, ast.AsyncFunctionDef):
             yield node.name, node
@@ -195,24 +582,63 @@ def _async_functions(
                     yield f"{node.name}.{method.name}", method
 
 
-def collect_async_blocking(root: Path = PROJECT_ROOT) -> dict[str, int]:
+def _scan_paths(root: Path, scan_roots: Sequence[str | Path]) -> Iterator[Path]:
+    """按稳定顺序产出存在的扫描目标。"""
+    for scan_root in scan_roots:
+        target = root / scan_root
+        if target.is_file():
+            yield target
+        elif target.is_dir():
+            yield from sorted(target.rglob("*.py"))
+
+
+def collect_async_blocking(
+    root: Path = PROJECT_ROOT,
+    scan_roots: Sequence[str | Path] = SCAN_ROOTS,
+) -> dict[str, int]:
     """扫描关键目录并以文件、函数、调用名聚合存量次数。"""
     debt: Counter[str] = Counter()
-    for scan_root in SCAN_ROOTS:
-        target = root / scan_root
-        paths = [target] if target.is_file() else sorted(target.rglob("*.py"))
-        for path in paths:
-            tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
-            relative = path.relative_to(root).as_posix()
-            for qualname, function in _async_functions(tree):
-                collector = _AsyncPathCollector(function)
-                for statement in function.body:
-                    collector.visit(statement)
-                visitor = _AsyncCallVisitor(collector.paths)
-                for statement in function.body:
-                    visitor.visit(statement)
+    oper_methods = _load_oper_methods(root)
+    for path in _scan_paths(root, scan_roots):
+        tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
+        imports = _ImportCollector()
+        for statement in tree.body:
+            imports.visit(statement)
+        relative = path.relative_to(root).as_posix()
+        pending: list[
+            tuple[
+                str,
+                ast.FunctionDef | ast.AsyncFunctionDef,
+                dict[str, _Binding],
+            ]
+        ] = [
+            (qualname, function, imports.bindings)
+            for qualname, function in _root_functions(tree)
+        ]
+        pending_index = 0
+        while pending_index < len(pending):
+            qualname, function, lexical_bindings = pending[pending_index]
+            pending_index += 1
+            is_async = isinstance(function, ast.AsyncFunctionDef)
+            visitor = _AsyncCallVisitor(
+                function,
+                lexical_bindings,
+                oper_methods,
+                record_calls=is_async,
+            )
+            for statement in function.body:
+                visitor.visit(statement)
+            if is_async:
                 for call_name, count in visitor.calls.items():
                     debt[f"{relative}:{qualname}:{call_name}"] += count
+            pending.extend(
+                (
+                    f"{qualname}.{nested.name}",
+                    nested,
+                    bindings,
+                )
+                for nested, bindings in visitor.nested_functions
+            )
     return dict(sorted(debt.items()))