diffcontext/impact/visualizer.py

"""
visualizer.py — Visual blast radius renderer.

Renders the blast radius as a colored, indented tree showing:
  - The changed symbol at the root
  - Direct callers (who calls this?)
  - Direct callees (what does this call?)
  - Transitive impact propagation
  - Proof chains: the actual code line creating each edge

Designed for terminal output with ANSI colors.
"""

import ast
import os
import re
from typing import Dict, List, Optional, Set, Tuple


# ANSI color codes
class _C:
    RED = "\033[91m"
    YELLOW = "\033[93m"
    GREEN = "\033[92m"
    CYAN = "\033[96m"
    MAGENTA = "\033[95m"
    BLUE = "\033[94m"
    DIM = "\033[2m"
    BOLD = "\033[1m"
    RESET = "\033[0m"
    WHITE = "\033[97m"


def render_blast_radius(
    graph: Dict[str, List[str]],
    changed_symbols: List[str],
    symbols: dict,
    max_depth: int = 3,
    show_proof: bool = False,
    repo_path: str = "",
) -> str:
    """
    Render a visual tree of the blast radius for changed symbols.

    Returns a formatted string ready for terminal output.
    """
    # Build reverse graph (callers)
    reverse: Dict[str, Set[str]] = {}
    for caller, callees in graph.items():
        for callee in callees:
            reverse.setdefault(callee, set()).add(caller)

    lines: List[str] = []

    lines.append("")
    lines.append(f"{_C.BOLD}{_C.WHITE}{'═' * 70}{_C.RESET}")
    lines.append(f"{_C.BOLD}{_C.WHITE}  BLAST RADIUS ANALYSIS{_C.RESET}")
    lines.append(f"{_C.BOLD}{_C.WHITE}{'═' * 70}{_C.RESET}")

    for sym_id in changed_symbols:
        lines.append("")
        lines.append(f"  {_C.BOLD}{_C.RED}⚡ CHANGED:{_C.RESET} {_C.BOLD}{sym_id}{_C.RESET}")

        # Show symbol location info
        if sym_id in symbols:
            sym = symbols[sym_id]
            lines.append(f"  {_C.DIM}   File: {sym.file}{_C.RESET}")
            lines.append(f"  {_C.DIM}   Line: {sym.lineno}{_C.RESET}")

        lines.append("")

        # ---- CALLERS (who is affected by this change?) ----
        direct_callers = sorted(reverse.get(sym_id, set()))
        lines.append(f"  {_C.BOLD}{_C.YELLOW}▲ WHO CALLS THIS? (directly affected){_C.RESET}")

        if not direct_callers:
            lines.append(f"  {_C.DIM}  (no direct callers found){_C.RESET}")
        else:
            for i, caller in enumerate(direct_callers):
                is_last = i == len(direct_callers) - 1
                prefix = "└──" if is_last else "├──"
                lines.append(f"  {_C.YELLOW}  {prefix} {caller}{_C.RESET}")

                if show_proof:
                    proof = _find_proof(caller, sym_id, symbols, graph)
                    if proof:
                        pad = "    " if is_last else "│   "
                        lines.append(f"  {_C.DIM}  {pad} proof: {proof}{_C.RESET}")

                # 2nd-level callers (transitive)
                if max_depth >= 2:
                    indirect_callers = sorted(reverse.get(caller, set()))
                    for j, caller2 in enumerate(indirect_callers[:5]):
                        is_last2 = j == len(indirect_callers[:5]) - 1
                        indent = "    " if is_last else "│   "
                        prefix2 = "└──" if is_last2 else "├──"
                        lines.append(
                            f"  {_C.DIM}  {indent}{prefix2} {caller2}{_C.RESET}"
                        )
                    if len(indirect_callers) > 5:
                        indent = "    " if is_last else "│   "
                        lines.append(
                            f"  {_C.DIM}  {indent}... +{len(indirect_callers) - 5} more{_C.RESET}"
                        )

        lines.append("")

        # ---- CALLEES (what does this function depend on?) ----
        direct_callees = sorted(graph.get(sym_id, []))
        lines.append(f"  {_C.BOLD}{_C.GREEN}▼ WHAT DOES THIS CALL? (dependencies){_C.RESET}")

        if not direct_callees:
            lines.append(f"  {_C.DIM}  (no outgoing calls resolved){_C.RESET}")
        else:
            for i, callee in enumerate(direct_callees):
                is_last = i == len(direct_callees) - 1
                prefix = "└──" if is_last else "├──"
                lines.append(f"  {_C.GREEN}  {prefix} {callee}{_C.RESET}")

                if show_proof:
                    proof = _find_proof(sym_id, callee, symbols, graph)
                    if proof:
                        pad = "    " if is_last else "│   "
                        lines.append(f"  {_C.DIM}  {pad} proof: {proof}{_C.RESET}")

                # 2nd-level callees
                if max_depth >= 2:
                    indirect_callees = sorted(graph.get(callee, []))
                    for j, callee2 in enumerate(indirect_callees[:5]):
                        is_last2 = j == len(indirect_callees[:5]) - 1
                        indent = "    " if is_last else "│   "
                        prefix2 = "└──" if is_last2 else "├──"
                        lines.append(
                            f"  {_C.DIM}  {indent}{prefix2} {callee2}{_C.RESET}"
                        )
                    if len(indirect_callees) > 5:
                        indent = "    " if is_last else "│   "
                        lines.append(
                            f"  {_C.DIM}  {indent}... +{len(indirect_callees) - 5} more{_C.RESET}"
                        )

        lines.append("")

        # ---- FULL TRANSITIVE BLAST RADIUS ----
        all_affected = _get_transitive_callers(reverse, sym_id, max_depth)
        lines.append(f"  {_C.BOLD}{_C.CYAN}◉ FULL BLAST RADIUS{_C.RESET}")
        lines.append(
            f"  {_C.CYAN}  {len(all_affected)} symbols transitively affected{_C.RESET}"
        )

        # Group by file
        by_file: Dict[str, List[str]] = {}
        for affected_sym in all_affected:
            parts = affected_sym.split(":", 1)
            filepath = parts[0] if len(parts) == 2 else "unknown"
            by_file.setdefault(filepath, []).append(affected_sym)

        for filepath, syms in sorted(by_file.items()):
            lines.append(f"  {_C.BLUE}  📄 {filepath} ({len(syms)} symbols){_C.RESET}")
            for sym in sorted(syms)[:8]:
                name = sym.split(":", 1)[1] if ":" in sym else sym
                lines.append(f"  {_C.DIM}     · {name}{_C.RESET}")
            if len(syms) > 8:
                lines.append(f"  {_C.DIM}     ... +{len(syms) - 8} more{_C.RESET}")

    # ---- SUMMARY ----
    lines.append("")
    lines.append(f"{_C.BOLD}{_C.WHITE}{'─' * 70}{_C.RESET}")

    total_blast = set()
    for sym_id in changed_symbols:
        total_blast.update(_get_transitive_callers(reverse, sym_id, max_depth))

    total_deps = set()
    for sym_id in changed_symbols:
        total_deps.update(_get_transitive_callees(graph, sym_id, max_depth))

    blast_files = set()
    for sym in total_blast:
        parts = sym.split(":", 1)
        if len(parts) == 2:
            blast_files.add(parts[0])

    lines.append(f"  {_C.BOLD}Summary:{_C.RESET}")
    lines.append(f"    Changed symbols    : {_C.RED}{len(changed_symbols)}{_C.RESET}")
    lines.append(f"    Direct callers     : {_C.YELLOW}{sum(len(reverse.get(s, set())) for s in changed_symbols)}{_C.RESET}")
    lines.append(f"    Direct dependencies: {_C.GREEN}{sum(len(graph.get(s, [])) for s in changed_symbols)}{_C.RESET}")
    lines.append(f"    Total blast radius : {_C.CYAN}{len(total_blast)} symbols across {len(blast_files)} files{_C.RESET}")
    lines.append(f"    Total dependencies : {_C.GREEN}{len(total_deps)} symbols{_C.RESET}")
    lines.append(f"{_C.BOLD}{_C.WHITE}{'═' * 70}{_C.RESET}")
    lines.append("")

    return "\n".join(lines)


def render_verification(
    graph: Dict[str, List[str]],
    changed_symbols: List[str],
    symbols: dict,
) -> str:
    """
    Render verification proof: for each edge in the blast radius,
    show the actual code line that creates the dependency.
    """
    reverse: Dict[str, Set[str]] = {}
    for caller, callees in graph.items():
        for callee in callees:
            reverse.setdefault(callee, set()).add(caller)

    lines: List[str] = []
    lines.append("")
    lines.append(f"{_C.BOLD}{_C.WHITE}{'═' * 70}{_C.RESET}")
    lines.append(f"{_C.BOLD}{_C.WHITE}  VERIFICATION: Proof of each connection{_C.RESET}")
    lines.append(f"{_C.BOLD}{_C.WHITE}{'═' * 70}{_C.RESET}")

    for sym_id in changed_symbols:
        lines.append("")
        lines.append(f"  {_C.BOLD}{_C.RED}⚡ {sym_id}{_C.RESET}")
        lines.append("")

        # Verify each caller
        callers = sorted(reverse.get(sym_id, set()))
        if callers:
            lines.append(f"  {_C.BOLD}{_C.YELLOW}  Callers (these functions call the changed code):{_C.RESET}")
            for caller in callers:
                lines.append(f"  {_C.YELLOW}    → {caller}{_C.RESET}")
                proof = _find_proof(caller, sym_id, symbols, graph)
                if proof:
                    lines.append(f"  {_C.DIM}      evidence: {proof}{_C.RESET}")
                else:
                    lines.append(f"  {_C.DIM}      evidence: (edge exists in call graph){_C.RESET}")
            lines.append("")

        # Verify each callee
        callees = sorted(graph.get(sym_id, []))
        if callees:
            lines.append(f"  {_C.BOLD}{_C.GREEN}  Callees (the changed code calls these):{_C.RESET}")
            for callee in callees:
                lines.append(f"  {_C.GREEN}    → {callee}{_C.RESET}")
                proof = _find_proof(sym_id, callee, symbols, graph)
                if proof:
                    lines.append(f"  {_C.DIM}      evidence: {proof}{_C.RESET}")
                else:
                    lines.append(f"  {_C.DIM}      evidence: (edge exists in call graph){_C.RESET}")

    lines.append("")
    lines.append(f"{_C.BOLD}{_C.WHITE}{'═' * 70}{_C.RESET}")
    lines.append("")
    return "\n".join(lines)


# ---------------------------------------------------------------------------
# Internal helpers
# ---------------------------------------------------------------------------


def _get_transitive_callers(
    reverse: Dict[str, Set[str]],
    start: str,
    max_depth: int,
) -> List[str]:
    """BFS up the reverse graph to find all transitively affected symbols."""
    visited: Set[str] = {start}
    result: List[str] = []
    frontier = [start]
    depth = 0

    while frontier and depth < max_depth:
        next_frontier = []
        for node in frontier:
            for caller in reverse.get(node, set()):
                if caller not in visited:
                    visited.add(caller)
                    result.append(caller)
                    next_frontier.append(caller)
        frontier = next_frontier
        depth += 1

    return result


def _get_transitive_callees(
    graph: Dict[str, List[str]],
    start: str,
    max_depth: int,
) -> List[str]:
    """BFS down the forward graph to find all dependencies."""
    visited: Set[str] = {start}
    result: List[str] = []
    frontier = [start]
    depth = 0

    while frontier and depth < max_depth:
        next_frontier = []
        for node in frontier:
            for callee in graph.get(node, []):
                if callee not in visited:
                    visited.add(callee)
                    result.append(callee)
                    next_frontier.append(callee)
        frontier = next_frontier
        depth += 1

    return result


def _find_proof(
    caller_id: str,
    callee_id: str,
    symbols: dict,
    graph: Dict[str, List[str]],
) -> Optional[str]:
    """
    Find the actual code line in `caller` that references `callee`.

    This is the "proof" that the edge is real — a grep through the caller's
    source code for the callee's function name.
    """
    if caller_id not in symbols:
        return None

    caller_sym = symbols[caller_id]
    callee_name = callee_id.split(":")[-1] if ":" in callee_id else callee_id

    # Strip class prefix for method calls: "ClassName.method" -> "method"
    if "." in callee_name:
        bare_name = callee_name.split(".")[-1]
    else:
        bare_name = callee_name

    # Search caller's code for the call
    for i, line in enumerate(caller_sym.code.splitlines(), start=1):
        stripped = line.strip()
        # Look for function/method call pattern
        if bare_name + "(" in stripped or f".{bare_name}(" in stripped:
            # Truncate long lines
            if len(stripped) > 80:
                stripped = stripped[:77] + "..."
            return f"line {caller_sym.lineno + i - 1}: {stripped}"

    return None