341 lines
13 KiB
Python
341 lines
13 KiB
Python
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<target>[^'\"]+)['\"]"
|
|
)
|
|
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],
|
|
)
|