from __future__ import annotations import ast import json import math import os import re from collections import Counter from pathlib import Path from .models import ArchitectureEdge, ArchitectureGraph, ArchitectureNode, Hotspot JS_IMPORT_RE = re.compile( r"(?:\bfrom\s+|\bimport\s*|\brequire\(\s*|\bimport\(\s*)['\"](?P[^'\"]+)['\"]" ) JS_EXTENSIONS = (".ts", ".tsx", ".js", ".jsx", ".mjs", ".cjs") SOURCE_ROOTS = {"src", "app", "backend", "frontend", "lib", "packages"} CONFIG_IGNORED_DIRS = {".git", ".venv", "node_modules", "dist", "build", "coverage", "target", "vendor"} def _bounded_config_files(root: Path, names: set[str], *, maximum: int = 20) -> list[Path]: files: list[Path] = [] for current, directories, candidates in os.walk(root, followlinks=False): directories[:] = [name for name in directories if name.casefold() not in CONFIG_IGNORED_DIRS] for name in candidates: if name in names: files.append(Path(current) / name) if len(files) >= maximum: return sorted(files) return sorted(files) def _ts_path_aliases(root: Path) -> list[tuple[str, str]]: """Read a small local subset of tsconfig paths without resolving packages or URLs.""" aliases: list[tuple[str, str]] = [] root_resolved = root.resolve() for config in _bounded_config_files(root, {"tsconfig.json", "jsconfig.json"}): try: if config.stat().st_size > 256_000: continue raw = config.read_text(encoding="utf-8") # Most project tsconfigs are JSON. Comments are deliberately not # interpreted as configuration to keep this scanner deterministic. data = json.loads(raw) options = data.get("compilerOptions", {}) paths = options.get("paths", {}) if isinstance(options, dict) else {} if not isinstance(paths, dict): continue base_url = options.get("baseUrl", ".") if isinstance(options, dict) else "." base = (config.parent / str(base_url)).resolve() if base != root_resolved and root_resolved not in base.parents: continue for alias, targets in paths.items(): if not isinstance(alias, str) or not isinstance(targets, list): continue for target in targets[:5]: if not isinstance(target, str) or target.startswith(("/", "\\")): continue resolved = (base / target.replace("*", "placeholder")).resolve() if resolved != root_resolved and root_resolved not in resolved.parents: continue aliases.append((alias, str((base / target).relative_to(root_resolved)).replace("\\", "/"))) except (OSError, ValueError, json.JSONDecodeError, TypeError): continue return aliases[:100] def _workspace_packages(root: Path) -> dict[str, Path]: packages: dict[str, Path] = {} for manifest in _bounded_config_files(root, {"package.json"}, maximum=100): try: if manifest.stat().st_size > 256_000: continue data = json.loads(manifest.read_text(encoding="utf-8")) name = data.get("name") if isinstance(data, dict) else None if isinstance(name, str) and name and not name.startswith((".", "/")): packages.setdefault(name, manifest.parent) except (OSError, ValueError, json.JSONDecodeError, TypeError): continue return packages def _module_names(path: str) -> list[str]: parts = list(Path(path).with_suffix("").parts) if parts and parts[-1] == "__init__": parts = parts[:-1] names: list[str] = [] if parts: names.append(".".join(parts)) if parts[0] in SOURCE_ROOTS and len(parts) > 1: names.append(".".join(parts[1:])) if len(parts) > 2 and parts[1] in SOURCE_ROOTS: names.append(".".join(parts[2:])) return [name for name in names if name] def _python_module_map(paths: list[str]) -> dict[str, str]: result: dict[str, str] = {} for path in paths: if not path.endswith(".py"): continue for module in _module_names(path): result.setdefault(module, path) return result def _resolve_python_target(module: str, module_map: dict[str, str]) -> str | None: candidate = module while candidate: if candidate in module_map: return module_map[candidate] if "." not in candidate: break candidate = candidate.rsplit(".", 1)[0] return None def _python_imports(path: str, text: str, module_map: dict[str, str]) -> set[str]: try: tree = ast.parse(text) except SyntaxError: return set() current_names = _module_names(path) current = current_names[0] if current_names else "" current_parts = current.split(".") if current else [] if not path.endswith("/__init__.py") and current_parts: package_parts = current_parts[:-1] else: package_parts = current_parts imports: set[str] = set() for node in ast.walk(tree): if isinstance(node, ast.Import): for alias in node.names: target = _resolve_python_target(alias.name, module_map) if target and target != path: imports.add(target) elif isinstance(node, ast.ImportFrom): module = node.module or "" if node.level: trim = max(0, node.level - 1) prefix = package_parts[:-trim] if trim else package_parts absolute = ".".join([*prefix, *([module] if module else [])]) else: absolute = module target = _resolve_python_target(absolute, module_map) if target and target != path: imports.add(target) return imports def _resolve_js_target( root: Path, source_path: str, specifier: str, known: set[str], aliases: list[tuple[str, str]], workspace_packages: dict[str, Path], ) -> str | None: if specifier.startswith("."): source_dir = (root / source_path).parent base = (source_dir / specifier).resolve() else: base = None for alias, target in aliases: if alias.endswith("/*") and specifier.startswith(alias[:-1]): base = (root / target.replace("*", specifier[len(alias) - 1 :])).resolve() break if alias == specifier: base = (root / target).resolve() break if base is None: package_name = next( (name for name in workspace_packages if specifier == name or specifier.startswith(f"{name}/")), None, ) if package_name: package_root = workspace_packages[package_name] suffix = specifier.removeprefix(package_name).lstrip("/") base = package_root / suffix if suffix else package_root / "src" / "index" if base is None: return None try: relative_base = base.relative_to(root.resolve()).as_posix() except ValueError: return None candidates = [relative_base] suffix = Path(relative_base).suffix.lower() if suffix in JS_EXTENSIONS: extensionless = relative_base[: -len(suffix)] candidates.extend(extensionless + extension for extension in JS_EXTENSIONS) candidates.extend(f"{extensionless}/index{extension}" for extension in JS_EXTENSIONS) elif not suffix: candidates.extend(relative_base + extension for extension in JS_EXTENSIONS) candidates.extend(f"{relative_base}/index{extension}" for extension in JS_EXTENSIONS) return next((candidate for candidate in candidates if candidate in known), None) def _js_imports( root: Path, path: str, text: str, known: set[str], aliases: list[tuple[str, str]], workspace_packages: dict[str, Path], ) -> set[str]: imports: set[str] = set() for match in JS_IMPORT_RE.finditer(text): target = _resolve_js_target(root, path, match.group("target"), known, aliases, workspace_packages) if target and target != path: imports.add(target) return imports def build_architecture( root: Path, source_texts: dict[str, str], language_by_path: dict[str, str], loc_by_path: dict[str, int], debt_by_path: dict[str, int], churn_by_path: dict[str, int], *, max_nodes: int = 160, max_edges: int = 600, ) -> tuple[ArchitectureGraph, list[Hotspot]]: max_nodes = max(1, max_nodes) max_edges = max(0, max_edges) known_paths = set(source_texts) module_map = _python_module_map(list(known_paths)) ts_aliases = _ts_path_aliases(root) workspace_packages = _workspace_packages(root) edge_pairs: set[tuple[str, str]] = set() total_inbound: Counter[str] = Counter() total_outbound: Counter[str] = Counter() total_edges = 0 truncated = len(known_paths) > max_nodes for path in sorted(source_texts): text = source_texts[path] language = language_by_path.get(path, "") if language == "Python": targets = _python_imports(path, text, module_map) elif language in {"TypeScript", "JavaScript"}: targets = _js_imports(root, path, text, known_paths, ts_aliases, workspace_packages) else: targets = set() for target in sorted(targets): if target not in known_paths: continue pair = (path, target) total_edges += 1 total_outbound[path] += 1 total_inbound[target] += 1 if len(edge_pairs) < max_edges: edge_pairs.add(pair) else: truncated = True hotspots: list[Hotspot] = [] for path in known_paths: loc = loc_by_path.get(path, 0) debt = debt_by_path.get(path, 0) churn = churn_by_path.get(path, 0) coupling = total_inbound[path] + total_outbound[path] score = min( 100, round( min(30, loc / 40) + min(20, debt * 4) + min(30, math.log10(churn + 1) * 10 if churn else 0) + min(20, coupling * 2.5) ), ) reasons: list[str] = [] if loc >= 500: reasons.append(f"large module ({loc} LOC)") if debt: reasons.append(f"{debt} debt markers") if churn >= 300: reasons.append(f"high recent churn ({churn} changed lines)") if coupling >= 8: reasons.append(f"high coupling ({coupling} local edges)") if not reasons and score >= 30: reasons.append("combined size, churn and coupling") hotspots.append( Hotspot( path=path, language=language_by_path.get(path, "Other"), lines_of_code=loc, debt_markers=debt, churn=churn, inbound=total_inbound[path], outbound=total_outbound[path], score=score, reasons=reasons, ) ) hotspots.sort(key=lambda item: (-item.score, -item.churn, -item.lines_of_code, item.path)) selected_paths = {item.path for item in hotspots[:max_nodes]} graph_edges = [ ArchitectureEdge(source=source, target=target) for source, target in sorted(edge_pairs) if source in selected_paths and target in selected_paths ][:max_edges] visible_inbound: Counter[str] = Counter() visible_outbound: Counter[str] = Counter() for edge in graph_edges: visible_outbound[edge.source] += 1 visible_inbound[edge.target] += 1 hotspot_scores = {item.path: item.score for item in hotspots} nodes = [ ArchitectureNode( id=path, label=Path(path).name, path=path, language=language_by_path.get(path, "Other"), inbound=total_inbound[path], outbound=total_outbound[path], visible_inbound=visible_inbound[path], visible_outbound=visible_outbound[path], hotspot_score=hotspot_scores[path], ) for path in sorted(selected_paths) ] clusters = Counter((Path(path).parts[0] if len(Path(path).parts) > 1 else "root") for path in selected_paths) return ( ArchitectureGraph( nodes=nodes, edges=graph_edges, clusters=dict(clusters.most_common()), truncated=truncated, total_nodes=len(known_paths), total_edges=total_edges, ), hotspots[:30], )