loogle-scripts/services/loogle-mcp/app/knowledge/indexer.py

# -*- 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]