# -*- coding: utf-8 -*- """Paperless → Qdrant indexing pipeline.""" import logging import zlib from concurrent.futures import ThreadPoolExecutor from typing import Optional LOGGER = logging.getLogger("loogle_mcp.indexer") # Indicizzazioni avviate dai tool MCP: fuori dalla richiesta, una alla volta (thermal gate) _background = ThreadPoolExecutor(max_workers=1, thread_name_prefix="bg-index") from ..db import get_conn from . import embeddings, gitea_indexer, paperless, qdrant_store from .text_chunk import chunk_text as _chunk_text def index_document(doc_id: int, force: bool = False, paperless_user: Optional[str] = None) -> dict: conn = get_conn() existing = conn.execute( "SELECT doc_id FROM indexed_documents WHERE doc_id=?", (doc_id,) ).fetchone() if existing and not force: return {"doc_id": doc_id, "skipped": True} doc = paperless.get_document(doc_id, username=paperless_user) text = paperless.download_document_text(doc_id, username=paperless_user) owner = paperless.document_owner(doc) visibility = paperless.document_visibility(doc, owner) chunks = _chunk_text(text) if not chunks: return {"doc_id": doc_id, "chunks": 0} vectors = embeddings.embed_texts(chunks) collection = qdrant_store.SHARED_COLLECTION if visibility == "family" else qdrant_store.kb_collection(owner) qdrant_store.delete_by_doc(collection, doc_id) ids = [] payloads = [] for i, chunk in enumerate(chunks): point_id = f"doc-{doc_id}-chunk-{i}" ids.append(point_id) payloads.append( { "doc_id": doc_id, "chunk_index": i, "title": doc.get("title") or f"Documento {doc_id}", "text": chunk, "owner": owner, "visibility": visibility, } ) qdrant_store.upsert_chunks(collection, ids, vectors, payloads) conn.execute( "INSERT INTO indexed_documents(doc_id,title,owner,visibility,chunk_count,indexed_at)" " VALUES (?,?,?,?,?,datetime('now'))" " ON CONFLICT(doc_id) DO UPDATE SET" " title=excluded.title, owner=excluded.owner, visibility=excluded.visibility," " chunk_count=excluded.chunk_count, indexed_at=datetime('now')", (doc_id, doc.get("title"), owner, visibility, len(chunks)), ) conn.commit() return {"doc_id": doc_id, "chunks": len(chunks), "collection": collection} def index_all(max_pages: int = 20) -> dict: indexed = 0 errors = 0 seen: set[int] = set() users = paperless.list_configured_users() or ["daniele"] for paperless_user in users: page = 1 while page <= max_pages: try: data = paperless.list_documents(page=page, page_size=25, username=paperless_user) except Exception as exc: LOGGER.error("Paperless list failed for %s: %s", paperless_user, exc) break results = data.get("results") or [] if not results: break for doc in results: doc_id = doc["id"] if doc_id in seen: continue seen.add(doc_id) try: index_document(doc_id, paperless_user=paperless_user) indexed += 1 except Exception as exc: LOGGER.warning("Index doc %s failed (%s): %s", doc_id, paperless_user, exc) errors += 1 if not data.get("next"): break page += 1 return {"indexed": indexed, "errors": errors, "users": users} def index_context_snippet( username: str, project_id: str, text: str, title: str, snippet_id: Optional[str] = None, ) -> None: chunks = _chunk_text(text) if not chunks: return vectors = embeddings.embed_texts(chunks) collection = qdrant_store.ctx_collection(username) doc_key = f"ctx-{username}-{project_id}" if snippet_id: doc_key = f"{doc_key}-{snippet_id}" # crc32 e non hash(): hash() cambia a ogni processo e il delete non trovava i vecchi chunk doc_id = zlib.crc32(doc_key.encode()) % (2**31) qdrant_store.delete_by_doc(collection, doc_id) ids = [] payloads = [] for i, chunk in enumerate(chunks): point_id = f"{doc_key}-chunk-{i}" ids.append(point_id) payloads.append( { "doc_id": doc_id, "project_id": project_id, "chunk_index": i, "title": title, "text": chunk, "owner": username, "visibility": "personal", "source": "context", "snippet_id": snippet_id or "latest", } ) qdrant_store.upsert_chunks(collection, ids, vectors, payloads) def index_project_context(username: str, project_id: str) -> dict: """Indicizza il context.md completo del progetto sostituendo la versione precedente. Personali → ctx_{utente}; condivisi → ctx_shared_family (visibili a tutta la famiglia). Così un contesto che cambia (es. inventario) non lascia in ricerca informazioni superate. """ from ..context import store as context_store data = context_store.get_project_context(username, project_id, session_limit=0, include_gitea=False) meta = data["meta"] shared = bool(meta.get("shared")) collection = qdrant_store.CTX_SHARED_COLLECTION if shared else qdrant_store.ctx_collection(username) namespace = context_store.SHARED_NS if shared else username drop_project_context(collection, project_id) chunks = _chunk_text(data["context_md"]) if not chunks: return {"project_id": project_id, "collection": collection, "chunks": 0} vectors = embeddings.embed_texts(chunks) doc_key = f"ctx-{namespace}-{project_id}" doc_id = zlib.crc32(doc_key.encode()) % (2**31) ids, payloads = [], [] for i, chunk in enumerate(chunks): ids.append(f"{doc_key}-chunk-{i}") payloads.append( { "doc_id": doc_id, "project_id": project_id, "chunk_index": i, "title": meta.get("title") or project_id, "text": chunk, "owner": meta.get("owner") or username, "visibility": "family" if shared else "personal", "shared": shared, "updated_by": meta.get("updated_by"), "source": "context", "snippet_id": "full", } ) qdrant_store.upsert_chunks(collection, ids, vectors, payloads) return {"project_id": project_id, "collection": collection, "chunks": len(chunks)} _DIGEST_SNIPPET_PREFIXES = ("daily-", "plan-") def drop_project_context(collection: str, project_id: str) -> int: """Rimuove tutti i chunk di un progetto (anche quelli indicizzati a pezzi in passato), tranne gli snippet del digest giornaliero (snippet_id daily-*/plan-*).""" return qdrant_store.delete_matching( collection, "project_id", project_id, lambda payload: not str(payload.get("snippet_id") or "").startswith(_DIGEST_SNIPPET_PREFIXES), ) def submit_background(label: str, fn, *args, **kwargs) -> None: """Esegue fn in background; gli errori finiscono nel log invece che al client.""" def _run(): try: fn(*args, **kwargs) LOGGER.info("Indicizzazione background completata: %s", label) except Exception: LOGGER.exception("Indicizzazione background fallita: %s", label) _background.submit(_run) def search_knowledge(username: str, query: str, limit: int = 8, is_admin: bool = False) -> list: vectors = [embeddings.embed_query(query)] collections = [ qdrant_store.SHARED_COLLECTION, qdrant_store.kb_collection(username), qdrant_store.ctx_collection(username), qdrant_store.CTX_SHARED_COLLECTION, ] collections.extend(gitea_indexer.gitea_collections_for_user(username, is_admin=is_admin)) collections.append(qdrant_store.APPS_SHARED_COLLECTION) if is_admin: for u in ("daniele", "lucia", "davide", "luca"): collections.append(qdrant_store.kb_collection(u)) collections = list(dict.fromkeys(collections)) hits = qdrant_store.search(collections, vectors[0], limit=limit) for hit in hits: if hit.get("repo") and not hit.get("source"): hit["source"] = "gitea" elif hit.get("source") in ("irrigazione", "turni"): hit["source_type"] = "apps" elif hit.get("doc_id") and not hit.get("source"): hit["source"] = "paperless" return hits def search_gitea_knowledge(username: str, query: str, limit: int = 8, is_admin: bool = False) -> list: return gitea_indexer.search_gitea_knowledge(username, query, limit=limit, is_admin=is_admin) def search_context(username: str, query: str, limit: int = 8) -> list: vectors = [embeddings.embed_query(query)] collections = [qdrant_store.ctx_collection(username), qdrant_store.CTX_SHARED_COLLECTION] return qdrant_store.search(collections, vectors[0], limit=limit) def list_recent_documents(limit: int = 20) -> list: rows = get_conn().execute( "SELECT * FROM indexed_documents ORDER BY indexed_at DESC LIMIT ?", (limit,) ).fetchall() return [dict(r) for r in rows]