content-atom/stage4/graph_outbox.py

#!/usr/bin/env python3
"""
MVP stage 4: graph/lineage — обход связей атомов через Apache AGE.

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 → 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")),
    "user": os.environ.get("POSTGRES_USER", "atom"),
    "dbname": os.environ.get("POSTGRES_DB", "content_atom"),
    "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"),
}

GRAPH_NAME = "content_atoms_graph"


def get_relations() -> list[dict]:
    """Связи из atom-db (Postgres 5434)."""
    conn = psycopg2.connect(**ATOM_DB)
    with conn.cursor() as cur:
        cur.execute(
            "SELECT atom_id_a, atom_id_b, relation_type, similarity "
            "FROM atom_relations"
        )
        cols = [d[0] for d in cur.description]
        rows = [dict(zip(cols, r)) for r in cur.fetchall()]
    conn.close()
    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, если нет."""
    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")

    # Удалить старые вершины/рёбра для идемпотентности
    # AGE: рёбра нужно удалить до вершин → DETACH DELETE
    _cypher(
        "SET search_path TO ag_catalog;"
        f"SELECT * FROM cypher('{GRAPH_NAME}', $$ "
        f"MATCH (v) DETACH 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) через Cypher."""
    relations = get_relations()
    print(f"  ✓ {len(relations)} связных атомов")
    for rel in relations:
        print(
            f"    • {rel['atom_id_a'][:8]} → "
            f"{rel['relation_type']} → {rel['atom_id_b'][:8]}"
        )


def main():
    action = sys.argv[1] if len(sys.argv) > 1 else "all"

    ensure_graph()

    if action in ("sync", "all", "--all"):
        sync_relations()
    if action in ("query", "all", "--all"):
        query_connected()

    print(f"\nMVP stage 4 graph OK: action={action}")
    return 0


if __name__ == "__main__":
    sys.exit(main())