MoviePilot/scripts/architecture/complexity.py

"""为 API/Application/Chain 公共入口维护只降不增的行数预算。"""

from __future__ import annotations

import argparse
import ast
import json
from dataclasses import dataclass
from pathlib import Path
from typing import Iterable

PROJECT_ROOT = Path(__file__).resolve().parents[2]
DEFAULT_BASELINE = PROJECT_ROOT / "tests/fixtures/architecture/complexity-baseline.json"
V2_BASELINE = PROJECT_ROOT / "tests/fixtures/architecture/complexity-v2-baseline.json"


@dataclass(frozen=True, slots=True)
class ComplexityRule:
    """描述目录、入口选择器与最大源代码行数。"""

    name: str
    root: str
    budget: int
    endpoint_only: bool = False


RULES = (
    ComplexityRule("api_endpoint", "app/api/endpoints", 80, endpoint_only=True),
    ComplexityRule("application_public", "app/application", 150),
    ComplexityRule("chain_public", "app/chain", 150),
)


def _iter_python_files(root: Path, roots: tuple[str, ...]) -> Iterable[Path]:
    """遍历 canonical 宿主源码，明确排除运行时插件副本。"""
    for scan_root in roots:
        path = root / scan_root
        if path.is_file() and path.suffix == ".py":
            yield path
            continue
        if not path.is_dir():
            continue
        for candidate in sorted(path.rglob("*.py")):
            relative = candidate.relative_to(root).as_posix()
            if relative == "app/plugins" or relative.startswith("app/plugins/"):
                continue
            yield candidate


def _walk_owned_nodes(
    nodes: Iterable[ast.AST],
    prefix: str = "",
) -> Iterable[tuple[str, ast.AST]]:
    """通过完整 AST 子节点遍历收集类、方法和任意控制流中的嵌套 owner。"""
    for node in nodes:
        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)):
            qualname = f"{prefix}.{node.name}" if prefix else node.name
            yield qualname, node
            yield from _walk_owned_nodes(ast.iter_child_nodes(node), qualname)
        else:
            yield from _walk_owned_nodes(ast.iter_child_nodes(node), prefix)


def _walk_functions(
    nodes: Iterable[ast.AST],
) -> Iterable[tuple[str, ast.FunctionDef | ast.AsyncFunctionDef]]:
    """收集含 dunder、Match、TryStar 在内的全部函数与方法 owner。"""
    for qualname, node in _walk_owned_nodes(nodes):
        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
            yield qualname, node


def _walk_classes(
    nodes: Iterable[ast.AST],
) -> Iterable[tuple[str, ast.ClassDef]]:
    """收集任意控制流和函数内部的类，并保留完整嵌套 owner。"""
    for qualname, node in _walk_owned_nodes(nodes):
        if isinstance(node, ast.ClassDef):
            yield qualname, node


def collect_complexity_v2(root: Path = PROJECT_ROOT) -> dict[str, dict[str, int]]:
    """收集私有方法、类、文件和 Scheduler 的可审计复杂度事实。"""
    roots = ("app/api/endpoints", "app/application", "app/chain", "app/scheduler")
    report: dict[str, dict[str, int]] = {"method": {}, "class": {}, "file": {}}
    for path in _iter_python_files(root, roots):
        relative = path.relative_to(root).as_posix()
        source_lines = path.read_text(encoding="utf-8").splitlines()
        if len(source_lines) > 1000:
            report["file"][relative] = len(source_lines)
        tree = ast.parse("\n".join(source_lines), filename=str(path))
        for qualname, method_node in _walk_functions(tree.body):
            line_count = (method_node.end_lineno or method_node.lineno) - method_node.lineno + 1
            if line_count > 150:
                report["method"][f"{relative}:{qualname}"] = line_count
        for qualname, class_node in _walk_classes(tree.body):
            line_count = (class_node.end_lineno or class_node.lineno) - class_node.lineno + 1
            if line_count > 500:
                report["class"][f"{relative}:{qualname}"] = line_count
    return report


def _is_endpoint(node: ast.FunctionDef | ast.AsyncFunctionDef) -> bool:
    """识别带常见 HTTP method decorator 的 API endpoint。"""
    methods = {"get", "post", "put", "patch", "delete", "options", "head"}
    for decorator in node.decorator_list:
        target = decorator.func if isinstance(decorator, ast.Call) else decorator
        if isinstance(target, ast.Attribute) and target.attr.lower() in methods:
            return True
    return False


def _public_entries(
    tree: ast.Module, endpoint_only: bool
) -> Iterable[tuple[str, ast.FunctionDef | ast.AsyncFunctionDef]]:
    """产出顶层函数和类直接拥有的公共方法，不把嵌套 helper 重复计数。"""
    for node in tree.body:
        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
            if not node.name.startswith("_") and (
                not endpoint_only or _is_endpoint(node)
            ):
                yield node.name, node
        elif isinstance(node, ast.ClassDef):
            for method in node.body:
                if isinstance(method, (ast.FunctionDef, ast.AsyncFunctionDef)):
                    if not method.name.startswith("_") and (
                        not endpoint_only or _is_endpoint(method)
                    ):
                        yield f"{node.name}.{method.name}", method


def collect_complexity(root: Path = PROJECT_ROOT) -> dict[str, dict[str, int]]:
    """收集每条规则下当前超过预算的入口及其精确源代码行数。"""
    report: dict[str, dict[str, int]] = {}
    for rule in RULES:
        debt: dict[str, int] = {}
        for path in sorted((root / rule.root).rglob("*.py")):
            tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
            relative = path.relative_to(root).as_posix()
            for qualname, node in _public_entries(tree, rule.endpoint_only):
                line_count = (node.end_lineno or node.lineno) - node.lineno + 1
                if line_count > rule.budget:
                    debt[f"{relative}:{qualname}"] = line_count
        report[rule.name] = debt
    return report


def compare_complexity(
    baseline: dict[str, dict[str, int]], current: dict[str, dict[str, int]]
) -> list[str]:
    """返回新增超限或既有超限增长问题；删除和缩短均合法。"""
    problems = []
    for rule in RULES:
        previous = baseline.get(rule.name, {})
        for entry, line_count in current.get(rule.name, {}).items():
            if entry not in previous:
                problems.append(f"{rule.name}: 新增超限 {entry}={line_count}>{rule.budget}")
            elif line_count > previous[entry]:
                problems.append(
                    f"{rule.name}: 既有超限增长 {entry}={line_count}>{previous[entry]}"
                )
    return problems


def compare_complexity_v2(
    baseline: dict[str, dict[str, int]], current: dict[str, dict[str, int]]
) -> list[str]:
    """返回 v2 私有方法、类和文件事实的新增或增长问题。"""
    problems: list[str] = []
    for category in baseline.keys() | current.keys():
        previous = baseline.get(category, {})
        for entry, line_count in current.get(category, {}).items():
            if entry not in previous:
                problems.append(f"{category}: 新增超限 {entry}={line_count}")
            elif line_count > previous[entry]:
                problems.append(
                    f"{category}: 既有超限增长 {entry}={line_count}>{previous[entry]}"
                )
    return problems


def main() -> int:
    """执行复杂度 baseline check 或显式 write。"""
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--write", action="store_true", help="写入当前超限基线")
    parser.add_argument("--v2", action="store_true", help="检查私有方法、类、文件和 Scheduler")
    parser.add_argument("--baseline", type=Path, default=DEFAULT_BASELINE)
    args = parser.parse_args()
    if args.v2:
        baseline_path = args.baseline if args.baseline != DEFAULT_BASELINE else V2_BASELINE
        current_v2 = collect_complexity_v2()
        if args.write:
            baseline_path.parent.mkdir(parents=True, exist_ok=True)
            baseline_path.write_text(
                json.dumps(current_v2, ensure_ascii=False, indent=2, sort_keys=True) + "\n",
                encoding="utf-8",
            )
            print(f"已写入 {baseline_path.relative_to(PROJECT_ROOT)}")
            return 0
        baseline_v2 = json.loads(baseline_path.read_text(encoding="utf-8"))
        problems_v2 = compare_complexity_v2(baseline_v2, current_v2)
        if problems_v2:
            print("\n".join(problems_v2))
            return 1
        print("复杂度 v2 ratchet 通过")
        return 0
    current = collect_complexity()
    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 = json.loads(args.baseline.read_text(encoding="utf-8"))
    problems = compare_complexity(baseline, current)
    if problems:
        print("\n".join(problems))
        return 1
    print("复杂度 ratchet 通过")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())