Compare commits

...
9 Commits
Author SHA1 Message Date
jpmschweitzerandClaude Opus 4.5 6d5760c297 release: v1.4.7 - Paperless custom field fix
Build and Push / build (release) Successful in 29s
🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2025-12-25 16:40:28 +01:00
jpmschweitzerandClaude Opus 4.5 867de65354 fix: use field ID for Paperless custom field updates
Paperless API requires field ID (integer) not field name (string)
when updating custom fields. Now looks up field ID by name before
updating library_indexed custom field.

Also includes webhook debugging endpoint for development.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2025-12-25 16:37:26 +01:00
jpmschweitzerandClaude Opus 4.5 f2b8c7d111 fix: update Paperless webhook payload to match include_document format
Build and Push / build (release) Successful in 30s
- Change model field from document_id to id (Paperless sends id)
- Add content, created, modified, added, original_file_name, owner fields
- Add extra="ignore" config to handle additional Paperless fields
- Update sync service to use content from webhook payload
- Skip Paperless API call when content already provided

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2025-12-25 14:38:40 +01:00
jpmschweitzer f4352841a2 Merge feature/document-storage: Paperless-ngx integration
Build and Push / build (release) Successful in 30s
2025-12-25 14:17:11 +01:00
jpmschweitzerandClaude Opus 4.5 4ff3fc4c7a 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>
2025-12-25 14:16:40 +01:00
jpmschweitzerandClaude Opus 4.5 e6e65d6d78 feat: add test data cleanup endpoint
Build and Push / build (release) Successful in 28s
Add POST /maintenance/cleanup/test-data endpoint to purge LLM test data
from wiki, graph, and vectors. Security-restricted to test user namespace
only (users/llm-tester/*, users/llm_tester/*).

- Supports dry_run=true (default) to preview before deleting
- Cleans vectors, graph nodes, and wiki pages
- Scheduler task configured for weekly cleanup (Sunday 3:00 AM)

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2025-12-24 21:12:16 +01:00
jpmschweitzerandClaude Opus 4.5 37f8e1819e feat: refactor volatile cache to vector storage with HybridRAG integration
Build and Push / build (release) Successful in 28s
- Migrate volatile backend from Redis to Qdrant for semantic search
- Add natural language conversion for structured data embedding
- Simplify API: /volatile/search, /volatile/store, /{namespace}/{key}
- Integrate volatile into HybridRAG with priority boost in RRF fusion
- Add POST /maintenance/cleanup/volatile for expiry purging
- Update tests for new Qdrant-based architecture (37/37 pass)

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2025-12-24 20:03:37 +01:00
jpmschweitzerandClaude Opus 4.5 1f848c4878 release: v1.4.2 - volatile cache system
Build and Push / build (release) Successful in 28s
Phase 2 of Memory Management System complete.
See CHANGELOG.md for details.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2025-12-24 17:32:17 +01:00
jpmschweitzerandClaude Opus 4.5 7297e6b9f1 feat: add volatile cache system for ephemeral data
Phase 2 of Memory Management System - volatile memory tier:

- VolatileCacheService: Redis-backed TTL storage
- Volatile router with full CRUD operations
- Predefined namespaces: weather, news, financial, transit, traffic,
  air_quality, sports, social, system, context, custom
- Each namespace has appropriate default TTL (1min to 1hr)
- Refresh schedule support via cron expressions
- Scheduler integration endpoint: GET /volatile/scheduled

Endpoints:
- GET/POST/DELETE /volatile/{namespace}/{key}
- GET/DELETE /volatile/{namespace}
- GET /volatile/stats
- GET /volatile/scheduled
- GET /volatile/namespaces

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2025-12-24 17:31:45 +01:00
21 changed files with 4179 additions and 113 deletions
+3 -1
View File
@@ -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:8091
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
SCHEDULER_API_KEY=key_here
PAPERLESS_TOKEN=key_here
+118
View File
@@ -5,6 +5,124 @@ 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.7] - 2025-12-25
### Fixed
- **Paperless Custom Field Update** - Fixed 400 error when marking documents as indexed
- Paperless API requires field ID (integer) not field name (string)
- Now looks up `library_indexed` field ID before updating
- Webhook params format: `doc_url` and `title` from Jinja templates
### Added
- **Webhook Debug Endpoint** - `POST /documents/webhook-capture` for development testing
## [1.4.6] - 2025-12-25
### Fixed
- **Paperless Webhook Payload Format** - Updated model to match Paperless `include_document=true` format
- Paperless sends `id` instead of `document_id`
- Paperless sends full document data including `content`, `title`, `tags`, etc.
- Webhook now uses content from payload, skipping extra Paperless API call
- Added `extra = "ignore"` to handle additional Paperless fields
## [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
- **Test Data Cleanup Endpoint** - `POST /maintenance/cleanup/test-data`
- Purges LLM test data from wiki, graph, and vectors
- Security-restricted to test user namespace only (`users/llm-tester/*`, `users/llm_tester/*`)
- Supports `dry_run=true` (default) to preview before deleting
- Scheduler task configured for weekly cleanup (Sunday 3:00 AM)
## [1.4.3] - 2025-12-24
### Changed
- **Volatile Cache System Refactored to Vector Storage**
- Backend migrated from Redis to Qdrant for semantic search capability
- Data converted to natural language for embedding and semantic retrieval
- Collection naming: `volatile_{user}` for per-user isolation
- TTL implemented via `ttl_expiry` timestamp in vector payload
- Simplified endpoints:
- `GET /volatile/search?q=...` - Semantic search across volatile data
- `POST /volatile/store?namespace=...&key=...` - Store with query params
- `GET /volatile/{namespace}/{key}` - Get specific record
- `DELETE /volatile/{namespace}/{key}` - Delete record
- Removed namespace-specific URL patterns (simpler API for LLM tool use)
### Added
- **HybridRAG Volatile Integration** - Volatile cache now included in multi-source search
- Volatile results get priority boost in RRF fusion (current data ranks higher)
- New config options: `enable_volatile`, `volatile_limit` (default 1), `volatile_threshold`
- Timing breakdown includes `volatile_ms`
- **Volatile Cleanup Endpoint** - `POST /maintenance/cleanup/volatile`
- Purges expired records across all `volatile_*` collections
- Scheduler task for every 10 minutes recommended
- Returns per-collection cleanup counts
- **Natural Language Conversion** - Structured data converted for embedding
- Template-based conversion for each namespace (weather, news, financial, etc.)
- Fallback for custom namespaces
## [1.4.2] - 2025-12-24
### Added
- **Volatile Cache System** - Ephemeral data storage with TTL
- `GET /volatile/{namespace}/{key}` - Retrieve cached record
- `POST /volatile/{namespace}/{key}` - Store/update record with TTL
- `DELETE /volatile/{namespace}/{key}` - Remove record
- `GET /volatile/{namespace}` - List keys in namespace
- `DELETE /volatile/{namespace}` - Clear all records in namespace
- `GET /volatile/stats` - Cache statistics by namespace
- `GET /volatile/scheduled` - Records needing refresh (for scheduler)
- `GET /volatile/namespaces` - List available namespaces with default TTLs
- **Volatile Namespaces** - Predefined categories with appropriate TTLs:
- `weather` (30min) - Weather conditions and forecasts
- `news` (1hr) - Headlines and breaking news
- `financial` (5min) - Stock prices, exchange rates
- `transit` (5min) - Train/bus schedules, delays
- `traffic` (10min) - Commute times, road conditions
- `air_quality` (1hr) - Pollution, pollen counts
- `sports` (1min) - Live scores, matches
- `social` (10min) - Social notifications
- `system` (1min) - Service health status
- `context` (1hr) - Session state
- `custom` (1hr) - User-defined data
- **Refresh Schedule Support** - Optional cron expressions for scheduler integration
## [1.4.1] - 2025-12-24
### Fixed
+509
View File
@@ -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 <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": "<span>matched</span> 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
<!-- In wiki catalog page -->
## Document Preview
<iframe
src="http://library-desk:8089/viewers/pdf/abc123"
width="100%" height="600px">
</iframe>
```
**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 |
+139 -90
View File
@@ -6,15 +6,24 @@ A three-tier memory architecture for Library Desk with intelligent orchestration
| Tier | Storage | Purpose | TTL |
|------|---------|---------|-----|
| **Volatile** | Redis | Weather, news, financial, ephemeral context | 5min - 2hr |
| **Documents** | TBD (research) | Git mirrors, PDFs, video, images | Permanent |
| **Volatile** | Qdrant (vectors) | Weather, news, financial, ephemeral context | 5min - 2hr |
| **Documents** | Paperless-ngx + ClamAV (host) | Git mirrors, PDFs, video, images | Permanent |
| **Knowledge** | Wiki + Neo4j | Personal dossiers, research, summaries | Permanent |
**Implementation Priority**: Cleanup → Volatile → Documents
**Implementation Priority**: Cleanup → Volatile → Documents → Test Data Cleanup
### Phase Status
| Phase | Status | Version |
|-------|--------|---------|
| Phase 1: Cleanup System | ✅ Complete | v1.4.0 |
| Phase 2: Volatile Memory | ✅ Complete | v1.4.3 |
| Phase 3: Document Storage | ✅ Planned | See [DOCUMENT_STORAGE_PLAN.md](DOCUMENT_STORAGE_PLAN.md) |
| Phase 4: Test Data Cleanup | ✅ Complete | v1.4.4 |
---
## Phase 1: Cleanup System Completion
## Phase 1: Cleanup System Completion
### Current State
- **COMPLETE** - All Phase 1 tasks implemented
@@ -57,9 +66,9 @@ A three-tier memory architecture for Library Desk with intelligent orchestration
---
## Phase 2: Volatile Memory System
## Phase 2: Volatile Memory System
### Architecture
### Architecture (Final Implementation)
```
┌─────────────────┐ ┌──────────────┐ ┌─────────────────┐
@@ -71,88 +80,46 @@ A three-tier memory architecture for Library Desk with intelligent orchestration
┌─────────────────┐
Redis
(DB 4, TTL)
Qdrant
(volatile_{user})
└─────────────────┘
```
### Data Model
**Key design decisions:**
- Vector storage in Qdrant (not Redis) for semantic search
- Collection per user: `volatile_{user}`
- TTL via `ttl_expiry` timestamp in payload
- Natural language conversion for embedding structured data
- Integrated into HybridRAG with priority boost
```python
class VolatileRecord(BaseModel):
key: str # e.g., "weather:rotterdam"
namespace: str # e.g., "weather", "news", "financial"
data: dict # Actual content
source: Optional[str] # Origin API/service
created_at: datetime
updated_at: datetime
ttl: int # Seconds until expiration
refresh_schedule: Optional[str] # Cron expression, if repeating
user: str # Multi-tenant isolation
```
**Key pattern**: `{user}:volatile:{namespace}:{key_hash}`
### Implementation Order: Integration-First
1. **Start with Consolidation Hook** - Understand data flow through existing system
2. **Build Service Layer** - VolatileCacheService with Redis operations
3. **Add API Endpoints** - REST interface for volatile data
4. **Biographer Integration** - Query user preferences for relevance
### Tasks
#### 2.1 Integrate with Consolidation (FIRST)
**New file**: `src/services/volatile_service.py`
```python
class VolatileCacheService:
async def get(user, namespace, key) -> Optional[VolatileRecord]
async def set(user, namespace, key, data, ttl, refresh_schedule=None)
async def delete(user, namespace, key)
async def list_namespace(user, namespace) -> List[str]
async def get_scheduled(user) -> List[VolatileRecord] # For scheduler
```
#### 2.2 Create Volatile API Router
**New file**: `src/routers/volatile.py`
### Endpoints (Implemented)
| Endpoint | Method | Purpose |
|----------|--------|---------|
| `/volatile/{namespace}/{key}` | GET | Retrieve record |
| `/volatile/{namespace}/{key}` | POST | Store/update record |
| `/volatile/search?q=...` | GET | Semantic search across volatile data |
| `/volatile/store?namespace=...&key=...` | POST | Store/update record |
| `/volatile/{namespace}/{key}` | GET | Retrieve specific record |
| `/volatile/{namespace}/{key}` | DELETE | Remove record |
| `/volatile/{namespace}` | GET | List keys in namespace |
| `/volatile/scheduled` | GET | List records needing refresh |
| `/volatile/stats` | GET | Cache statistics |
| `/volatile/scheduled` | GET | Records needing refresh |
| `/volatile/namespaces` | GET | List available namespaces |
| `/maintenance/cleanup/volatile` | POST | Purge expired records |
#### 2.3 Integrate with Consolidation
**File**: `src/services/consolidation_service.py`
### Namespaces
Add relevance trigger detection:
1. During consolidation, analyze search results for location/interest patterns
2. Query tatlock's Biographer collection for user preferences
3. If match found, create/update volatile refresh schedule
#### 2.4 Biographer Integration
**File**: `src/core/dependencies.py`
```python
def get_biographer_qdrant() -> QdrantClientWrapper:
"""Direct access to tatlock's Biographer collection."""
# Configure to connect to tatlock's Qdrant
```
#### 2.5 Scheduler-Side Configuration
Document required scheduler tasks:
```json
{
"task_name": "volatile_refresh",
"schedule": "*/15 * * * *",
"endpoint": "GET /volatile/scheduled",
"follow_up": "For each record, call refresh endpoint with record.refresh_schedule"
}
```
| Namespace | Default TTL | Use Case |
|-----------|-------------|----------|
| weather | 30 min | Current conditions, forecasts |
| news | 1 hour | Headlines, breaking news |
| financial | 5 min | Stock prices, exchange rates |
| transit | 5 min | Train/bus schedules, delays |
| traffic | 10 min | Commute times, road conditions |
| air_quality | 1 hour | Pollution, pollen counts |
| sports | 1 min | Live scores, matches |
| social | 10 min | Social notifications |
| system | 1 min | Service health status |
| context | 1 hour | Session state |
| custom | 1 hour | User-defined data |
---
@@ -219,34 +186,116 @@ Add LLM-powered category descriptor generation:
---
## Files to Modify/Create
## Phase 4: LLM Tester Data Cleanup ✅
### Phase 1 (Cleanup)
- `src/routers/maintenance.py` - Add timestamp tracking
### Problem
LLM testing creates accumulated cruft across the system:
- Wiki.js pages under `llm-tester/` and `llm_tester/` paths
- Graph nodes (Document, Entity) linked to test pages
- Vector chunks in Qdrant for test content
This data accumulates over time and clutters Wiki.js visually (no separate tenant scope for tests).
### Solution
Add a maintenance endpoint to purge all LLM tester artifacts across wiki, graph, and vectors.
### Tasks
#### 4.1 Identify Test Data Patterns ✅
**Patterns matched** (security-restricted to test user namespace):
- `users/llm-tester/*`
- `users/llm_tester/*`
#### 4.2 Add Cleanup Endpoint ✅
**File**: `src/routers/maintenance.py`
```python
@router.post("/cleanup/test-data")
async def cleanup_test_data(
dry_run: bool = Query(default=True),
wiki: WikiJSDep = None,
vector_service: VectorServiceDep = None,
graph_service: GraphServiceDep = None,
api_key: str = Depends(verify_api_key)
):
"""
Purge LLM tester data from wiki, graph, and vectors.
**Security**: Only deletes pages in the test user namespace:
- users/llm-tester/*
- users/llm_tester/*
Use dry_run=true to preview what would be deleted.
"""
```
#### 4.3 Implementation Steps ✅
1. **Wiki cleanup**: Delete pages via GraphQL mutation
2. **Graph cleanup**: Delete Document nodes using `delete_page()` method
3. **Vector cleanup**: Delete chunks using `delete_page_chunks()` method
#### 4.4 Scheduler Integration ✅
**Recommended schedule**: Weekly (Sunday 3:00 AM)
```json
{
"task_name": "test_data_cleanup",
"schedule": "0 3 * * 0",
"endpoint": "POST /maintenance/cleanup/test-data?dry_run=false",
"description": "Weekly cleanup of LLM test data"
}
```
### Files to Modify
- `src/routers/maintenance.py` - Add cleanup endpoint
- `src/services/wiki_service.py` - Add bulk delete by path pattern (if needed)
- `src/services/graph_service.py` - May need pattern-based node deletion
- `src/services/vector_service.py` - Add pattern-based chunk deletion
---
## Files Modified/Created
### Phase 1 (Cleanup) ✅
- `src/routers/maintenance.py` - Timestamp tracking, cleanup endpoints
- `src/services/graph_service.py` - Bidirectional validation
- `src/services/vector_service.py` - Cross-reference checks
- `LIBRARIAN_INTEGRATION.md` - Scheduler config docs
### Phase 2 (Volatile)
- `src/services/volatile_service.py` - **NEW**
- `src/routers/volatile.py` - **NEW**
- `src/models/volatile.py` - **NEW**
- `src/core/dependencies.py` - Add Biographer client
- `src/services/consolidation_service.py` - Relevance triggers
- `tests/test_volatile.py` - **NEW**
### Phase 2 (Volatile)
- `src/services/volatile_service.py` - Qdrant-based volatile cache
- `src/routers/volatile.py` - Simplified endpoints
- `src/models/volatile.py` - Namespaces and models
- `src/models/hybrid_rag.py` - Volatile config options
- `src/services/hybrid_rag_service.py` - Volatile integration
- `src/clients/qdrant_client.py` - Expiry filter methods
- `tests/test_volatile.py` - 37 tests
### Phase 3 (Documents)
- `docs/DOCUMENT_STORAGE_RESEARCH.md` - **NEW**
- `src/services/document_store_service.py` - **NEW** (post-research)
- `src/routers/documents.py` - **NEW** (post-research)
### Phase 4 (Test Data Cleanup)
- `src/routers/maintenance.py` - Add cleanup endpoint
- `src/services/wiki_service.py` - Bulk delete by path pattern
- `src/services/graph_service.py` - Pattern-based node deletion
- `src/services/vector_service.py` - Pattern-based chunk deletion
---
## Resolved Design Decisions
1. **Biographer Qdrant**: Same Qdrant instance, different collection. Library-Desk queries directly.
2. **Scheduler API**: Has REST API for task registration. Library-Desk can programmatically create refresh schedules.
3. **External API calls**: Library-Desk routes through SearXNG for web search. Consider dedicated API integrations for high-value volatiles (weather, financial) for consistent quality.
1. **Volatile Storage**: Qdrant vectors (not Redis) for semantic search capability
2. **Collection Naming**: `volatile_{user}` for per-user isolation
3. **TTL Mechanism**: `ttl_expiry` timestamp in payload, background cleanup job
4. **HybridRAG Integration**: Volatile as third source with RRF priority boost
5. **Biographer Qdrant**: Same Qdrant instance, different collection
6. **Scheduler API**: Has REST API for task registration
---
+1 -1
View File
@@ -1,6 +1,6 @@
[project]
name = "library-desk"
version = "1.4.1"
version = "1.4.7"
description = "Coordination service for The Library system - HybridRAG queries, document ingestion, entity extraction, and knowledge consolidation"
readme = "README.md"
requires-python = ">=3.12"
+488
View File
@@ -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", []),
)
+131 -1
View File
@@ -11,7 +11,7 @@ Provides async vector operations with:
from qdrant_client import QdrantClient
from qdrant_client.models import (
Distance, VectorParams, PointStruct,
Filter, FieldCondition, MatchValue
Filter, FieldCondition, MatchValue, Range
)
from typing import List, Dict, Any, Optional
import uuid
@@ -676,4 +676,134 @@ class QdrantClientWrapper:
except Exception as e:
logger.error(f"Failed to list collections: {e}", exc_info=True)
return []
# ========== Volatile Data Methods ==========
async def search_with_expiry_filter(
self,
collection_name: str,
query_vector: List[float],
current_timestamp: int,
limit: int = 10,
score_threshold: float = 0.7
) -> List[Dict[str, Any]]:
"""
Search vectors filtering out expired records.
Args:
collection_name: Collection name
query_vector: Query embedding vector
current_timestamp: Current time in milliseconds
limit: Maximum results
score_threshold: Minimum similarity score
Returns:
List of non-expired search results
"""
# Filter: ttl_expiry > current_timestamp (not expired)
expiry_filter = Filter(
must=[
FieldCondition(
key="ttl_expiry",
range=Range(gt=current_timestamp)
)
]
)
try:
response = self.client.query_points(
collection_name=collection_name,
query=query_vector,
limit=limit,
score_threshold=score_threshold,
query_filter=expiry_filter,
with_payload=True
)
return [
{
"id": str(point.id),
"score": point.score,
"payload": dict(point.payload)
}
for point in response.points
]
except Exception as e:
logger.error(f"Volatile search failed: {e}", exc_info=True)
return []
async def delete_expired_vectors(
self,
collection_name: str,
current_timestamp: int
) -> int:
"""
Delete all vectors where ttl_expiry < current_timestamp.
Args:
collection_name: Collection name
current_timestamp: Current time in milliseconds
Returns:
Number of points deleted (approximate)
"""
# Filter: ttl_expiry < current_timestamp (expired)
expiry_filter = Filter(
must=[
FieldCondition(
key="ttl_expiry",
range=Range(lt=current_timestamp)
)
]
)
try:
# First count how many will be deleted (scroll to count)
count = 0
offset = None
while True:
points, next_offset = self.client.scroll(
collection_name=collection_name,
scroll_filter=expiry_filter,
limit=100,
offset=offset,
with_payload=False
)
count += len(points)
if next_offset is None:
break
offset = next_offset
if count == 0:
return 0
# Delete expired points
self.client.delete(
collection_name=collection_name,
points_selector=expiry_filter
)
logger.info(f"Deleted {count} expired vectors from {collection_name}")
return count
except Exception as e:
logger.error(f"Failed to delete expired vectors: {e}", exc_info=True)
return 0
async def get_volatile_collections(self) -> List[str]:
"""
Get all volatile collections (prefixed with 'volatile_').
Returns:
List of volatile collection names
"""
try:
collections = self.client.get_collections()
return [
c.name for c in collections.collections
if c.name.startswith("volatile_")
]
except Exception as e:
logger.error(f"Failed to list volatile collections: {e}", exc_info=True)
return []
+5
View File
@@ -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")
+53 -1
View File
@@ -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
+3 -1
View File
@@ -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
maintenance, volatile, documents
)
app.include_router(wiki.router)
@@ -66,6 +66,8 @@ app.include_router(webhooks.router)
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"
+207
View File
@@ -0,0 +1,207 @@
"""
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.
Supports Jinja template format:
- doc_url: Contains document ID in URL path (e.g., http://paperless:8000/documents/123/)
- title: Document title from {{ doc_title }}
"""
doc_url: str = Field(..., description="Paperless document URL containing ID")
title: Optional[str] = Field(None, description="Document title")
class Config:
extra = "ignore" # Ignore extra fields
@property
def document_id(self) -> int:
"""Extract document ID from doc_url."""
import re
match = re.search(r'/documents/(\d+)/?', self.doc_url)
if match:
return int(match.group(1))
raise ValueError(f"Cannot extract document ID from URL: {self.doc_url}")
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")
+5 -1
View File
@@ -14,13 +14,16 @@ class HybridRAGConfig(BaseModel):
vector_limit: int = Field(default=10, ge=1, le=50, description="Max vector results")
graph_limit: int = Field(default=10, ge=1, le=50, description="Max graph results")
web_limit: int = Field(default=5, ge=1, le=20, description="Max web results")
volatile_limit: int = Field(default=1, ge=1, le=5, description="Max volatile results (typically 1)")
enable_vector: bool = Field(default=True, description="Enable vector search")
enable_graph: bool = Field(default=True, description="Enable graph search")
enable_web: bool = Field(default=True, description="Enable web search")
enable_volatile: bool = Field(default=True, description="Enable volatile cache search")
enable_reranking: bool = Field(default=True, description="Enable LLM re-ranking")
enable_enrichment: bool = Field(default=True, description="Enable graph enrichment")
final_result_count: int = Field(default=10, ge=1, le=50, description="Final results to return")
rrf_k: int = Field(default=60, ge=1, le=100, description="RRF constant")
volatile_threshold: float = Field(default=0.8, ge=0.5, le=1.0, description="Volatile similarity threshold")
class RelatedDossier(BaseModel):
@@ -34,7 +37,7 @@ class RelatedDossier(BaseModel):
class HybridRAGResult(BaseModel):
"""Single result from HybridRAG query."""
source_type: str = Field(..., description="Source: 'vector', 'graph', 'web'")
source_type: str = Field(..., description="Source: 'wiki', 'web', 'volatile'")
title: str
content: str
url: Optional[str] = Field(None, description="URL for web results")
@@ -53,6 +56,7 @@ class TimingBreakdown(BaseModel):
vector_ms: float = Field(..., description="Phase 1: Vector search")
graph_ms: float = Field(..., description="Phase 1: Graph search")
web_ms: float = Field(..., description="Phase 1: Web search")
volatile_ms: float = Field(default=0, description="Phase 1: Volatile cache search")
fusion_ms: float = Field(..., description="Phase 2: RRF fusion")
enrichment_ms: float = Field(..., description="Phase 3: Graph enrichment")
reranking_ms: float = Field(..., description="Phase 4: LLM re-ranking")
+130
View File
@@ -0,0 +1,130 @@
"""
Volatile memory models for Library Desk.
Provides models for ephemeral cached data with TTL - weather, news, financial data,
transit schedules, and other time-sensitive external information.
"""
from pydantic import BaseModel, Field
from typing import Dict, Any, Optional, List
from datetime import datetime
from enum import Enum
class VolatileNamespace(str, Enum):
"""
Predefined namespaces for volatile data.
Each namespace can have different default TTLs and refresh schedules.
"""
# Real-time external data
WEATHER = "weather" # Current conditions, forecasts
NEWS = "news" # Headlines, breaking news
FINANCIAL = "financial" # Stock prices, exchange rates, crypto
TRANSIT = "transit" # Train/bus schedules, delays, disruptions
TRAFFIC = "traffic" # Commute times, road conditions
AIR_QUALITY = "air_quality" # Pollution levels, pollen counts
SPORTS = "sports" # Live scores, upcoming matches
# System/integration data
SOCIAL = "social" # Social media mentions, notifications
SYSTEM = "system" # Service health, infrastructure status
# Ephemeral context
CONTEXT = "context" # Conversation context, session state
CUSTOM = "custom" # User-defined volatile data
# Default TTLs per namespace (in seconds)
NAMESPACE_DEFAULT_TTL: Dict[str, int] = {
VolatileNamespace.WEATHER: 1800, # 30 min - weather changes slowly
VolatileNamespace.NEWS: 3600, # 1 hour - news cycles
VolatileNamespace.FINANCIAL: 300, # 5 min - markets move fast
VolatileNamespace.TRANSIT: 300, # 5 min - schedules update frequently
VolatileNamespace.TRAFFIC: 600, # 10 min - traffic patterns
VolatileNamespace.AIR_QUALITY: 3600, # 1 hour - air quality stable
VolatileNamespace.SPORTS: 60, # 1 min - live scores
VolatileNamespace.SOCIAL: 600, # 10 min - social notifications
VolatileNamespace.SYSTEM: 60, # 1 min - system health
VolatileNamespace.CONTEXT: 3600, # 1 hour - session context
VolatileNamespace.CUSTOM: 3600, # 1 hour - default for custom
}
class VolatileRecord(BaseModel):
"""
A volatile cache record with TTL.
Volatile records are ephemeral data stored in Redis with automatic expiration.
Used for weather, news, financial data, and other time-sensitive information.
"""
key: str = Field(..., description="Record key (e.g., 'rotterdam', 'nos-headlines')")
namespace: str = Field(..., description="Namespace (e.g., 'weather', 'news', 'financial')")
data: Dict[str, Any] = Field(..., description="Actual content/payload")
source: Optional[str] = Field(None, description="Origin API/service (e.g., 'openweathermap', 'nos.nl')")
created_at: datetime = Field(default_factory=datetime.utcnow, description="When record was created")
updated_at: datetime = Field(default_factory=datetime.utcnow, description="When record was last updated")
ttl: int = Field(..., ge=60, le=604800, description="Time-to-live in seconds (max 7 days)")
refresh_schedule: Optional[str] = Field(None, description="Cron expression for scheduled refresh")
user: str = Field(..., description="User identifier for multi-tenancy")
class VolatileRecordCreate(BaseModel):
"""Request model for creating/updating a volatile record."""
data: Dict[str, Any] = Field(..., description="Content to store")
source: Optional[str] = Field(None, description="Origin API/service")
ttl: Optional[int] = Field(None, ge=60, le=604800, description="TTL in seconds (uses namespace default if not set)")
refresh_schedule: Optional[str] = Field(None, description="Cron expression for scheduled refresh")
class VolatileRecordResponse(BaseModel):
"""Response model for a volatile record."""
key: str = Field(..., description="Record key")
namespace: str = Field(..., description="Namespace")
data: Dict[str, Any] = Field(..., description="Stored content")
source: Optional[str] = Field(None, description="Origin API/service")
created_at: datetime = Field(..., description="Creation timestamp")
updated_at: datetime = Field(..., description="Last update timestamp")
ttl: int = Field(..., description="TTL in seconds")
ttl_remaining: int = Field(..., description="Seconds until expiration")
refresh_schedule: Optional[str] = Field(None, description="Cron expression if scheduled")
user: str = Field(..., description="User identifier")
class VolatileListResponse(BaseModel):
"""Response model for listing volatile records."""
namespace: str = Field(..., description="Namespace queried")
keys: List[str] = Field(..., description="List of keys in namespace")
count: int = Field(..., description="Number of keys")
user: str = Field(..., description="User identifier")
class VolatileScheduledResponse(BaseModel):
"""Response model for records needing refresh."""
records: List[VolatileRecordResponse] = Field(..., description="Records with refresh schedules")
count: int = Field(..., description="Number of scheduled records")
user: str = Field(..., description="User identifier")
class VolatileStatsResponse(BaseModel):
"""Response model for volatile cache statistics."""
total_records: int = Field(..., description="Total volatile records for user")
by_namespace: Dict[str, int] = Field(..., description="Record count per namespace")
scheduled_count: int = Field(..., description="Records with refresh schedules")
total_memory_bytes: Optional[int] = Field(None, description="Approximate memory usage")
user: str = Field(..., description="User identifier")
class VolatileDeleteResponse(BaseModel):
"""Response model for delete operation."""
key: str = Field(..., description="Deleted key")
namespace: str = Field(..., description="Namespace")
deleted: bool = Field(..., description="Whether record was found and deleted")
user: str = Field(..., description="User identifier")
class VolatileBulkDeleteResponse(BaseModel):
"""Response model for bulk delete operations."""
namespace: Optional[str] = Field(None, description="Namespace if namespace-wide delete")
deleted_count: int = Field(..., description="Number of records deleted")
user: str = Field(..., description="User identifier")
+424
View File
@@ -0,0 +1,424 @@
"""
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, Request
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
doc_id = payload.document_id
logger.info(f"Webhook received: document_id={doc_id}, title={payload.title}")
settings = get_settings()
if not settings.document_store_enabled:
return WebhookResponse(
document_id=doc_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
)
# Fetch content from Paperless (template only provides doc_url and title)
result = await sync_service.index_document(
document_id=doc_id,
user=user,
)
return WebhookResponse(
document_id=doc_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 {doc_id}: {e}", exc_info=True)
return WebhookResponse(
document_id=doc_id,
status="error",
indexed=False,
message=str(e)
)
# =============================================================================
# Debug Capture Endpoint
# =============================================================================
@router.post("/webhook-capture")
async def capture_webhook(request: Request):
"""Capture raw webhook payload for debugging."""
import json
from pathlib import Path
from datetime import datetime
# Get raw body
body = await request.body()
headers = dict(request.headers)
query_params = dict(request.query_params)
# Build capture data
capture = {
"timestamp": datetime.now().isoformat(),
"method": request.method,
"url": str(request.url),
"query_params": query_params,
"headers": headers,
"content_type": headers.get("content-type", "unknown"),
"body_raw": body.decode("utf-8", errors="replace"),
}
# Try to parse as JSON
try:
capture["body_json"] = json.loads(body)
except:
capture["body_json"] = None
# Write to file
capture_file = Path("logs/webhook_capture.json")
capture_file.parent.mkdir(exist_ok=True)
with open(capture_file, "w") as f:
json.dump(capture, f, indent=2, default=str)
logger.info(f"Captured webhook: {capture['body_raw'][:200]}")
return {"status": "captured", "file": str(capture_file)}
# =============================================================================
# Simple Webhook (URL parameters only)
# =============================================================================
@router.post("/webhook-simple", response_model=WebhookResponse)
async def receive_webhook_simple(
doc_url: str = Query(..., description="Paperless document URL containing ID"),
title: str = Query(default="", description="Document title"),
user: str = Query(default=DEFAULT_USER, description="User identifier"),
paperless: PaperlessDep = None,
qdrant: QdrantDep = None,
ollama: OllamaDep = None,
neo4j: Neo4jDep = None,
wiki: WikiJSDep = None,
):
"""
Simple webhook endpoint accepting URL parameters.
Used when Paperless Jinja templates don't work with JSON body.
URL format: /webhook-simple?doc_url=http://...&title=...&user=...
"""
from src.services.document_sync_service import DocumentSyncService
import re
# Extract document ID from URL
match = re.search(r'/documents/(\d+)/?', doc_url)
if not match:
return WebhookResponse(
document_id=0,
status="error",
indexed=False,
message=f"Cannot extract document ID from URL: {doc_url}"
)
doc_id = int(match.group(1))
logger.info(f"Webhook-simple received: document_id={doc_id}, title={title}")
settings = get_settings()
if not settings.document_store_enabled:
return WebhookResponse(
document_id=doc_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=doc_id,
user=user,
)
return WebhookResponse(
document_id=doc_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-simple failed for document {doc_id}: {e}", exc_info=True)
return WebhookResponse(
document_id=doc_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
)
+13 -8
View File
@@ -1,7 +1,7 @@
"""
HybridRAG router for multi-source search API.
Provides endpoint for combining vector, graph, and web search
Provides endpoint for combining vector, graph, volatile cache, and web search
with RRF fusion and LLM re-ranking.
"""
@@ -34,10 +34,12 @@ def get_hybrid_rag_service(
"""Get HybridRAG service instance with all dependencies."""
from src.services.vector_service import VectorService
from src.services.graph_service import GraphService
from src.services.volatile_service import VolatileCacheService
# Create component services
vector_service = VectorService(qdrant_client, wiki_client, ollama_client)
graph_service = GraphService(neo4j_client, wiki_client)
volatile_service = VolatileCacheService(qdrant_client, ollama_client, settings)
# Create HybridRAG service
return HybridRAGService(
@@ -46,7 +48,8 @@ def get_hybrid_rag_service(
searxng_client=searxng_client,
ollama_client=ollama_client,
content_extractor=content_extractor,
settings=settings
settings=settings,
volatile_service=volatile_service
)
@@ -58,26 +61,28 @@ async def hybrid_search(
api_key: str = Depends(verify_api_key)
):
"""
Execute HybridRAG query combining vector, graph, and web search.
Execute HybridRAG query combining vector, graph, volatile cache, and web search.
**6-Phase Pipeline:**
1. **Query Enhancement**: Extract keywords/synonyms with LLM
2. **Parallel Retrieval**: Search vector (Qdrant), graph (Neo4j), web (SearXNG)
3. **RRF Fusion**: Merge results with Reciprocal Rank Fusion
2. **Parallel Retrieval**: Search vector (Qdrant), graph (Neo4j), volatile cache, web (SearXNG)
3. **RRF Fusion**: Merge results with Reciprocal Rank Fusion (volatile gets priority boost)
4. **Enrichment**: Add related documents via shared entities
5. **LLM Re-ranking**: Re-rank with mistral-nemo for relevance
5. **LLM Re-ranking**: Re-rank with configured model for relevance
6. **Context Formatting**: Format for LLM consumption
7. **Persistence**: Store for Librarian knowledge consolidation
**Example Request:**
```json
{
"query": "How does Docker orchestration work with Kubernetes?",
"query": "What's the weather in Rotterdam?",
"user": "jpmschweitzer",
"config": {
"vector_limit": 10,
"graph_limit": 10,
"web_limit": 5,
"volatile_limit": 5,
"enable_volatile": true,
"enable_reranking": true,
"final_result_count": 10
}
@@ -85,7 +90,7 @@ async def hybrid_search(
```
**Returns:**
- Ranked results from all sources
- Ranked results from all sources (wiki, volatile, web)
- Extracted keywords/synonyms
- Related dossiers (via graph)
- Formatted context for LLM
+186 -1
View File
@@ -16,10 +16,13 @@ import time
from src.services.vector_service import VectorService
from src.services.graph_service import GraphService
from src.services.volatile_service import VolatileCacheService
from src.core.dependencies import (
VectorServiceDep, GraphServiceDep, WikiJSDep, RedisDep,
verify_api_key
QdrantDep, OllamaDep, verify_api_key
)
from src.config import get_settings
from src.core.multi_tenancy import DEFAULT_USER
from datetime import datetime, timezone
logger = logging.getLogger(__name__)
@@ -209,6 +212,44 @@ class ReconcileIndexResponse(BaseModel):
total_duration_ms: float
class VolatileCleanupResponse(BaseModel):
"""Response from volatile cache cleanup operation."""
success: bool
collections_processed: int
total_expired_purged: int
by_collection: Dict[str, int] = Field(default_factory=dict)
duration_ms: float
class TestDataCleanupResponse(BaseModel):
"""Response from test data cleanup operation."""
success: bool
dry_run: bool
wiki_pages_deleted: int
graph_nodes_deleted: int
vector_chunks_deleted: int
pages_found: List[Dict[str, Any]] = Field(default_factory=list)
duration_ms: float
# Test data path patterns - restricted to test user namespace only
# These are the only paths that can be cleaned up for safety
TEST_USER_PATH_PREFIXES = [
"users/llm-tester/",
"users/llm_tester/",
]
def _matches_test_user_path(path: str) -> bool:
"""Check if a path is in the test user namespace.
Only matches paths that START with test user prefixes for safety.
This prevents accidental deletion of non-test data.
"""
path_lower = path.lower()
return any(path_lower.startswith(prefix) for prefix in TEST_USER_PATH_PREFIXES)
# ========== Endpoints ==========
@router.post("/cleanup/vectors", response_model=VectorCleanupResponse)
@@ -490,6 +531,150 @@ async def cleanup_all(
raise HTTPException(status_code=500, detail=str(e))
@router.post("/cleanup/volatile", response_model=VolatileCleanupResponse)
async def cleanup_volatile(
qdrant: QdrantDep = None,
ollama: OllamaDep = None,
api_key: str = Depends(verify_api_key)
):
"""
Purge expired volatile cache records across all users.
Loops through all volatile_* collections and removes records where
ttl_expiry < current_timestamp.
**Scheduler Task** - Recommended to run every 10 minutes.
**Scheduler Integration:**
```json
{
"task_name": "volatile_cleanup",
"schedule": "*/10 * * * *",
"endpoint": "POST /maintenance/cleanup/volatile",
"description": "Purge expired volatile cache records"
}
```
"""
start_time = time.time()
try:
settings = get_settings()
service = VolatileCacheService(
qdrant_client=qdrant,
ollama_client=ollama,
settings=settings
)
# Purge expired from all volatile collections
results = await service.purge_all_expired()
total_purged = sum(results.values())
duration_ms = (time.time() - start_time) * 1000
logger.info(f"Volatile cleanup complete: {total_purged} expired records purged from {len(results)} collections")
return VolatileCleanupResponse(
success=True,
collections_processed=len(results),
total_expired_purged=total_purged,
by_collection=results,
duration_ms=duration_ms
)
except Exception as e:
logger.error(f"Volatile cleanup failed: {e}", exc_info=True)
raise HTTPException(status_code=500, detail=str(e))
@router.post("/cleanup/test-data", response_model=TestDataCleanupResponse)
async def cleanup_test_data(
dry_run: bool = Query(default=True, description="Preview only, don't delete"),
wiki: WikiJSDep = None,
vector_service: VectorServiceDep = None,
graph_service: GraphServiceDep = None,
api_key: str = Depends(verify_api_key)
):
"""
Purge LLM tester data from wiki, graph, and vectors.
**Security**: Only deletes pages in the test user namespace:
- users/llm-tester/*
- users/llm_tester/*
This endpoint cannot delete data outside these paths.
**Use dry_run=true (default) to preview what would be deleted.**
**Scheduler Integration:**
```json
{
"task_name": "test_data_cleanup",
"schedule": "0 3 * * 0",
"endpoint": "POST /maintenance/cleanup/test-data?dry_run=false",
"description": "Weekly cleanup of LLM test data"
}
```
"""
start_time = time.time()
try:
# List all wiki pages
all_pages = await wiki.list_all_pages(batch_size=500)
# Filter for test user paths only (security: restricted to test namespace)
test_pages = [
{"id": p["id"], "path": p["path"], "title": p.get("title", "")}
for p in all_pages
if _matches_test_user_path(p.get("path", ""))
]
logger.info(f"Found {len(test_pages)} test pages matching patterns: {TEST_USER_PATH_PREFIXES}")
wiki_deleted = 0
graph_deleted = 0
vector_deleted = 0
if not dry_run and test_pages:
for page in test_pages:
page_id = page["id"]
page_path = page["path"]
try:
# Delete vector chunks for this page (using DEFAULT_USER collection)
chunks_removed = await vector_service.delete_page_chunks(page_id, DEFAULT_USER)
vector_deleted += chunks_removed
# Delete graph node for this page (returns count, may be 0 if no node)
graph_removed = await graph_service.delete_page(page_id, DEFAULT_USER)
graph_deleted += graph_removed
# Delete wiki page (raises exception on failure, returns None on success)
await wiki.delete_page(page_id)
wiki_deleted += 1
logger.info(f"Deleted test page: {page_path} (id={page_id})")
except Exception as e:
logger.error(f"Failed to delete page {page_path}: {e}")
continue
duration_ms = (time.time() - start_time) * 1000
return TestDataCleanupResponse(
success=True,
dry_run=dry_run,
wiki_pages_deleted=wiki_deleted,
graph_nodes_deleted=graph_deleted,
vector_chunks_deleted=vector_deleted,
pages_found=test_pages,
duration_ms=duration_ms
)
except Exception as e:
logger.error(f"Test data cleanup failed: {e}", exc_info=True)
raise HTTPException(status_code=500, detail=str(e))
@router.get("/health", response_model=HealthCheckResponse)
async def maintenance_health(
user: str = Query(..., description="User identifier"),
+273
View File
@@ -0,0 +1,273 @@
"""
Volatile cache router for Library Desk API.
Endpoints for ephemeral cached data with TTL - weather, news, financial, etc.
Data is stored as vectors in Qdrant for semantic search retrieval.
"""
from fastapi import APIRouter, HTTPException, Depends, Query
import logging
from src.models.volatile import (
VolatileRecordCreate,
VolatileRecordResponse,
VolatileListResponse,
VolatileScheduledResponse,
VolatileStatsResponse,
VolatileDeleteResponse,
VolatileNamespace,
NAMESPACE_DEFAULT_TTL,
)
from src.services.volatile_service import VolatileCacheService
from src.core.dependencies import verify_api_key, QdrantDep, OllamaDep
from src.core.multi_tenancy import DEFAULT_USER
from src.config import get_settings
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/volatile", tags=["Volatile Cache"])
def get_volatile_service(qdrant: QdrantDep, ollama: OllamaDep) -> VolatileCacheService:
"""Get volatile cache service instance."""
settings = get_settings()
return VolatileCacheService(
qdrant_client=qdrant,
ollama_client=ollama,
settings=settings
)
@router.get("/stats", response_model=VolatileStatsResponse)
async def get_stats(
user: str = Query(default=DEFAULT_USER, description="User identifier"),
qdrant: QdrantDep = None,
ollama: OllamaDep = None,
api_key: str = Depends(verify_api_key)
):
"""
Get volatile cache statistics.
Returns counts of records by namespace and scheduled refresh info.
"""
service = get_volatile_service(qdrant, ollama)
stats = await service.get_stats(user)
return VolatileStatsResponse(
total_records=stats["total_records"],
by_namespace=stats["by_namespace"],
scheduled_count=stats["scheduled_count"],
total_memory_bytes=None,
user=user,
)
@router.get("/scheduled", response_model=VolatileScheduledResponse)
async def get_scheduled(
user: str = Query(default=DEFAULT_USER, description="User identifier"),
qdrant: QdrantDep = None,
ollama: OllamaDep = None,
api_key: str = Depends(verify_api_key)
):
"""
Get records with refresh schedules.
Used by scheduler to determine what volatile data needs refreshing.
Returns all records that have a refresh_schedule cron expression set.
"""
service = get_volatile_service(qdrant, ollama)
records = await service.get_scheduled(user)
return VolatileScheduledResponse(
records=records,
count=len(records),
user=user,
)
@router.get("/namespaces")
async def list_namespaces(
api_key: str = Depends(verify_api_key)
):
"""
List available namespaces and their default TTLs.
Returns predefined namespaces with their default TTL values.
"""
return {
"namespaces": [
{
"name": ns.value,
"default_ttl": NAMESPACE_DEFAULT_TTL.get(ns, 3600),
"description": _get_namespace_description(ns),
}
for ns in VolatileNamespace
]
}
def _get_namespace_description(ns: VolatileNamespace) -> str:
"""Get human-readable description for namespace."""
descriptions = {
VolatileNamespace.WEATHER: "Weather conditions and forecasts",
VolatileNamespace.NEWS: "Headlines and breaking news",
VolatileNamespace.FINANCIAL: "Stock prices, exchange rates, crypto",
VolatileNamespace.TRANSIT: "Train/bus schedules, delays",
VolatileNamespace.TRAFFIC: "Commute times, road conditions",
VolatileNamespace.AIR_QUALITY: "Pollution levels, pollen counts",
VolatileNamespace.SPORTS: "Live scores, upcoming matches",
VolatileNamespace.SOCIAL: "Social media mentions, notifications",
VolatileNamespace.SYSTEM: "Service health, infrastructure status",
VolatileNamespace.CONTEXT: "Conversation context, session state",
VolatileNamespace.CUSTOM: "User-defined volatile data",
}
return descriptions.get(ns, "Custom namespace")
@router.get("/search")
async def search_volatile(
q: str = Query(..., min_length=1, description="Search query"),
user: str = Query(default=DEFAULT_USER, description="User identifier"),
limit: int = Query(default=5, ge=1, le=20, description="Maximum results"),
threshold: float = Query(default=0.75, ge=0.5, le=1.0, description="Minimum similarity score"),
qdrant: QdrantDep = None,
ollama: OllamaDep = None,
api_key: str = Depends(verify_api_key)
):
"""
Semantic search across volatile data.
Searches all volatile data for semantically similar content.
Higher threshold = stricter matching.
**Example:**
```
GET /volatile/search?q=weather%20rotterdam&user=jpmschweitzer
```
"""
service = get_volatile_service(qdrant, ollama)
results = await service.search(user, q, limit=limit, score_threshold=threshold)
return {
"query": q,
"results": results,
"count": len(results),
"user": user,
}
@router.post("/store", response_model=VolatileRecordResponse)
async def store_volatile(
namespace: str = Query(..., description="Data namespace (weather, news, etc.)"),
key: str = Query(..., description="Record key (e.g., 'rotterdam', 'nos-headlines')"),
request: VolatileRecordCreate = None,
user: str = Query(default=DEFAULT_USER, description="User identifier"),
qdrant: QdrantDep = None,
ollama: OllamaDep = None,
api_key: str = Depends(verify_api_key)
):
"""
Store volatile data.
Data is converted to natural language and embedded for semantic search.
If the same namespace+key already exists, it will be updated.
**Example Request:**
```json
POST /volatile/store?namespace=weather&key=rotterdam
{
"data": {
"temperature": 8,
"conditions": "Cloudy",
"humidity": 85
},
"source": "openweathermap",
"ttl": 1800,
"refresh_schedule": "0 * * * *"
}
```
**Refresh Schedule:**
Optional cron expression for automatic refresh. The scheduler
will query `/volatile/scheduled` and trigger refreshes.
"""
# Validate namespace if not custom
if namespace != VolatileNamespace.CUSTOM:
try:
VolatileNamespace(namespace)
except ValueError:
valid = [ns.value for ns in VolatileNamespace]
raise HTTPException(
status_code=400,
detail=f"Invalid namespace '{namespace}'. Valid: {valid}"
)
service = get_volatile_service(qdrant, ollama)
try:
record = await service.store(
user=user,
namespace=namespace,
key=key,
data=request.data,
source=request.source,
ttl=request.ttl,
refresh_schedule=request.refresh_schedule,
)
return record
except Exception as e:
logger.error(f"Failed to store volatile record: {e}")
raise HTTPException(status_code=500, detail=f"Failed to store record: {str(e)}")
@router.get("/{namespace}/{key}", response_model=VolatileRecordResponse)
async def get_record(
namespace: str,
key: str,
user: str = Query(default=DEFAULT_USER, description="User identifier"),
qdrant: QdrantDep = None,
ollama: OllamaDep = None,
api_key: str = Depends(verify_api_key)
):
"""
Get a specific volatile record by namespace and key.
**Example:**
```
GET /volatile/weather/rotterdam?user=jpmschweitzer
```
"""
service = get_volatile_service(qdrant, ollama)
record = await service.get(user, namespace, key)
if not record:
raise HTTPException(
status_code=404,
detail=f"Record '{key}' not found in namespace '{namespace}'"
)
return record
@router.delete("/{namespace}/{key}", response_model=VolatileDeleteResponse)
async def delete_record(
namespace: str,
key: str,
user: str = Query(default=DEFAULT_USER, description="User identifier"),
qdrant: QdrantDep = None,
ollama: OllamaDep = None,
api_key: str = Depends(verify_api_key)
):
"""
Delete a specific volatile record.
"""
service = get_volatile_service(qdrant, ollama)
deleted = await service.delete(user, namespace, key)
return VolatileDeleteResponse(
key=key,
namespace=namespace,
deleted=deleted,
user=user,
)
+293
View File
@@ -0,0 +1,293 @@
"""
Document sync service for Library Desk.
Handles indexing of Paperless-ngx documents into vectors and graph.
Called by webhook when Paperless completes document processing.
"""
import logging
import re
import hashlib
import uuid
from typing import Optional, List
from dataclasses import dataclass
from src.clients.paperless_client import PaperlessClient
from src.clients.qdrant_client import QdrantClientWrapper
from src.clients.ollama_client import OllamaClient
from src.clients.neo4j_client import Neo4jClient
from src.clients.wikijs_client import WikiJSClient
from src.core.multi_tenancy import get_qdrant_collection_name
from src.config import Settings
logger = logging.getLogger(__name__)
@dataclass
class IndexResult:
"""Result of indexing a single document."""
success: bool
document_id: int
title: str = ""
chunks_created: int = 0
error: Optional[str] = None
class DocumentSyncService:
"""
Service for syncing Paperless documents to Library Desk indexes.
Handles:
- Fetching document content from Paperless API
- Chunking and embedding into Qdrant
- Creating graph nodes in Neo4j
"""
def __init__(
self,
paperless_client: PaperlessClient,
qdrant_client: QdrantClientWrapper,
ollama_client: OllamaClient,
neo4j_client: Neo4jClient,
wiki_client: WikiJSClient,
settings: Settings,
chunk_size: int = 500,
chunk_overlap: int = 50
):
self.paperless = paperless_client
self.qdrant = qdrant_client
self.ollama = ollama_client
self.neo4j = neo4j_client
self.wiki = wiki_client
self.settings = settings
self.chunk_size = chunk_size
self.chunk_overlap = chunk_overlap
def _chunk_text(self, text: str) -> List[str]:
"""Chunk text into overlapping segments."""
text = re.sub(r'\s+', ' ', text).strip()
words = text.split()
if len(words) <= self.chunk_size:
return [text] if text else []
chunks = []
start = 0
while start < len(words):
end = start + self.chunk_size
chunk_words = words[start:end]
chunks.append(' '.join(chunk_words))
start = end - self.chunk_overlap
return chunks
async def index_document(
self,
document_id: int,
user: str,
content: Optional[str] = None,
title: Optional[str] = None,
) -> IndexResult:
"""
Index a single document from Paperless into vectors and graph.
Args:
document_id: Paperless document ID
user: User identifier for multi-tenancy
content: Optional document content (if provided, skip Paperless API call)
title: Optional document title (if provided, skip Paperless API call)
Returns:
IndexResult with success status and details
"""
logger.info(f"Indexing document {document_id} for user {user}")
try:
# If content and title provided (from webhook), skip API call
if content is not None and title is not None:
doc_title = title
doc_content = content
original_filename = None
correspondent = None
document_type = None
tags = []
else:
# Fetch document from Paperless
doc = await self.paperless.get_document(document_id)
if not doc:
return IndexResult(
success=False,
document_id=document_id,
error="Document not found in Paperless"
)
doc_title = doc.title
doc_content = doc.content or ""
original_filename = doc.original_file_name
correspondent = doc.correspondent
document_type = doc.document_type
tags = doc.tags
if not doc_content.strip():
logger.warning(f"Document {document_id} has no text content")
return IndexResult(
success=True,
document_id=document_id,
title=doc_title,
chunks_created=0,
error="No text content (possibly image/video only)"
)
# Index vectors
chunks_created = await self._index_vectors(
document_id=document_id,
title=doc_title,
content=doc_content,
user=user,
metadata={
"paperless_id": document_id,
"original_filename": original_filename,
"correspondent": correspondent,
"document_type": document_type,
"tags": tags,
}
)
# Index graph node
await self._index_graph(
document_id=document_id,
title=doc_title,
content=doc_content,
user=user,
)
# Mark as indexed in Paperless (optional - if custom field exists)
try:
await self._mark_indexed(document_id)
except Exception as e:
logger.debug(f"Could not mark document as indexed: {e}")
logger.info(f"Successfully indexed document {document_id}: {chunks_created} chunks")
return IndexResult(
success=True,
document_id=document_id,
title=doc_title,
chunks_created=chunks_created
)
except Exception as e:
logger.error(f"Failed to index document {document_id}: {e}", exc_info=True)
return IndexResult(
success=False,
document_id=document_id,
error=str(e)
)
async def _index_vectors(
self,
document_id: int,
title: str,
content: str,
user: str,
metadata: dict,
) -> int:
"""Create vector embeddings for document content."""
collection = get_qdrant_collection_name(user)
self.qdrant.ensure_collection(collection)
# Delete existing chunks for this document
try:
self.qdrant.client.delete(
collection_name=collection,
points_selector={
"filter": {
"must": [
{"key": "doc_type", "match": {"value": "document"}},
{"key": "paperless_id", "match": {"value": document_id}},
]
}
}
)
except Exception as e:
logger.debug(f"No existing chunks to delete: {e}")
# Chunk content
chunks = self._chunk_text(content)
if not chunks:
return 0
# Generate embeddings
embeddings = await self.ollama.embed_batch(chunks)
# Build points
points = []
for i, (chunk, embedding) in enumerate(zip(chunks, embeddings)):
point_id = str(uuid.uuid4())
content_hash = hashlib.md5(chunk.encode()).hexdigest()
points.append({
"id": point_id,
"vector": embedding,
"payload": {
"doc_type": "document",
"paperless_id": document_id,
"title": title,
"chunk_text": chunk,
"chunk_index": i,
"content_hash": content_hash,
**metadata
}
})
# Upsert to Qdrant
if points:
self.qdrant.client.upsert(
collection_name=collection,
points=points
)
return len(points)
async def _index_graph(
self,
document_id: int,
title: str,
content: str,
user: str,
):
"""Create graph node for document."""
# Create Document node in Neo4j
query = """
MERGE (d:Document {paperless_id: $paperless_id, user: $user})
SET d.title = $title,
d.doc_type = 'document',
d.updated_at = datetime()
RETURN d
"""
await self.neo4j.execute_query(
query,
{
"paperless_id": document_id,
"user": user,
"title": title,
}
)
# TODO: Extract entities from content and create relationships
# This could use the same entity extraction as wiki pages
async def _mark_indexed(self, document_id: int):
"""Mark document as indexed in Paperless custom field."""
# Try to update library_indexed custom field if it exists
try:
# Look up field ID by name (Paperless requires ID, not name)
field = await self.paperless.get_custom_field_by_name("library_indexed")
if field:
await self.paperless.update_document(
document_id=document_id,
custom_fields=[{"field": field["id"], "value": True}]
)
except Exception:
# Field might not exist, that's OK
pass
+67 -8
View File
@@ -20,6 +20,7 @@ import logging
from src.services.vector_service import VectorService
from src.services.graph_service import GraphService
from src.services.volatile_service import VolatileCacheService
from src.clients.searxng_client import SearXNGClient
from src.clients.ollama_client import OllamaClient
from src.clients.content_extractor import ContentExtractor
@@ -46,7 +47,8 @@ class HybridRAGService:
searxng_client: SearXNGClient,
ollama_client: OllamaClient,
content_extractor: ContentExtractor,
settings: Settings
settings: Settings,
volatile_service: Optional[VolatileCacheService] = None
):
"""
Initialize HybridRAG service.
@@ -58,6 +60,7 @@ class HybridRAGService:
ollama_client: Client for LLM (keyword extraction, re-ranking)
content_extractor: Client for extracting full content from URLs
settings: Application settings
volatile_service: Service for volatile cache search (optional)
"""
self.vector = vector_service
self.graph = graph_service
@@ -65,6 +68,7 @@ class HybridRAGService:
self.ollama = ollama_client
self.content_extractor = content_extractor
self.settings = settings
self.volatile = volatile_service
self.reranker_model = settings.ollama_model
async def search(
@@ -104,8 +108,9 @@ class HybridRAGService:
timing["vector_ms"] = raw_results.get("timing", {}).get("vector_ms", 0)
timing["graph_ms"] = raw_results.get("timing", {}).get("graph_ms", 0)
timing["web_ms"] = raw_results.get("timing", {}).get("web_ms", 0)
timing["volatile_ms"] = raw_results.get("timing", {}).get("volatile_ms", 0)
# Phase 2: Two-Stage RRF Fusion
# Phase 2: Three-Source RRF Fusion
phase2_start = time.time()
# Stage 1: Merge wiki sources (vector + graph) into single ranking
@@ -115,10 +120,12 @@ class HybridRAGService:
k=config.rrf_k
)
# Stage 2: Final RRF between wiki and web (equal footing)
# Stage 2: Final RRF between wiki, volatile, and web
# Volatile gets priority boost (smaller k = higher contribution per rank)
fused_results = self._reciprocal_rank_fusion(
wiki_results=wiki_merged,
web_results=raw_results.get("web", []),
volatile_results=raw_results.get("volatile", []),
k=config.rrf_k
)
timing["fusion_ms"] = (time.time() - phase2_start) * 1000
@@ -389,6 +396,37 @@ JSON:"""
tasks["web"] = web_search()
# Volatile cache search
if config.enable_volatile and self.volatile:
async def volatile_search():
start = time.time()
try:
results = await self.volatile.search(
user=user,
query=query,
limit=config.volatile_limit,
score_threshold=config.volatile_threshold
)
formatted = [
{
"key": r.key,
"namespace": r.namespace,
"title": f"{r.namespace}: {r.key}",
"content": r.data.get("text", "") if isinstance(r.data, dict) else str(r.data),
"raw_data": r.data,
"source_api": r.source,
"ttl_remaining": r.ttl_remaining,
"source": "volatile"
}
for r in results
]
return formatted, (time.time() - start) * 1000
except Exception as e:
logger.error(f"Volatile search failed: {e}", exc_info=True)
return [], (time.time() - start) * 1000
tasks["volatile"] = volatile_search()
# Execute all searches in parallel
results_dict = await asyncio.gather(*tasks.values())
@@ -401,7 +439,8 @@ JSON:"""
logger.info(
f"Parallel retrieval: vector={len(output.get('vector', []))}, "
f"graph={len(output.get('graph', []))}, web={len(output.get('web', []))}"
f"graph={len(output.get('graph', []))}, web={len(output.get('web', []))}, "
f"volatile={len(output.get('volatile', []))}"
)
return output
@@ -491,23 +530,42 @@ JSON:"""
self,
wiki_results: List[Dict],
web_results: List[Dict],
volatile_results: Optional[List[Dict]] = None,
k: int = 60
) -> List[Dict[str, Any]]:
"""
Stage 2: Final RRF between wiki (single source) and web.
Stage 2: Final RRF between wiki, volatile, and web.
Wiki results are pre-merged from vector+graph, so wiki and web
now compete on equal footing.
Wiki results are pre-merged from vector+graph. Volatile results
get a priority boost (smaller effective k) since they represent
current, time-sensitive information.
Args:
wiki_results: Pre-merged wiki results from _merge_wiki_sources()
web_results: Results from web search
volatile_results: Results from volatile cache (fresh data)
k: RRF constant (default 60)
Returns:
Final merged and sorted results
"""
rrf_scores = {}
volatile_results = volatile_results or []
# Volatile results get priority boost (k/2 = stronger score per rank)
volatile_k = k // 2
for rank, result in enumerate(volatile_results, start=1):
key = result.get("key")
namespace = result.get("namespace", "unknown")
if not key:
continue
result_id = f"volatile_{namespace}_{key}"
rrf_scores[result_id] = {
"result": result,
"rrf_score": 1 / (volatile_k + rank), # Priority boost
"sources": ["volatile"],
"source_type": "volatile"
}
# Wiki results (single source, already merged)
for rank, result in enumerate(wiki_results, start=1):
@@ -542,7 +600,8 @@ JSON:"""
reverse=True
)
logger.info(f"Final RRF: {len(sorted_results)} results (wiki + web)")
volatile_count = len([r for r in sorted_results if r["source_type"] == "volatile"])
logger.info(f"Final RRF: {len(sorted_results)} results (wiki + volatile[{volatile_count}] + web)")
return sorted_results
+564
View File
@@ -0,0 +1,564 @@
"""
Volatile Cache service for Library Desk.
Provides ephemeral data storage with TTL using Qdrant vectors:
- Weather, news, financial data
- Transit schedules, traffic conditions
- System status, social notifications
Data is stored as embedded vectors for semantic search retrieval.
"""
import hashlib
import logging
import time
from datetime import datetime
from typing import List, Optional, Dict, Any
from src.clients.qdrant_client import QdrantClientWrapper
from src.clients.ollama_client import OllamaClient
from src.config import Settings
from src.models.volatile import (
VolatileRecordResponse,
VolatileNamespace,
NAMESPACE_DEFAULT_TTL,
)
logger = logging.getLogger(__name__)
class VolatileCacheService:
"""
Service for volatile data with TTL stored in Qdrant.
Stores ephemeral data as vectors for semantic search retrieval.
Each user has an isolated volatile collection.
"""
COLLECTION_PREFIX = "volatile_"
def __init__(
self,
qdrant_client: QdrantClientWrapper,
ollama_client: OllamaClient,
settings: Settings
):
"""
Initialize volatile cache service.
Args:
qdrant_client: Qdrant client for vector storage
ollama_client: Ollama client for embeddings
settings: Application settings
"""
self.qdrant = qdrant_client
self.ollama = ollama_client
self.settings = settings
logger.info("Initialized VolatileCacheService (Qdrant backend)")
def _collection_name(self, user: str) -> str:
"""Get volatile collection name for user."""
return f"{self.COLLECTION_PREFIX}{user}"
def _make_vector_id(self, namespace: str, key: str) -> str:
"""
Generate deterministic vector ID for namespace/key.
Same namespace+key always produces same ID for upsert behavior.
"""
combined = f"{namespace}:{key}"
return hashlib.md5(combined.encode()).hexdigest()
def _get_default_ttl(self, namespace: str) -> int:
"""Get default TTL for a namespace."""
try:
ns = VolatileNamespace(namespace)
return NAMESPACE_DEFAULT_TTL.get(ns, self.settings.volatile_default_ttl)
except ValueError:
return self.settings.volatile_default_ttl
def _current_timestamp_ms(self) -> int:
"""Get current timestamp in milliseconds."""
return int(time.time() * 1000)
def _to_natural_language(
self,
namespace: str,
key: str,
data: Dict[str, Any]
) -> str:
"""
Convert structured data to natural language for embedding.
This creates a text representation that embeds well semantically.
"""
# Template-based conversion for known namespaces
if namespace == VolatileNamespace.WEATHER:
temp = data.get("temperature", data.get("temp", "unknown"))
conditions = data.get("conditions", data.get("weather", ""))
humidity = data.get("humidity", "")
text = f"Current weather in {key}: {temp}°C"
if conditions:
text += f", {conditions}"
if humidity:
text += f", humidity {humidity}%"
return text
elif namespace == VolatileNamespace.NEWS:
title = data.get("title", data.get("headline", ""))
summary = data.get("summary", data.get("description", ""))
source = data.get("source", "")
text = f"News: {title}"
if summary:
text += f". {summary}"
if source:
text += f" (Source: {source})"
return text
elif namespace == VolatileNamespace.FINANCIAL:
symbol = data.get("symbol", key)
price = data.get("price", "")
change = data.get("change", data.get("change_percent", ""))
text = f"Financial data for {symbol}"
if price:
text += f": price {price}"
if change:
text += f", change {change}%"
return text
elif namespace == VolatileNamespace.TRANSIT:
route = data.get("route", data.get("line", key))
status = data.get("status", "")
delay = data.get("delay", data.get("delay_minutes", ""))
text = f"Transit {route}"
if status:
text += f": {status}"
if delay:
text += f", delay {delay} minutes"
return text
elif namespace == VolatileNamespace.TRAFFIC:
location = data.get("location", key)
duration = data.get("duration", data.get("travel_time", ""))
congestion = data.get("congestion", "")
text = f"Traffic for {location}"
if duration:
text += f": {duration} minutes"
if congestion:
text += f", congestion level {congestion}"
return text
elif namespace == VolatileNamespace.AIR_QUALITY:
location = data.get("location", key)
aqi = data.get("aqi", data.get("index", ""))
quality = data.get("quality", "")
text = f"Air quality in {location}"
if aqi:
text += f": AQI {aqi}"
if quality:
text += f" ({quality})"
return text
elif namespace == VolatileNamespace.SPORTS:
event = data.get("event", data.get("match", key))
score = data.get("score", "")
status = data.get("status", "")
text = f"Sports: {event}"
if score:
text += f" - Score: {score}"
if status:
text += f" ({status})"
return text
elif namespace == VolatileNamespace.SYSTEM:
service = data.get("service", key)
status = data.get("status", "unknown")
message = data.get("message", "")
text = f"System status for {service}: {status}"
if message:
text += f". {message}"
return text
# Fallback: serialize key fields
text_parts = [f"{namespace} data for {key}:"]
for k, v in data.items():
if isinstance(v, (str, int, float, bool)):
text_parts.append(f"{k}: {v}")
return " ".join(text_parts)
async def store(
self,
user: str,
namespace: str,
key: str,
data: Dict[str, Any],
source: Optional[str] = None,
ttl: Optional[int] = None,
refresh_schedule: Optional[str] = None
) -> VolatileRecordResponse:
"""
Store volatile data as an embedded vector.
Args:
user: User identifier
namespace: Data namespace (from controlled list)
key: Record key (normalized slug)
data: Structured data to store
source: Origin API/service
ttl: TTL in seconds (uses namespace default if not set)
refresh_schedule: Optional cron expression for refresh
Returns:
The stored record
"""
collection = self._collection_name(user)
# Ensure collection exists
await self.qdrant.ensure_collection(collection)
# Calculate TTL and expiry
effective_ttl = ttl if ttl is not None else self._get_default_ttl(namespace)
now_ms = self._current_timestamp_ms()
expiry_ms = now_ms + (effective_ttl * 1000)
# Convert to natural language for embedding
text = self._to_natural_language(namespace, key, data)
# Generate embedding
embedding = await self.ollama.embed(text)
if not embedding:
raise ValueError("Failed to generate embedding for volatile data")
# Build payload
now = datetime.utcnow()
payload = {
"doc_type": "volatile",
"namespace": namespace,
"key": key,
"text": text,
"raw_data": data,
"source": source,
"created_at": now.isoformat(),
"updated_at": now.isoformat(),
"ttl": effective_ttl,
"ttl_expiry": expiry_ms,
"refresh_schedule": refresh_schedule,
"user": user,
}
# Upsert vector (same namespace+key = same ID = update)
vector_id = self._make_vector_id(namespace, key)
success = await self.qdrant.upsert_vector(
collection_name=collection,
vector_id=vector_id,
vector=embedding,
payload=payload
)
if not success:
raise ValueError("Failed to store volatile vector")
logger.debug(f"Stored volatile {namespace}:{key} with TTL {effective_ttl}s")
return VolatileRecordResponse(
key=key,
namespace=namespace,
data=data,
source=source,
created_at=now,
updated_at=now,
ttl=effective_ttl,
ttl_remaining=effective_ttl,
refresh_schedule=refresh_schedule,
user=user,
)
async def search(
self,
user: str,
query: str,
limit: int = 5,
score_threshold: float = 0.75
) -> List[VolatileRecordResponse]:
"""
Semantic search across volatile data.
Args:
user: User identifier
query: Search query
limit: Maximum results
score_threshold: Minimum similarity score (higher = stricter)
Returns:
List of matching volatile records
"""
collection = self._collection_name(user)
# Check if collection exists
if not await self.qdrant.collection_exists(collection):
return []
# Generate query embedding
query_embedding = await self.ollama.embed(query)
if not query_embedding:
logger.error("Failed to embed query for volatile search")
return []
# Search with expiry filter
now_ms = self._current_timestamp_ms()
results = await self.qdrant.search_with_expiry_filter(
collection_name=collection,
query_vector=query_embedding,
current_timestamp=now_ms,
limit=limit,
score_threshold=score_threshold
)
# Convert to response models
responses = []
for result in results:
payload = result["payload"]
ttl_expiry = payload.get("ttl_expiry", 0)
ttl_remaining = max(0, (ttl_expiry - now_ms) // 1000)
responses.append(VolatileRecordResponse(
key=payload["key"],
namespace=payload["namespace"],
data=payload.get("raw_data", {}),
source=payload.get("source"),
created_at=datetime.fromisoformat(payload["created_at"]),
updated_at=datetime.fromisoformat(payload["updated_at"]),
ttl=payload.get("ttl", 0),
ttl_remaining=ttl_remaining,
refresh_schedule=payload.get("refresh_schedule"),
user=payload["user"],
))
return responses
async def get(
self,
user: str,
namespace: str,
key: str
) -> Optional[VolatileRecordResponse]:
"""
Get a specific volatile record by namespace and key.
Args:
user: User identifier
namespace: Data namespace
key: Record key
Returns:
Record if found and not expired, None otherwise
"""
# Use search with high threshold to find exact match
query = self._to_natural_language(namespace, key, {"key": key})
results = await self.search(user, query, limit=10, score_threshold=0.5)
# Find exact namespace+key match
for result in results:
if result.namespace == namespace and result.key == key:
return result
return None
async def delete(
self,
user: str,
namespace: str,
key: str
) -> bool:
"""
Delete a specific volatile record.
Args:
user: User identifier
namespace: Data namespace
key: Record key
Returns:
True if deleted, False if not found
"""
collection = self._collection_name(user)
if not await self.qdrant.collection_exists(collection):
return False
vector_id = self._make_vector_id(namespace, key)
try:
deleted = await self.qdrant.delete_by_ids(
collection_name=collection,
point_ids=[vector_id]
)
return deleted > 0
except Exception as e:
logger.error(f"Failed to delete volatile {namespace}:{key}: {e}")
return False
async def get_scheduled(
self,
user: str
) -> List[VolatileRecordResponse]:
"""
Get all records with refresh schedules.
Used by scheduler to determine what needs refreshing.
Args:
user: User identifier
Returns:
List of records with refresh_schedule set
"""
collection = self._collection_name(user)
if not await self.qdrant.collection_exists(collection):
return []
now_ms = self._current_timestamp_ms()
scheduled = []
# Scroll through all non-expired records
try:
all_points = await self.qdrant.scroll_all_points(
collection_name=collection,
with_payload=True
)
for point in all_points:
payload = point.get("payload", {})
ttl_expiry = payload.get("ttl_expiry", 0)
# Skip expired
if ttl_expiry <= now_ms:
continue
# Only include if has refresh schedule
if payload.get("refresh_schedule"):
ttl_remaining = max(0, (ttl_expiry - now_ms) // 1000)
scheduled.append(VolatileRecordResponse(
key=payload["key"],
namespace=payload["namespace"],
data=payload.get("raw_data", {}),
source=payload.get("source"),
created_at=datetime.fromisoformat(payload["created_at"]),
updated_at=datetime.fromisoformat(payload["updated_at"]),
ttl=payload.get("ttl", 0),
ttl_remaining=ttl_remaining,
refresh_schedule=payload["refresh_schedule"],
user=payload["user"],
))
return scheduled
except Exception as e:
logger.error(f"Failed to get scheduled volatile records: {e}")
return []
async def get_stats(
self,
user: str
) -> Dict[str, Any]:
"""
Get cache statistics for user.
Args:
user: User identifier
Returns:
Statistics dict
"""
collection = self._collection_name(user)
if not await self.qdrant.collection_exists(collection):
return {
"total_records": 0,
"by_namespace": {},
"scheduled_count": 0,
"expired_count": 0,
}
now_ms = self._current_timestamp_ms()
by_namespace: Dict[str, int] = {}
total = 0
scheduled = 0
expired = 0
try:
all_points = await self.qdrant.scroll_all_points(
collection_name=collection,
with_payload=True
)
for point in all_points:
payload = point.get("payload", {})
namespace = payload.get("namespace", "unknown")
ttl_expiry = payload.get("ttl_expiry", 0)
if ttl_expiry <= now_ms:
expired += 1
else:
total += 1
by_namespace[namespace] = by_namespace.get(namespace, 0) + 1
if payload.get("refresh_schedule"):
scheduled += 1
return {
"total_records": total,
"by_namespace": by_namespace,
"scheduled_count": scheduled,
"expired_count": expired,
}
except Exception as e:
logger.error(f"Failed to get volatile stats: {e}")
return {
"total_records": 0,
"by_namespace": {},
"scheduled_count": 0,
"expired_count": 0,
}
async def purge_expired(
self,
user: str
) -> int:
"""
Purge all expired volatile records for user.
Args:
user: User identifier
Returns:
Number of records purged
"""
collection = self._collection_name(user)
if not await self.qdrant.collection_exists(collection):
return 0
now_ms = self._current_timestamp_ms()
return await self.qdrant.delete_expired_vectors(collection, now_ms)
async def purge_all_expired(self) -> Dict[str, int]:
"""
Purge expired records from all volatile collections.
Returns:
Dict of collection -> purged count
"""
collections = await self.qdrant.get_volatile_collections()
results = {}
now_ms = self._current_timestamp_ms()
for collection in collections:
purged = await self.qdrant.delete_expired_vectors(collection, now_ms)
if purged > 0:
results[collection] = purged
logger.info(f"Purged {purged} expired from {collection}")
return results
+567
View File
@@ -0,0 +1,567 @@
"""
Tests for volatile cache router and service (Qdrant backend).
Tests:
- Volatile record CRUD operations
- Namespace listing and management
- Scheduled record retrieval
- TTL behavior and expiry filtering
- Semantic search
- Natural language conversion
"""
import pytest
from datetime import datetime
from unittest.mock import AsyncMock, MagicMock, patch
from src.models.volatile import (
VolatileRecord,
VolatileRecordCreate,
VolatileRecordResponse,
VolatileListResponse,
VolatileScheduledResponse,
VolatileStatsResponse,
VolatileDeleteResponse,
VolatileBulkDeleteResponse,
VolatileNamespace,
NAMESPACE_DEFAULT_TTL,
)
class TestVolatileModels:
"""Test volatile data models."""
def test_volatile_record_creation(self):
"""Test VolatileRecord model creation."""
record = VolatileRecord(
key="rotterdam",
namespace="weather",
data={"temperature": 18, "conditions": "Cloudy"},
source="openweathermap",
ttl=1800,
user="jpmschweitzer",
)
assert record.key == "rotterdam"
assert record.namespace == "weather"
assert record.data["temperature"] == 18
assert record.ttl == 1800
assert record.refresh_schedule is None
def test_volatile_record_with_schedule(self):
"""Test VolatileRecord with refresh schedule."""
record = VolatileRecord(
key="nos-headlines",
namespace="news",
data={"headlines": ["Test headline"]},
source="nos.nl",
ttl=3600,
refresh_schedule="0 * * * *",
user="jpmschweitzer",
)
assert record.refresh_schedule == "0 * * * *"
def test_volatile_record_create(self):
"""Test VolatileRecordCreate model."""
create = VolatileRecordCreate(
data={"price": 150.50, "change": 2.3},
source="alpha_vantage",
ttl=300,
)
assert create.data["price"] == 150.50
assert create.ttl == 300
def test_volatile_record_response(self):
"""Test VolatileRecordResponse model."""
response = VolatileRecordResponse(
key="rotterdam",
namespace="weather",
data={"temperature": 18},
source="openweathermap",
created_at=datetime.utcnow(),
updated_at=datetime.utcnow(),
ttl=1800,
ttl_remaining=1500,
user="jpmschweitzer",
)
assert response.ttl_remaining == 1500
assert response.ttl == 1800
class TestVolatileNamespaces:
"""Test volatile namespaces and defaults."""
def test_all_namespaces_have_default_ttl(self):
"""Verify all namespaces have default TTLs defined."""
for ns in VolatileNamespace:
assert ns in NAMESPACE_DEFAULT_TTL, f"Missing TTL for {ns}"
assert NAMESPACE_DEFAULT_TTL[ns] > 0
def test_weather_default_ttl(self):
"""Test weather namespace default TTL."""
assert NAMESPACE_DEFAULT_TTL[VolatileNamespace.WEATHER] == 1800 # 30 min
def test_financial_default_ttl(self):
"""Test financial namespace default TTL."""
assert NAMESPACE_DEFAULT_TTL[VolatileNamespace.FINANCIAL] == 300 # 5 min
def test_sports_default_ttl(self):
"""Test sports namespace default TTL (fast updates)."""
assert NAMESPACE_DEFAULT_TTL[VolatileNamespace.SPORTS] == 60 # 1 min
def test_namespace_count(self):
"""Test we have the expected number of namespaces."""
assert len(VolatileNamespace) == 11
class TestVolatileListResponse:
"""Test list response models."""
def test_list_response(self):
"""Test VolatileListResponse model."""
response = VolatileListResponse(
namespace="weather",
keys=["rotterdam", "amsterdam", "utrecht"],
count=3,
user="jpmschweitzer",
)
assert response.count == 3
assert "rotterdam" in response.keys
class TestVolatileScheduledResponse:
"""Test scheduled records response."""
def test_scheduled_response_empty(self):
"""Test empty scheduled response."""
response = VolatileScheduledResponse(
records=[],
count=0,
user="jpmschweitzer",
)
assert response.count == 0
assert response.records == []
def test_scheduled_response_with_records(self):
"""Test scheduled response with records."""
record = VolatileRecordResponse(
key="nos-headlines",
namespace="news",
data={"headlines": []},
source="nos.nl",
created_at=datetime.utcnow(),
updated_at=datetime.utcnow(),
ttl=3600,
ttl_remaining=3000,
refresh_schedule="0 */6 * * *",
user="jpmschweitzer",
)
response = VolatileScheduledResponse(
records=[record],
count=1,
user="jpmschweitzer",
)
assert response.count == 1
assert response.records[0].refresh_schedule == "0 */6 * * *"
class TestVolatileStatsResponse:
"""Test stats response model."""
def test_stats_response(self):
"""Test VolatileStatsResponse model."""
response = VolatileStatsResponse(
total_records=15,
by_namespace={"weather": 3, "news": 5, "financial": 7},
scheduled_count=2,
total_memory_bytes=None,
user="jpmschweitzer",
)
assert response.total_records == 15
assert response.by_namespace["weather"] == 3
assert response.scheduled_count == 2
class TestVolatileDeleteResponses:
"""Test delete response models."""
def test_delete_response(self):
"""Test VolatileDeleteResponse model."""
response = VolatileDeleteResponse(
key="rotterdam",
namespace="weather",
deleted=True,
user="jpmschweitzer",
)
assert response.deleted is True
def test_delete_not_found(self):
"""Test delete response when record not found."""
response = VolatileDeleteResponse(
key="nonexistent",
namespace="weather",
deleted=False,
user="jpmschweitzer",
)
assert response.deleted is False
def test_bulk_delete_response(self):
"""Test VolatileBulkDeleteResponse model."""
response = VolatileBulkDeleteResponse(
namespace="weather",
deleted_count=5,
user="jpmschweitzer",
)
assert response.deleted_count == 5
assert response.namespace == "weather"
class TestVolatileService:
"""Test VolatileCacheService functionality (Qdrant backend)."""
@pytest.fixture
def mock_qdrant(self):
"""Create mock Qdrant client."""
qdrant = AsyncMock()
qdrant.ensure_collection = AsyncMock()
qdrant.collection_exists = AsyncMock(return_value=True)
qdrant.upsert_vector = AsyncMock(return_value=True)
qdrant.delete_by_ids = AsyncMock(return_value=1)
qdrant.search_with_expiry_filter = AsyncMock(return_value=[])
qdrant.scroll_all_points = AsyncMock(return_value=[])
qdrant.delete_expired_vectors = AsyncMock(return_value=0)
qdrant.get_volatile_collections = AsyncMock(return_value=[])
return qdrant
@pytest.fixture
def mock_ollama(self):
"""Create mock Ollama client."""
ollama = AsyncMock()
ollama.embed = AsyncMock(return_value=[0.1] * 768) # Return 768-dim embedding
return ollama
@pytest.fixture
def mock_settings(self):
"""Create mock settings."""
settings = MagicMock()
settings.volatile_default_ttl = 3600
return settings
@pytest.fixture
def volatile_service(self, mock_qdrant, mock_ollama, mock_settings):
"""Create VolatileCacheService with mocks."""
from src.services.volatile_service import VolatileCacheService
return VolatileCacheService(
qdrant_client=mock_qdrant,
ollama_client=mock_ollama,
settings=mock_settings
)
def test_collection_name(self, volatile_service):
"""Test collection naming pattern."""
name = volatile_service._collection_name("jpmschweitzer")
assert name == "volatile_jpmschweitzer"
def test_make_vector_id(self, volatile_service):
"""Test deterministic vector ID generation."""
id1 = volatile_service._make_vector_id("weather", "rotterdam")
id2 = volatile_service._make_vector_id("weather", "rotterdam")
id3 = volatile_service._make_vector_id("weather", "amsterdam")
assert id1 == id2 # Same namespace+key = same ID
assert id1 != id3 # Different key = different ID
assert len(id1) == 32 # MD5 hex length
def test_get_default_ttl_known_namespace(self, volatile_service):
"""Test default TTL for known namespace."""
ttl = volatile_service._get_default_ttl("weather")
assert ttl == 1800 # Weather namespace default
def test_get_default_ttl_unknown_namespace(self, volatile_service):
"""Test default TTL for unknown namespace."""
ttl = volatile_service._get_default_ttl("unknown_namespace")
assert ttl == 3600 # Falls back to settings default
def test_to_natural_language_weather(self, volatile_service):
"""Test natural language conversion for weather data."""
text = volatile_service._to_natural_language(
namespace="weather",
key="rotterdam",
data={"temperature": 18, "conditions": "Cloudy", "humidity": 75}
)
assert "rotterdam" in text.lower()
assert "18" in text
assert "Cloudy" in text
assert "75" in text
def test_to_natural_language_news(self, volatile_service):
"""Test natural language conversion for news data."""
text = volatile_service._to_natural_language(
namespace="news",
key="nos-headlines",
data={"title": "Breaking News", "summary": "Something happened", "source": "NOS"}
)
assert "Breaking News" in text
assert "Something happened" in text
assert "NOS" in text
def test_to_natural_language_financial(self, volatile_service):
"""Test natural language conversion for financial data."""
text = volatile_service._to_natural_language(
namespace="financial",
key="AAPL",
data={"symbol": "AAPL", "price": 150.50, "change": 2.3}
)
assert "AAPL" in text
assert "price" in text.lower()
assert "change" in text.lower()
def test_to_natural_language_transit(self, volatile_service):
"""Test natural language conversion for transit data."""
text = volatile_service._to_natural_language(
namespace="transit",
key="ns-intercity",
data={"route": "Amsterdam-Rotterdam", "status": "On time", "delay": 0}
)
assert "Amsterdam-Rotterdam" in text or "ns-intercity" in text.lower()
assert "On time" in text
def test_to_natural_language_fallback(self, volatile_service):
"""Test natural language fallback for unknown namespace."""
text = volatile_service._to_natural_language(
namespace="custom",
key="test-key",
data={"foo": "bar", "count": 42}
)
assert "custom" in text.lower()
assert "foo" in text or "bar" in text
@pytest.mark.asyncio
async def test_store_success(self, volatile_service, mock_qdrant, mock_ollama):
"""Test successful store operation."""
result = await volatile_service.store(
user="jpmschweitzer",
namespace="weather",
key="rotterdam",
data={"temperature": 18, "conditions": "Sunny"},
source="openweathermap",
ttl=1800
)
assert result.key == "rotterdam"
assert result.namespace == "weather"
assert result.ttl == 1800
mock_qdrant.ensure_collection.assert_called_once()
mock_ollama.embed.assert_called_once()
mock_qdrant.upsert_vector.assert_called_once()
@pytest.mark.asyncio
async def test_store_uses_namespace_default_ttl(self, volatile_service, mock_qdrant, mock_ollama):
"""Test store uses namespace default TTL when not specified."""
result = await volatile_service.store(
user="jpmschweitzer",
namespace="weather",
key="amsterdam",
data={"temperature": 16},
source="openweathermap",
ttl=None # Not specified
)
assert result.ttl == 1800 # Weather default
@pytest.mark.asyncio
async def test_search_empty_collection(self, volatile_service, mock_qdrant, mock_ollama):
"""Test search when collection doesn't exist."""
mock_qdrant.collection_exists.return_value = False
results = await volatile_service.search(
user="jpmschweitzer",
query="weather rotterdam"
)
assert results == []
mock_ollama.embed.assert_not_called()
@pytest.mark.asyncio
async def test_search_with_results(self, volatile_service, mock_qdrant, mock_ollama):
"""Test search returns results."""
import time
now_ms = int(time.time() * 1000)
mock_qdrant.search_with_expiry_filter.return_value = [
{
"score": 0.95,
"payload": {
"key": "rotterdam",
"namespace": "weather",
"raw_data": {"temperature": 18},
"source": "openweathermap",
"created_at": datetime.utcnow().isoformat(),
"updated_at": datetime.utcnow().isoformat(),
"ttl": 1800,
"ttl_expiry": now_ms + 900000, # 15 min remaining
"refresh_schedule": None,
"user": "jpmschweitzer"
}
}
]
results = await volatile_service.search(
user="jpmschweitzer",
query="weather rotterdam"
)
assert len(results) == 1
assert results[0].key == "rotterdam"
assert results[0].namespace == "weather"
@pytest.mark.asyncio
async def test_delete_success(self, volatile_service, mock_qdrant):
"""Test successful delete."""
mock_qdrant.delete_by_ids.return_value = 1
result = await volatile_service.delete("jpmschweitzer", "weather", "rotterdam")
assert result is True
mock_qdrant.delete_by_ids.assert_called_once()
@pytest.mark.asyncio
async def test_delete_not_found(self, volatile_service, mock_qdrant):
"""Test delete when record not found."""
mock_qdrant.delete_by_ids.return_value = 0
result = await volatile_service.delete("jpmschweitzer", "weather", "nonexistent")
assert result is False
@pytest.mark.asyncio
async def test_get_stats_empty(self, volatile_service, mock_qdrant):
"""Test stats with no records."""
mock_qdrant.collection_exists.return_value = False
stats = await volatile_service.get_stats("jpmschweitzer")
assert stats["total_records"] == 0
assert stats["by_namespace"] == {}
assert stats["scheduled_count"] == 0
@pytest.mark.asyncio
async def test_get_stats_with_records(self, volatile_service, mock_qdrant):
"""Test stats with records."""
import time
now_ms = int(time.time() * 1000)
mock_qdrant.scroll_all_points.return_value = [
{"payload": {"namespace": "weather", "ttl_expiry": now_ms + 100000}},
{"payload": {"namespace": "weather", "ttl_expiry": now_ms + 100000, "refresh_schedule": "0 * * * *"}},
{"payload": {"namespace": "news", "ttl_expiry": now_ms + 100000}},
{"payload": {"namespace": "weather", "ttl_expiry": now_ms - 100000}}, # Expired
]
stats = await volatile_service.get_stats("jpmschweitzer")
assert stats["total_records"] == 3 # Excludes expired
assert stats["by_namespace"]["weather"] == 2
assert stats["by_namespace"]["news"] == 1
assert stats["scheduled_count"] == 1
assert stats["expired_count"] == 1
@pytest.mark.asyncio
async def test_purge_expired(self, volatile_service, mock_qdrant):
"""Test purging expired records."""
mock_qdrant.delete_expired_vectors.return_value = 5
result = await volatile_service.purge_expired("jpmschweitzer")
assert result == 5
mock_qdrant.delete_expired_vectors.assert_called_once()
@pytest.mark.asyncio
async def test_purge_all_expired(self, volatile_service, mock_qdrant):
"""Test purging expired from all collections."""
mock_qdrant.get_volatile_collections.return_value = [
"volatile_user1",
"volatile_user2"
]
mock_qdrant.delete_expired_vectors.side_effect = [3, 2]
results = await volatile_service.purge_all_expired()
assert results["volatile_user1"] == 3
assert results["volatile_user2"] == 2
@pytest.mark.asyncio
async def test_get_scheduled(self, volatile_service, mock_qdrant):
"""Test getting scheduled records."""
import time
now_ms = int(time.time() * 1000)
mock_qdrant.scroll_all_points.return_value = [
{
"payload": {
"key": "nos-headlines",
"namespace": "news",
"raw_data": {"headlines": []},
"source": "nos.nl",
"created_at": datetime.utcnow().isoformat(),
"updated_at": datetime.utcnow().isoformat(),
"ttl": 3600,
"ttl_expiry": now_ms + 1800000,
"refresh_schedule": "0 */6 * * *",
"user": "jpmschweitzer"
}
},
{
"payload": {
"key": "rotterdam",
"namespace": "weather",
"raw_data": {"temperature": 18},
"source": "openweathermap",
"created_at": datetime.utcnow().isoformat(),
"updated_at": datetime.utcnow().isoformat(),
"ttl": 1800,
"ttl_expiry": now_ms + 900000,
"refresh_schedule": None, # Not scheduled
"user": "jpmschweitzer"
}
}
]
scheduled = await volatile_service.get_scheduled("jpmschweitzer")
assert len(scheduled) == 1
assert scheduled[0].key == "nos-headlines"
assert scheduled[0].refresh_schedule == "0 */6 * * *"
class TestVolatileCleanupEndpoint:
"""Test volatile cleanup in maintenance router."""
@pytest.mark.asyncio
async def test_cleanup_volatile(self):
"""Test volatile cleanup endpoint."""
from src.routers.maintenance import cleanup_volatile, VolatileCleanupResponse
mock_qdrant = AsyncMock()
mock_qdrant.get_volatile_collections = AsyncMock(return_value=[
"volatile_user1",
"volatile_user2"
])
mock_qdrant.delete_expired_vectors = AsyncMock(side_effect=[3, 2])
mock_ollama = AsyncMock()
mock_settings = MagicMock()
mock_settings.volatile_default_ttl = 3600
with patch('src.routers.maintenance.get_settings', return_value=mock_settings):
result = await cleanup_volatile(
qdrant=mock_qdrant,
ollama=mock_ollama,
api_key="test"
)
assert result.success is True
assert result.collections_processed == 2
assert result.total_expired_purged == 5
assert result.by_collection["volatile_user1"] == 3
assert result.by_collection["volatile_user2"] == 2