# File Objective : Build the Vantage Knowledge Graph end-to-end. # Scope : Dev script — run after ingest + tag_topics to populate Neo4j. # What it does : 1. Coverage — reads topic_coverage.csv (written by tag_topics.py). # 2. Signals — Google Trends volume + velocity per topic (anchor=calculator). # 3. Gap score — Demand × (1 − Coverage) per topic. # 4. Push — (:Topic){gap metadata} + (:Topic)-[:COVERED_BY]->(:Document). # 5. Prints top gaps with covered documents. # What it does not: Wipe the graph (run wipe_graph.py first) or re-ingest. # # Usage : python scripts/build_graph.py # all topics # SAMPLE=10 python scripts/build_graph.py # first N topics only from __future__ import annotations import csv import json import os from collections import defaultdict from pathlib import Path from vantage_core.adapters.graph_neo4j import Neo4jGraphRepo from vantage_core.adapters.signals_pytrends import fetch_signals from vantage_core.domain.gap import compute_gaps from vantage_core.models import SignalRecord from vantage_core.topic_seeds import load_topics SAMPLE = int(os.environ.get("SAMPLE", 0)) # >0 → first N topics TOPICS_ENV = os.environ.get("TOPICS", "") # comma-separated override _SIG_CACHE = Path("data/signals.json") _COVERAGE_CSV = Path("data/topic_coverage.csv") def load_coverage() -> dict[str, list[str]]: """Read topic_coverage.csv → {topic: [doc_id, ...]}. Written by tag_topics.py.""" coverage: dict[str, list[str]] = defaultdict(list) with _COVERAGE_CSV.open(encoding="utf-8") as f: for row in csv.DictReader(f): coverage[row["topic"]].append(row["doc_id"]) return dict(coverage) def load_signals(topics: list[str]) -> list[SignalRecord]: """Return signals from disk cache if present; otherwise fetch live and save cache.""" if _SIG_CACHE.exists(): print(f" Using cached signals ({_SIG_CACHE})") data = json.loads(_SIG_CACHE.read_text(encoding="utf-8")) cached = {r["keyword"]: SignalRecord(**r) for r in data["signals"]} # fetch any topics missing from cache missing = [t for t in topics if t not in cached] if missing: print(f" {len(missing)} topics not in cache — fetching live...") fresh = fetch_signals(missing) for s in fresh: cached[s.keyword] = s _save_signals_cache(list(cached.values())) return [cached[t] for t in topics if t in cached] signals = fetch_signals(topics) _save_signals_cache(signals) return signals def _save_signals_cache(signals: list[SignalRecord]) -> None: _SIG_CACHE.parent.mkdir(parents=True, exist_ok=True) _SIG_CACHE.write_text( json.dumps({"signals": [s.model_dump() for s in signals]}, indent=2), encoding="utf-8", ) print(f" Signals cached → {_SIG_CACHE} ({len(signals)} records)") def main() -> None: graph = Neo4jGraphRepo() # ── 0. Load coverage CSV, pre-filter to topics with ≥1 doc ─────────────── print(f"Loading coverage from {_COVERAGE_CSV}...") raw_coverage = load_coverage() # topic → [doc_id] if TOPICS_ENV: candidates = [t.strip() for t in TOPICS_ENV.split(",")] elif SAMPLE: candidates = load_topics()[:SAMPLE] else: candidates = load_topics() topics = [t for t in candidates if t in raw_coverage] print(f" {len(topics)}/{len(candidates)} topics have ≥1 doc\n") # cover: topic → list of {doc_id} dicts cover: dict[str, list[dict]] = {t: [{"doc_id": d} for d in raw_coverage[t]] for t in topics} print(f"Building graph for {len(topics)} topics\n") # ── 1. Signals ──────────────────────────────────────────────────────────── print("Loading signals...") signals = load_signals(topics) # ── 2. Gap scores ───────────────────────────────────────────────────────── counts = {t: len(docs) for t, docs in cover.items()} gaps = compute_gaps(signals, counts) # ── 3. Push to Neo4j: Topic nodes + COVERED_BY edges ────────────────────── print("\nPushing to Neo4j...") graph.build(gaps) edges = [ {"topic": g.topic, "doc_id": d["doc_id"], "title": ""} for g in gaps for d in cover.get(g.topic, []) ] graph.link_coverage(edges) print(f" {len(gaps)} Topic nodes + {len(edges)} COVERED_BY edges\n") # ── 4. Sample output ────────────────────────────────────────────────────── print("── Top 10 gaps ──────────────────────────────────────────────────────") for g in gaps[:10]: print(f" gap={g.gap_score:.4f} {g.label:<13} vol={g.volume:.2f} vel={g.velocity:+.2f} " f"cov={g.coverage_count}docs {g.topic}") for d in graph.docs_for_topic(g.topic, limit=2): print(f" └─ [{d['doc_id']}] {d['title'][:60]}") graph.close() if __name__ == "__main__": main()