Function analyze_changes

code_review_graph/changes.py:277–409

Analyze changes and produce risk-scored review guidance. Args: store: The graph store. changed_files: List of changed file paths. changed_ranges: Optional pre-parsed diff ranges. If not provided and ``repo_root`` is given, they are computed via the detected V

(
    store: GraphStore,
    changed_files: list[str],
    changed_ranges: dict[str, list[tuple[int, int]]] | None = None,
    repo_root: str | None = None,
    base: str = "HEAD~1",
)

def analyze_changes(
    store: GraphStore,
    changed_files: list[str],
    changed_ranges: dict[str, list[tuple[int, int]]] | None = None,
    repo_root: str | None = None,
    base: str = "HEAD~1",
) -> dict[str, Any]:
    """Analyze changes and produce risk-scored review guidance.

    Args:
        store: The graph store.
        changed_files: List of changed file paths.
        changed_ranges: Optional pre-parsed diff ranges. If not provided and
            ``repo_root`` is given, they are computed via the detected VCS
            (Git or SVN).
        repo_root: Repository root (for git/svn diff).
        base: Git ref or SVN revision range to diff against.

    Returns:
        Dict with ``summary``, ``risk_score``, ``changed_functions``,
        ``affected_flows``, ``test_gaps``, and ``review_priorities``.
    """
    # Compute changed ranges if not provided.
    if changed_ranges is None and repo_root is not None:
        # Diff keys are forward-slash paths relative to the repo root, but
        # the graph stores absolute native paths. Remap so lookups work on
        # Windows, where the LIKE-suffix fallback cannot bridge
        # "src/app.py" to "C:\repo\src\app.py" (#528). Keys that are
        # already absolute pass through pathlib joining unchanged. The
        # explicit changed_ranges path (MCP) is untouched — tools/review.py
        # remaps before calling, and remapping twice would corrupt keys.
        root_path = Path(repo_root)
        changed_ranges = {
            str(root_path / key): ranges
            for key, ranges in parse_diff_ranges(repo_root, base).items()
        }

    # Map changes to nodes.
    if changed_ranges:
        changed_nodes = map_changes_to_nodes(store, changed_ranges)
    else:
        # Fallback: all nodes in changed files.
        changed_nodes = []
        for fp in changed_files:
            changed_nodes.extend(store.get_nodes_by_file(fp))

    # Filter to functions/tests for risk scoring (skip File nodes).
    changed_funcs = [
        n for n in changed_nodes
        if n.kind in ("Function", "Test", "Class")
    ]

    # Cap to prevent O(N*M) query explosion on large PRs.
    _max_funcs = int(os.environ.get("CRG_MAX_CHANGED_FUNCS", "500"))
    funcs_truncated = len(changed_funcs) > _max_funcs
    if funcs_truncated:
        changed_funcs = changed_funcs[:_max_funcs]