reuse-surface/reuse_surface/graph.py

from __future__ import annotations

import re
from dataclasses import dataclass
from pathlib import Path
from typing import Any

from reuse_surface.federation import FEDERATED_INDEX_PATH, compose_federated_index
from reuse_surface.registry import ROOT, load_index, parse_front_matter

GRAPH_PATH = ROOT / "docs" / "graph" / "capability-graph.mmd"
RELATION_TYPES = [
    "depends_on",
    "supports",
    "used_by",
    "related_to",
    "specializes",
    "generalizes",
    "replaces",
    "wraps",
]


@dataclass
class RelationEdge:
    source_id: str
    target_id: str
    relation_type: str


def _node_id(capability_id: str) -> str:
    return re.sub(r"[^a-zA-Z0-9_]", "_", capability_id)


def _load_local_relations() -> dict[str, dict[str, list[str]]]:
    index = load_index()
    relations_by_id: dict[str, dict[str, list[str]]] = {}
    for item in index.get("capabilities", []):
        path = ROOT / item["path"]
        if not path.exists():
            continue
        entry = parse_front_matter(path)
        relations = entry.get("relations") or {}
        relations_by_id[entry["id"]] = {
            relation_type: list(targets)
            for relation_type, targets in relations.items()
            if isinstance(targets, list)
        }
    return relations_by_id


def _known_ids() -> set[str]:
    if FEDERATED_INDEX_PATH.exists():
        import yaml

        data = yaml.safe_load(FEDERATED_INDEX_PATH.read_text(encoding="utf-8"))
    else:
        data, _ = compose_federated_index()
    return {item["id"] for item in data.get("capabilities", [])}


def collect_edges() -> list[RelationEdge]:
    relations_by_id = _load_local_relations()
    edges: list[RelationEdge] = []
    for source_id, relation_map in relations_by_id.items():
        for relation_type, targets in relation_map.items():
            for target_id in targets:
                edges.append(
                    RelationEdge(
                        source_id=source_id,
                        target_id=target_id,
                        relation_type=relation_type,
                    )
                )
    return edges


def find_depends_on_cycles() -> list[list[str]]:
    relations_by_id = _load_local_relations()
    graph: dict[str, list[str]] = {
        cap_id: list(relation_map.get("depends_on", []))
        for cap_id, relation_map in relations_by_id.items()
    }
    cycles: list[list[str]] = []
    visited: set[str] = set()
    stack: set[str] = set()
    path: list[str] = []

    def dfs(node: str) -> None:
        visited.add(node)
        stack.add(node)
        path.append(node)
        for neighbor in graph.get(node, []):
            if neighbor not in visited:
                dfs(neighbor)
            elif neighbor in stack:
                start = path.index(neighbor)
                cycles.append(path[start:] + [neighbor])
        path.pop()
        stack.remove(node)

    for node in graph:
        if node not in visited:
            dfs(node)
    return cycles


def find_broken_references(known: set[str] | None = None) -> list[str]:
    known = known or _known_ids()
    warnings: list[str] = []
    for edge in collect_edges():
        if edge.target_id not in known:
            warnings.append(
                f"broken relation: {edge.source_id} "
                f"{edge.relation_type} -> {edge.target_id}"
            )
    return warnings


def check_relations() -> list[str]:
    warnings: list[str] = []
    for cycle in find_depends_on_cycles():
        warnings.append(f"depends_on cycle: {' -> '.join(cycle)}")
    warnings.extend(find_broken_references())
    return warnings


def _node_labels() -> dict[str, str]:
    index = load_index()
    labels: dict[str, str] = {}
    for item in index.get("capabilities", []):
        labels[item["id"]] = f"{item['id']}<br/>{item['vector']}"
    return labels


def render_mermaid() -> str:
    labels = _node_labels()
    edges = collect_edges()
    lines = ["graph LR"]
    for cap_id, label in sorted(labels.items()):
        lines.append(f'  {_node_id(cap_id)}["{label}"]')
    for edge in edges:
        lines.append(
            f"  {_node_id(edge.source_id)} -->|{edge.relation_type}| {_node_id(edge.target_id)}"
        )
    return "\n".join(lines) + "\n"


def write_graph(path: Path | None = None) -> Path:
    target = path or GRAPH_PATH
    target.parent.mkdir(parents=True, exist_ok=True)
    target.write_text(render_mermaid(), encoding="utf-8")
    return target