feat: add Paperless-ngx document storage integration
- Add /documents router with webhook, upload, search, health endpoints - Create DocumentSyncService for indexing documents to vectors/graph - Add PaperlessClient for REST API integration - Configure dependency injection for Paperless client - Add document models for webhook payloads and responses - Event-driven architecture via Paperless workflow webhooks 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
Claude Opus 4.5
@@ -0,0 +1,488 @@
|
||||
"""
|
||||
Paperless-ngx API client for Library Desk.
|
||||
|
||||
Provides async document management via Paperless-ngx:
|
||||
- Document upload and retrieval
|
||||
- Search and filtering
|
||||
- Custom field management
|
||||
- Task status tracking
|
||||
"""
|
||||
|
||||
import httpx
|
||||
from typing import Optional, List, Dict, Any
|
||||
from dataclasses import dataclass
|
||||
import logging
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
@dataclass
|
||||
class PaperlessDocument:
|
||||
"""Represents a document from Paperless-ngx."""
|
||||
id: int
|
||||
title: str
|
||||
content: str
|
||||
created: Optional[str] = None
|
||||
modified: Optional[str] = None
|
||||
added: Optional[str] = None
|
||||
correspondent: Optional[int] = None
|
||||
document_type: Optional[int] = None
|
||||
storage_path: Optional[int] = None
|
||||
tags: List[int] = None
|
||||
archive_serial_number: Optional[int] = None
|
||||
original_file_name: Optional[str] = None
|
||||
archived_file_name: Optional[str] = None
|
||||
custom_fields: List[Dict[str, Any]] = None
|
||||
|
||||
def __post_init__(self):
|
||||
if self.tags is None:
|
||||
self.tags = []
|
||||
if self.custom_fields is None:
|
||||
self.custom_fields = []
|
||||
|
||||
|
||||
@dataclass
|
||||
class SearchHit:
|
||||
"""Search result with relevance info."""
|
||||
document: PaperlessDocument
|
||||
score: float
|
||||
rank: int
|
||||
highlights: Optional[str] = None
|
||||
|
||||
|
||||
class PaperlessClient:
|
||||
"""
|
||||
Paperless-ngx REST API client.
|
||||
|
||||
Documentation: https://docs.paperless-ngx.com/api/
|
||||
"""
|
||||
|
||||
def __init__(self, base_url: str, token: str, timeout: int = 30):
|
||||
"""
|
||||
Initialize Paperless-ngx client.
|
||||
|
||||
Args:
|
||||
base_url: Paperless-ngx base URL (e.g., "http://paperless:8000")
|
||||
token: API token for authentication
|
||||
timeout: Request timeout in seconds
|
||||
"""
|
||||
self.base_url = base_url.rstrip("/")
|
||||
self.api_url = f"{self.base_url}/api"
|
||||
self.headers = {
|
||||
"Authorization": f"Token {token}",
|
||||
"Accept": "application/json",
|
||||
}
|
||||
self.client = httpx.AsyncClient(timeout=float(timeout), headers=self.headers)
|
||||
logger.info(f"Initialized Paperless client: {base_url}")
|
||||
|
||||
async def close(self):
|
||||
"""Close HTTP client."""
|
||||
await self.client.aclose()
|
||||
|
||||
# =========================================================================
|
||||
# Document Operations
|
||||
# =========================================================================
|
||||
|
||||
async def get_document(self, document_id: int) -> Optional[PaperlessDocument]:
|
||||
"""
|
||||
Get a document by ID.
|
||||
|
||||
Args:
|
||||
document_id: Paperless document ID
|
||||
|
||||
Returns:
|
||||
PaperlessDocument or None if not found
|
||||
"""
|
||||
try:
|
||||
response = await self.client.get(f"{self.api_url}/documents/{document_id}/")
|
||||
response.raise_for_status()
|
||||
data = response.json()
|
||||
return self._parse_document(data)
|
||||
except httpx.HTTPStatusError as e:
|
||||
if e.response.status_code == 404:
|
||||
return None
|
||||
logger.error(f"Failed to get document {document_id}: {e}")
|
||||
raise
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to get document {document_id}: {e}")
|
||||
raise
|
||||
|
||||
async def get_document_content(self, document_id: int) -> Optional[str]:
|
||||
"""
|
||||
Get extracted text content of a document.
|
||||
|
||||
Args:
|
||||
document_id: Paperless document ID
|
||||
|
||||
Returns:
|
||||
Text content or None if not found
|
||||
"""
|
||||
doc = await self.get_document(document_id)
|
||||
return doc.content if doc else None
|
||||
|
||||
async def list_documents(
|
||||
self,
|
||||
page: int = 1,
|
||||
page_size: int = 25,
|
||||
ordering: str = "-added",
|
||||
correspondent: Optional[int] = None,
|
||||
document_type: Optional[int] = None,
|
||||
tags: Optional[List[int]] = None,
|
||||
) -> Dict[str, Any]:
|
||||
"""
|
||||
List documents with pagination and filtering.
|
||||
|
||||
Args:
|
||||
page: Page number (starts at 1)
|
||||
page_size: Results per page
|
||||
ordering: Sort order (prefix with - for descending)
|
||||
correspondent: Filter by correspondent ID
|
||||
document_type: Filter by document type ID
|
||||
tags: Filter by tag IDs
|
||||
|
||||
Returns:
|
||||
Paginated response with count, next, previous, results
|
||||
"""
|
||||
params = {
|
||||
"page": page,
|
||||
"page_size": page_size,
|
||||
"ordering": ordering,
|
||||
}
|
||||
if correspondent:
|
||||
params["correspondent__id"] = correspondent
|
||||
if document_type:
|
||||
params["document_type__id"] = document_type
|
||||
if tags:
|
||||
params["tags__id__in"] = ",".join(str(t) for t in tags)
|
||||
|
||||
try:
|
||||
response = await self.client.get(f"{self.api_url}/documents/", params=params)
|
||||
response.raise_for_status()
|
||||
data = response.json()
|
||||
return {
|
||||
"count": data.get("count", 0),
|
||||
"next": data.get("next"),
|
||||
"previous": data.get("previous"),
|
||||
"results": [self._parse_document(d) for d in data.get("results", [])],
|
||||
}
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to list documents: {e}")
|
||||
raise
|
||||
|
||||
async def search_documents(
|
||||
self,
|
||||
query: str,
|
||||
page: int = 1,
|
||||
page_size: int = 25,
|
||||
) -> List[SearchHit]:
|
||||
"""
|
||||
Full-text search documents.
|
||||
|
||||
Args:
|
||||
query: Search query string
|
||||
page: Page number
|
||||
page_size: Results per page
|
||||
|
||||
Returns:
|
||||
List of SearchHit with document and relevance info
|
||||
"""
|
||||
params = {
|
||||
"query": query,
|
||||
"page": page,
|
||||
"page_size": page_size,
|
||||
}
|
||||
|
||||
try:
|
||||
response = await self.client.get(f"{self.api_url}/documents/", params=params)
|
||||
response.raise_for_status()
|
||||
data = response.json()
|
||||
|
||||
results = []
|
||||
for item in data.get("results", []):
|
||||
doc = self._parse_document(item)
|
||||
hit_info = item.get("__search_hit__", {})
|
||||
results.append(SearchHit(
|
||||
document=doc,
|
||||
score=hit_info.get("score", 0.0),
|
||||
rank=hit_info.get("rank", 0),
|
||||
highlights=hit_info.get("highlights"),
|
||||
))
|
||||
return results
|
||||
except Exception as e:
|
||||
logger.error(f"Search failed for '{query}': {e}")
|
||||
raise
|
||||
|
||||
async def upload_document(
|
||||
self,
|
||||
file_content: bytes,
|
||||
filename: str,
|
||||
title: Optional[str] = None,
|
||||
correspondent: Optional[int] = None,
|
||||
document_type: Optional[int] = None,
|
||||
tags: Optional[List[int]] = None,
|
||||
custom_fields: Optional[List[Dict[str, Any]]] = None,
|
||||
) -> str:
|
||||
"""
|
||||
Upload a document to Paperless-ngx.
|
||||
|
||||
Args:
|
||||
file_content: File bytes
|
||||
filename: Original filename
|
||||
title: Document title (optional, derived from filename if not set)
|
||||
correspondent: Correspondent ID
|
||||
document_type: Document type ID
|
||||
tags: List of tag IDs
|
||||
custom_fields: List of custom field values
|
||||
|
||||
Returns:
|
||||
Task UUID for tracking consumption status
|
||||
"""
|
||||
files = {"document": (filename, file_content)}
|
||||
data = {}
|
||||
|
||||
if title:
|
||||
data["title"] = title
|
||||
if correspondent:
|
||||
data["correspondent"] = correspondent
|
||||
if document_type:
|
||||
data["document_type"] = document_type
|
||||
if tags:
|
||||
# Tags need to be sent multiple times for multiple values
|
||||
data["tags"] = tags
|
||||
if custom_fields:
|
||||
data["custom_fields"] = custom_fields
|
||||
|
||||
try:
|
||||
response = await self.client.post(
|
||||
f"{self.api_url}/documents/post_document/",
|
||||
files=files,
|
||||
data=data,
|
||||
)
|
||||
response.raise_for_status()
|
||||
result = response.json()
|
||||
task_id = result.get("task_id", "")
|
||||
logger.info(f"Uploaded document '{filename}', task_id: {task_id}")
|
||||
return task_id
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to upload document '{filename}': {e}")
|
||||
raise
|
||||
|
||||
async def get_task_status(self, task_id: str) -> Dict[str, Any]:
|
||||
"""
|
||||
Get status of a consumption task.
|
||||
|
||||
Args:
|
||||
task_id: Task UUID from upload
|
||||
|
||||
Returns:
|
||||
Task status with state, result, etc.
|
||||
"""
|
||||
try:
|
||||
response = await self.client.get(
|
||||
f"{self.api_url}/tasks/",
|
||||
params={"task_id": task_id},
|
||||
)
|
||||
response.raise_for_status()
|
||||
data = response.json()
|
||||
results = data.get("results", [])
|
||||
if results:
|
||||
return results[0]
|
||||
return {"status": "NOT_FOUND"}
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to get task status {task_id}: {e}")
|
||||
raise
|
||||
|
||||
async def update_document(
|
||||
self,
|
||||
document_id: int,
|
||||
title: Optional[str] = None,
|
||||
correspondent: Optional[int] = None,
|
||||
document_type: Optional[int] = None,
|
||||
tags: Optional[List[int]] = None,
|
||||
custom_fields: Optional[List[Dict[str, Any]]] = None,
|
||||
) -> PaperlessDocument:
|
||||
"""
|
||||
Update a document's metadata.
|
||||
|
||||
Args:
|
||||
document_id: Document ID to update
|
||||
title: New title
|
||||
correspondent: New correspondent ID
|
||||
document_type: New document type ID
|
||||
tags: New tag IDs (replaces existing)
|
||||
custom_fields: New custom field values
|
||||
|
||||
Returns:
|
||||
Updated document
|
||||
"""
|
||||
data = {}
|
||||
if title is not None:
|
||||
data["title"] = title
|
||||
if correspondent is not None:
|
||||
data["correspondent"] = correspondent
|
||||
if document_type is not None:
|
||||
data["document_type"] = document_type
|
||||
if tags is not None:
|
||||
data["tags"] = tags
|
||||
if custom_fields is not None:
|
||||
data["custom_fields"] = custom_fields
|
||||
|
||||
try:
|
||||
response = await self.client.patch(
|
||||
f"{self.api_url}/documents/{document_id}/",
|
||||
json=data,
|
||||
)
|
||||
response.raise_for_status()
|
||||
return self._parse_document(response.json())
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to update document {document_id}: {e}")
|
||||
raise
|
||||
|
||||
# =========================================================================
|
||||
# Custom Fields
|
||||
# =========================================================================
|
||||
|
||||
async def list_custom_fields(self) -> List[Dict[str, Any]]:
|
||||
"""
|
||||
List all custom fields.
|
||||
|
||||
Returns:
|
||||
List of custom field definitions
|
||||
"""
|
||||
try:
|
||||
response = await self.client.get(f"{self.api_url}/custom_fields/")
|
||||
response.raise_for_status()
|
||||
return response.json().get("results", [])
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to list custom fields: {e}")
|
||||
raise
|
||||
|
||||
async def get_custom_field_by_name(self, name: str) -> Optional[Dict[str, Any]]:
|
||||
"""
|
||||
Get a custom field by name.
|
||||
|
||||
Args:
|
||||
name: Custom field name
|
||||
|
||||
Returns:
|
||||
Custom field definition or None
|
||||
"""
|
||||
fields = await self.list_custom_fields()
|
||||
for field in fields:
|
||||
if field.get("name") == name:
|
||||
return field
|
||||
return None
|
||||
|
||||
# =========================================================================
|
||||
# Tags, Correspondents, Document Types
|
||||
# =========================================================================
|
||||
|
||||
async def list_tags(self) -> List[Dict[str, Any]]:
|
||||
"""List all tags."""
|
||||
try:
|
||||
response = await self.client.get(f"{self.api_url}/tags/")
|
||||
response.raise_for_status()
|
||||
return response.json().get("results", [])
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to list tags: {e}")
|
||||
raise
|
||||
|
||||
async def list_correspondents(self) -> List[Dict[str, Any]]:
|
||||
"""List all correspondents."""
|
||||
try:
|
||||
response = await self.client.get(f"{self.api_url}/correspondents/")
|
||||
response.raise_for_status()
|
||||
return response.json().get("results", [])
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to list correspondents: {e}")
|
||||
raise
|
||||
|
||||
async def list_document_types(self) -> List[Dict[str, Any]]:
|
||||
"""List all document types."""
|
||||
try:
|
||||
response = await self.client.get(f"{self.api_url}/document_types/")
|
||||
response.raise_for_status()
|
||||
return response.json().get("results", [])
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to list document types: {e}")
|
||||
raise
|
||||
|
||||
# =========================================================================
|
||||
# Bulk Operations
|
||||
# =========================================================================
|
||||
|
||||
async def bulk_edit(
|
||||
self,
|
||||
document_ids: List[int],
|
||||
method: str,
|
||||
parameters: Optional[Dict[str, Any]] = None,
|
||||
) -> Dict[str, Any]:
|
||||
"""
|
||||
Bulk edit documents.
|
||||
|
||||
Args:
|
||||
document_ids: List of document IDs
|
||||
method: Operation (add_tag, remove_tag, set_correspondent, etc.)
|
||||
parameters: Operation parameters
|
||||
|
||||
Returns:
|
||||
Operation result
|
||||
"""
|
||||
data = {
|
||||
"documents": document_ids,
|
||||
"method": method,
|
||||
}
|
||||
if parameters:
|
||||
data["parameters"] = parameters
|
||||
|
||||
try:
|
||||
response = await self.client.post(
|
||||
f"{self.api_url}/documents/bulk_edit/",
|
||||
json=data,
|
||||
)
|
||||
response.raise_for_status()
|
||||
return response.json()
|
||||
except Exception as e:
|
||||
logger.error(f"Bulk edit failed: {e}")
|
||||
raise
|
||||
|
||||
# =========================================================================
|
||||
# Health Check
|
||||
# =========================================================================
|
||||
|
||||
async def health_check(self) -> bool:
|
||||
"""
|
||||
Check if Paperless-ngx is responding.
|
||||
|
||||
Returns:
|
||||
True if service is healthy
|
||||
"""
|
||||
try:
|
||||
response = await self.client.get(f"{self.api_url}/", timeout=5.0)
|
||||
return response.status_code < 400
|
||||
except Exception as e:
|
||||
logger.error(f"Paperless health check failed: {e}")
|
||||
return False
|
||||
|
||||
# =========================================================================
|
||||
# Helpers
|
||||
# =========================================================================
|
||||
|
||||
def _parse_document(self, data: Dict[str, Any]) -> PaperlessDocument:
|
||||
"""Parse API response into PaperlessDocument."""
|
||||
return PaperlessDocument(
|
||||
id=data.get("id", 0),
|
||||
title=data.get("title", ""),
|
||||
content=data.get("content", ""),
|
||||
created=data.get("created"),
|
||||
modified=data.get("modified"),
|
||||
added=data.get("added"),
|
||||
correspondent=data.get("correspondent"),
|
||||
document_type=data.get("document_type"),
|
||||
storage_path=data.get("storage_path"),
|
||||
tags=data.get("tags", []),
|
||||
archive_serial_number=data.get("archive_serial_number"),
|
||||
original_file_name=data.get("original_file_name"),
|
||||
archived_file_name=data.get("archived_file_name"),
|
||||
custom_fields=data.get("custom_fields", []),
|
||||
)
|
||||
@@ -101,6 +101,11 @@ class Settings(BaseSettings):
|
||||
content_extraction_timeout: int = Field(default=5, ge=1, le=30, description="Trafilatura per-URL timeout in seconds")
|
||||
content_max_length: int = Field(default=2000, ge=500, le=10000, description="Max extracted content length per result")
|
||||
|
||||
# Paperless-ngx Configuration
|
||||
paperless_url: str = Field(default="http://paperless:8000", description="Paperless-ngx URL")
|
||||
paperless_token: str = Field(default="", description="Paperless-ngx API token")
|
||||
paperless_timeout: int = Field(default=30, ge=5, le=120, description="Paperless API timeout in seconds")
|
||||
|
||||
# Document Store Configuration
|
||||
document_store_enabled: bool = Field(default=True, description="Enable document store feature")
|
||||
document_catalog_path_prefix: str = Field(default="docs", description="Wiki path prefix for catalog pages")
|
||||
|
||||
@@ -22,6 +22,7 @@ from src.clients.wikijs_client import WikiJSClient
|
||||
from src.clients.searxng_client import SearXNGClient
|
||||
from src.clients.ollama_client import OllamaClient
|
||||
from src.clients.content_extractor import ContentExtractor
|
||||
from src.clients.paperless_client import PaperlessClient
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
@@ -155,6 +156,28 @@ def get_content_extractor() -> ContentExtractor:
|
||||
return extractor
|
||||
|
||||
|
||||
@lru_cache
|
||||
def get_paperless_client() -> PaperlessClient:
|
||||
"""
|
||||
Get Paperless-ngx client singleton.
|
||||
|
||||
Returns:
|
||||
Initialized Paperless-ngx REST API client
|
||||
|
||||
Note: Returns None-like client if paperless_token is not configured
|
||||
"""
|
||||
settings = get_settings()
|
||||
if not settings.paperless_token:
|
||||
logger.warning("Paperless token not configured - document storage disabled")
|
||||
client = PaperlessClient(
|
||||
base_url=settings.paperless_url,
|
||||
token=settings.paperless_token,
|
||||
timeout=settings.paperless_timeout
|
||||
)
|
||||
logger.debug(f"Created Paperless client: {settings.paperless_url}")
|
||||
return client
|
||||
|
||||
|
||||
# Type aliases for FastAPI endpoint dependencies
|
||||
# Usage: def my_endpoint(neo4j: Neo4jDep):
|
||||
Neo4jDep = Annotated[Neo4jClient, Depends(get_neo4j_client)]
|
||||
@@ -164,6 +187,7 @@ SearXNGDep = Annotated[SearXNGClient, Depends(get_searxng_client)]
|
||||
OllamaDep = Annotated[OllamaClient, Depends(get_ollama_client)]
|
||||
RedisDep = Annotated[aioredis.Redis, Depends(get_redis_client)]
|
||||
ContentExtractorDep = Annotated[ContentExtractor, Depends(get_content_extractor)]
|
||||
PaperlessDep = Annotated[PaperlessClient, Depends(get_paperless_client)]
|
||||
|
||||
|
||||
# Lifecycle management functions
|
||||
@@ -202,6 +226,21 @@ async def startup_clients():
|
||||
logger.error(f"✗ Ollama health check failed: {e}")
|
||||
pass
|
||||
|
||||
# Check Paperless availability
|
||||
settings = get_settings()
|
||||
if settings.paperless_token:
|
||||
try:
|
||||
paperless = get_paperless_client()
|
||||
is_healthy = await paperless.health_check()
|
||||
if is_healthy:
|
||||
logger.info(f"✓ Paperless-ngx ready: {settings.paperless_url}")
|
||||
else:
|
||||
logger.warning("✗ Paperless-ngx not responding")
|
||||
except Exception as e:
|
||||
logger.error(f"✗ Paperless health check failed: {e}")
|
||||
else:
|
||||
logger.info("○ Paperless-ngx not configured (document storage disabled)")
|
||||
|
||||
# Qdrant, Wiki.js, SearXNG are lazy-initialized
|
||||
logger.info("Service clients startup complete")
|
||||
|
||||
@@ -230,7 +269,8 @@ async def shutdown_clients():
|
||||
clients_to_close = [
|
||||
("Wiki.js", get_wikijs_client()),
|
||||
("SearXNG", get_searxng_client()),
|
||||
("Ollama", get_ollama_client())
|
||||
("Ollama", get_ollama_client()),
|
||||
("Paperless", get_paperless_client()),
|
||||
]
|
||||
|
||||
for name, client in clients_to_close:
|
||||
@@ -312,6 +352,18 @@ async def check_service_health() -> dict:
|
||||
logger.error(f"Ollama health check failed: {e}")
|
||||
health["ollama"] = False
|
||||
|
||||
# Paperless-ngx
|
||||
settings = get_settings()
|
||||
if settings.paperless_token:
|
||||
try:
|
||||
paperless = get_paperless_client()
|
||||
health["paperless"] = await paperless.health_check()
|
||||
except Exception as e:
|
||||
logger.error(f"Paperless health check failed: {e}")
|
||||
health["paperless"] = False
|
||||
else:
|
||||
health["paperless"] = None # Not configured
|
||||
|
||||
return health
|
||||
|
||||
|
||||
|
||||
+2
-1
@@ -51,7 +51,7 @@ app.add_middleware(
|
||||
from src.routers import (
|
||||
wiki, tools, graph, vector, hybrid_rag, consolidation,
|
||||
ingestion, entity_linking, webhooks, rag_search, content,
|
||||
maintenance, volatile
|
||||
maintenance, volatile, documents
|
||||
)
|
||||
|
||||
app.include_router(wiki.router)
|
||||
@@ -67,6 +67,7 @@ app.include_router(rag_search.router)
|
||||
app.include_router(content.router)
|
||||
app.include_router(maintenance.router)
|
||||
app.include_router(volatile.router)
|
||||
app.include_router(documents.router)
|
||||
|
||||
# Mount static files directory for Wiki.js integration scripts
|
||||
static_dir = Path(__file__).parent.parent / "static"
|
||||
|
||||
@@ -0,0 +1,193 @@
|
||||
"""
|
||||
Document storage models for Library Desk.
|
||||
|
||||
Models for Paperless-ngx document management, virus scanning,
|
||||
and document sync operations.
|
||||
"""
|
||||
|
||||
from pydantic import BaseModel, Field
|
||||
from typing import Dict, Any, Optional, List
|
||||
from datetime import datetime
|
||||
from enum import Enum
|
||||
|
||||
|
||||
class DocumentType(str, Enum):
|
||||
"""Types of documents supported in the document store."""
|
||||
PDF = "pdf"
|
||||
IMAGE = "image"
|
||||
VIDEO = "video"
|
||||
TEXT = "text"
|
||||
ARCHIVE = "archive"
|
||||
OTHER = "other"
|
||||
|
||||
|
||||
class SyncStatus(str, Enum):
|
||||
"""Status of document sync with Library Desk."""
|
||||
PENDING = "pending"
|
||||
INDEXED = "indexed"
|
||||
FAILED = "failed"
|
||||
SKIPPED = "skipped"
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# Document Models
|
||||
# =============================================================================
|
||||
|
||||
|
||||
class DocumentMetadata(BaseModel):
|
||||
"""Metadata for a document in Paperless-ngx."""
|
||||
paperless_id: int = Field(..., description="Paperless-ngx document ID")
|
||||
title: str = Field(..., description="Document title")
|
||||
filename: Optional[str] = Field(None, description="Original filename")
|
||||
content: Optional[str] = Field(None, description="Extracted text content")
|
||||
created: Optional[datetime] = Field(None, description="Document creation date")
|
||||
modified: Optional[datetime] = Field(None, description="Last modification date")
|
||||
added: Optional[datetime] = Field(None, description="Date added to Paperless")
|
||||
correspondent: Optional[str] = Field(None, description="Correspondent name")
|
||||
document_type: Optional[str] = Field(None, description="Document type name")
|
||||
tags: List[str] = Field(default_factory=list, description="Tag names")
|
||||
custom_fields: Dict[str, Any] = Field(default_factory=dict, description="Custom field values")
|
||||
|
||||
|
||||
class DocumentRecord(BaseModel):
|
||||
"""A document record with sync status."""
|
||||
metadata: DocumentMetadata = Field(..., description="Document metadata from Paperless")
|
||||
sync_status: SyncStatus = Field(default=SyncStatus.PENDING, description="Library Desk sync status")
|
||||
indexed_at: Optional[datetime] = Field(None, description="When indexed in Library Desk")
|
||||
collection: Optional[str] = Field(None, description="Collection name (e.g., 'fastapi-docs')")
|
||||
source_url: Optional[str] = Field(None, description="Original source URL if uploaded via HybridRAG")
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# Upload Request/Response Models
|
||||
# =============================================================================
|
||||
|
||||
|
||||
class DocumentUploadRequest(BaseModel):
|
||||
"""Request to upload a document to Paperless-ngx."""
|
||||
url: Optional[str] = Field(None, description="URL to download document from")
|
||||
title: Optional[str] = Field(None, description="Document title (derived from filename if not set)")
|
||||
collection: Optional[str] = Field(None, description="Collection to add document to")
|
||||
tags: List[str] = Field(default_factory=list, description="Tags to apply")
|
||||
correspondent: Optional[str] = Field(None, description="Correspondent name")
|
||||
document_type: Optional[str] = Field(None, description="Document type name")
|
||||
|
||||
|
||||
class DocumentUploadResponse(BaseModel):
|
||||
"""Response from document upload."""
|
||||
task_id: str = Field(..., description="Paperless task ID for tracking")
|
||||
filename: str = Field(..., description="Uploaded filename")
|
||||
message: str = Field(..., description="Status message")
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# Webhook Models
|
||||
# =============================================================================
|
||||
|
||||
|
||||
class PaperlessWebhookPayload(BaseModel):
|
||||
"""Payload from Paperless-ngx webhook."""
|
||||
document_id: int = Field(..., description="Paperless document ID")
|
||||
event: str = Field(..., description="Event type (document_added, document_updated)")
|
||||
title: Optional[str] = Field(None, description="Document title")
|
||||
correspondent: Optional[int] = Field(None, description="Correspondent ID")
|
||||
document_type: Optional[int] = Field(None, description="Document type ID")
|
||||
tags: List[int] = Field(default_factory=list, description="Tag IDs")
|
||||
|
||||
|
||||
class WebhookResponse(BaseModel):
|
||||
"""Response to webhook processing."""
|
||||
document_id: int = Field(..., description="Processed document ID")
|
||||
status: str = Field(..., description="Processing status")
|
||||
indexed: bool = Field(..., description="Whether document was indexed")
|
||||
message: Optional[str] = Field(None, description="Additional details")
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# Sync Models
|
||||
# =============================================================================
|
||||
|
||||
|
||||
class SyncRequest(BaseModel):
|
||||
"""Request to sync documents from Paperless-ngx."""
|
||||
since: Optional[datetime] = Field(None, description="Only sync documents modified after this time")
|
||||
collection: Optional[str] = Field(None, description="Only sync documents in this collection")
|
||||
limit: int = Field(default=100, ge=1, le=1000, description="Maximum documents to sync")
|
||||
force_reindex: bool = Field(default=False, description="Re-index already indexed documents")
|
||||
|
||||
|
||||
class SyncResult(BaseModel):
|
||||
"""Result of a sync operation."""
|
||||
documents_found: int = Field(..., description="Total documents matching criteria")
|
||||
documents_indexed: int = Field(..., description="Successfully indexed")
|
||||
documents_skipped: int = Field(..., description="Skipped (already indexed)")
|
||||
documents_failed: int = Field(..., description="Failed to index")
|
||||
errors: List[str] = Field(default_factory=list, description="Error messages")
|
||||
duration_seconds: float = Field(..., description="Sync duration")
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# Collection Models
|
||||
# =============================================================================
|
||||
|
||||
|
||||
class Collection(BaseModel):
|
||||
"""A logical grouping of documents."""
|
||||
name: str = Field(..., description="Collection name (e.g., 'fastapi-docs')")
|
||||
description: Optional[str] = Field(None, description="Collection description")
|
||||
document_count: int = Field(default=0, description="Number of documents")
|
||||
source: Optional[str] = Field(None, description="Source (e.g., 'github.com/tiangolo/fastapi')")
|
||||
last_sync: Optional[datetime] = Field(None, description="Last sync timestamp")
|
||||
wiki_page: Optional[str] = Field(None, description="Wiki catalog page path")
|
||||
|
||||
|
||||
class CollectionListResponse(BaseModel):
|
||||
"""Response listing all collections."""
|
||||
collections: List[Collection] = Field(..., description="List of collections")
|
||||
total_documents: int = Field(..., description="Total documents across all collections")
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# Search Models
|
||||
# =============================================================================
|
||||
|
||||
|
||||
class DocumentSearchRequest(BaseModel):
|
||||
"""Request to search documents."""
|
||||
query: str = Field(..., min_length=1, description="Search query")
|
||||
collection: Optional[str] = Field(None, description="Limit to collection")
|
||||
document_type: Optional[DocumentType] = Field(None, description="Filter by type")
|
||||
limit: int = Field(default=10, ge=1, le=50, description="Maximum results")
|
||||
include_content: bool = Field(default=False, description="Include full text content")
|
||||
|
||||
|
||||
class DocumentSearchHit(BaseModel):
|
||||
"""A document search result."""
|
||||
paperless_id: int = Field(..., description="Paperless document ID")
|
||||
title: str = Field(..., description="Document title")
|
||||
score: float = Field(..., description="Relevance score")
|
||||
highlights: Optional[str] = Field(None, description="Highlighted matching text")
|
||||
collection: Optional[str] = Field(None, description="Collection name")
|
||||
document_type: Optional[str] = Field(None, description="Document type")
|
||||
content_preview: Optional[str] = Field(None, description="Content preview if requested")
|
||||
|
||||
|
||||
class DocumentSearchResponse(BaseModel):
|
||||
"""Response from document search."""
|
||||
query: str = Field(..., description="Original query")
|
||||
hits: List[DocumentSearchHit] = Field(..., description="Search results")
|
||||
total: int = Field(..., description="Total matching documents")
|
||||
duration_ms: int = Field(..., description="Search duration in milliseconds")
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# Health Check Models
|
||||
# =============================================================================
|
||||
|
||||
|
||||
class DocumentStoreHealth(BaseModel):
|
||||
"""Health status of document storage components."""
|
||||
paperless_healthy: bool = Field(..., description="Paperless-ngx responding")
|
||||
paperless_version: Optional[str] = Field(None, description="Paperless version")
|
||||
total_documents: Optional[int] = Field(None, description="Total documents in Paperless")
|
||||
indexed_documents: Optional[int] = Field(None, description="Documents indexed in Library Desk")
|
||||
@@ -0,0 +1,298 @@
|
||||
"""
|
||||
Document storage router for Library Desk API.
|
||||
|
||||
Event-driven integration with Paperless-ngx:
|
||||
- Webhook receiver triggers indexing after Paperless virus scan passes
|
||||
- Upload endpoint sends files to Paperless for processing
|
||||
- Search across indexed documents
|
||||
"""
|
||||
|
||||
from fastapi import APIRouter, HTTPException, Depends, Query, UploadFile, File
|
||||
from typing import Optional
|
||||
import logging
|
||||
import time
|
||||
|
||||
from src.models.document import (
|
||||
PaperlessWebhookPayload,
|
||||
WebhookResponse,
|
||||
DocumentUploadRequest,
|
||||
DocumentUploadResponse,
|
||||
DocumentSearchRequest,
|
||||
DocumentSearchResponse,
|
||||
DocumentStoreHealth,
|
||||
)
|
||||
from src.core.dependencies import (
|
||||
verify_api_key,
|
||||
PaperlessDep,
|
||||
QdrantDep,
|
||||
OllamaDep,
|
||||
Neo4jDep,
|
||||
WikiJSDep,
|
||||
)
|
||||
from src.core.multi_tenancy import DEFAULT_USER
|
||||
from src.config import get_settings
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
router = APIRouter(prefix="/documents", tags=["Documents"])
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# Webhook Endpoint (primary integration - event-driven)
|
||||
# =============================================================================
|
||||
|
||||
|
||||
@router.post("/webhook", response_model=WebhookResponse)
|
||||
async def receive_webhook(
|
||||
payload: PaperlessWebhookPayload,
|
||||
paperless: PaperlessDep,
|
||||
qdrant: QdrantDep,
|
||||
ollama: OllamaDep,
|
||||
neo4j: Neo4jDep,
|
||||
wiki: WikiJSDep,
|
||||
user: str = Query(default=DEFAULT_USER, description="User identifier"),
|
||||
):
|
||||
"""
|
||||
Receive webhook events from Paperless-ngx.
|
||||
|
||||
This is the primary integration point. Configure Paperless workflow:
|
||||
1. Trigger: Document Added (after consumption completes)
|
||||
2. Condition: Document passed virus scan (ClamAV in Paperless)
|
||||
3. Action: Webhook POST to this endpoint
|
||||
|
||||
Library Desk indexes the document into vectors and graph.
|
||||
"""
|
||||
from src.services.document_sync_service import DocumentSyncService
|
||||
|
||||
logger.info(f"Webhook received: document_id={payload.document_id}, event={payload.event}")
|
||||
|
||||
settings = get_settings()
|
||||
if not settings.document_store_enabled:
|
||||
return WebhookResponse(
|
||||
document_id=payload.document_id,
|
||||
status="skipped",
|
||||
indexed=False,
|
||||
message="Document store is disabled"
|
||||
)
|
||||
|
||||
try:
|
||||
sync_service = DocumentSyncService(
|
||||
paperless_client=paperless,
|
||||
qdrant_client=qdrant,
|
||||
ollama_client=ollama,
|
||||
neo4j_client=neo4j,
|
||||
wiki_client=wiki,
|
||||
settings=settings
|
||||
)
|
||||
|
||||
result = await sync_service.index_document(
|
||||
document_id=payload.document_id,
|
||||
user=user
|
||||
)
|
||||
|
||||
return WebhookResponse(
|
||||
document_id=payload.document_id,
|
||||
status="indexed" if result.success else "failed",
|
||||
indexed=result.success,
|
||||
message=result.error if not result.success else f"Indexed: {result.title}"
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Webhook processing failed for document {payload.document_id}: {e}", exc_info=True)
|
||||
return WebhookResponse(
|
||||
document_id=payload.document_id,
|
||||
status="error",
|
||||
indexed=False,
|
||||
message=str(e)
|
||||
)
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# Upload Endpoints
|
||||
# =============================================================================
|
||||
|
||||
|
||||
@router.post("/upload", response_model=DocumentUploadResponse)
|
||||
async def upload_document(
|
||||
file: UploadFile = File(...),
|
||||
title: Optional[str] = Query(None, description="Document title"),
|
||||
collection: Optional[str] = Query(None, description="Collection name"),
|
||||
paperless: PaperlessDep = None,
|
||||
api_key: str = Depends(verify_api_key),
|
||||
):
|
||||
"""
|
||||
Upload a document to Paperless-ngx.
|
||||
|
||||
Paperless handles virus scanning. If clean, Paperless webhook
|
||||
triggers indexing back to Library Desk.
|
||||
"""
|
||||
settings = get_settings()
|
||||
if not settings.document_store_enabled:
|
||||
raise HTTPException(status_code=503, detail="Document store is disabled")
|
||||
|
||||
content = await file.read()
|
||||
filename = file.filename or "document"
|
||||
|
||||
custom_fields = []
|
||||
if collection:
|
||||
custom_fields.append({"field": "collection", "value": collection})
|
||||
|
||||
try:
|
||||
task_id = await paperless.upload_document(
|
||||
file_content=content,
|
||||
filename=filename,
|
||||
title=title,
|
||||
custom_fields=custom_fields if custom_fields else None,
|
||||
)
|
||||
|
||||
return DocumentUploadResponse(
|
||||
task_id=task_id,
|
||||
filename=filename,
|
||||
message=f"Uploaded to Paperless, task {task_id}. Indexing via webhook after scan."
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Upload failed for '{filename}': {e}")
|
||||
raise HTTPException(status_code=500, detail=f"Upload failed: {e}")
|
||||
|
||||
|
||||
@router.post("/upload-url", response_model=DocumentUploadResponse)
|
||||
async def upload_from_url(
|
||||
request: DocumentUploadRequest,
|
||||
paperless: PaperlessDep = None,
|
||||
api_key: str = Depends(verify_api_key),
|
||||
):
|
||||
"""
|
||||
Download document from URL and upload to Paperless-ngx.
|
||||
|
||||
Used by HybridRAG to save discovered PDFs. Paperless scans and
|
||||
webhooks back for indexing.
|
||||
"""
|
||||
import httpx
|
||||
|
||||
settings = get_settings()
|
||||
if not settings.document_store_enabled:
|
||||
raise HTTPException(status_code=503, detail="Document store is disabled")
|
||||
|
||||
if not request.url:
|
||||
raise HTTPException(status_code=400, detail="URL is required")
|
||||
|
||||
try:
|
||||
async with httpx.AsyncClient(timeout=60.0) as client:
|
||||
response = await client.get(request.url, follow_redirects=True)
|
||||
response.raise_for_status()
|
||||
content = response.content
|
||||
filename = request.url.split("/")[-1].split("?")[0] or "document"
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Download failed from {request.url}: {e}")
|
||||
raise HTTPException(status_code=400, detail=f"Download failed: {e}")
|
||||
|
||||
try:
|
||||
custom_fields = [{"field": "source_url", "value": request.url}]
|
||||
if request.collection:
|
||||
custom_fields.append({"field": "collection", "value": request.collection})
|
||||
|
||||
task_id = await paperless.upload_document(
|
||||
file_content=content,
|
||||
filename=filename,
|
||||
title=request.title,
|
||||
custom_fields=custom_fields,
|
||||
)
|
||||
|
||||
return DocumentUploadResponse(
|
||||
task_id=task_id,
|
||||
filename=filename,
|
||||
message=f"Uploaded from URL, task {task_id}. Indexing via webhook after scan."
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Upload failed for URL '{request.url}': {e}")
|
||||
raise HTTPException(status_code=500, detail=f"Upload failed: {e}")
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# Search
|
||||
# =============================================================================
|
||||
|
||||
|
||||
@router.post("/search", response_model=DocumentSearchResponse)
|
||||
async def search_documents(
|
||||
request: DocumentSearchRequest,
|
||||
qdrant: QdrantDep,
|
||||
ollama: OllamaDep,
|
||||
user: str = Query(default=DEFAULT_USER, description="User identifier"),
|
||||
api_key: str = Depends(verify_api_key),
|
||||
):
|
||||
"""
|
||||
Semantic search across indexed documents.
|
||||
"""
|
||||
from src.services.vector_service import VectorService
|
||||
from src.core.dependencies import get_wikijs_client
|
||||
from src.models.document import DocumentSearchHit
|
||||
|
||||
settings = get_settings()
|
||||
if not settings.document_store_enabled:
|
||||
raise HTTPException(status_code=503, detail="Document store is disabled")
|
||||
|
||||
start_time = time.time()
|
||||
|
||||
try:
|
||||
wiki = get_wikijs_client()
|
||||
vector_service = VectorService(qdrant, wiki, ollama)
|
||||
|
||||
results = await vector_service.search(
|
||||
query=request.query,
|
||||
user=user,
|
||||
limit=request.limit,
|
||||
score_threshold=0.5,
|
||||
doc_type="document"
|
||||
)
|
||||
|
||||
hits = []
|
||||
for result in results.get("results", []):
|
||||
hits.append(DocumentSearchHit(
|
||||
paperless_id=result.get("metadata", {}).get("paperless_id", 0),
|
||||
title=result.get("title", ""),
|
||||
score=result.get("score", 0.0),
|
||||
highlights=result.get("chunk_text", "")[:200] if request.include_content else None,
|
||||
collection=result.get("metadata", {}).get("collection"),
|
||||
document_type=result.get("metadata", {}).get("document_type"),
|
||||
content_preview=result.get("chunk_text", "")[:500] if request.include_content else None,
|
||||
))
|
||||
|
||||
return DocumentSearchResponse(
|
||||
query=request.query,
|
||||
hits=hits,
|
||||
total=len(hits),
|
||||
duration_ms=int((time.time() - start_time) * 1000)
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Document search failed: {e}", exc_info=True)
|
||||
raise HTTPException(status_code=500, detail=str(e))
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# Health
|
||||
# =============================================================================
|
||||
|
||||
|
||||
@router.get("/health", response_model=DocumentStoreHealth)
|
||||
async def document_store_health(paperless: PaperlessDep):
|
||||
"""Check Paperless-ngx connectivity."""
|
||||
settings = get_settings()
|
||||
|
||||
paperless_healthy = False
|
||||
if settings.paperless_token:
|
||||
try:
|
||||
paperless_healthy = await paperless.health_check()
|
||||
except Exception as e:
|
||||
logger.error(f"Paperless health check failed: {e}")
|
||||
|
||||
return DocumentStoreHealth(
|
||||
paperless_healthy=paperless_healthy,
|
||||
paperless_version="connected" if paperless_healthy else None,
|
||||
total_documents=None,
|
||||
indexed_documents=None
|
||||
)
|
||||
@@ -0,0 +1,274 @@
|
||||
"""
|
||||
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,
|
||||
) -> IndexResult:
|
||||
"""
|
||||
Index a single document from Paperless into vectors and graph.
|
||||
|
||||
Args:
|
||||
document_id: Paperless document ID
|
||||
user: User identifier for multi-tenancy
|
||||
|
||||
Returns:
|
||||
IndexResult with success status and details
|
||||
"""
|
||||
logger.info(f"Indexing document {document_id} for user {user}")
|
||||
|
||||
try:
|
||||
# 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"
|
||||
)
|
||||
|
||||
title = doc.title
|
||||
content = doc.content or ""
|
||||
|
||||
if not content.strip():
|
||||
logger.warning(f"Document {document_id} has no text content")
|
||||
return IndexResult(
|
||||
success=True,
|
||||
document_id=document_id,
|
||||
title=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=title,
|
||||
content=content,
|
||||
user=user,
|
||||
metadata={
|
||||
"paperless_id": document_id,
|
||||
"original_filename": doc.original_file_name,
|
||||
"correspondent": doc.correspondent,
|
||||
"document_type": doc.document_type,
|
||||
"tags": doc.tags,
|
||||
}
|
||||
)
|
||||
|
||||
# Index graph node
|
||||
await self._index_graph(
|
||||
document_id=document_id,
|
||||
title=title,
|
||||
content=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=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:
|
||||
await self.paperless.update_document(
|
||||
document_id=document_id,
|
||||
custom_fields=[{"field": "library_indexed", "value": True}]
|
||||
)
|
||||
except Exception:
|
||||
# Field might not exist, that's OK
|
||||
pass
|
||||
Reference in New Issue
Block a user