Spaces:
Sleeping
Sleeping
| """Build a heterogeneous networkx graph over the candidate pool. | |
| Vectorised: we build long-format edge DataFrames and use | |
| `nx.from_edgelist` to add edges in bulk. This is roughly 50x faster | |
| than the iterrows approach on 100k candidates. | |
| """ | |
| from __future__ import annotations | |
| import pickle | |
| from collections import defaultdict | |
| from pathlib import Path | |
| from typing import Dict, List, Tuple | |
| import networkx as nx | |
| import numpy as np | |
| import pandas as pd | |
| from .. import config | |
| from ..data import load_or_build_parquet | |
| def _safe_label(prefix: str, name: str) -> str: | |
| name = (name or "").strip() | |
| if not name: | |
| name = "unknown" | |
| return f"{prefix}::{name}" | |
| def _explode_lists(df: pd.DataFrame, col: str, prefix: str) -> pd.DataFrame: | |
| """Return a DataFrame with columns (candidate_id, neighbour).""" | |
| s = df[["candidate_id", col]].copy() | |
| s[col] = s[col].apply(lambda v: list(v) if v is not None else []) | |
| s = s.explode(col).dropna() | |
| s["neighbour"] = s[col].apply(lambda x: _safe_label(prefix, x)) | |
| s = s.drop(columns=[col]) | |
| s = s[s["neighbour"].str.split("::", n=1).str[1] != "unknown"] | |
| return s | |
| def build_graph(force: bool = False) -> nx.Graph: | |
| if not force and config.GRAPH_PICKLE.exists(): | |
| with open(config.GRAPH_PICKLE, "rb") as f: | |
| return pickle.load(f) | |
| df = load_or_build_parquet() | |
| G = nx.Graph() | |
| cand_ids = df["candidate_id"].tolist() | |
| cand_nodes = [f"cand::{c}" for c in cand_ids] | |
| G.add_nodes_from((n, {"kind": "candidate"}) for n in cand_nodes) | |
| # Skill edges | |
| print("building skill edges...") | |
| skill_df = _explode_lists(df, "skill_names", "skill") | |
| skill_edges = list(zip(skill_df["candidate_id"].map(lambda c: f"cand::{c}"), | |
| skill_df["neighbour"])) | |
| G.add_edges_from(skill_edges, kind="has_skill", weight=1.0) | |
| # Capture unique skill nodes | |
| print(f" skill edges: {len(skill_edges):,}") | |
| # Title edges | |
| print("building title edges...") | |
| title_df = _explode_lists(df, "career_titles", "title") | |
| title_edges = list(zip(title_df["candidate_id"].map(lambda c: f"cand::{c}"), | |
| title_df["neighbour"])) | |
| G.add_edges_from(title_edges, kind="held_title", weight=1.0) | |
| print(f" title edges: {len(title_edges):,}") | |
| # Company edges | |
| print("building company edges...") | |
| comp_df = _explode_lists(df, "career_companies", "company") | |
| comp_edges = list(zip(comp_df["candidate_id"].map(lambda c: f"cand::{c}"), | |
| comp_df["neighbour"])) | |
| G.add_edges_from(comp_edges, kind="worked_at", weight=1.0) | |
| print(f" company edges: {len(comp_edges):,}") | |
| # Industry edges | |
| print("building industry edges...") | |
| ind_df = _explode_lists(df, "career_industries", "industry") | |
| ind_edges = list(zip(ind_df["candidate_id"].map(lambda c: f"cand::{c}"), | |
| ind_df["neighbour"])) | |
| G.add_edges_from(ind_edges, kind="in_industry", weight=1.0) | |
| print(f" industry edges: {len(ind_edges):,}") | |
| # Skill-skill co-occurrence within same candidate | |
| print("building skill cooccurrence edges...") | |
| cand_skill: Dict[str, List[str]] = defaultdict(list) | |
| for cid, sn in skill_edges: | |
| cand_skill[cid].append(sn) | |
| cooccur_counts: Dict[Tuple[str, str], int] = defaultdict(int) | |
| for c, skills in cand_skill.items(): | |
| # dedupe within candidate | |
| skills = list(set(skills)) | |
| for i in range(len(skills)): | |
| for j in range(i + 1, len(skills)): | |
| a, b = skills[i], skills[j] | |
| if a == b: | |
| continue | |
| if a > b: | |
| a, b = b, a | |
| cooccur_counts[(a, b)] += 1 | |
| cooccur_edges = [(a, b, {"kind": "skill_cooccur", "weight": float(c)}) | |
| for (a, b), c in cooccur_counts.items()] | |
| G.add_edges_from(cooccur_edges) | |
| print(f" cooccurrence edges: {len(cooccur_edges):,}") | |
| with open(config.GRAPH_PICKLE, "wb") as f: | |
| pickle.dump(G, f, protocol=pickle.HIGHEST_PROTOCOL) | |
| return G | |
| if __name__ == "__main__": | |
| import sys, time | |
| t = time.time() | |
| G = build_graph(force=True) | |
| print(f"graph built: {G.number_of_nodes():,} nodes, {G.number_of_edges():,} edges in {time.time()-t:.1f}s") | |