Function detect_communities

code_review_graph/communities.py:585–632

Detect communities in the code graph. Uses the Leiden algorithm via igraph if available, otherwise falls back to file-based grouping. Args: store: The GraphStore instance. min_size: Minimum number of nodes for a community to be included. Returns: List of co

(
    store: GraphStore, min_size: int = 2
)

def detect_communities(
    store: GraphStore, min_size: int = 2
) -> list[dict[str, Any]]:
    """Detect communities in the code graph.

    Uses the Leiden algorithm via igraph if available, otherwise falls back to
    file-based grouping.

    Args:
        store: The GraphStore instance.
        min_size: Minimum number of nodes for a community to be included.

    Returns:
        List of community dicts with keys: name, level, size, cohesion,
        dominant_language, description, members, member_qns.
    """
    # Gather all nodes (exclude File nodes to focus on code entities)
    all_edges = store.get_all_edges()
    unique_nodes = store.get_all_nodes(exclude_files=True)

    # Build adjacency index once for fast cohesion computation
    adj = _build_adjacency(all_edges)

    logger.info(
        "Loaded %d unique nodes, %d edges",
        len(unique_nodes), len(all_edges),
    )

    if IGRAPH_AVAILABLE:
        logger.info("Detecting communities with Leiden algorithm (igraph)")
        results = _detect_leiden(unique_nodes, all_edges, min_size, adj=adj)
    else:
        logger.info("igraph not available, using file-based community detection")
        results = _detect_file_based(unique_nodes, all_edges, min_size, adj=adj)

    # Split oversized communities
    results = _split_oversized(
        results, unique_nodes, all_edges,
    )

    # Convert member_qns (internal set) to a list for serialization safety,
    # then strip it from the returned dicts to avoid leaking internal state.
    for comm in results:
        if "member_qns" in comm:
            comm["member_qns"] = list(comm["member_qns"])
            del comm["member_qns"]

    return results


def incremental_detect_communities(