mirror of
https://github.com/allaunthefox/Research-Stack.git
synced 2026-07-31 03:05:21 +00:00
allaun
513d2484a3
- distribute_extraction.py: add gremlin_escape() for special chars, add --batch-size async mode with submitAsync() - load_cornfield.py: pipe SQL via stdin to avoid shell metachar issues - populate_ene_tables.py: same stdin pipe fix for neon SSH calls
291 lines
11 KiB
Python
291 lines
11 KiB
Python
#!/usr/bin/env -S uv run
|
|
# INFRA:LIVE postgres neon-64gb
|
|
# INFRA:LIVE gremlin azure-cosmos
|
|
# /// script
|
|
# requires-python = ">=3.11"
|
|
# dependencies = ["gremlinpython", "python-dotenv"]
|
|
# ///
|
|
"""
|
|
load_cornfield.py — Load cornfield_concepts.json into Postgres math_objects
|
|
and Gremlin concepts graph.
|
|
|
|
Postgres: INSERT INTO arxiv.math_objects (upsert on id)
|
|
Gremlin: upsert vertex (label=concept, cohort=cornfield) + depends_on edges
|
|
to existing modules where lean_name resolves.
|
|
|
|
Run: uv run scripts/load_cornfield.py
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
from pathlib import Path
|
|
from dotenv import load_dotenv
|
|
|
|
ROOT = Path(__file__).resolve().parent.parent
|
|
CORNFIELD_FILE = ROOT / "extraction" / "cornfield_concepts.json"
|
|
ENV_FILE = ROOT / ".env.gremlin"
|
|
load_dotenv(ENV_FILE)
|
|
|
|
NEON_HOST = "neon-64gb"
|
|
CONTAINER = "arxiv-pg"
|
|
|
|
# ── Postgres helpers ──────────────────────────────────────────────────────────
|
|
|
|
def neon(sql: str, timeout: int = 60) -> str:
|
|
"""Run SQL on neon-64gb arxiv DB, return stdout. Uses stdin pipe to avoid shell escaping."""
|
|
r = subprocess.run(
|
|
["ssh", NEON_HOST,
|
|
f"podman exec -i {CONTAINER} psql -U postgres -d arxiv -t -A"],
|
|
input=sql, capture_output=True, text=True, timeout=timeout,
|
|
)
|
|
if r.returncode != 0:
|
|
print(f" NEON ERR: {r.stderr[:200]}")
|
|
return r.stdout.strip()
|
|
|
|
|
|
def pg_array(lst: list) -> str:
|
|
"""Format a Python list as Postgres text[] literal."""
|
|
if not lst:
|
|
return "ARRAY[]::text[]"
|
|
escaped = [s.replace("'", "''") for s in lst]
|
|
return "ARRAY['" + "','".join(escaped) + "']"
|
|
|
|
|
|
def upsert_math_object(concept: dict):
|
|
cid = concept["id"].replace("'", "''")
|
|
ctype = concept["type"].replace("'", "''")
|
|
name = concept["name"].replace("'", "''")
|
|
expr = concept.get("expression", "").replace("'", "''")
|
|
latex = concept.get("latex", "").replace("'", "''")
|
|
desc = concept.get("description", "").replace("'", "''")
|
|
src = concept.get("source_file", "").replace("'", "''")
|
|
lean_name = concept.get("lean_name", "").replace("'", "''")
|
|
lean_st = concept.get("lean_status", "").replace("'", "''")
|
|
tags = pg_array(concept.get("tags", []))
|
|
related = pg_array(concept.get("related", []))
|
|
|
|
sql = f"""
|
|
INSERT INTO math_objects (id, type, name, expression, latex, description,
|
|
source_file, lean_name, lean_status, tags, related)
|
|
VALUES (
|
|
'{cid}', '{ctype}', '{name}', '{expr}', '{latex}', '{desc}',
|
|
'{src}', '{lean_name}', '{lean_st}', {tags}, {related}
|
|
)
|
|
ON CONFLICT (id) DO UPDATE SET
|
|
type = EXCLUDED.type,
|
|
name = EXCLUDED.name,
|
|
expression = EXCLUDED.expression,
|
|
latex = EXCLUDED.latex,
|
|
description = EXCLUDED.description,
|
|
source_file = EXCLUDED.source_file,
|
|
lean_name = EXCLUDED.lean_name,
|
|
lean_status = EXCLUDED.lean_status,
|
|
tags = EXCLUDED.tags,
|
|
related = EXCLUDED.related;
|
|
""".replace("\n", " ").strip()
|
|
|
|
result = neon(sql)
|
|
return result
|
|
|
|
|
|
# ── Gremlin helpers ───────────────────────────────────────────────────────────
|
|
|
|
def make_gremlin_client():
|
|
from gremlin_python.driver import client as gc, serializer
|
|
return gc.Client(
|
|
os.environ["GREMLIN_ENDPOINT"], "g",
|
|
username=os.environ["GREMLIN_USERNAME"],
|
|
password=os.environ["GREMLIN_PASSWORD"],
|
|
message_serializer=serializer.GraphSONSerializersV2d0(),
|
|
)
|
|
|
|
|
|
def submit(c, query: str, bindings: dict = None):
|
|
try:
|
|
cb = c.submitAsync(query, bindings or {})
|
|
return cb.result(timeout=30).all().result()
|
|
except Exception as e:
|
|
print(f" GREMLIN ERR: {e!s:.120}")
|
|
return None
|
|
|
|
|
|
def safe_id(s: str) -> str:
|
|
return s.replace("/", ".").replace("\\", ".").replace("|", ".").replace(" ", "_")
|
|
|
|
|
|
def upsert_gremlin_vertex(c, concept: dict):
|
|
vid = safe_id(concept["id"])
|
|
label = "concept"
|
|
name = concept["name"].replace("'", "\\'")
|
|
cohort = "cornfield"
|
|
status = concept.get("lean_status", "").replace("'", "\\'")
|
|
action = concept.get("silversight_action", "").replace("'", "\\'")
|
|
wow = concept.get("wow_note", "").replace("'", "\\'")[:200]
|
|
tags = ",".join(concept.get("tags", []))
|
|
|
|
q = (
|
|
f"g.V().has('{label}','id',idv).fold()"
|
|
f".coalesce(unfold(),"
|
|
f" addV('{label}').property('id',idv).property('pk',idv)"
|
|
f")"
|
|
f".property('name',namev)"
|
|
f".property('cohort',cohortv)"
|
|
f".property('lean_status',statusv)"
|
|
f".property('silversight_action',actionv)"
|
|
f".property('wow_note',wowv)"
|
|
f".property('tags',tagsv)"
|
|
)
|
|
b = {
|
|
"idv": vid, "namev": name, "cohortv": cohort,
|
|
"statusv": status, "actionv": action, "wowv": wow, "tagsv": tags,
|
|
}
|
|
return submit(c, q, b)
|
|
|
|
|
|
def add_gremlin_edge(c, src_id: str, dst_id: str, label: str):
|
|
src = safe_id(src_id)
|
|
dst = safe_id(dst_id)
|
|
lbl = label.replace("'", "\\'")
|
|
q = (
|
|
f"g.V().has('id','{src}').as('s')"
|
|
f".V().has('id','{dst}').as('d')"
|
|
f".coalesce("
|
|
f" select('s').outE('{lbl}').where(inV().as('d')),"
|
|
f" addE('{lbl}').from('s').to('d')"
|
|
f")"
|
|
)
|
|
return submit(c, q)
|
|
|
|
|
|
# ── ENE table loader ──────────────────────────────────────────────────────────
|
|
|
|
def upsert_ene_relation(concept: dict):
|
|
"""Add concept to ene.relations with upstream/downstream edges."""
|
|
cid = concept["id"].replace("'", "''")
|
|
name = concept["name"].replace("'", "''")
|
|
desc = concept.get("description", "")[:300].replace("'", "''")
|
|
tags = pg_array(concept.get("tags", []))
|
|
|
|
# Insert into ene.relations (id, name, description, tags, kind)
|
|
sql = f"""
|
|
INSERT INTO ene.relations (id, name, description, tags, kind)
|
|
VALUES ('{cid}', '{name}', '{desc}', {tags}, 'cornfield_concept')
|
|
ON CONFLICT (id) DO UPDATE SET
|
|
name = EXCLUDED.name,
|
|
description = EXCLUDED.description,
|
|
tags = EXCLUDED.tags,
|
|
kind = EXCLUDED.kind;
|
|
""".replace("\n", " ").strip()
|
|
|
|
# Check if ene.relations has the right columns first
|
|
check = neon("SELECT column_name FROM information_schema.columns WHERE table_schema='ene' AND table_name='relations' LIMIT 5")
|
|
if "id" not in check:
|
|
print(f" SKIP ene.relations — schema mismatch (cols: {check[:80]})")
|
|
return
|
|
return neon(sql)
|
|
|
|
|
|
# ── Main ──────────────────────────────────────────────────────────────────────
|
|
|
|
def main():
|
|
print("=" * 60)
|
|
print(" Cornfield Concept Loader")
|
|
print("=" * 60)
|
|
|
|
data = json.loads(CORNFIELD_FILE.read_text())
|
|
concepts = data["concepts"]
|
|
print(f"\n {len(concepts)} concepts to load\n")
|
|
|
|
# ── Phase 1: Postgres math_objects ────────────────────────────────────────
|
|
print("── Phase 1: Postgres math_objects ──")
|
|
pg_ok = 0
|
|
for c in concepts:
|
|
result = upsert_math_object(c)
|
|
status = "OK" if "INSERT" in result or "UPDATE" in result or result == "" else f"?({result[:40]})"
|
|
print(f" [{status:6s}] {c['id']}")
|
|
if "ERR" not in status:
|
|
pg_ok += 1
|
|
print(f" {pg_ok}/{len(concepts)} rows upserted to math_objects\n")
|
|
|
|
# Verify
|
|
count = neon("SELECT COUNT(*) FROM math_objects WHERE id LIKE 'cf_%'")
|
|
print(f" math_objects WHERE id LIKE 'cf_%': {count} rows\n")
|
|
|
|
# ── Phase 2: Gremlin vertices + edges ─────────────────────────────────────
|
|
print("── Phase 2: Gremlin concept vertices ──")
|
|
try:
|
|
c = make_gremlin_client()
|
|
except Exception as e:
|
|
print(f" SKIP Gremlin (no client): {e}")
|
|
c = None
|
|
|
|
if c:
|
|
g_ok = 0
|
|
for concept in concepts:
|
|
r = upsert_gremlin_vertex(c, concept)
|
|
status = "OK" if r is not None else "ERR"
|
|
print(f" [{status}] {concept['id']}")
|
|
if status == "OK":
|
|
g_ok += 1
|
|
time.sleep(0.1) # RU budget
|
|
|
|
print(f"\n {g_ok}/{len(concepts)} vertices upserted to Gremlin\n")
|
|
|
|
# ── Phase 3: depends_on edges (cornfield → active modules) ────────────
|
|
print("── Phase 3: depends_on edges ──")
|
|
# Map lean_name prefixes to known active module IDs
|
|
lean_module_map = {
|
|
"Semantics.NNonEuclideanGeometry": "Semantics.Semantics.NNonEuclideanGeometry",
|
|
"Semantics.SSMS_nD": "Semantics.Semantics.SSMS_nD",
|
|
"Semantics.GraphRank": "Semantics.Semantics.GraphRank",
|
|
"Semantics.Spectrum": "Semantics.Semantics.Spectrum",
|
|
"Semantics.LandauerCompression": "Semantics.Semantics.LandauerCompression",
|
|
"Semantics.MorphicDSP": "Semantics.Semantics.MorphicDSP",
|
|
"Semantics.AngrySphinx": "Semantics.Semantics.AngrySphinx",
|
|
}
|
|
edge_count = 0
|
|
for concept in concepts:
|
|
src = safe_id(concept["id"])
|
|
# Edge to active module if lean_name resolves
|
|
for prefix, module_id in lean_module_map.items():
|
|
if prefix in concept.get("lean_name", ""):
|
|
r = add_gremlin_edge(c, src, module_id, "depends_on")
|
|
if r is not None:
|
|
print(f" {src} --depends_on--> {module_id}")
|
|
edge_count += 1
|
|
# Edges to related concepts
|
|
for rel_id in concept.get("related", []):
|
|
if rel_id.startswith("cf_"):
|
|
r = add_gremlin_edge(c, src, safe_id(rel_id), "relates_to")
|
|
if r is not None:
|
|
edge_count += 1
|
|
elif rel_id.startswith("math_"):
|
|
r = add_gremlin_edge(c, src, rel_id, "relates_to")
|
|
if r is not None:
|
|
edge_count += 1
|
|
time.sleep(0.05)
|
|
|
|
print(f"\n {edge_count} edges added\n")
|
|
|
|
# Final count
|
|
count = submit(c, "g.V().has('cohort','cornfield').count()")
|
|
print(f" Gremlin cornfield vertices: {count}\n")
|
|
c.close()
|
|
|
|
# ── Phase 4: ENE relations table ──────────────────────────────────────────
|
|
print("── Phase 4: ENE relations table ──")
|
|
ene_ok = 0
|
|
for concept in concepts:
|
|
result = upsert_ene_relation(concept)
|
|
if result is not None and "SKIP" not in str(result):
|
|
ene_ok += 1
|
|
print(f" {ene_ok}/{len(concepts)} rows attempted in ene.relations\n")
|
|
|
|
print("Done.")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|