diff --git a/.env.example b/.env.example index 0285f94..4f75923 100644 --- a/.env.example +++ b/.env.example @@ -7,6 +7,7 @@ QDRANT_PORT=6333 OLLAMA_URL=http://192.168.86.149:11434 SEARXNG_URL=http://192.168.86.149:8080 REDIS_HOST=192.168.86.149 +PAPERLESS_URL=http://192.168.86.149:8000 OLLAMA_MODEL=mistral-nemo-large:latest OLLAMA_EMBEDDING_MODEL=nomic-embed-text @@ -20,4 +21,5 @@ WIKI_GRAPHQL_API=your_jwt_token_here LIBRARY_API_KEY=key_here NEO4J_PASSWORD=key_here WIKIJS_DB_PASSWORD=key_here -SCHEDULER_API_KEY=key_here \ No newline at end of file +SCHEDULER_API_KEY=key_here +PAPERLESS_TOKEN=key_here \ No newline at end of file diff --git a/CHANGELOG.md b/CHANGELOG.md index d9924de..2393725 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,34 @@ All notable changes to Library Desk will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [1.4.5] - 2025-12-25 + +### Added + +- **Document Storage Integration** - Paperless-ngx integration for PDFs, images, and documents + - Event-driven architecture via Paperless webhooks + - `POST /documents/webhook` - Receive document events from Paperless workflows + - `POST /documents/upload` - Upload files directly to Paperless + - `POST /documents/upload-url` - Download and upload documents from URL + - `POST /documents/search` - Semantic search across indexed documents + - `GET /documents/health` - Paperless connectivity health check +- **DocumentSyncService** - Indexes Paperless documents into vectors and graph + - Fetches document content via Paperless API + - Chunks text and generates embeddings for Qdrant + - Creates Document nodes in Neo4j knowledge graph + - Supports multi-tenancy via user parameter in webhook URL +- **PaperlessClient** - REST API client for Paperless-ngx + - Document retrieval, upload, and update operations + - Health check support +- **Paperless Workflow Configuration** + - Production workflow: Document Added (NOT tagged llm-test) → webhook to Library Desk + - Test workflow: Document Added (tagged llm-test) → webhook with test user + +### Changed + +- Updated `src/config.py` with Paperless configuration settings +- Added `PaperlessDep` dependency injection for document endpoints + ## [1.4.4] - 2025-12-24 ### Added diff --git a/docs/DOCUMENT_STORAGE_PLAN.md b/docs/DOCUMENT_STORAGE_PLAN.md new file mode 100644 index 0000000..235c5fe --- /dev/null +++ b/docs/DOCUMENT_STORAGE_PLAN.md @@ -0,0 +1,509 @@ +# Phase 3: Document Storage System - Implementation Plan + +## Overview + +Document storage tier for Library Desk - storing and indexing PDFs, images, videos, and git documentation mirrors. + +**User Decisions:** +- Paperless-ngx container for OCR +- Ebooks deferred to future phase +- Video.js player deferred to after core implementation + +| Phase | Status | Version | +|-------|--------|---------| +| Phase 1: Cleanup System | Complete | v1.4.0 | +| Phase 2: Volatile Memory | Complete | v1.4.3 | +| Phase 3: Document Storage | Planning | - | +| Phase 4: Test Data Cleanup | Complete | v1.4.4 | + +--- + +## Architecture + +**Paperless-ngx as primary document store** (no SeaweedFS needed): + +``` +┌─────────────────────────────────────────────────────────────────┐ +│ External Sources │ +│ ┌─────────┐ ┌────────────┐ ┌──────────────┐ │ +│ │ GitHub │ │ Direct │ │ Email/Folder │ │ +│ │ Docs │ │ Upload │ │ Ingestion │ │ +│ └────┬────┘ └─────┬──────┘ └──────┬───────┘ │ +└───────┼─────────────┼────────────────┼──────────────────────────┘ + │ │ │ + ▼ ▼ ▼ +┌─────────────────────────────────────────────────────────────────┐ +│ Paperless-ngx │ +│ ┌───────────────────────────────────────────────────────────┐ │ +│ │ - Document storage (PDFs, images, videos) │ │ +│ │ - OCR via Tesseract (PDFs, images) │ │ +│ │ - Web UI for browsing/tagging │ │ +│ │ - REST API for integration │ │ +│ └─────────────────────────┬─────────────────────────────────┘ │ +└────────────────────────────┼────────────────────────────────────┘ + │ REST API (sync) + ▼ +┌─────────────────────────────────────────────────────────────────┐ +│ Library Desk │ +│ ┌───────────────────────────────────────────────────────────┐ │ +│ │ DocumentSyncService │ │ +│ │ - Polls Paperless for new/updated docs │ │ +│ │ - Extracts text + metadata via API │ │ +│ │ - Sends to vector/graph pipelines │ │ +│ └─────────────────────────┬─────────────────────────────────┘ │ +│ │ │ +│ ┌────────────────┼────────────────┐ │ +│ ▼ ▼ ▼ │ +│ ┌──────────┐ ┌──────────┐ ┌──────────┐ │ +│ │ Qdrant │ │ Neo4j │ │ Wiki.js │ │ +│ │ (vectors)│ │ (graph) │ │ (catalog)│ │ +│ └──────────┘ └──────────┘ └──────────┘ │ +└─────────────────────────────────────────────────────────────────┘ +``` + +**File handling by type:** + +| File Type | Paperless | Library Desk | +|-----------|-----------|--------------| +| PDFs | OCR → text | Index text → vectors/graph | +| Images | OCR → text | Index text → vectors/graph | +| Videos | Storage only | Index metadata → vectors/graph | + +--- + +## Technology Stack + +| Component | Purpose | Rationale | +|-----------|---------|-----------| +| **Paperless-ngx** | Document storage + OCR | All-in-one: storage, OCR, web UI, REST API | +| **ClamAV** | Virus scanning | Host OS install, pyclamd integration, better isolation | +| **PDF.js** | PDF viewer | Embeddable in Wiki.js (deferred) | + +**Why Paperless-ngx as primary store:** +- Eliminates need for separate blob storage (SeaweedFS/MinIO) +- Built-in web UI for browsing and tagging +- Tesseract OCR with 100+ language support +- REST API for Library Desk integration +- Handles videos as raw files (no OCR, but stored) +- Email and folder watching for automatic ingestion +- Active community, well-maintained + +--- + +## Paperless-ngx API Deep Dive + +### Authentication +``` +POST /api/token/ +Body: {"username": "...", "password": "..."} +Response: {"token": "..."} + +Header: Authorization: Token +``` + +### Document Upload (for HybridRAG → Paperless) +``` +POST /api/documents/post_document/ +Content-Type: multipart/form-data + +Fields: +- document (file, required) +- title (string) +- created (datetime) +- correspondent (ID) +- document_type (ID) +- storage_path (ID) +- tags (repeatable IDs) +- custom_fields (JSON array) + +Response: {"task_id": "uuid"} +``` + +Track consumption: `GET /api/tasks/?task_id={uuid}` → returns document ID when complete + +### Document Search +``` +GET /api/documents/?query=search+terms # Full-text search +GET /api/documents/?more_like_id=123 # Similarity search + +Response includes __search_hit__: +{ + "score": 0.95, + "highlights": "matched text", + "rank": 0 +} +``` + +### Custom Field Filtering +``` +GET /api/documents/?custom_field_query=field_name__operation=value + +Operations: +- exact, in, isnull, exists (all types) +- icontains, istartswith, iendswith (text) +- gt, gte, lt, lte, range (numeric/date) +- contains (document links) +``` + +### Bulk Operations +``` +POST /api/documents/bulk_edit/ +{ + "documents": [1, 2, 3], + "method": "add_tag|remove_tag|set_correspondent|set_document_type|merge|split|...", + "parameters": {...} +} +``` + +### Webhooks (Push to Library Desk!) +Paperless workflows can trigger webhooks on document events: + +| Trigger | When | Available Data | +|---------|------|----------------| +| Consumption Started | Before OCR | file_path, source, filename | +| Document Added | After OCR | content, tags, doc_type, correspondent, `{doc_url}` | +| Document Updated | On change | Same as Added | +| Scheduled | Time-based | Date offsets from document dates | + +**Webhook Action**: POST to Library Desk endpoint with document data + +### Organization Features + +| Feature | Purpose | API Endpoint | +|---------|---------|--------------| +| Tags | Nested labels (5 levels deep) | `/api/tags/` | +| Correspondents | Source/destination | `/api/correspondents/` | +| Document Types | Classification | `/api/document_types/` | +| Storage Paths | File organization | `/api/storage_paths/` | +| Custom Fields | Extensible metadata | `/api/custom_fields/` | + +### Custom Fields We Should Create +| Field Name | Type | Purpose | +|------------|------|---------| +| `source_url` | URL | Original download URL (for HybridRAG uploads) | +| `library_indexed` | Boolean | Sync status with Library Desk | +| `library_doc_id` | Text | Library Desk document reference | +| `collection` | Text | Logical grouping (e.g., "fastapi-docs") | + +### External LLM Add-ons (Optional) +Community tools exist for Ollama integration: +- **[paperless-ai](https://github.com/clusterzx/paperless-ai)** - Auto-tagging, RAG chat +- **[paperless-gpt](https://github.com/icereed/paperless-gpt)** - LLM-enhanced OCR, auto-titling + +**Recommendation:** Skip these - Library Desk already has Ollama integration for: +- Embedding (nomic-embed-text) +- LLM analysis (mistral-nemo) +- Entity extraction +- HybridRAG + +We'll do our own classification/tagging via Library Desk after sync. + +--- + +## Virus Scanning Integration + +**ClamAV daemon + pyclamd** (no third-party REST wrappers): + +``` +┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐ +│ File Upload │────►│ Library Desk │────►│ ClamAV Daemon │ +│ (URL or file) │ │ (pyclamd) │ │ (clamd:3310) │ +└─────────────────┘ └────────┬────────┘ └─────────────────┘ + │ + ┌────────────┴────────────┐ + ▼ ▼ + ┌──────────┐ ┌──────────┐ + │ Clean │ │ Infected │ + │ ✓ │ │ ✗ │ + └────┬─────┘ └────┬─────┘ + │ │ + ▼ ▼ + Upload to Paperless Reject + Log +``` + +### ClamAV Deployment (Host OS) + +ClamAV runs on the host OS (not containerized) for better security isolation: + +```bash +# Installed via apt on Ubuntu/Debian +# Config: /etc/clamav/clamd.conf +# TCPSocket 3310 +# TCPAddr 0.0.0.0 +``` + +Benefits: scans outside container isolation, single virus DB, survives container restarts. + +### Library Desk Integration +```python +# src/clients/clamav_client.py +import pyclamd + +class ClamAVClient: + def __init__(self, host: str, port: int = 3310): + self.cd = pyclamd.ClamdNetworkSocket(host, port) + + async def scan_bytes(self, data: bytes) -> ScanResult: + """Scan file bytes, return clean/infected status.""" + result = self.cd.scan_stream(data) + if result is None: + return ScanResult(clean=True) + return ScanResult(clean=False, virus_name=result['stream'][1]) + + def ping(self) -> bool: + """Health check.""" + return self.cd.ping() +``` + +### Scan Points +| Location | When | Action on Infected | +|----------|------|-------------------| +| `/documents/upload` | Before Paperless upload | Reject with 400, log threat | +| HybridRAG web fetch | Before saving PDF | Skip file, log threat | +| `/documents/webhook` | Optional re-scan | Quarantine in Paperless | + +### Config Settings +```python +# src/config.py +CLAMAV_HOST: str = "192.168.86.149" # Host OS IP (not container) +CLAMAV_PORT: int = 3310 +CLAMAV_ENABLED: bool = True # Bypass for testing +CLAMAV_TIMEOUT: int = 30 # seconds +``` + +--- + +## Integration Strategy + +### Option A: Webhook Push (Preferred) +``` +Paperless Workflow → POST webhook → Library Desk /documents/webhook +``` +- Real-time indexing when documents added/updated +- Configure in Paperless: Workflow → Document Added → Webhook Action +- Library Desk receives document ID, fetches content via API + +### Option B: Polling Pull (Fallback) +``` +Scheduler → POST /documents/sync → Library Desk polls Paperless +``` +- Periodic sync for missed webhooks or initial bulk import +- Track `library_indexed` custom field to skip already-processed docs + +### Option C: HybridRAG Upload (New!) +``` +HybridRAG web search → finds PDF → POST to Paperless → webhook → indexed +``` +- When HybridRAG finds a relevant PDF/document in web results +- Download and upload to Paperless with `source_url` custom field +- Paperless OCRs it, triggers webhook, Library Desk indexes + +--- + +## Library Desk API Design + +### Documents Router (`/documents`) + +| Endpoint | Method | Purpose | +|----------|--------|---------| +| `/documents/webhook` | POST | Receive Paperless webhook (Document Added/Updated) | +| `/documents/sync` | POST | Pull new/updated docs from Paperless → index | +| `/documents/upload` | POST | Upload file to Paperless (for HybridRAG) | +| `/documents/sync-from-git` | POST | Pull docs from Gitea → upload to Paperless → index | +| `/documents/{document_id}` | GET | Get document metadata | +| `/documents/{document_id}/text` | GET | Get extracted text | +| `/documents/search` | POST | Semantic search across documents | +| `/documents/collection/{name}` | GET | List documents in collection | +| `/documents/collection/{name}/catalog` | POST | Generate wiki catalog page | + +**Upload flow (HybridRAG → Paperless):** +1. HybridRAG finds PDF in web results +2. POST `/documents/upload` with URL or file +3. Library Desk downloads, uploads to Paperless with metadata +4. Returns task_id for async tracking +5. Paperless webhook triggers indexing when OCR complete + +### Viewers Router (`/viewers`) - Deferred + +| Endpoint | Method | Purpose | +|----------|--------|---------| +| `/viewers/pdf/{document_id}` | GET | Serve PDF.js viewer | +| `/viewers/image/{document_id}` | GET | Serve image lightbox | +| `/viewers/video/{document_id}` | GET | Serve Video.js player | + +--- + +## Data Flow: Document Processing Pipeline + +``` +1. INTAKE (Paperless-ngx handles this) + └─ Upload via Paperless UI, email, or folder watch + └─ Paperless assigns document ID and stores file + +2. OCR EXTRACTION (Paperless-ngx handles this) + ├─ PDFs → Tesseract → Plain text + ├─ Images → Tesseract → Plain text + └─ Videos → Metadata only (no OCR) + +3. SYNC TO LIBRARY DESK (scheduled or manual) + └─ Poll Paperless API for new/updated documents + └─ Fetch text content + metadata + +4. TEXT CHUNKING + └─ VectorService._chunk_text() (existing) + +5. EMBEDDING + └─ OllamaClient.embed() (existing) + +6. VECTOR STORAGE (Qdrant) + └─ Payload: {doc_type: "document", paperless_id, ...} + +7. GRAPH STORAGE (Neo4j) + └─ Document node + MENTIONS relationships + +8. WIKI CATALOG (optional) + └─ Auto-generate catalog page via ConsolidationService +``` + +--- + +## Git Docs Integration + +Extends existing `scheduler/src/executors/doc_sync_executor.py`: + +1. **Scheduler** syncs docs from GitHub → Gitea (existing) +2. **Post-sync hook** calls `POST /documents/sync-from-git` +3. **Library Desk** indexes docs into vectors/graph +4. **Auto-generate** wiki catalog page for collection + +--- + +## Wiki.js Viewer Integration + +Since Wiki.js v2 requires disabled HTML sanitization for iframes: + +```markdown + +## Document Preview + + +``` + +**Wiki.js Settings Required:** +- `Administration > Security > Allowed HTML Elements: iframe` +- `Content Security Policy: frame-src http://library-desk:8089` + +--- + +## Implementation Phases + +### Phase 3.1: Infrastructure Setup +- [ ] Deploy Paperless-ngx container (Docker Compose) +- [x] ClamAV installed on host OS (port 3310) +- [ ] Configure Paperless: storage path, OCR settings, API token +- [ ] Create custom fields in Paperless: `source_url`, `library_indexed`, `library_doc_id`, `collection` +- [ ] Create `src/clients/paperless_client.py` +- [ ] Create `src/clients/clamav_client.py` (pyclamd wrapper) +- [ ] Create `src/models/document.py` +- [ ] Add config settings to `src/config.py` (PAPERLESS_*, CLAMAV_*) + +### Phase 3.2: Webhook Integration (Push) +- [ ] Create `src/routers/documents.py` +- [ ] Implement `/documents/webhook` endpoint (receives Paperless events) +- [ ] Configure Paperless Workflow: Document Added → Webhook → Library Desk +- [ ] Create `src/services/document_sync_service.py` +- [ ] Implement document indexing pipeline (fetch text → chunk → embed → graph) + +### Phase 3.3: Polling Sync (Pull Fallback) +- [ ] Implement `/documents/sync` endpoint +- [ ] Poll Paperless for docs where `library_indexed=false` +- [ ] Track sync state (last_sync timestamp in Redis) +- [ ] Update `library_indexed` after successful indexing + +### Phase 3.4: HybridRAG Upload Integration +- [ ] Implement `/documents/upload` endpoint +- [ ] Download file from URL +- [ ] **Virus scan before upload** (reject if infected, log threat) +- [ ] Upload clean files to Paperless with metadata +- [ ] Set `source_url` custom field +- [ ] Extend HybridRAG service to detect and upload relevant PDFs +- [ ] Add `save_to_documents` option to HybridRAG config + +### Phase 3.5: Indexing Pipeline +- [ ] Extend VectorService for `doc_type: "document"` +- [ ] Extend GraphService for Document nodes (link to Paperless ID) +- [ ] Implement `/documents/search` endpoint +- [ ] Add dependency injection + +### Phase 3.6: Git Docs Integration +- [ ] Create `src/clients/gitea_client.py` +- [ ] Implement `/documents/sync-from-git` → bulk upload to Paperless +- [ ] Create collection auto-cataloging (wiki pages) +- [ ] Add scheduler task for periodic git sync + +### Phase 3.7: Viewers (Deferred) +*After core implementation is working* +- [ ] Create `static/pdf-viewer.html` (PDF.js) +- [ ] Create `static/image-viewer.html` +- [ ] Create `static/video-player.html` (Video.js) +- [ ] Create `src/routers/viewers.py` + +### Phase 3.8: Maintenance & Testing +- [ ] Extend cleanup for document orphans +- [ ] Add document orphan detection (Paperless deleted but still in Qdrant/Neo4j) +- [ ] Create `tests/test_document_sync.py` +- [ ] Create `tests/test_paperless_client.py` + +--- + +## Files to Create + +| Path | Purpose | +|------|---------| +| `src/clients/paperless_client.py` | Paperless-ngx REST API client | +| `src/clients/clamav_client.py` | ClamAV scanner (pyclamd wrapper) | +| `src/clients/gitea_client.py` | Gitea repo access | +| `src/models/document.py` | Document/Collection/ScanResult models | +| `src/services/document_sync_service.py` | Sync orchestrator | +| `src/routers/documents.py` | Document endpoints (webhook, sync, upload, search) | +| `tests/test_document_sync.py` | Sync service tests | +| `tests/test_paperless_client.py` | API client tests | +| `tests/test_clamav_client.py` | Virus scanner tests | +| `docker/docker-compose.documents.yml` | Paperless + ClamAV deployment | + +**Deferred files (Phase 3.7):** + +| Path | Purpose | +|------|---------| +| `src/routers/viewers.py` | Viewer endpoints | +| `static/pdf-viewer.html` | PDF.js viewer | +| `static/image-viewer.html` | Image lightbox | +| `static/video-player.html` | Video.js player | + +## Files to Modify + +| Path | Changes | +|------|---------| +| `src/config.py` | `PAPERLESS_*`, `CLAMAV_*` settings | +| `src/core/dependencies.py` | DocumentSyncService, PaperlessClient, ClamAVClient DI | +| `src/main.py` | Register documents router | +| `src/services/vector_service.py` | `doc_type: "document"` handling | +| `src/services/graph_service.py` | Document node with Paperless ID | +| `src/services/hybrid_rag_service.py` | Add `save_to_documents` option + virus scan | +| `src/models/hybrid_rag.py` | Add `save_to_documents` config | +| `src/routers/maintenance.py` | Document orphan cleanup, ClamAV health check | +| `requirements.txt` | Add `pyclamd` | + +## Paperless Custom Fields Setup + +Create these in Paperless UI (Administration → Custom Fields): + +| Field | Type | Purpose | +|-------|------|---------| +| `source_url` | URL | Original download URL | +| `library_indexed` | Boolean | Sync status | +| `library_doc_id` | Text | Library Desk reference | +| `collection` | Text | Logical grouping | diff --git a/docs/MEMORY_SYSTEM_PLAN.md b/docs/MEMORY_SYSTEM_PLAN.md index ef10ef3..3833af3 100644 --- a/docs/MEMORY_SYSTEM_PLAN.md +++ b/docs/MEMORY_SYSTEM_PLAN.md @@ -7,7 +7,7 @@ A three-tier memory architecture for Library Desk with intelligent orchestration | Tier | Storage | Purpose | TTL | |------|---------|---------|-----| | **Volatile** | Qdrant (vectors) | Weather, news, financial, ephemeral context | 5min - 2hr | -| **Documents** | TBD (research) | Git mirrors, PDFs, video, images | Permanent | +| **Documents** | Paperless-ngx + ClamAV (host) | Git mirrors, PDFs, video, images | Permanent | | **Knowledge** | Wiki + Neo4j | Personal dossiers, research, summaries | Permanent | **Implementation Priority**: Cleanup → Volatile → Documents → Test Data Cleanup @@ -18,7 +18,7 @@ A three-tier memory architecture for Library Desk with intelligent orchestration |-------|--------|---------| | Phase 1: Cleanup System | ✅ Complete | v1.4.0 | | Phase 2: Volatile Memory | ✅ Complete | v1.4.3 | -| Phase 3: Document Storage | ⏳ Pending | - | +| Phase 3: Document Storage | ✅ Planned | See [DOCUMENT_STORAGE_PLAN.md](DOCUMENT_STORAGE_PLAN.md) | | Phase 4: Test Data Cleanup | ✅ Complete | v1.4.4 | --- diff --git a/pyproject.toml b/pyproject.toml index 81e12d0..bc0c42d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "library-desk" -version = "1.4.4" +version = "1.4.5" description = "Coordination service for The Library system - HybridRAG queries, document ingestion, entity extraction, and knowledge consolidation" readme = "README.md" requires-python = ">=3.12" diff --git a/src/clients/paperless_client.py b/src/clients/paperless_client.py new file mode 100644 index 0000000..dc2c4dd --- /dev/null +++ b/src/clients/paperless_client.py @@ -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", []), + ) diff --git a/src/config.py b/src/config.py index fe11c14..b64aa6e 100644 --- a/src/config.py +++ b/src/config.py @@ -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") diff --git a/src/core/dependencies.py b/src/core/dependencies.py index b3e6492..a8cd2e7 100644 --- a/src/core/dependencies.py +++ b/src/core/dependencies.py @@ -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 diff --git a/src/main.py b/src/main.py index 6cfb8b0..2855c65 100644 --- a/src/main.py +++ b/src/main.py @@ -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" diff --git a/src/models/document.py b/src/models/document.py new file mode 100644 index 0000000..dcdbc33 --- /dev/null +++ b/src/models/document.py @@ -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") diff --git a/src/routers/documents.py b/src/routers/documents.py new file mode 100644 index 0000000..004e53f --- /dev/null +++ b/src/routers/documents.py @@ -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 + ) diff --git a/src/services/document_sync_service.py b/src/services/document_sync_service.py new file mode 100644 index 0000000..ece34a2 --- /dev/null +++ b/src/services/document_sync_service.py @@ -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