""" Document sync service for Library Desk. Handles indexing of Paperless-ngx documents into vectors and graph. Called by webhook when Paperless completes document processing. """ import logging import re import hashlib import uuid from typing import Optional, List from dataclasses import dataclass from src.clients.paperless_client import PaperlessClient from src.clients.qdrant_client import QdrantClientWrapper from src.clients.ollama_client import OllamaClient from src.clients.neo4j_client import Neo4jClient from src.clients.wikijs_client import WikiJSClient from src.core.multi_tenancy import get_qdrant_collection_name from src.config import Settings logger = logging.getLogger(__name__) @dataclass class IndexResult: """Result of indexing a single document.""" success: bool document_id: int title: str = "" chunks_created: int = 0 error: Optional[str] = None class DocumentSyncService: """ Service for syncing Paperless documents to Library Desk indexes. Handles: - Fetching document content from Paperless API - Chunking and embedding into Qdrant - Creating graph nodes in Neo4j """ def __init__( self, paperless_client: PaperlessClient, qdrant_client: QdrantClientWrapper, ollama_client: OllamaClient, neo4j_client: Neo4jClient, wiki_client: WikiJSClient, settings: Settings, chunk_size: int = 500, chunk_overlap: int = 50 ): self.paperless = paperless_client self.qdrant = qdrant_client self.ollama = ollama_client self.neo4j = neo4j_client self.wiki = wiki_client self.settings = settings self.chunk_size = chunk_size self.chunk_overlap = chunk_overlap def _chunk_text(self, text: str) -> List[str]: """Chunk text into overlapping segments.""" text = re.sub(r'\s+', ' ', text).strip() words = text.split() if len(words) <= self.chunk_size: return [text] if text else [] chunks = [] start = 0 while start < len(words): end = start + self.chunk_size chunk_words = words[start:end] chunks.append(' '.join(chunk_words)) start = end - self.chunk_overlap return chunks async def index_document( self, document_id: int, user: str, content: Optional[str] = None, title: Optional[str] = None, ) -> IndexResult: """ Index a single document from Paperless into vectors and graph. Args: document_id: Paperless document ID user: User identifier for multi-tenancy content: Optional document content (if provided, skip Paperless API call) title: Optional document title (if provided, skip Paperless API call) Returns: IndexResult with success status and details """ logger.info(f"Indexing document {document_id} for user {user}") try: # If content and title provided (from webhook), skip API call if content is not None and title is not None: doc_title = title doc_content = content original_filename = None correspondent = None document_type = None tags = [] else: # Fetch document from Paperless doc = await self.paperless.get_document(document_id) if not doc: return IndexResult( success=False, document_id=document_id, error="Document not found in Paperless" ) doc_title = doc.title doc_content = doc.content or "" original_filename = doc.original_file_name correspondent = doc.correspondent document_type = doc.document_type tags = doc.tags if not doc_content.strip(): logger.warning(f"Document {document_id} has no text content") return IndexResult( success=True, document_id=document_id, title=doc_title, chunks_created=0, error="No text content (possibly image/video only)" ) # Index vectors chunks_created = await self._index_vectors( document_id=document_id, title=doc_title, content=doc_content, user=user, metadata={ "paperless_id": document_id, "original_filename": original_filename, "correspondent": correspondent, "document_type": document_type, "tags": tags, } ) # Index graph node await self._index_graph( document_id=document_id, title=doc_title, content=doc_content, user=user, ) # Mark as indexed in Paperless (optional - if custom field exists) try: await self._mark_indexed(document_id) except Exception as e: logger.debug(f"Could not mark document as indexed: {e}") logger.info(f"Successfully indexed document {document_id}: {chunks_created} chunks") return IndexResult( success=True, document_id=document_id, title=doc_title, chunks_created=chunks_created ) except Exception as e: logger.error(f"Failed to index document {document_id}: {e}", exc_info=True) return IndexResult( success=False, document_id=document_id, error=str(e) ) async def _index_vectors( self, document_id: int, title: str, content: str, user: str, metadata: dict, ) -> int: """Create vector embeddings for document content.""" collection = get_qdrant_collection_name(user) self.qdrant.ensure_collection(collection) # Delete existing chunks for this document try: self.qdrant.client.delete( collection_name=collection, points_selector={ "filter": { "must": [ {"key": "doc_type", "match": {"value": "document"}}, {"key": "paperless_id", "match": {"value": document_id}}, ] } } ) except Exception as e: logger.debug(f"No existing chunks to delete: {e}") # Chunk content chunks = self._chunk_text(content) if not chunks: return 0 # Generate embeddings embeddings = await self.ollama.embed_batch(chunks) # Build points points = [] for i, (chunk, embedding) in enumerate(zip(chunks, embeddings)): point_id = str(uuid.uuid4()) content_hash = hashlib.md5(chunk.encode()).hexdigest() points.append({ "id": point_id, "vector": embedding, "payload": { "doc_type": "document", "paperless_id": document_id, "title": title, "chunk_text": chunk, "chunk_index": i, "content_hash": content_hash, **metadata } }) # Upsert to Qdrant if points: self.qdrant.client.upsert( collection_name=collection, points=points ) return len(points) async def _index_graph( self, document_id: int, title: str, content: str, user: str, ): """Create graph node for document.""" # Create Document node in Neo4j query = """ MERGE (d:Document {paperless_id: $paperless_id, user: $user}) SET d.title = $title, d.doc_type = 'document', d.updated_at = datetime() RETURN d """ await self.neo4j.execute_query( query, { "paperless_id": document_id, "user": user, "title": title, } ) # TODO: Extract entities from content and create relationships # This could use the same entity extraction as wiki pages async def _mark_indexed(self, document_id: int): """Mark document as indexed in Paperless custom field.""" # Try to update library_indexed custom field if it exists try: # Look up field ID by name (Paperless requires ID, not name) field = await self.paperless.get_custom_field_by_name("library_indexed") if field: await self.paperless.update_document( document_id=document_id, custom_fields=[{"field": field["id"], "value": True}] ) except Exception: # Field might not exist, that's OK pass