Monorepo Navigator

#!/usr/bin/env python3
"""
Dependency Graph Generator - Generate Mermaid diagrams of monorepo package dependencies.

Scans a monorepo for internal cross-package dependencies and outputs a Mermaid
flowchart showing the dependency structure. Supports filtering by package,
direction control, depth limits, and multiple layout orientations.

Author: Claude Skills Engineering Team
License: MIT
"""

import argparse
import glob
import json
import os
import sys
from collections import deque
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Dict, List, Optional, Set, Tuple


@dataclass
class PackageNode:
    """Represents a package in the dependency graph."""
    name: str
    path: str
    category: str  # apps, packages, libs, etc.
    internal_deps: List[str] = field(default_factory=list)
    dep_count: int = 0
    dependent_count: int = 0


@dataclass
class GraphResult:
    """Result containing the generated graph data."""
    mermaid: str
    node_count: int
    edge_count: int
    packages: List[Dict[str, Any]]
    categories: Dict[str, int]
    circular_deps: List[List[str]]
    warnings: List[str]


class DependencyGraphBuilder:
    """Builds a dependency graph from monorepo package structure."""

    def __init__(self, root_path: str):
        self.root_path = Path(root_path).resolve()
        self.nodes: Dict[str, PackageNode] = {}
        self.edges: List[Tuple[str, str]] = []
        self.reverse_deps: Dict[str, Set[str]] = {}
        self.warnings: List[str] = []

    def build(self) -> None:
        """Discover packages and build the graph."""
        workspace_globs = self._detect_workspaces()
        if not workspace_globs:
            workspace_globs = ["packages/*", "apps/*", "libs/*", "modules/*"]
            self.warnings.append("No workspace config found; using default patterns.")

        self._discover_packages(workspace_globs)
        self._resolve_edges()

    def _detect_workspaces(self) -> List[str]:
        """Detect workspace globs from config files."""
        pnpm_ws = self.root_path / "pnpm-workspace.yaml"
        if pnpm_ws.exists():
            return self._parse_pnpm_workspace(pnpm_ws)

        root_pkg = self.root_path / "package.json"
        if root_pkg.exists():
            try:
                with open(root_pkg, "r") as f:
                    data = json.load(f)
                ws = data.get("workspaces")
                if isinstance(ws, list):
                    return ws
                if isinstance(ws, dict) and "packages" in ws:
                    return ws["packages"]
            except (json.JSONDecodeError, OSError):
                pass

        lerna_cfg = self.root_path / "lerna.json"
        if lerna_cfg.exists():
            try:
                with open(lerna_cfg, "r") as f:
                    return json.load(f).get("packages", ["packages/*"])
            except (json.JSONDecodeError, OSError):
                pass

        return []

    def _parse_pnpm_workspace(self, filepath: Path) -> List[str]:
        """Parse pnpm-workspace.yaml without PyYAML."""
        globs = []
        try:
            with open(filepath, "r") as f:
                in_packages = False
                for line in f:
                    stripped = line.strip()
                    if stripped == "packages:":
                        in_packages = True
                        continue
                    if in_packages:
                        if stripped.startswith("- "):
                            globs.append(stripped[2:].strip().strip("'\""))
                        elif stripped and not stripped.startswith("#"):
                            break
        except OSError:
            pass
        return globs

    def _discover_packages(self, workspace_globs: List[str]) -> None:
        """Find all packages matching workspace globs."""
        for ws_glob in workspace_globs:
            if ws_glob.startswith("!"):
                continue
            pattern = str(self.root_path / ws_glob / "package.json")
            for pkg_path in sorted(glob.glob(pattern)):
                pkg_json = Path(pkg_path)
                try:
                    with open(pkg_json, "r") as f:
                        data = json.load(f)
                except (json.JSONDecodeError, OSError):
                    continue
                name = data.get("name")
                if not name:
                    continue
                rel_path = str(pkg_json.parent.relative_to(self.root_path))
                category = rel_path.split("/")[0] if "/" in rel_path else "root"
                self.nodes[name] = PackageNode(
                    name=name,
                    path=rel_path,
                    category=category,
                )

    def _resolve_edges(self) -> None:
        """Resolve internal dependency edges."""
        internal_names = set(self.nodes.keys())
        for name, node in self.nodes.items():
            pkg_json = self.root_path / node.path / "package.json"
            try:
                with open(pkg_json, "r") as f:
                    data = json.load(f)
            except (json.JSONDecodeError, OSError):
                continue

            dep_sections = [
                data.get("dependencies", {}),
                data.get("devDependencies", {}),
                data.get("peerDependencies", {}),
            ]
            deps = set()
            for section in dep_sections:
                for dep_name in section:
                    if dep_name in internal_names and dep_name != name:
                        deps.add(dep_name)

            node.internal_deps = sorted(deps)
            node.dep_count = len(deps)
            for dep in deps:
                self.edges.append((name, dep))
                self.reverse_deps.setdefault(dep, set()).add(name)

        # Count dependents
        for name in self.nodes:
            self.nodes[name].dependent_count = len(self.reverse_deps.get(name, set()))

    def detect_circular_deps(self) -> List[List[str]]:
        """Detect circular dependencies using DFS."""
        cycles: List[List[str]] = []
        visited: Set[str] = set()
        rec_stack: Set[str] = set()
        path: List[str] = []

        def dfs(node: str) -> None:
            visited.add(node)
            rec_stack.add(node)
            path.append(node)

            for dep in self.nodes.get(node, PackageNode("", "", "")).internal_deps:
                if dep not in visited:
                    dfs(dep)
                elif dep in rec_stack:
                    cycle_start = path.index(dep)
                    cycle = path[cycle_start:] + [dep]
                    normalized = self._normalize_cycle(cycle)
                    if normalized not in [self._normalize_cycle(c) for c in cycles]:
                        cycles.append(cycle)

            path.pop()
            rec_stack.discard(node)

        for name in sorted(self.nodes.keys()):
            if name not in visited:
                dfs(name)

        return cycles

    @staticmethod
    def _normalize_cycle(cycle: List[str]) -> Tuple[str, ...]:
        """Normalize a cycle for deduplication."""
        if len(cycle) <= 1:
            return tuple(cycle)
        min_idx = cycle.index(min(cycle[:-1]))
        normalized = cycle[min_idx:-1] + cycle[:min_idx]
        return tuple(normalized)

    def filter_by_package(
        self, focus: str, depth: int = -1, direction: str = "both"
    ) -> "DependencyGraphBuilder":
        """Create a filtered graph centered on a specific package."""
        if focus not in self.nodes:
            self.warnings.append(f"Package '{focus}' not found in graph.")
            return self

        relevant: Set[str] = {focus}

        if direction in ("deps", "both"):
            self._collect_deps(focus, relevant, depth)
        if direction in ("dependents", "both"):
            self._collect_dependents(focus, relevant, depth)

        # Filter nodes and edges
        filtered = DependencyGraphBuilder(str(self.root_path))
        filtered.nodes = {k: v for k, v in self.nodes.items() if k in relevant}
        filtered.edges = [(a, b) for a, b in self.edges if a in relevant and b in relevant]
        filtered.reverse_deps = {
            k: v & relevant for k, v in self.reverse_deps.items() if k in relevant
        }
        filtered.warnings = self.warnings
        return filtered

    def _collect_deps(self, start: str, result: Set[str], max_depth: int) -> None:
        """BFS to collect dependencies down to max_depth."""
        queue: deque = deque([(start, 0)])
        while queue:
            current, depth = queue.popleft()
            if max_depth >= 0 and depth >= max_depth:
                continue
            for dep in self.nodes.get(current, PackageNode("", "", "")).internal_deps:
                if dep not in result:
                    result.add(dep)
                    queue.append((dep, depth + 1))

    def _collect_dependents(self, start: str, result: Set[str], max_depth: int) -> None:
        """BFS to collect dependents up to max_depth."""
        queue: deque = deque([(start, 0)])
        while queue:
            current, depth = queue.popleft()
            if max_depth >= 0 and depth >= max_depth:
                continue
            for dep in self.reverse_deps.get(current, set()):
                if dep not in result:
                    result.add(dep)
                    queue.append((dep, depth + 1))

    def generate_mermaid(self, orientation: str = "TD", show_categories: bool = True) -> str:
        """Generate Mermaid flowchart syntax."""
        lines = [f"graph {orientation}"]

        # Sanitize node names for Mermaid (replace special chars)
        def node_id(name: str) -> str:
            return name.replace("@", "").replace("/", "_").replace("-", "_")

        def node_label(name: str) -> str:
            return name

        # Group by category with subgraphs
        if show_categories and len(set(n.category for n in self.nodes.values())) > 1:
            categories: Dict[str, List[PackageNode]] = {}
            for node in self.nodes.values():
                categories.setdefault(node.category, []).append(node)

            for category, nodes in sorted(categories.items()):
                lines.append(f"    subgraph {category}")
                for node in sorted(nodes, key=lambda n: n.name):
                    nid = node_id(node.name)
                    label = node_label(node.name)
                    if node.dependent_count == 0:
                        lines.append(f"        {nid}[{label}]")
                    elif node.dep_count == 0:
                        lines.append(f"        {nid}({label})")
                    else:
                        lines.append(f"        {nid}[{label}]")
                lines.append("    end")
        else:
            for node in sorted(self.nodes.values(), key=lambda n: n.name):
                nid = node_id(node.name)
                label = node_label(node.name)
                if node.dependent_count == 0:
                    lines.append(f"    {nid}[{label}]")
                elif node.dep_count == 0:
                    lines.append(f"    {nid}({label})")
                else:
                    lines.append(f"    {nid}[{label}]")

        # Add edges
        lines.append("")
        for source, target in sorted(self.edges):
            lines.append(f"    {node_id(source)} --> {node_id(target)}")

        return "\n".join(lines)


def build_graph_result(builder: DependencyGraphBuilder, orientation: str, show_categories: bool) -> GraphResult:
    """Build the complete graph result."""
    circular = builder.detect_circular_deps()
    mermaid = builder.generate_mermaid(orientation, show_categories)

    categories: Dict[str, int] = {}
    for node in builder.nodes.values():
        categories[node.category] = categories.get(node.category, 0) + 1

    packages_info = []
    for node in sorted(builder.nodes.values(), key=lambda n: n.name):
        packages_info.append({
            "name": node.name,
            "path": node.path,
            "category": node.category,
            "internal_deps": node.internal_deps,
            "dep_count": node.dep_count,
            "dependent_count": node.dependent_count,
        })

    return GraphResult(
        mermaid=mermaid,
        node_count=len(builder.nodes),
        edge_count=len(builder.edges),
        packages=packages_info,
        categories=categories,
        circular_deps=[cycle for cycle in circular],
        warnings=builder.warnings,
    )


def format_human(result: GraphResult) -> str:
    """Format graph result for human-readable output."""
    lines = []
    lines.append("=" * 60)
    lines.append("DEPENDENCY GRAPH")
    lines.append("=" * 60)
    lines.append(f"Packages: {result.node_count}  Edges: {result.edge_count}")
    lines.append(f"Categories: {', '.join(f'{k}({v})' for k, v in sorted(result.categories.items()))}")
    lines.append("")

    if result.circular_deps:
        lines.append("-" * 60)
        lines.append("CIRCULAR DEPENDENCIES DETECTED")
        lines.append("-" * 60)
        for cycle in result.circular_deps:
            lines.append(f"  ! {' -> '.join(cycle)}")
        lines.append("")

    # Package connectivity summary
    lines.append("-" * 60)
    lines.append("PACKAGE CONNECTIVITY")
    lines.append("-" * 60)
    for pkg in result.packages:
        deps_str = f"deps={pkg['dep_count']}"
        dependents_str = f"dependents={pkg['dependent_count']}"
        lines.append(f"  {pkg['name']:40s} {deps_str:10s} {dependents_str}")
    lines.append("")

    # Mermaid diagram
    lines.append("-" * 60)
    lines.append("MERMAID DIAGRAM")
    lines.append("-" * 60)
    lines.append("")
    lines.append("```mermaid")
    lines.append(result.mermaid)
    lines.append("```")
    lines.append("")

    if result.warnings:
        lines.append("-" * 60)
        lines.append("WARNINGS")
        lines.append("-" * 60)
        for w in result.warnings:
            lines.append(f"  ! {w}")
        lines.append("")

    return "\n".join(lines)


def main() -> None:
    parser = argparse.ArgumentParser(
        description="Generate a Mermaid dependency graph of monorepo internal packages.",
        epilog="Example: python dependency_graph.py /path/to/monorepo --focus @repo/ui --depth 2",
    )
    parser.add_argument(
        "path",
        nargs="?",
        default=".",
        help="Path to monorepo root (default: current directory)",
    )
    parser.add_argument(
        "--json",
        action="store_true",
        dest="json_output",
        help="Output results as JSON",
    )
    parser.add_argument(
        "--mermaid-only",
        action="store_true",
        help="Output only the raw Mermaid diagram (no wrapper)",
    )
    parser.add_argument(
        "--orientation",
        choices=["TD", "LR", "BT", "RL"],
        default="TD",
        help="Graph orientation: TD (top-down), LR (left-right), BT, RL (default: TD)",
    )
    parser.add_argument(
        "--focus",
        default=None,
        help="Focus on a specific package and its connections",
    )
    parser.add_argument(
        "--depth",
        type=int,
        default=-1,
        help="Max depth from focus package (-1 for unlimited, default: -1)",
    )
    parser.add_argument(
        "--direction",
        choices=["deps", "dependents", "both"],
        default="both",
        help="Direction from focus: deps, dependents, or both (default: both)",
    )
    parser.add_argument(
        "--no-categories",
        action="store_true",
        help="Do not group packages into category subgraphs",
    )

    args = parser.parse_args()

    root = Path(args.path).resolve()
    if not root.is_dir():
        print(f"Error: {root} is not a valid directory", file=sys.stderr)
        sys.exit(1)

    builder = DependencyGraphBuilder(str(root))
    builder.build()

    if not builder.nodes:
        print("No packages found. Is this a monorepo with workspace configuration?", file=sys.stderr)
        sys.exit(1)

    if args.focus:
        builder = builder.filter_by_package(args.focus, args.depth, args.direction)

    result = build_graph_result(builder, args.orientation, not args.no_categories)

    if args.json_output:
        output = {
            "mermaid": result.mermaid,
            "node_count": result.node_count,
            "edge_count": result.edge_count,
            "packages": result.packages,
            "categories": result.categories,
            "circular_deps": result.circular_deps,
            "warnings": result.warnings,
        }
        print(json.dumps(output, indent=2))
    elif args.mermaid_only:
        print(result.mermaid)
    else:
        print(format_human(result))


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Impact Detector - Determine which monorepo packages are affected by file changes.

Given a list of changed files (or a git ref to diff against), this tool resolves
which packages are directly affected, then computes transitive dependents to
show the full blast radius. Useful for selective CI builds and PR impact summaries.

Author: Claude Skills Engineering Team
License: MIT
"""

import argparse
import glob
import json
import os
import subprocess
import sys
from collections import deque
from dataclasses import dataclass, field, asdict
from pathlib import Path
from typing import Any, Dict, List, Optional, Set, Tuple


@dataclass
class ImpactResult:
    """Result of impact analysis."""
    changed_files: List[str]
    directly_affected: List[str]
    transitively_affected: List[str]
    total_affected: List[str]
    unaffected: List[str]
    blast_radius_percent: float
    impact_chains: Dict[str, List[str]]
    root_config_changed: bool
    warnings: List[str]


class MonorepoGraph:
    """Builds and queries the internal dependency graph."""

    def __init__(self, root_path: str):
        self.root_path = Path(root_path).resolve()
        self.packages: Dict[str, str] = {}  # name -> relative path
        self.paths_to_names: Dict[str, str] = {}  # relative path -> name
        self.deps: Dict[str, Set[str]] = {}  # name -> set of internal deps
        self.reverse_deps: Dict[str, Set[str]] = {}  # name -> set of dependents
        self.warnings: List[str] = []

    def build(self) -> None:
        """Discover packages and build the dependency graph."""
        workspace_globs = self._detect_workspaces()
        if not workspace_globs:
            workspace_globs = ["packages/*", "apps/*", "libs/*", "modules/*"]
            self.warnings.append("No workspace config found; using default glob patterns.")

        self._discover_packages(workspace_globs)
        self._build_graph()

    def _detect_workspaces(self) -> List[str]:
        """Detect workspace globs from config files."""
        # pnpm-workspace.yaml
        pnpm_ws = self.root_path / "pnpm-workspace.yaml"
        if pnpm_ws.exists():
            return self._parse_pnpm_workspace(pnpm_ws)

        # package.json workspaces
        root_pkg = self.root_path / "package.json"
        if root_pkg.exists():
            try:
                with open(root_pkg, "r") as f:
                    data = json.load(f)
                ws = data.get("workspaces")
                if isinstance(ws, list):
                    return ws
                if isinstance(ws, dict) and "packages" in ws:
                    return ws["packages"]
            except (json.JSONDecodeError, OSError):
                pass

        # lerna.json
        lerna_cfg = self.root_path / "lerna.json"
        if lerna_cfg.exists():
            try:
                with open(lerna_cfg, "r") as f:
                    return json.load(f).get("packages", ["packages/*"])
            except (json.JSONDecodeError, OSError):
                pass

        return []

    def _parse_pnpm_workspace(self, filepath: Path) -> List[str]:
        """Parse pnpm-workspace.yaml without external YAML library."""
        globs = []
        try:
            with open(filepath, "r") as f:
                in_packages = False
                for line in f:
                    stripped = line.strip()
                    if stripped == "packages:":
                        in_packages = True
                        continue
                    if in_packages:
                        if stripped.startswith("- "):
                            globs.append(stripped[2:].strip().strip("'\""))
                        elif stripped and not stripped.startswith("#"):
                            break
        except OSError:
            pass
        return globs

    def _discover_packages(self, workspace_globs: List[str]) -> None:
        """Find all packages matching workspace globs."""
        for ws_glob in workspace_globs:
            if ws_glob.startswith("!"):
                continue
            pattern = str(self.root_path / ws_glob / "package.json")
            for pkg_path in sorted(glob.glob(pattern)):
                pkg_json = Path(pkg_path)
                try:
                    with open(pkg_json, "r") as f:
                        data = json.load(f)
                except (json.JSONDecodeError, OSError):
                    continue
                name = data.get("name")
                if not name:
                    continue
                rel_path = str(pkg_json.parent.relative_to(self.root_path))
                self.packages[name] = rel_path
                self.paths_to_names[rel_path] = name

    def _build_graph(self) -> None:
        """Build dependency and reverse-dependency maps."""
        internal_names = set(self.packages.keys())
        for name, rel_path in self.packages.items():
            pkg_json = self.root_path / rel_path / "package.json"
            try:
                with open(pkg_json, "r") as f:
                    data = json.load(f)
            except (json.JSONDecodeError, OSError):
                continue

            all_dep_sections = [
                data.get("dependencies", {}),
                data.get("devDependencies", {}),
                data.get("peerDependencies", {}),
            ]
            internal_deps = set()
            for section in all_dep_sections:
                for dep_name in section:
                    if dep_name in internal_names:
                        internal_deps.add(dep_name)

            self.deps[name] = internal_deps
            for dep in internal_deps:
                self.reverse_deps.setdefault(dep, set()).add(name)

    def get_package_for_file(self, filepath: str) -> Optional[str]:
        """Determine which package a file belongs to based on path prefix."""
        # Normalize path
        normalized = filepath.replace("\\", "/").lstrip("./")

        # Sort paths longest first for most specific match
        sorted_paths = sorted(self.paths_to_names.keys(), key=len, reverse=True)
        for pkg_path in sorted_paths:
            if normalized.startswith(pkg_path + "/") or normalized == pkg_path:
                return self.paths_to_names[pkg_path]
        return None

    def get_transitive_dependents(self, package_name: str) -> List[str]:
        """BFS to find all transitive dependents of a package."""
        visited: Set[str] = set()
        queue = deque([package_name])
        result = []

        while queue:
            current = queue.popleft()
            for dependent in self.reverse_deps.get(current, set()):
                if dependent not in visited:
                    visited.add(dependent)
                    result.append(dependent)
                    queue.append(dependent)

        return sorted(result)

    def get_impact_chain(self, source: str, target: str) -> List[str]:
        """Find the shortest dependency chain from source to target (via reverse deps)."""
        if source == target:
            return [source]
        visited: Set[str] = {source}
        queue: deque = deque([(source, [source])])
        while queue:
            current, path = queue.popleft()
            for dep in self.reverse_deps.get(current, set()):
                if dep == target:
                    return path + [dep]
                if dep not in visited:
                    visited.add(dep)
                    queue.append((dep, path + [dep]))
        return []


ROOT_CONFIG_FILES = {
    "turbo.json", "pnpm-workspace.yaml", "lerna.json", "nx.json",
    "tsconfig.json", "tsconfig.base.json", ".eslintrc.js", ".eslintrc.json",
    "jest.config.js", "vitest.config.ts", "pnpm-lock.yaml", "yarn.lock",
    "package-lock.json",
}


def get_changed_files_from_git(root_path: str, ref: str) -> List[str]:
    """Get list of changed files compared to a git ref."""
    try:
        result = subprocess.run(
            ["git", "diff", "--name-only", ref],
            capture_output=True,
            text=True,
            cwd=root_path,
            timeout=30,
        )
        if result.returncode != 0:
            # Try with merge-base
            merge_base = subprocess.run(
                ["git", "merge-base", ref, "HEAD"],
                capture_output=True,
                text=True,
                cwd=root_path,
                timeout=15,
            )
            if merge_base.returncode == 0:
                base = merge_base.stdout.strip()
                result = subprocess.run(
                    ["git", "diff", "--name-only", base],
                    capture_output=True,
                    text=True,
                    cwd=root_path,
                    timeout=30,
                )
        return [f for f in result.stdout.strip().split("\n") if f]
    except (subprocess.TimeoutExpired, FileNotFoundError):
        return []


def detect_impact(
    root_path: str,
    changed_files: List[str],
) -> ImpactResult:
    """Run impact detection on the given changed files."""
    graph = MonorepoGraph(root_path)
    graph.build()

    all_package_names = set(graph.packages.keys())
    directly_affected: Set[str] = set()
    root_config_changed = False

    for filepath in changed_files:
        basename = os.path.basename(filepath)
        if basename in ROOT_CONFIG_FILES or filepath.startswith(".github/"):
            root_config_changed = True

        pkg_name = graph.get_package_for_file(filepath)
        if pkg_name:
            directly_affected.add(pkg_name)

    # Compute transitive dependents
    transitively_affected: Set[str] = set()
    for pkg in directly_affected:
        for dep in graph.get_transitive_dependents(pkg):
            if dep not in directly_affected:
                transitively_affected.add(dep)

    total_affected = directly_affected | transitively_affected

    # If root config changed, all packages are affected
    if root_config_changed:
        total_affected = all_package_names
        transitively_affected = all_package_names - directly_affected

    unaffected = sorted(all_package_names - total_affected)

    # Build impact chains for transitively affected packages
    impact_chains: Dict[str, List[str]] = {}
    for target in sorted(transitively_affected):
        for source in sorted(directly_affected):
            chain = graph.get_impact_chain(source, target)
            if chain:
                impact_chains[target] = chain
                break

    total_count = len(all_package_names) if all_package_names else 1
    blast_pct = round(len(total_affected) / total_count * 100, 1)

    return ImpactResult(
        changed_files=changed_files,
        directly_affected=sorted(directly_affected),
        transitively_affected=sorted(transitively_affected),
        total_affected=sorted(total_affected),
        unaffected=unaffected,
        blast_radius_percent=blast_pct,
        impact_chains=impact_chains,
        root_config_changed=root_config_changed,
        warnings=graph.warnings,
    )


def format_human(result: ImpactResult) -> str:
    """Format impact result for human-readable output."""
    lines = []
    lines.append("=" * 60)
    lines.append("IMPACT ANALYSIS")
    lines.append("=" * 60)
    lines.append(f"Changed files:     {len(result.changed_files)}")
    lines.append(f"Blast radius:      {result.blast_radius_percent}%")
    if result.root_config_changed:
        lines.append("Root config changed: YES (all packages affected)")
    lines.append("")

    lines.append("-" * 60)
    lines.append(f"DIRECTLY AFFECTED ({len(result.directly_affected)})")
    lines.append("-" * 60)
    for pkg in result.directly_affected:
        lines.append(f"  * {pkg}")
    lines.append("")

    if result.transitively_affected:
        lines.append("-" * 60)
        lines.append(f"TRANSITIVELY AFFECTED ({len(result.transitively_affected)})")
        lines.append("-" * 60)
        for pkg in result.transitively_affected:
            chain = result.impact_chains.get(pkg, [])
            chain_str = f"  (via {' -> '.join(chain)})" if chain else ""
            lines.append(f"  ~ {pkg}{chain_str}")
        lines.append("")

    if result.unaffected:
        lines.append("-" * 60)
        lines.append(f"UNAFFECTED ({len(result.unaffected)})")
        lines.append("-" * 60)
        for pkg in result.unaffected:
            lines.append(f"    {pkg}")
        lines.append("")

    if result.warnings:
        lines.append("-" * 60)
        lines.append("WARNINGS")
        lines.append("-" * 60)
        for w in result.warnings:
            lines.append(f"  ! {w}")
        lines.append("")

    return "\n".join(lines)


def main() -> None:
    parser = argparse.ArgumentParser(
        description="Detect which monorepo packages are affected by file changes.",
        epilog="Example: python impact_detector.py --ref origin/main --json",
    )
    parser.add_argument(
        "path",
        nargs="?",
        default=".",
        help="Path to monorepo root (default: current directory)",
    )
    parser.add_argument(
        "--files",
        nargs="+",
        help="Explicit list of changed file paths (relative to monorepo root)",
    )
    parser.add_argument(
        "--ref",
        default=None,
        help="Git ref to diff against (e.g., origin/main, HEAD~3)",
    )
    parser.add_argument(
        "--json",
        action="store_true",
        dest="json_output",
        help="Output results as JSON",
    )
    parser.add_argument(
        "--affected-only",
        action="store_true",
        help="Only print names of affected packages (one per line)",
    )

    args = parser.parse_args()

    root = Path(args.path).resolve()
    if not root.is_dir():
        print(f"Error: {root} is not a valid directory", file=sys.stderr)
        sys.exit(1)

    # Determine changed files
    changed_files: List[str] = []
    if args.files:
        changed_files = args.files
    elif args.ref:
        changed_files = get_changed_files_from_git(str(root), args.ref)
        if not changed_files:
            print(f"No changed files found compared to {args.ref}", file=sys.stderr)
            sys.exit(0)
    else:
        # Read from stdin
        if not sys.stdin.isatty():
            changed_files = [line.strip() for line in sys.stdin if line.strip()]
        else:
            print("Error: Provide --files, --ref, or pipe file list via stdin", file=sys.stderr)
            parser.print_help(sys.stderr)
            sys.exit(1)

    result = detect_impact(str(root), changed_files)

    if args.json_output:
        print(json.dumps(asdict(result), indent=2))
    elif args.affected_only:
        for pkg in result.total_affected:
            print(pkg)
    else:
        print(format_human(result))


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Package Analyzer - Analyze monorepo package structure and dependencies.

Detects packages in a monorepo by scanning workspace configuration files
(pnpm-workspace.yaml, package.json workspaces, lerna.json), resolves
internal cross-package dependencies, identifies shared external dependencies,
and reports on package health metrics.

Author: Claude Skills Engineering Team
License: MIT
"""

import argparse
import glob
import json
import os
import re
import sys
from dataclasses import dataclass, field, asdict
from pathlib import Path
from typing import Any, Dict, List, Optional, Set, Tuple


@dataclass
class Package:
    """Represents a monorepo package."""
    name: str
    path: str
    version: str
    private: bool
    internal_deps: List[str] = field(default_factory=list)
    external_deps: Dict[str, str] = field(default_factory=dict)
    dev_deps: Dict[str, str] = field(default_factory=dict)
    peer_deps: Dict[str, str] = field(default_factory=dict)
    scripts: List[str] = field(default_factory=list)
    has_tests: bool = False
    has_build: bool = False
    entry_point: Optional[str] = None


@dataclass
class AnalysisResult:
    """Full analysis result for the monorepo."""
    root_path: str
    workspace_tool: str
    total_packages: int
    packages: List[Dict[str, Any]]
    internal_dependency_count: int
    shared_external_deps: List[Dict[str, Any]]
    orphan_packages: List[str]
    leaf_packages: List[str]
    root_packages: List[str]
    warnings: List[str]


class PackageAnalyzer:
    """Analyzes monorepo package structure and dependencies."""

    def __init__(self, root_path: str):
        self.root_path = Path(root_path).resolve()
        self.packages: Dict[str, Package] = {}
        self.workspace_tool = "unknown"
        self.warnings: List[str] = []

    def analyze(self) -> AnalysisResult:
        """Run full analysis on the monorepo."""
        workspace_globs = self._detect_workspace_config()
        if not workspace_globs:
            self.warnings.append("No workspace configuration found. Scanning for package.json files.")
            workspace_globs = ["packages/*", "apps/*", "libs/*", "modules/*"]

        self._discover_packages(workspace_globs)
        self._resolve_internal_deps()
        self._detect_package_features()

        shared_deps = self._find_shared_external_deps()
        orphans = self._find_orphan_packages()
        leaves = self._find_leaf_packages()
        roots = self._find_root_packages()

        return AnalysisResult(
            root_path=str(self.root_path),
            workspace_tool=self.workspace_tool,
            total_packages=len(self.packages),
            packages=[asdict(p) for p in self.packages.values()],
            internal_dependency_count=sum(len(p.internal_deps) for p in self.packages.values()),
            shared_external_deps=shared_deps,
            orphan_packages=orphans,
            leaf_packages=leaves,
            root_packages=roots,
            warnings=self.warnings,
        )

    def _detect_workspace_config(self) -> List[str]:
        """Detect workspace configuration and return package globs."""
        # Try pnpm-workspace.yaml
        pnpm_ws = self.root_path / "pnpm-workspace.yaml"
        if pnpm_ws.exists():
            self.workspace_tool = "pnpm"
            return self._parse_pnpm_workspace(pnpm_ws)

        # Try root package.json workspaces field
        root_pkg = self.root_path / "package.json"
        if root_pkg.exists():
            try:
                with open(root_pkg, "r") as f:
                    data = json.load(f)
                workspaces = data.get("workspaces", None)
                if isinstance(workspaces, list):
                    self.workspace_tool = "npm/yarn"
                    return workspaces
                if isinstance(workspaces, dict) and "packages" in workspaces:
                    self.workspace_tool = "yarn"
                    return workspaces["packages"]
            except (json.JSONDecodeError, OSError):
                pass

        # Try lerna.json
        lerna_cfg = self.root_path / "lerna.json"
        if lerna_cfg.exists():
            self.workspace_tool = "lerna"
            try:
                with open(lerna_cfg, "r") as f:
                    data = json.load(f)
                return data.get("packages", ["packages/*"])
            except (json.JSONDecodeError, OSError):
                return ["packages/*"]

        return []

    def _parse_pnpm_workspace(self, filepath: Path) -> List[str]:
        """Parse pnpm-workspace.yaml without PyYAML (stdlib only)."""
        globs = []
        try:
            with open(filepath, "r") as f:
                in_packages = False
                for line in f:
                    stripped = line.strip()
                    if stripped == "packages:":
                        in_packages = True
                        continue
                    if in_packages:
                        if stripped.startswith("- "):
                            val = stripped[2:].strip().strip("'\"")
                            globs.append(val)
                        elif stripped and not stripped.startswith("#"):
                            break
        except OSError:
            self.warnings.append(f"Could not read {filepath}")
        return globs

    def _discover_packages(self, workspace_globs: List[str]) -> None:
        """Find all packages matching workspace globs."""
        for ws_glob in workspace_globs:
            # Skip negation patterns
            if ws_glob.startswith("!"):
                continue
            pattern = str(self.root_path / ws_glob / "package.json")
            for pkg_path in sorted(glob.glob(pattern)):
                self._load_package(Path(pkg_path))

    def _load_package(self, pkg_json_path: Path) -> None:
        """Load a single package from its package.json."""
        try:
            with open(pkg_json_path, "r") as f:
                data = json.load(f)
        except (json.JSONDecodeError, OSError) as e:
            self.warnings.append(f"Could not parse {pkg_json_path}: {e}")
            return

        name = data.get("name")
        if not name:
            self.warnings.append(f"Package at {pkg_json_path.parent} has no name field")
            return

        pkg = Package(
            name=name,
            path=str(pkg_json_path.parent.relative_to(self.root_path)),
            version=data.get("version", "0.0.0"),
            private=data.get("private", False),
            external_deps=dict(data.get("dependencies", {})),
            dev_deps=dict(data.get("devDependencies", {})),
            peer_deps=dict(data.get("peerDependencies", {})),
            scripts=list(data.get("scripts", {}).keys()),
            entry_point=data.get("main") or data.get("module") or data.get("exports"),
        )
        self.packages[name] = pkg

    def _resolve_internal_deps(self) -> None:
        """Separate internal workspace deps from external deps."""
        internal_names = set(self.packages.keys())
        for pkg in self.packages.values():
            internal = []
            external_only = {}
            for dep_name, dep_version in pkg.external_deps.items():
                if dep_name in internal_names:
                    internal.append(dep_name)
                else:
                    external_only[dep_name] = dep_version
            pkg.internal_deps = sorted(internal)
            pkg.external_deps = external_only

            # Also check devDependencies for internal refs
            dev_external = {}
            for dep_name, dep_version in pkg.dev_deps.items():
                if dep_name in internal_names and dep_name not in pkg.internal_deps:
                    pkg.internal_deps.append(dep_name)
                else:
                    dev_external[dep_name] = dep_version
            pkg.dev_deps = dev_external
            pkg.internal_deps = sorted(set(pkg.internal_deps))

    def _detect_package_features(self) -> None:
        """Detect if packages have tests, build scripts, etc."""
        for pkg in self.packages.values():
            pkg.has_tests = any(s in pkg.scripts for s in ("test", "test:unit", "test:e2e", "vitest"))
            pkg.has_build = any(s in pkg.scripts for s in ("build", "compile", "bundle"))

    def _find_shared_external_deps(self) -> List[Dict[str, Any]]:
        """Find external dependencies shared across multiple packages."""
        dep_usage: Dict[str, List[Tuple[str, str]]] = {}
        for pkg in self.packages.values():
            for dep_name, dep_version in pkg.external_deps.items():
                dep_usage.setdefault(dep_name, []).append((pkg.name, dep_version))

        shared = []
        for dep_name, users in sorted(dep_usage.items()):
            if len(users) < 2:
                continue
            versions = list(set(v for _, v in users))
            entry = {
                "name": dep_name,
                "used_by_count": len(users),
                "packages": [u[0] for u in users],
                "versions": versions,
                "version_mismatch": len(versions) > 1,
            }
            shared.append(entry)

        shared.sort(key=lambda x: x["used_by_count"], reverse=True)
        return shared

    def _find_orphan_packages(self) -> List[str]:
        """Find packages that nobody depends on and that depend on nothing internal."""
        all_depended_on: Set[str] = set()
        for pkg in self.packages.values():
            all_depended_on.update(pkg.internal_deps)

        orphans = []
        for name, pkg in self.packages.items():
            if name not in all_depended_on and not pkg.internal_deps:
                orphans.append(name)
        return sorted(orphans)

    def _find_leaf_packages(self) -> List[str]:
        """Find packages with no internal dependencies (leaves of the dep tree)."""
        return sorted(name for name, pkg in self.packages.items() if not pkg.internal_deps)

    def _find_root_packages(self) -> List[str]:
        """Find packages that nothing else depends on (roots/apps)."""
        all_depended_on: Set[str] = set()
        for pkg in self.packages.values():
            all_depended_on.update(pkg.internal_deps)
        return sorted(name for name in self.packages if name not in all_depended_on)


def format_human(result: AnalysisResult) -> str:
    """Format analysis result for human-readable output."""
    lines = []
    lines.append("=" * 60)
    lines.append("MONOREPO PACKAGE ANALYSIS")
    lines.append("=" * 60)
    lines.append(f"Root:            {result.root_path}")
    lines.append(f"Workspace Tool:  {result.workspace_tool}")
    lines.append(f"Total Packages:  {result.total_packages}")
    lines.append(f"Internal Deps:   {result.internal_dependency_count}")
    lines.append("")

    # Package summary table
    lines.append("-" * 60)
    lines.append("PACKAGES")
    lines.append("-" * 60)
    for pkg in result.packages:
        private_tag = " [private]" if pkg["private"] else ""
        lines.append(f"  {pkg['name']}@{pkg['version']}{private_tag}")
        lines.append(f"    Path: {pkg['path']}")
        if pkg["internal_deps"]:
            lines.append(f"    Internal deps: {', '.join(pkg['internal_deps'])}")
        ext_count = len(pkg["external_deps"])
        dev_count = len(pkg["dev_deps"])
        lines.append(f"    External deps: {ext_count}  Dev deps: {dev_count}")
        features = []
        if pkg["has_build"]:
            features.append("build")
        if pkg["has_tests"]:
            features.append("tests")
        if features:
            lines.append(f"    Features: {', '.join(features)}")
        lines.append("")

    # Leaf packages
    if result.leaf_packages:
        lines.append("-" * 60)
        lines.append("LEAF PACKAGES (no internal deps)")
        lines.append("-" * 60)
        for name in result.leaf_packages:
            lines.append(f"  - {name}")
        lines.append("")

    # Root packages
    if result.root_packages:
        lines.append("-" * 60)
        lines.append("ROOT PACKAGES (nothing depends on them)")
        lines.append("-" * 60)
        for name in result.root_packages:
            lines.append(f"  - {name}")
        lines.append("")

    # Orphan packages
    if result.orphan_packages:
        lines.append("-" * 60)
        lines.append("ORPHAN PACKAGES (isolated - no internal connections)")
        lines.append("-" * 60)
        for name in result.orphan_packages:
            lines.append(f"  - {name}")
        lines.append("")

    # Shared deps
    if result.shared_external_deps:
        lines.append("-" * 60)
        lines.append("SHARED EXTERNAL DEPENDENCIES")
        lines.append("-" * 60)
        for dep in result.shared_external_deps[:15]:
            mismatch = " [VERSION MISMATCH]" if dep["version_mismatch"] else ""
            lines.append(f"  {dep['name']} (used by {dep['used_by_count']} packages){mismatch}")
            if dep["version_mismatch"]:
                lines.append(f"    Versions: {', '.join(dep['versions'])}")
        lines.append("")

    # Warnings
    if result.warnings:
        lines.append("-" * 60)
        lines.append("WARNINGS")
        lines.append("-" * 60)
        for w in result.warnings:
            lines.append(f"  ! {w}")
        lines.append("")

    return "\n".join(lines)


def main() -> None:
    parser = argparse.ArgumentParser(
        description="Analyze monorepo package structure, dependencies, and health.",
        epilog="Example: python package_analyzer.py /path/to/monorepo --json",
    )
    parser.add_argument(
        "path",
        nargs="?",
        default=".",
        help="Path to monorepo root (default: current directory)",
    )
    parser.add_argument(
        "--json",
        action="store_true",
        dest="json_output",
        help="Output results as JSON",
    )
    parser.add_argument(
        "--only-shared",
        action="store_true",
        help="Only show shared external dependencies",
    )
    parser.add_argument(
        "--only-orphans",
        action="store_true",
        help="Only show orphan packages",
    )

    args = parser.parse_args()

    root = Path(args.path).resolve()
    if not root.is_dir():
        print(f"Error: {root} is not a valid directory", file=sys.stderr)
        sys.exit(1)

    analyzer = PackageAnalyzer(str(root))
    result = analyzer.analyze()

    if result.total_packages == 0:
        print("No packages found. Is this a monorepo with workspace configuration?", file=sys.stderr)
        sys.exit(1)

    if args.json_output:
        print(json.dumps(asdict(result), indent=2))
    elif args.only_shared:
        if not result.shared_external_deps:
            print("No shared external dependencies found.")
        else:
            for dep in result.shared_external_deps:
                mismatch = " [MISMATCH]" if dep["version_mismatch"] else ""
                print(f"{dep['name']} ({dep['used_by_count']} packages){mismatch}")
    elif args.only_orphans:
        if not result.orphan_packages:
            print("No orphan packages found.")
        else:
            for name in result.orphan_packages:
                print(name)
    else:
        print(format_human(result))


if __name__ == "__main__":
    main()