sysops
131 lines
4.8 KiB
Python
131 lines
4.8 KiB
Python
"""Automatische NAS + second-brain synchronisatie."""
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import logging
|
|
import os
|
|
from typing import Any
|
|
|
|
import httpx
|
|
|
|
from app.db import execute, fetch_all, fetch_one
|
|
|
|
log = logging.getLogger("cockpit.auto_ingest")
|
|
|
|
DOC_INGEST_URL = os.getenv("DOC_INGEST_URL", "http://10.4.7.19:8750")
|
|
TOOLS_API_URL = os.getenv("TOOLS_API_URL", "http://tools-api:8700")
|
|
|
|
|
|
def _log_run(source: str, status: str, details: dict[str, Any]) -> None:
|
|
try:
|
|
execute(
|
|
"INSERT INTO ingest_automation_log (source, status, details) VALUES (%s, %s, %s::jsonb)",
|
|
(source, status, json.dumps(details)),
|
|
)
|
|
except Exception as exc:
|
|
log.warning("ingest log failed: %s", exc)
|
|
|
|
|
|
def _brain_chat_id_for_client(client_id: int) -> int:
|
|
return -int(client_id)
|
|
|
|
|
|
async def sync_nas_to_brain(limit: int = 40) -> dict[str, Any]:
|
|
"""Indexeer geanalyseerde documenten in second brain (pgvector) per klant."""
|
|
rows = fetch_all(
|
|
"""
|
|
SELECT da.storage_path, da.filename, da.doc_type, da.sentiment_label,
|
|
da.word_count, dl.client_id, c.name AS client_name
|
|
FROM document_analytics da
|
|
LEFT JOIN document_links dl ON (
|
|
da.storage_path = dl.storage_path
|
|
OR da.storage_path LIKE dl.storage_path || '/%'
|
|
) AND dl.is_folder = TRUE
|
|
LEFT JOIN document_links dl2 ON da.storage_path = dl2.storage_path AND dl2.is_folder = FALSE
|
|
LEFT JOIN clients c ON c.id = COALESCE(dl2.client_id, dl.client_id)
|
|
WHERE da.storage_path IS NOT NULL
|
|
ORDER BY da.analyzed_at DESC NULLS LAST
|
|
LIMIT %s
|
|
""",
|
|
(max(1, min(limit, 200)),),
|
|
)
|
|
synced = 0
|
|
errors = 0
|
|
async with httpx.AsyncClient(timeout=120.0) as client:
|
|
for row in rows:
|
|
path = row.get("storage_path") or ""
|
|
if not path:
|
|
continue
|
|
client_id = row.get("client_id")
|
|
chat_id = _brain_chat_id_for_client(client_id) if client_id else 888_000_001
|
|
text = (
|
|
f"NAS document: {row.get('filename') or path}\n"
|
|
f"Pad: {path}\nType: {row.get('doc_type')}\n"
|
|
f"Sentiment: {row.get('sentiment_label')}\nWoorden: {row.get('word_count')}"
|
|
)
|
|
msg_hash = int(hashlib.md5(path.encode()).hexdigest()[:8], 16)
|
|
try:
|
|
r = await client.post(
|
|
f"{TOOLS_API_URL.rstrip('/')}/brain/messages",
|
|
json={
|
|
"chat_id": chat_id,
|
|
"direction": "inbound",
|
|
"role": "system",
|
|
"content_type": "document",
|
|
"content_text": text[:4000],
|
|
"agent_name": "auto-ingest",
|
|
"embed": True,
|
|
"chat_type": "client_brain" if client_id else "nas_corpus",
|
|
"user_name": row.get("client_name") or "NAS",
|
|
"content_json": {"storage_path": path, "client_id": client_id},
|
|
"telegram_message_id": msg_hash,
|
|
},
|
|
)
|
|
if r.status_code < 400:
|
|
synced += 1
|
|
else:
|
|
errors += 1
|
|
except Exception as exc:
|
|
log.warning("brain sync %s: %s", path, exc)
|
|
errors += 1
|
|
return {"synced": synced, "errors": errors, "candidates": len(rows)}
|
|
|
|
|
|
async def run_full_auto_sync(force_scan: bool = False) -> dict[str, Any]:
|
|
"""Volledige pipeline: NAS scan → Chroma RAG → brain embeddings."""
|
|
out: dict[str, Any] = {"ok": True, "steps": {}}
|
|
try:
|
|
async with httpx.AsyncClient(timeout=300.0) as client:
|
|
r = await client.post(
|
|
f"{DOC_INGEST_URL.rstrip('/')}/ingest/scan",
|
|
params={"force": "true" if force_scan else "false"},
|
|
)
|
|
r.raise_for_status()
|
|
out["steps"]["nas_scan"] = r.json()
|
|
except Exception as exc:
|
|
out["steps"]["nas_scan"] = {"error": str(exc)}
|
|
brain = await sync_nas_to_brain()
|
|
out["steps"]["brain_sync"] = brain
|
|
status = "ok" if not out["steps"].get("nas_scan", {}).get("error") else "partial"
|
|
_log_run("auto_sync", status, out)
|
|
out["last_logged"] = True
|
|
return out
|
|
|
|
|
|
def last_auto_sync() -> dict[str, Any] | None:
|
|
row = fetch_one(
|
|
"SELECT source, status, details, created_at FROM ingest_automation_log ORDER BY created_at DESC LIMIT 1"
|
|
)
|
|
if not row:
|
|
return None
|
|
d = dict(row)
|
|
if hasattr(d.get("created_at"), "isoformat"):
|
|
d["created_at"] = d["created_at"].isoformat()
|
|
if isinstance(d.get("details"), str):
|
|
try:
|
|
d["details"] = json.loads(d["details"])
|
|
except Exception:
|
|
pass
|
|
return d
|