"""Extension marketplace — Arch Mapper : cartographie d’architecture code. Outils : - ``arch_import_graph`` — graphe d’imports Python + cycles ; - ``arch_hotspots`` — hotspots complexité (AST) ; - ``arch_module_map`` — carte modules / couches / fan-in fan-out. """ from __future__ import annotations import ast import json from collections import defaultdict from pathlib import Path _SKIP = {".git", "node_modules", "__pycache__", ".venv", "venv", "dist", "build", ".tox", ".mypy_cache"} def register(workspace: Path) -> list: """Enregistre les outils Arch Mapper.""" def _root(where: str) -> Path: p = Path(where).expanduser() root = p.resolve() if p.is_absolute() else (workspace / where).resolve() if not root.is_dir(): raise FileNotFoundError(f"répertoire introuvable: {root}") return root def _py_files(root: Path, max_files: int = 400) -> list[Path]: files: list[Path] = [] for path in root.rglob("*.py"): if any(part in _SKIP for part in path.parts): continue files.append(path) if len(files) >= max_files: break return files def _mod_name(path: Path, root: Path) -> str: try: rel = path.relative_to(root).with_suffix("") except ValueError: rel = path.with_suffix("") parts = list(rel.parts) if parts and parts[-1] == "__init__": parts = parts[:-1] return ".".join(parts) or path.stem def _complexity(tree: ast.AST) -> dict: fns = 0 classes = 0 branches = 0 max_depth = 0 class V(ast.NodeVisitor): def __init__(self) -> None: self.depth = 0 def generic_visit(self, node: ast.AST) -> None: nonlocal max_depth, branches, fns, classes self.depth += 1 max_depth = max(max_depth, self.depth) if isinstance(node, (ast.If, ast.For, ast.While, ast.Try, ast.With, ast.BoolOp, ast.Match)): branches += 1 if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)): fns += 1 if isinstance(node, ast.ClassDef): classes += 1 super().generic_visit(node) self.depth -= 1 V().visit(tree) return {"functions": fns, "classes": classes, "branches": branches, "ast_depth": max_depth} def arch_import_graph(where: str = ".", max_files: int = 300) -> str: """Construit le graphe d’imports Python et détecte les cycles simples. Args: where: Racine à cartographier. max_files: Plafond de fichiers Python. """ try: root = _root(where) except Exception as exc: # noqa: BLE001 return json.dumps({"error": str(exc)}, ensure_ascii=False) edges: list[dict] = [] nodes: set[str] = set() adj: dict[str, set[str]] = defaultdict(set) for path in _py_files(root, max_files): src = _mod_name(path, root) nodes.add(src) try: tree = ast.parse(path.read_text(encoding="utf-8", errors="replace")) except SyntaxError: continue for node in ast.walk(tree): if isinstance(node, ast.Import): for alias in node.names: dst = alias.name.split(".")[0] edges.append({"from": src, "to": alias.name, "kind": "import"}) adj[src].add(alias.name) nodes.add(alias.name) elif isinstance(node, ast.ImportFrom) and node.module: edges.append({"from": src, "to": node.module, "kind": "from"}) adj[src].add(node.module) nodes.add(node.module) # Cycles via DFS sur sous-graphe interne (préfixe commun). internal = {n for n in nodes if n.split(".")[0] in {m.split(".")[0] for m in nodes}} cycles: list[list[str]] = [] visiting: set[str] = set() visited: set[str] = set() stack: list[str] = [] def dfs(u: str) -> None: visiting.add(u) stack.append(u) for v in adj.get(u, ()): if v not in internal: continue if v in visiting: if v in stack: i = stack.index(v) cyc = stack[i:] + [v] if len(cyc) <= 8: cycles.append(cyc) continue if v not in visited: dfs(v) stack.pop() visiting.discard(u) visited.add(u) for n in sorted(internal): if n not in visited: dfs(n) return json.dumps( { "root": str(root), "nodes": len(nodes), "edges": len(edges), "edge_sample": edges[:80], "cycles_found": cycles[:20], "hint": "Les cycles internes signalent souvent un couplage à casser (ports/adapters, façade).", }, ensure_ascii=False, indent=2, ) def arch_hotspots(where: str = ".", limit: int = 25) -> str: """Classe les fichiers Python par score de hotspot (taille + branches AST). Args: where: Racine. limit: Top N hotspots. """ try: root = _root(where) except Exception as exc: # noqa: BLE001 return json.dumps({"error": str(exc)}, ensure_ascii=False) rows: list[dict] = [] for path in _py_files(root): try: text = path.read_text(encoding="utf-8", errors="replace") tree = ast.parse(text) except (OSError, SyntaxError): continue meta = _complexity(tree) loc = text.count("\n") + 1 score = loc * 0.15 + meta["branches"] * 2.2 + meta["functions"] * 1.1 + meta["ast_depth"] * 0.8 rel = str(path.relative_to(workspace)) if path.is_relative_to(workspace) else str(path) rows.append({"path": rel, "loc": loc, "score": round(score, 2), **meta}) rows.sort(key=lambda r: r["score"], reverse=True) return json.dumps( { "root": str(root), "hotspots": rows[: max(1, limit)], "advice": "Découper les hotspots (fonctions >50L, modules god-object) avant d’ajouter des features.", }, ensure_ascii=False, indent=2, ) def arch_module_map(where: str = ".", max_files: int = 300) -> str: """Carte modules : fan-in / fan-out et hypothèse de couches. Args: where: Racine. max_files: Plafond fichiers. """ try: root = _root(where) except Exception as exc: # noqa: BLE001 return json.dumps({"error": str(exc)}, ensure_ascii=False) fan_out: dict[str, set[str]] = defaultdict(set) fan_in: dict[str, set[str]] = defaultdict(set) layers_guess = { "cli": [], "agent": [], "domain": [], "infra": [], "other": [], } for path in _py_files(root, max_files): src = _mod_name(path, root) bucket = "other" low = src.lower() if any(k in low for k in ("cli", "cmd", "main", "ui")): bucket = "cli" elif any(k in low for k in ("agent", "llm", "tool")): bucket = "agent" elif any(k in low for k in ("domain", "model", "core", "service")): bucket = "domain" elif any(k in low for k in ("db", "http", "fs", "filesystem", "io", "plugin", "adapter")): bucket = "infra" layers_guess[bucket].append(src) try: tree = ast.parse(path.read_text(encoding="utf-8", errors="replace")) except SyntaxError: continue for node in ast.walk(tree): targets: list[str] = [] if isinstance(node, ast.Import): targets = [a.name for a in node.names] elif isinstance(node, ast.ImportFrom) and node.module: targets = [node.module] for t in targets: fan_out[src].add(t) fan_in[t].add(src) ranking = [] for mod, outs in fan_out.items(): ranking.append( { "module": mod, "fan_out": len(outs), "fan_in": len(fan_in.get(mod, ())), "coupling": len(outs) + len(fan_in.get(mod, ())), } ) ranking.sort(key=lambda r: r["coupling"], reverse=True) return json.dumps( { "root": str(root), "layers_guess": {k: v[:40] for k, v in layers_guess.items()}, "most_coupled": ranking[:30], "notes": [ "fan_out élevé = module trop dépendant", "fan_in élevé = point central (OK s’il est une façade stable)", "Les couches sont heuristiques — à valider avec le domaine métier", ], }, ensure_ascii=False, indent=2, ) return [arch_import_graph, arch_hotspots, arch_module_map]