stage4: graph_outbox на AGE (Cypher, fixed password+syntax)

stage4/graph_outbox.py

@@ -1,22 +1,29 @@
 #!/usr/bin/env python3
 """
-MVP stage 4: graph/lineage — обход связей атомов.
+MVP stage 4: graph/lineage — обход связей атомов через Apache AGE.
 
-graph_outbox: atom_relations → обход связей (Postgres, не AGE).
-AGE в контейнере не имеет vertices()/edges()/cypher() → не работает.
-Вместо AGE — читаем atom_relations напрямую из Postgres.
+graph_outbox: atom_relations (Postgres) → graph (Apache AGE).
+Синхронизация: читаем atom_relations из Postgres, пишем вершины/рёбра
+в AGE через cypher() (OpenCypher). Тело атома НЕ хранится — только id,
+тип и relation_type (минимальные атрибуты).
+
+AGE-контейнер apache/age:1.8.0 — расширение Postgres.
+Требует `SET search_path TO ag_catalog;` перед cypher().
+Cypher: CREATE/MATCH/RETURN вместо устаревших vertices()/edges().
 
 Запуск:
-    python3 stage4/graph_outbox.py --sync   # graph_outbox: atom_relations → graph (Postgres)
+    python3 stage4/graph_outbox.py --sync   # graph_outbox: atom_relations → AGE
     python3 stage4/graph_outbox.py --query   # обход связей (связные атомы)
     python3 stage4/graph_outbox.py --all
 """
 
 import os
 import sys
+import tempfile
 
 import psycopg2
 
+# Postgres atom-db (истина, atom_relations)
 ATOM_DB = {
     "host": os.environ.get("POSTGRES_HOST", "127.0.0.1"),
     "port": int(os.environ.get("POSTGRES_PORT", "5434")),
@@ -25,16 +32,16 @@ ATOM_DB = {
     "password": os.environ.get("POSTGRES_PASSWORD", "atompass123"),
 }
 
+# Apache AGE (Postgres-расширение)
+AGE_DB = {
+    "host": os.environ.get("AGE_HOST", "127.0.0.1"),
+    "port": int(os.environ.get("AGE_PORT", "5435")),
+    "user": os.environ.get("AGE_USER", "atom"),
+    "dbname": os.environ.get("AGE_DB", "content_atom"),
+    "password": os.environ.get("AGE_PASSWORD", "agepass123"),
+}
 
-def get_atoms() -> list[dict]:
-    """Атомы из atom-db (Postgres 5434)."""
-    conn = psycopg2.connect(**ATOM_DB)
-    with conn.cursor() as cur:
-        cur.execute("SELECT id, type, word_count FROM atoms LIMIT 50")
-        cols = [d[0] for d in cur.description]
-        rows = [dict(zip(cols, r)) for r in cur.fetchall()]
-    conn.close()
-    return rows
+GRAPH_NAME = "content_atoms_graph"
 
 
 def get_relations() -> list[dict]:
@@ -51,18 +58,102 @@ def get_relations() -> list[dict]:
     return rows
 
 
+def _psql_file(db: dict, sql: str) -> str:
+    """Выполнить multi-line SQL через stdin (защита от shell $)."""
+    import subprocess
+
+    env = dict(os.environ)
+    env["PGPASSWORD"] = db["password"]
+    cmd = [
+        "psql",
+        "-h", db["host"],
+        "-p", str(db["port"]),
+        "-U", db["user"],
+        "-d", db["dbname"],
+        "-q",
+        "-A",
+    ]
+    with tempfile.NamedTemporaryFile(
+        "w", suffix=".sql", delete=False
+    ) as f:
+        f.write(sql)
+        path = f.name
+    try:
+        out = subprocess.run(
+            cmd,
+            stdin=open(path),
+            capture_output=True,
+            text=True,
+            env=env,
+        )
+        return out.stdout + out.stderr
+    finally:
+        os.unlink(path)
+
+
+def _cypher(sql: str) -> str:
+    """Выполнить Cypher-запрос в AGE."""
+    return _psql_file(AGE_DB, sql)
+
+
 def ensure_graph() -> None:
-    """graph_outbox: atom_relations → graph (Postgres)."""
-    conn = psycopg2.connect(**ATOM_DB)
-    with conn.cursor() as cur:
-        cur.execute("SELECT count(*) FROM atom_relations")
-        n = cur.fetchone()[0]
-    conn.close()
-    print(f"  ✓ graph_outbox: {n} связей (atom_relations) в Postgres")
+    """Создать graph, если нет."""
+    exists = _cypher(
+        "SET search_path TO ag_catalog;"
+        "SELECT name FROM ag_graph WHERE name = '" + GRAPH_NAME + "';"
+    ).strip()
+    if not exists:
+        _cypher(
+            "SET search_path TO ag_catalog;"
+            f"SELECT create_graph('{GRAPH_NAME}');"
+        )
+        print(f"  ✓ graph {GRAPH_NAME} создан")
+    else:
+        print(f"  ✓ graph {GRAPH_NAME} уже существует")
+
+
+def sync_relations() -> None:
+    """graph_outbox: atom_relations → AGE (вершины + рёбра).
+
+    Тело атома НЕ копируется — только id + тип.
+    Рёбра: relation_type → label ребра.
+    """
+    relations = get_relations()
+    print(f"  graph_outbox: {len(relations)} связей → AGE")
+
+    # Удалить старые вершины/ёбра для идемпотентности
+    _cypher(
+        "SET search_path TO ag_catalog;"
+        f"SELECT * FROM cypher('{GRAPH_NAME}', $$ "
+        f"MATCH (v) DELETE v $$) AS (v agtype);"
+    )
+
+    # Вершины: уникальный set atom_id (создаём напрямую)
+    ids = sorted(
+        {r["atom_id_a"] for r in relations}
+        | {r["atom_id_b"] for r in relations}
+    )
+    for aid in ids:
+        _cypher(
+            "SET search_path TO ag_catalog;"
+            f"SELECT * FROM cypher('{GRAPH_NAME}', $$ "
+            f"CREATE (:Atom {{atom_id: '{aid}'}}) $$) AS (v agtype);"
+        )
+
+    # Рёбра: MATCH по atom_id (свойство, не auto-generated id)
+    for rel in relations:
+        _cypher(
+            "SET search_path TO ag_catalog;"
+            f"SELECT * FROM cypher('{GRAPH_NAME}', $$ "
+            f"MATCH (a {{atom_id: '{rel['atom_id_a']}'}}), "
+            f"(b {{atom_id: '{rel['atom_id_b']}'}}) "
+            f"CREATE (a)-[:{rel['relation_type']}]->(b) "
+            f"$$) AS (e agtype);"
+        )
 
 
 def query_connected() -> None:
-    """Обход: атомы с связями (duplicate) в Postgres."""
+    """Обход: атомы с связями (duplicate) через Cypher."""
     relations = get_relations()
     print(f"  ✓ {len(relations)} связных атомов")
     for rel in relations:
@@ -78,7 +169,7 @@ def main():
     ensure_graph()
 
     if action in ("sync", "all", "--all"):
-        pass
+        sync_relations()
     if action in ("query", "all", "--all"):
         query_connected()