#!/usr/bin/env python3 """Topic 10 Run 01 — better Cluster ↔ PDFGroup graphs (static SVG + interactive HTML) Why this exists --------------- The spring-layout PNG is useful as a quick sanity check, but it becomes "hairball-y" when many clusters share the same few PDFGroups. This script produces: (A) a clean *bipartite* graph (clusters on the left, PDFGroups on the right) (B) an optional *focal* view (ego graph around a chosen cluster, e.g. "MATTHEW GUERTIN") (C) optional *thresholding* (hide edges below N docs) to reduce clutter (D) optional *interactive* HTML (pan/zoom/drag) with PyVis Inputs (from exports) --------------------- reports/10_font_tracking/10r1_tracked_docs__summary.csv - must contain: cluster_id OR cluster_name, pdf_groups (DuckDB list-like) reports/10_font_tracking/10r1_pdf_group_inventory.csv - used to attach n_docs per PDFGroup (for sizing / hover text) Outputs (written to reports/10_font_tracking/) ---------------------------------------------- 10r1_graph_bipartite.svg (static SVG, if Graphviz available) 10r1_graph_bipartite.png (static PNG fallback) 10r1_graph_focal_.svg/png (focal ego version) 10r1_graph_bipartite_interactive.html (interactive HTML, if pyvis installed) Quick start ----------- 1) Run from project root: python3 sql/topics/10_font_tracking/10_font_tracking__run01__graph_plus.py 2) Focal view (recommended): python3 sql/topics/10_font_tracking/10_font_tracking__run01__graph_plus.py \ --focal "MATTHEW GUERTIN" --min_edge_weight 2 Dependencies (optional, recommended) ------------------------------------ Interactive HTML: - pyvis python3 -m pip install pyvis Clean SVG layout: - graphviz system package + python graphviz (optional) sudo apt-get update sudo apt-get install -y graphviz python3 -m pip install graphviz If you don't have pip: sudo apt-get install -y python3-pip python3-venv Notes ----- - This is a *bipartite* plot (two columns) so the structure is readable: clusters connect to PDFGroups; PDFGroups are labeled like "A1", "B1", etc. - Edge weight is doc count for that cluster↔PDFGroup. Use --min_edge_weight to declutter. """ from __future__ import annotations import argparse from pathlib import Path import pandas as pd import matplotlib.pyplot as plt try: import networkx as nx # type: ignore except Exception as e: raise SystemExit( "networkx is required for this script. Install with: python3 -m pip install networkx" ) from e OUT_DIR = Path("reports/10_font_tracking") DOC_SUMMARY = OUT_DIR / "10r1_tracked_docs__summary.csv" GROUP_INV = OUT_DIR / "10r1_pdf_group_inventory.csv" def _read_csv(p: Path) -> pd.DataFrame: if not p.exists(): raise FileNotFoundError(f"Missing expected input: {p}") return pd.read_csv(p) def _parse_listish(x) -> list[str]: """Parse DuckDB-exported list columns like: "['A1', 'B1']" or "[]" or NaN """ if pd.isna(x): return [] s = str(x).strip() if s in ("[]", ""): return [] s = s.strip("[]") if not s: return [] parts = [p.strip().strip("'").strip('"') for p in s.split(",")] return [p for p in parts if p] def build_edge_table(doc_df: pd.DataFrame) -> pd.DataFrame: cluster_col = "cluster_name" if "cluster_name" in doc_df.columns else "cluster_id" if "pdf_groups" not in doc_df.columns: raise ValueError(f"{DOC_SUMMARY} missing required column: pdf_groups") rows: list[tuple[str, str]] = [] for _, r in doc_df.iterrows(): cluster = str(r.get(cluster_col)) for g in _parse_listish(r.get("pdf_groups")): rows.append((cluster, str(g))) if not rows: raise ValueError("No edges found (pdf_groups column may be empty or not list-like).") edges = pd.DataFrame(rows, columns=["cluster", "pdf_group"]) edges = edges.value_counts().reset_index(name="weight") return edges def build_graph(edges: pd.DataFrame, min_edge_weight: int = 1) -> nx.Graph: e = edges.copy() e = e[e["weight"] >= min_edge_weight] if e.empty: raise ValueError(f"No edges survive min_edge_weight={min_edge_weight}") G = nx.Graph() for _, r in e.iterrows(): c = r["cluster"] g = f"PDFGroup:{r['pdf_group']}" G.add_node(c, bipartite="cluster") G.add_node(g, bipartite="pdf_group") G.add_edge(c, g, weight=float(r["weight"])) return G def ego_subgraph(G: nx.Graph, focal: str, hops: int = 1) -> nx.Graph: if focal not in G: raise ValueError(f"Focal node not found: {focal!r}") nodes = {focal} frontier = {focal} for _ in range(hops): nxt = set() for n in frontier: nxt |= set(G.neighbors(n)) nodes |= nxt frontier = nxt return G.subgraph(nodes).copy() def to_dot_bipartite(G: nx.Graph, group_sizes: dict[str, int] | None = None) -> str: """Create a Graphviz DOT with clusters left, PDFGroups right.""" clusters = [n for n, d in G.nodes(data=True) if d.get("bipartite") == "cluster"] groups = [n for n, d in G.nodes(data=True) if d.get("bipartite") == "pdf_group"] clusters = sorted(clusters) groups = sorted(groups) def glabel(n: str) -> str: return n.replace("PDFGroup:", "") lines: list[str] = [] lines.append("graph G {") lines.append(' graph [bgcolor="white", overlap=false, splines=true, ranksep=1.0, nodesep=0.6];') lines.append(' node [style=filled, fontname="Helvetica"];') lines.append(' edge [color="#2b2b2b", penwidth=1.0, fontname="Helvetica"];') # left rank lines.append(" { rank=min;") for c in clusters: safe = str(c).replace('"', "'") lines.append(f' "{safe}" [shape=circle, width=0.45, fixedsize=true, fontsize=10, fillcolor="#2E6F9E"];') lines.append(" }") # right rank lines.append(" { rank=max;") for g in groups: lbl = glabel(g).replace('"', "'") if group_sizes and g in group_sizes: n = group_sizes[g] width = 0.55 + min(1.30, (n / 40.0)) else: width = 0.75 lines.append(f' "{g}" [label="{lbl}", shape=box, width={width:.2f}, height=0.35, fixedsize=true, fontsize=10, fillcolor="#F2B134"];') lines.append(" }") for u, v, d in G.edges(data=True): w = float(d.get("weight", 1.0)) pw = 0.8 + 0.22 * w if w >= 5: lines.append(f' "{u}" -- "{v}" [penwidth={pw:.2f}, label="{int(w)}", fontsize=9];') else: lines.append(f' "{u}" -- "{v}" [penwidth={pw:.2f}];') lines.append("}") return "\n".join(lines) def render_graphviz_svg(dot: str, out_svg: Path) -> bool: """Render DOT to SVG using system graphviz (dot). Returns True on success.""" import subprocess out_svg.parent.mkdir(parents=True, exist_ok=True) try: subprocess.run(["dot", "-Tsvg", "-o", str(out_svg)], input=dot.encode("utf-8"), check=True) return True except Exception: return False def render_matplotlib_bipartite(G: nx.Graph, out_png: Path) -> None: out_png.parent.mkdir(parents=True, exist_ok=True) clusters = sorted([n for n, d in G.nodes(data=True) if d.get("bipartite") == "cluster"]) groups = sorted([n for n, d in G.nodes(data=True) if d.get("bipartite") == "pdf_group"]) pos = {} for i, c in enumerate(clusters): y = 1.0 - (i / max(1, len(clusters) - 1)) pos[c] = (0.0, y) for i, g in enumerate(groups): y = 1.0 - (i / max(1, len(groups) - 1)) pos[g] = (1.0, y) plt.figure(figsize=(20, 11)) nx.draw_networkx_nodes(G, pos, nodelist=clusters, node_size=560, alpha=0.95) nx.draw_networkx_nodes(G, pos, nodelist=groups, node_size=430, alpha=0.95) widths = [0.6 + 0.22 * G[u][v]["weight"] for u, v in G.edges()] nx.draw_networkx_edges(G, pos, width=widths, alpha=0.45) labels = {c: c for c in clusters} labels.update({g: g.replace("PDFGroup:", "") for g in groups}) nx.draw_networkx_labels(G, pos, labels=labels, font_size=9) plt.title("Topic 10 Run 01 — Cluster ↔ PDFGroup bipartite layout") plt.axis("off") plt.tight_layout() plt.savefig(out_png, dpi=240) plt.close() def render_pyvis_html(G: nx.Graph, out_html: Path) -> bool: try: from pyvis.network import Network # type: ignore except Exception: return False out_html.parent.mkdir(parents=True, exist_ok=True) net = Network(height="900px", width="100%", bgcolor="#ffffff", font_color="#111", notebook=False, directed=False) net.barnes_hut(gravity=-20000, central_gravity=0.25, spring_length=140, spring_strength=0.02, damping=0.09) for n, d in G.nodes(data=True): if d.get("bipartite") == "pdf_group": label = n.replace("PDFGroup:", "") net.add_node(n, label=label, shape="box") else: net.add_node(n, label=str(n), shape="dot") for u, v, d in G.edges(data=True): w = float(d.get("weight", 1.0)) net.add_edge(u, v, value=w, title=f"docs={int(w)}") net.show(str(out_html)) return True def main() -> None: ap = argparse.ArgumentParser() ap.add_argument("--min_edge_weight", type=int, default=1, help="Hide edges below this doc-count.") ap.add_argument("--focal", type=str, default=None, help='Focal cluster label (e.g. "MATTHEW GUERTIN").') ap.add_argument("--hops", type=int, default=1, help="Ego hops (1 = clusters↔PDFGroups only).") ap.add_argument("--interactive", action="store_true", help="Also write interactive HTML (requires pyvis).") args = ap.parse_args() doc_df = _read_csv(DOC_SUMMARY) inv_df = _read_csv(GROUP_INV) edges = build_edge_table(doc_df) G = build_graph(edges, min_edge_weight=args.min_edge_weight) group_sizes: dict[str, int] = {} if "pdf_group" in inv_df.columns and "n_docs" in inv_df.columns: for _, r in inv_df.iterrows(): group_sizes[f"PDFGroup:{r['pdf_group']}"] = int(r["n_docs"]) # global outputs dot = to_dot_bipartite(G, group_sizes=group_sizes) svg_ok = render_graphviz_svg(dot, OUT_DIR / "10r1_graph_bipartite.svg") if not svg_ok: render_matplotlib_bipartite(G, OUT_DIR / "10r1_graph_bipartite.png") if args.interactive: html_ok = render_pyvis_html(G, OUT_DIR / "10r1_graph_bipartite_interactive.html") if not html_ok: print("pyvis not installed; skipping interactive HTML. Install with: python3 -m pip install pyvis") # focal outputs if args.focal: focal = args.focal if focal not in G: m = [n for n in G.nodes() if str(n).lower() == focal.lower()] if m: focal = m[0] H = ego_subgraph(G, focal, hops=args.hops) safe = str(focal).replace(" ", "_").replace("/", "_").replace("PDFGroup:", "PDFGroup_") dot2 = to_dot_bipartite(H, group_sizes=group_sizes) svg_ok2 = render_graphviz_svg(dot2, OUT_DIR / f"10r1_graph_focal_{safe}.svg") if not svg_ok2: render_matplotlib_bipartite(H, OUT_DIR / f"10r1_graph_focal_{safe}.png") if args.interactive: render_pyvis_html(H, OUT_DIR / f"10r1_graph_focal_{safe}.html") print(f"Wrote graph outputs to: {OUT_DIR.resolve()}") if __name__ == "__main__": main()