Aggregate full graph data into file-level nodes. Each file becomes a node sized by symbol count. Edges between files represent aggregated cross-file dependencies.
(data: dict)
| 279 | |
| 280 | |
| 281 | def _aggregate_file(data: dict) -> dict: |
| 282 | """Aggregate full graph data into file-level nodes. |
| 283 | |
| 284 | Each file becomes a node sized by symbol count. |
| 285 | Edges between files represent aggregated cross-file dependencies. |
| 286 | """ |
| 287 | nodes = data["nodes"] |
| 288 | edges = data["edges"] |
| 289 | |
| 290 | # Count symbols per file |
| 291 | file_symbol_count: Counter[str] = Counter() |
| 292 | qn_to_file: dict[str, str] = {} |
| 293 | file_languages: dict[str, str] = {} |
| 294 | |
| 295 | for n in nodes: |
| 296 | fp = n.get("file_path", "") |
| 297 | if not fp: |
| 298 | continue |
| 299 | qn_to_file[n["qualified_name"]] = fp |
| 300 | if n["kind"] != "File": |
| 301 | file_symbol_count[fp] += 1 |
| 302 | else: |
| 303 | file_symbol_count.setdefault(fp, 0) |
| 304 | if n.get("language"): |
| 305 | file_languages[fp] = n["language"] |
| 306 | |
| 307 | # Build file nodes |
| 308 | file_nodes = [] |
| 309 | for fp, count in file_symbol_count.items(): |
| 310 | parts = fp.replace("\\", "/").split("/") |
| 311 | short = parts[-1] if parts else fp |
| 312 | parent = parts[-2] if len(parts) >= 2 else "" |
| 313 | label = f"{parent}/{short}" if parent else short |
| 314 | # Recover community_id from the majority of symbols in this file |
| 315 | cid = None |
| 316 | for n in nodes: |
| 317 | if n.get("file_path") == fp and n.get("community_id") is not None: |
| 318 | cid = n["community_id"] |
| 319 | break |
| 320 | file_nodes.append({ |
| 321 | "qualified_name": fp, |
| 322 | "name": label, |
| 323 | "kind": "File", |
| 324 | "file_path": fp, |
| 325 | "line_start": None, |
| 326 | "line_end": None, |
| 327 | "language": file_languages.get(fp, ""), |
| 328 | "community_id": cid, |
| 329 | "symbol_count": count, |
| 330 | }) |
| 331 | |
| 332 | # Aggregate cross-file edges |
| 333 | cross_file_counts: Counter[tuple[str, str]] = Counter() |
| 334 | for e in edges: |
| 335 | src_fp = qn_to_file.get(e["source"]) |
| 336 | tgt_fp = qn_to_file.get(e["target"]) |
| 337 | if src_fp and tgt_fp and src_fp != tgt_fp: |
| 338 | pair = (src_fp, tgt_fp) |