mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-10-31 20:34:48 +08:00
455 lines
16 KiB
Python
455 lines
16 KiB
Python
"""
|
||
枚举宿主原生并发原语并维护稳定归属事实。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import ast
|
||
import json
|
||
from dataclasses import dataclass
|
||
from pathlib import Path
|
||
from typing import Iterable, TypedDict, cast
|
||
|
||
PROJECT_ROOT = Path(__file__).resolve().parents[2]
|
||
DEFAULT_BASELINE = PROJECT_ROOT / "tests/fixtures/architecture/concurrency-baseline.json"
|
||
SCAN_ROOTS = ("app",)
|
||
# 插件副本、SDK 和兼容层由各自的兼容契约治理，不计入宿主调用债务。
|
||
EXCLUDED_ROOTS = ("app/plugins", "app/sdk", "app/runtime/compat")
|
||
|
||
# 按规范来源登记受治理的任务、线程和进程创建入口，避免按短名称误报。
|
||
_RECORDED_CALLS = frozenset({
|
||
"app.runtime.execution.run_in_threadpool",
|
||
"app.runtime.execution.run_in_threadpool_to_completion",
|
||
"asyncio.TaskGroup",
|
||
"asyncio.create_task",
|
||
"asyncio.ensure_future",
|
||
"asyncio.to_thread",
|
||
"concurrent.futures.ProcessPoolExecutor",
|
||
"concurrent.futures.ThreadPoolExecutor",
|
||
"multiprocessing.Process",
|
||
"starlette.concurrency.run_in_threadpool",
|
||
"threading.Thread",
|
||
"threading.Timer",
|
||
})
|
||
_LOOP_FACTORIES = frozenset({
|
||
"asyncio.get_event_loop",
|
||
"asyncio.get_running_loop",
|
||
"asyncio.new_event_loop",
|
||
})
|
||
_LOOP_TYPES = frozenset({"asyncio.AbstractEventLoop", "asyncio.BaseEventLoop"})
|
||
_EXECUTOR_TYPES = frozenset({
|
||
"concurrent.futures.Executor",
|
||
"concurrent.futures.ProcessPoolExecutor",
|
||
"concurrent.futures.ThreadPoolExecutor",
|
||
})
|
||
_EVENT_LOOP_KIND = "asyncio.AbstractEventLoop"
|
||
_TASK_GROUP_KIND = "asyncio.TaskGroup"
|
||
_EXECUTOR_KIND = "concurrent.futures.Executor"
|
||
# 实例方法统一映射到稳定接口路径，避免具体事件循环或执行器类型影响基线。
|
||
_RECEIVER_CALLS = {
|
||
(_EVENT_LOOP_KIND, "create_task"): "asyncio.AbstractEventLoop.create_task",
|
||
(_EVENT_LOOP_KIND, "run_in_executor"): "asyncio.AbstractEventLoop.run_in_executor",
|
||
(_TASK_GROUP_KIND, "create_task"): "asyncio.TaskGroup.create_task",
|
||
(_EXECUTOR_KIND, "map"): "concurrent.futures.Executor.map",
|
||
(_EXECUTOR_KIND, "submit"): "concurrent.futures.Executor.submit",
|
||
}
|
||
|
||
|
||
class ConcurrencyRecord(TypedDict):
|
||
"""
|
||
描述一种原生并发调用在文件内的归属计数。
|
||
|
||
:ivar target: 并发原语的规范导入路径，用于区分同名调用
|
||
:ivar owners: 所属函数或类限定名到静态调用次数的映射
|
||
"""
|
||
|
||
target: str
|
||
owners: dict[str, int]
|
||
|
||
|
||
@dataclass(frozen=True, slots=True)
|
||
class ConcurrencyFact:
|
||
"""
|
||
描述一个已确认来源的原生并发调用及其静态归属。
|
||
|
||
:param path: 相对于仓库根目录的源码路径
|
||
:param target: 已确认来源的并发原语规范导入路径
|
||
:param owner: 调用所在函数或类的限定名；模块级调用记为 <module>
|
||
"""
|
||
|
||
path: str
|
||
target: str
|
||
owner: str
|
||
|
||
@property
|
||
def key(self) -> str:
|
||
"""
|
||
返回不依赖物理行号的事实键。
|
||
"""
|
||
return f"{self.path}:{self.target}"
|
||
|
||
|
||
def _iter_python_files(root: Path, scan_roots: Iterable[str]) -> Iterable[Path]:
|
||
"""
|
||
遍历完整宿主源码，并只排除有独立治理契约的兼容表面。
|
||
"""
|
||
for scan_root in scan_roots:
|
||
path = root / scan_root
|
||
if path.is_file():
|
||
candidates = [path]
|
||
elif path.is_dir():
|
||
candidates = sorted(path.rglob("*.py"))
|
||
else:
|
||
candidates = []
|
||
for candidate in candidates:
|
||
relative = candidate.relative_to(root).as_posix()
|
||
if any(
|
||
relative == excluded or relative.startswith(f"{excluded}/")
|
||
for excluded in EXCLUDED_ROOTS
|
||
):
|
||
continue
|
||
yield candidate
|
||
|
||
|
||
class _ConcurrencyVisitor(ast.NodeVisitor):
|
||
"""
|
||
按规范来源导入、局部来源和词法归属收集并发调用。
|
||
"""
|
||
|
||
def __init__(self, relative_path: str) -> None:
|
||
"""
|
||
初始化文件位置、归属栈和逐作用域名称绑定。
|
||
"""
|
||
self._relative_path = relative_path
|
||
self._owners: list[str] = []
|
||
self._class_owners: list[str] = []
|
||
self._bindings: list[dict[str, str | None]] = [{}]
|
||
self._attribute_bindings: dict[tuple[str, str], str] = {}
|
||
self.facts: list[ConcurrencyFact] = []
|
||
|
||
@property
|
||
def _owner(self) -> str:
|
||
"""
|
||
返回最具体的可定位归属。
|
||
"""
|
||
return ".".join(self._owners) if self._owners else "<module>"
|
||
|
||
@property
|
||
def _class_owner(self) -> str:
|
||
"""
|
||
返回当前类路径，用于区分不同类的实例属性。
|
||
"""
|
||
return ".".join(self._class_owners)
|
||
|
||
def _bind_name(self, name: str, origin: str | None) -> None:
|
||
"""
|
||
在当前作用域绑定名称；None 表示显式遮蔽外层导入。
|
||
"""
|
||
self._bindings[-1][name] = origin
|
||
|
||
def _lookup_name(self, name: str) -> str | None:
|
||
"""
|
||
按 Python 词法作用域顺序解析名称来源。
|
||
"""
|
||
for scope in reversed(self._bindings):
|
||
if name in scope:
|
||
return scope[name]
|
||
return None
|
||
|
||
def _attribute_key(self, node: ast.Attribute) -> tuple[str, str]:
|
||
"""
|
||
为实例属性生成带类归属的稳定绑定键。
|
||
"""
|
||
return self._class_owner, ast.unparse(node)
|
||
|
||
def _resolve_expression(self, node: ast.AST) -> str | None:
|
||
"""
|
||
解析名称、属性或构造结果的规范来源。
|
||
"""
|
||
if isinstance(node, ast.Name):
|
||
return self._lookup_name(node.id)
|
||
if isinstance(node, ast.Attribute):
|
||
bound = self._attribute_bindings.get(self._attribute_key(node))
|
||
if bound:
|
||
return bound
|
||
base = self._resolve_expression(node.value)
|
||
return f"{base}.{node.attr}" if base else None
|
||
if isinstance(node, ast.Call):
|
||
target = self._resolve_expression(node.func)
|
||
if target in _LOOP_FACTORIES:
|
||
return _EVENT_LOOP_KIND
|
||
if target == _TASK_GROUP_KIND:
|
||
return _TASK_GROUP_KIND
|
||
if target in _EXECUTOR_TYPES:
|
||
return _EXECUTOR_KIND
|
||
return None
|
||
|
||
def _resolve_call_target(self, node: ast.Call) -> str | None:
|
||
"""
|
||
仅在调用来源可证明时返回受治理的规范目标。
|
||
"""
|
||
direct = self._resolve_expression(node.func)
|
||
if direct in _RECORDED_CALLS:
|
||
return direct
|
||
if isinstance(node.func, ast.Attribute):
|
||
receiver = self._resolve_expression(node.func.value)
|
||
return _RECEIVER_CALLS.get((receiver or "", node.func.attr))
|
||
return None
|
||
|
||
def _bind_target(self, target: ast.expr, origin: str | None) -> None:
|
||
"""
|
||
把赋值目标关联到可证明来源，并显式遮蔽未知名称。
|
||
"""
|
||
if isinstance(target, ast.Name):
|
||
self._bind_name(target.id, origin)
|
||
elif isinstance(target, ast.Attribute):
|
||
key = self._attribute_key(target)
|
||
if origin:
|
||
self._attribute_bindings[key] = origin
|
||
else:
|
||
self._attribute_bindings.pop(key, None)
|
||
elif isinstance(target, (ast.Tuple, ast.List)):
|
||
for item in target.elts:
|
||
self._bind_target(item, None)
|
||
|
||
def _bind_annotated_arguments(self, node: ast.FunctionDef | ast.AsyncFunctionDef) -> None:
|
||
"""
|
||
根据参数注解标记事件循环、TaskGroup 和 Executor 来源。
|
||
"""
|
||
arguments = [*node.args.posonlyargs, *node.args.args, *node.args.kwonlyargs]
|
||
if node.args.vararg:
|
||
arguments.append(node.args.vararg)
|
||
if node.args.kwarg:
|
||
arguments.append(node.args.kwarg)
|
||
for argument in arguments:
|
||
if argument.annotation is None:
|
||
continue
|
||
origins = {
|
||
origin
|
||
for item in ast.walk(argument.annotation)
|
||
if (origin := self._resolve_expression(item))
|
||
}
|
||
if origins & _LOOP_TYPES:
|
||
self._bind_name(argument.arg, _EVENT_LOOP_KIND)
|
||
elif _TASK_GROUP_KIND in origins:
|
||
self._bind_name(argument.arg, _TASK_GROUP_KIND)
|
||
elif origins & _EXECUTOR_TYPES:
|
||
self._bind_name(argument.arg, _EXECUTOR_KIND)
|
||
|
||
def _visit_function(self, node: ast.FunctionDef | ast.AsyncFunctionDef) -> None:
|
||
"""
|
||
进入函数归属和局部作用域后扫描全部子节点。
|
||
"""
|
||
self._bind_name(node.name, None)
|
||
self._owners.append(node.name)
|
||
self._bindings.append({})
|
||
self._bind_annotated_arguments(node)
|
||
self.generic_visit(node)
|
||
self._bindings.pop()
|
||
self._owners.pop()
|
||
|
||
def visit_Import(self, node: ast.Import) -> None:
|
||
"""
|
||
登记模块导入及其别名。
|
||
"""
|
||
for alias in node.names:
|
||
bound_name = alias.asname or alias.name.split(".", 1)[0]
|
||
origin = alias.name if alias.asname else bound_name
|
||
self._bind_name(bound_name, origin)
|
||
|
||
def visit_ImportFrom(self, node: ast.ImportFrom) -> None:
|
||
"""
|
||
登记直接导入符号及其别名。
|
||
"""
|
||
if not node.module:
|
||
return
|
||
for alias in node.names:
|
||
if alias.name == "*":
|
||
continue
|
||
self._bind_name(alias.asname or alias.name, f"{node.module}.{alias.name}")
|
||
|
||
def visit_ClassDef(self, node: ast.ClassDef) -> None:
|
||
"""
|
||
把类名纳入归属路径并隔离类体名称绑定。
|
||
"""
|
||
self._bind_name(node.name, None)
|
||
self._owners.append(node.name)
|
||
self._class_owners.append(node.name)
|
||
self._bindings.append({})
|
||
self.generic_visit(node)
|
||
self._bindings.pop()
|
||
self._class_owners.pop()
|
||
self._owners.pop()
|
||
|
||
def visit_FunctionDef(self, node: ast.FunctionDef) -> None:
|
||
"""
|
||
扫描同步函数及其嵌套归属。
|
||
"""
|
||
self._visit_function(node)
|
||
|
||
def visit_AsyncFunctionDef(self, node: ast.AsyncFunctionDef) -> None:
|
||
"""
|
||
扫描异步函数及其嵌套归属。
|
||
"""
|
||
self._visit_function(node)
|
||
|
||
def visit_Assign(self, node: ast.Assign) -> None:
|
||
"""
|
||
记录赋值右侧调用并传播可证明的运行时来源。
|
||
"""
|
||
self.visit(node.value)
|
||
origin = self._resolve_expression(node.value)
|
||
for target in node.targets:
|
||
self._bind_target(target, origin)
|
||
|
||
def visit_AnnAssign(self, node: ast.AnnAssign) -> None:
|
||
"""
|
||
处理带注解赋值并传播右侧来源。
|
||
"""
|
||
self.visit(node.annotation)
|
||
if node.value:
|
||
self.visit(node.value)
|
||
self._bind_target(node.target, self._resolve_expression(node.value) if node.value else None)
|
||
|
||
def visit_NamedExpr(self, node: ast.NamedExpr) -> None:
|
||
"""
|
||
处理海象表达式中的来源传播。
|
||
"""
|
||
self.visit(node.value)
|
||
self._bind_target(node.target, self._resolve_expression(node.value))
|
||
|
||
def _visit_with(self, node: ast.With | ast.AsyncWith) -> None:
|
||
"""
|
||
先解析上下文管理器，再让其绑定在语句体内可见。
|
||
"""
|
||
for item in node.items:
|
||
self.visit(item.context_expr)
|
||
if item.optional_vars:
|
||
self._bind_target(item.optional_vars, self._resolve_expression(item.context_expr))
|
||
for statement in node.body:
|
||
self.visit(statement)
|
||
|
||
def visit_With(self, node: ast.With) -> None:
|
||
"""
|
||
扫描同步上下文管理器及其绑定。
|
||
"""
|
||
self._visit_with(node)
|
||
|
||
def visit_AsyncWith(self, node: ast.AsyncWith) -> None:
|
||
"""
|
||
扫描异步上下文管理器及其绑定。
|
||
"""
|
||
self._visit_with(node)
|
||
|
||
def visit_Call(self, node: ast.Call) -> None:
|
||
"""
|
||
记录来源明确的原语并继续访问参数中的嵌套调用。
|
||
"""
|
||
target = self._resolve_call_target(node)
|
||
if target:
|
||
self.facts.append(
|
||
ConcurrencyFact(
|
||
path=self._relative_path,
|
||
target=target,
|
||
owner=self._owner,
|
||
)
|
||
)
|
||
self.generic_visit(node)
|
||
|
||
|
||
def collect_concurrency(
|
||
root: Path = PROJECT_ROOT,
|
||
*,
|
||
scan_roots: Iterable[str] = SCAN_ROOTS,
|
||
) -> dict[str, ConcurrencyRecord]:
|
||
"""
|
||
按文件、规范目标和归属计数收集原生并发事实。
|
||
|
||
:param root: 用于解析相对源码路径的仓库根目录
|
||
:param scan_roots: 仓库内待扫描的文件或目录路径
|
||
:return: 文件与原语组成的稳定键到调用目标及所属位置计数的映射
|
||
"""
|
||
records: dict[str, ConcurrencyRecord] = {}
|
||
for path in _iter_python_files(root, scan_roots):
|
||
relative = path.relative_to(root).as_posix()
|
||
tree = ast.parse(path.read_text(encoding="utf-8-sig"), filename=str(path))
|
||
visitor = _ConcurrencyVisitor(relative)
|
||
visitor.visit(tree)
|
||
for fact in visitor.facts:
|
||
record = records.setdefault(fact.key, {"target": fact.target, "owners": {}})
|
||
record["owners"][fact.owner] = record["owners"].get(fact.owner, 0) + 1
|
||
return dict(sorted(records.items()))
|
||
|
||
|
||
def compare_concurrency(
|
||
baseline: dict[str, ConcurrencyRecord],
|
||
current: dict[str, ConcurrencyRecord],
|
||
) -> list[str]:
|
||
"""
|
||
拒绝新增/增长，并要求事实减少时同步下调低水位基线。
|
||
"""
|
||
problems: list[str] = []
|
||
for key in sorted(baseline.keys() | current.keys()):
|
||
previous = baseline.get(key)
|
||
record = current.get(key)
|
||
if previous is None and record is not None:
|
||
problems.append(f"新增原生并发事实：{key} owners={record['owners']}")
|
||
continue
|
||
if record is None and previous is not None:
|
||
problems.append(f"并发低水位已下降，请刷新基线：{key}")
|
||
continue
|
||
if previous is None or record is None:
|
||
continue
|
||
if record.get("target") != previous.get("target"):
|
||
problems.append(f"并发 target 漂移：{key}")
|
||
previous_owners = previous.get("owners", {})
|
||
current_owners = record.get("owners", {})
|
||
for owner in sorted(previous_owners.keys() | current_owners.keys()):
|
||
old_count = previous_owners.get(owner, 0)
|
||
new_count = current_owners.get(owner, 0)
|
||
if not owner:
|
||
problems.append(f"并发事实缺少 owner：{key}")
|
||
elif new_count > old_count:
|
||
problems.append(
|
||
f"并发调用新增或增长：{key} owner={owner} {new_count}>{old_count}"
|
||
)
|
||
elif new_count < old_count:
|
||
problems.append(
|
||
f"并发低水位已下降，请刷新基线：{key} owner={owner} {new_count}<{old_count}"
|
||
)
|
||
return problems
|
||
|
||
|
||
def main() -> int:
|
||
"""
|
||
执行原生并发事实只降不增检查。
|
||
"""
|
||
parser = argparse.ArgumentParser(description=__doc__)
|
||
parser.add_argument("--write", action="store_true", help="写入当前并发事实基线")
|
||
parser.add_argument("--baseline", type=Path, default=DEFAULT_BASELINE)
|
||
args = parser.parse_args()
|
||
current = collect_concurrency()
|
||
if args.write:
|
||
args.baseline.parent.mkdir(parents=True, exist_ok=True)
|
||
args.baseline.write_text(
|
||
json.dumps(current, ensure_ascii=False, indent=2, sort_keys=True) + "\n",
|
||
encoding="utf-8",
|
||
)
|
||
print(f"已写入 {args.baseline.relative_to(PROJECT_ROOT)}")
|
||
return 0
|
||
baseline = cast(
|
||
dict[str, ConcurrencyRecord],
|
||
json.loads(args.baseline.read_text(encoding="utf-8")),
|
||
)
|
||
problems = compare_concurrency(baseline, current)
|
||
if problems:
|
||
print("\n".join(problems))
|
||
return 1
|
||
print("原生并发事实 ratchet 通过")
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|