70 lines
2.6 KiB
Python
70 lines
2.6 KiB
Python
"""Outbox-синк: PG → Qdrant + Memgraph."""
|
|
from __future__ import annotations
|
|
import json
|
|
from . import config, db, qdrant_store, memgraph_store
|
|
|
|
def sync_batch(conn, limit: int = 100) -> dict:
|
|
"""Обработать пачку необработанных outbox-записей."""
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
"SELECT id, entity_type, entity_id, op FROM pm_outbox "
|
|
"WHERE processed = false ORDER BY id LIMIT %s FOR UPDATE SKIP LOCKED",
|
|
(limit,),
|
|
)
|
|
rows = cur.fetchall()
|
|
|
|
stats = {"atom": 0, "link": 0, "error": 0}
|
|
for row in rows:
|
|
try:
|
|
if row["entity_type"] == "atom":
|
|
_sync_atom(conn, row)
|
|
stats["atom"] += 1
|
|
elif row["entity_type"] == "link":
|
|
_sync_link(conn, row)
|
|
stats["link"] += 1
|
|
with conn.cursor() as cur:
|
|
cur.execute("UPDATE pm_outbox SET processed = true WHERE id = %s", (row["id"],))
|
|
except Exception as e:
|
|
stats["error"] += 1
|
|
print(f"[outbox] error id={row['id']}: {e}")
|
|
conn.commit()
|
|
return stats
|
|
|
|
def _sync_atom(conn, row) -> None:
|
|
with conn.cursor() as cur:
|
|
if row["op"] == "delete":
|
|
qdrant_store.delete_atom(row["entity_id"])
|
|
return
|
|
cur.execute("SELECT id, type, claim, content, confidence, status, doc_id FROM pm_atom WHERE id = %s", (row["entity_id"],))
|
|
a = cur.fetchone()
|
|
if not a:
|
|
return
|
|
# 1) узел в графе (Memgraph)
|
|
memgraph_store.upsert_atom(a["id"], a["type"], a["claim"], a["status"])
|
|
# 2) вектор в Qdrant (эмбеддим claim+content — claim семантически сфокусирован)
|
|
embed_text = (a["claim"] or "") + " " + a["content"][:1500]
|
|
payload = {
|
|
"type": a["type"], "claim": a["claim"] or "",
|
|
"confidence": a["confidence"], "status": a["status"],
|
|
"doc_id": a["doc_id"],
|
|
}
|
|
qdrant_store.upsert_atom(a["id"], embed_text, payload)
|
|
|
|
def _sync_link(conn, row) -> None:
|
|
with conn.cursor() as cur:
|
|
cur.execute("SELECT src, dst, kind FROM pm_atom_link WHERE id = %s", (row["entity_id"],))
|
|
l = cur.fetchone()
|
|
if not l:
|
|
return
|
|
memgraph_store.upsert_link(l["src"], l["dst"], l["kind"])
|
|
|
|
def sync_all(conn, batch: int = 100, max_batches: int = 100) -> dict:
|
|
total = {"atom": 0, "link": 0, "error": 0}
|
|
for _ in range(max_batches):
|
|
s = sync_batch(conn, batch)
|
|
for k in total:
|
|
total[k] += s[k]
|
|
if s["atom"] + s["link"] + s["error"] == 0:
|
|
break
|
|
return total
|