Compare commits

..
7 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
20 changed files with 3401 additions and 484 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
+91
View File
@@ -5,6 +5,97 @@ 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
+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.2"
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
+2 -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, volatile
maintenance, volatile, documents
)
app.include_router(wiki.router)
@@ -67,6 +67,7 @@ app.include_router(rag_search.router)
app.include_router(content.router)
app.include_router(maintenance.router)
app.include_router(volatile.router)
app.include_router(documents.router)
# Mount static files directory for Wiki.js integration scripts
static_dir = Path(__file__).parent.parent / "static"
+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")
+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"),
+96 -94
View File
@@ -2,6 +2,7 @@
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
@@ -14,11 +15,11 @@ from src.models.volatile import (
VolatileScheduledResponse,
VolatileStatsResponse,
VolatileDeleteResponse,
VolatileBulkDeleteResponse,
VolatileNamespace,
NAMESPACE_DEFAULT_TTL,
)
from src.services.volatile_service import VolatileCacheService
from src.core.dependencies import verify_api_key, RedisDep
from src.core.dependencies import verify_api_key, QdrantDep, OllamaDep
from src.core.multi_tenancy import DEFAULT_USER
from src.config import get_settings
@@ -27,16 +28,21 @@ logger = logging.getLogger(__name__)
router = APIRouter(prefix="/volatile", tags=["Volatile Cache"])
def get_volatile_service(redis: RedisDep) -> VolatileCacheService:
def get_volatile_service(qdrant: QdrantDep, ollama: OllamaDep) -> VolatileCacheService:
"""Get volatile cache service instance."""
settings = get_settings()
return VolatileCacheService(redis_client=redis, settings=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"),
redis: RedisDep = None,
qdrant: QdrantDep = None,
ollama: OllamaDep = None,
api_key: str = Depends(verify_api_key)
):
"""
@@ -44,14 +50,14 @@ async def get_stats(
Returns counts of records by namespace and scheduled refresh info.
"""
service = get_volatile_service(redis)
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=stats.get("total_memory_bytes"),
total_memory_bytes=None,
user=user,
)
@@ -59,7 +65,8 @@ async def get_stats(
@router.get("/scheduled", response_model=VolatileScheduledResponse)
async def get_scheduled(
user: str = Query(default=DEFAULT_USER, description="User identifier"),
redis: RedisDep = None,
qdrant: QdrantDep = None,
ollama: OllamaDep = None,
api_key: str = Depends(verify_api_key)
):
"""
@@ -68,7 +75,7 @@ async def get_scheduled(
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(redis)
service = get_volatile_service(qdrant, ollama)
records = await service.get_scheduled(user)
return VolatileScheduledResponse(
@@ -86,10 +93,7 @@ async def list_namespaces(
List available namespaces and their default TTLs.
Returns predefined namespaces with their default TTL values.
Custom namespaces can also be used with the default TTL.
"""
from src.models.volatile import NAMESPACE_DEFAULT_TTL
return {
"namespaces": [
{
@@ -120,103 +124,62 @@ def _get_namespace_description(ns: VolatileNamespace) -> str:
return descriptions.get(ns, "Custom namespace")
@router.get("/{namespace}", response_model=VolatileListResponse)
async def list_keys(
namespace: str,
@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"),
redis: RedisDep = None,
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)
):
"""
List all keys in a namespace.
Semantic search across volatile data.
Returns the list of keys stored in the specified namespace.
"""
service = get_volatile_service(redis)
keys = await service.list_namespace(user, namespace)
return VolatileListResponse(
namespace=namespace,
keys=keys,
count=len(keys),
user=user,
)
@router.delete("/{namespace}", response_model=VolatileBulkDeleteResponse)
async def delete_namespace(
namespace: str,
user: str = Query(default=DEFAULT_USER, description="User identifier"),
redis: RedisDep = None,
api_key: str = Depends(verify_api_key)
):
"""
Delete all records in a namespace.
Removes all volatile data for the specified namespace.
"""
service = get_volatile_service(redis)
deleted = await service.delete_namespace(user, namespace)
return VolatileBulkDeleteResponse(
namespace=namespace,
deleted_count=deleted,
user=user,
)
@router.get("/{namespace}/{key}", response_model=VolatileRecordResponse)
async def get_record(
namespace: str,
key: str,
user: str = Query(default=DEFAULT_USER, description="User identifier"),
redis: RedisDep = None,
api_key: str = Depends(verify_api_key)
):
"""
Get a volatile record.
Returns the record if it exists and has not expired.
Searches all volatile data for semantically similar content.
Higher threshold = stricter matching.
**Example:**
```
GET /volatile/weather/rotterdam?user=jpmschweitzer
GET /volatile/search?q=weather%20rotterdam&user=jpmschweitzer
```
"""
service = get_volatile_service(redis)
record = await service.get(user, namespace, key)
service = get_volatile_service(qdrant, ollama)
results = await service.search(user, q, limit=limit, score_threshold=threshold)
if not record:
raise HTTPException(
status_code=404,
detail=f"Record '{key}' not found in namespace '{namespace}'"
)
return record
return {
"query": q,
"results": results,
"count": len(results),
"user": user,
}
@router.post("/{namespace}/{key}", response_model=VolatileRecordResponse)
async def set_record(
namespace: str,
key: str,
request: VolatileRecordCreate,
@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"),
redis: RedisDep = None,
qdrant: QdrantDep = None,
ollama: OllamaDep = None,
api_key: str = Depends(verify_api_key)
):
"""
Store or update a volatile record.
Store volatile data.
Creates or updates a record with the specified TTL.
If TTL is not provided, the namespace default is used.
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": 18,
"conditions": "Partly cloudy",
"humidity": 65
"temperature": 8,
"conditions": "Cloudy",
"humidity": 85
},
"source": "openweathermap",
"ttl": 1800,
@@ -228,10 +191,21 @@ async def set_record(
Optional cron expression for automatic refresh. The scheduler
will query `/volatile/scheduled` and trigger refreshes.
"""
service = get_volatile_service(redis)
# 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.set(
record = await service.store(
user=user,
namespace=namespace,
key=key,
@@ -244,7 +218,36 @@ async def set_record(
except Exception as e:
logger.error(f"Failed to store volatile record: {e}")
raise HTTPException(status_code=500, detail="Failed to store record")
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)
@@ -252,15 +255,14 @@ async def delete_record(
namespace: str,
key: str,
user: str = Query(default=DEFAULT_USER, description="User identifier"),
redis: RedisDep = None,
qdrant: QdrantDep = None,
ollama: OllamaDep = None,
api_key: str = Depends(verify_api_key)
):
"""
Delete a volatile record.
Removes the record from the cache.
Delete a specific volatile record.
"""
service = get_volatile_service(redis)
service = get_volatile_service(qdrant, ollama)
deleted = await service.delete(user, namespace, key)
return VolatileDeleteResponse(
+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
+371 -240
View File
@@ -1,23 +1,24 @@
"""
Volatile Cache service for Library Desk.
Provides ephemeral data storage with TTL for time-sensitive information:
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 json
import logging
import hashlib
import logging
import time
from datetime import datetime
from typing import List, Optional, Dict, Any
import redis.asyncio as aioredis
from src.clients.qdrant_client import QdrantClientWrapper
from src.clients.ollama_client import OllamaClient
from src.config import Settings
from src.models.volatile import (
VolatileRecord,
VolatileRecordResponse,
VolatileNamespace,
NAMESPACE_DEFAULT_TTL,
@@ -28,47 +29,46 @@ logger = logging.getLogger(__name__)
class VolatileCacheService:
"""
Service for volatile data with TTL.
Service for volatile data with TTL stored in Qdrant.
Stores ephemeral data in Redis with automatic expiration.
Supports multiple namespaces with configurable TTLs.
Stores ephemeral data as vectors for semantic search retrieval.
Each user has an isolated volatile collection.
"""
# Redis key prefix for volatile data
KEY_PREFIX = "volatile"
COLLECTION_PREFIX = "volatile_"
def __init__(
self,
redis_client: aioredis.Redis,
qdrant_client: QdrantClientWrapper,
ollama_client: OllamaClient,
settings: Settings
):
"""
Initialize volatile cache service.
Args:
redis_client: Async Redis client
qdrant_client: Qdrant client for vector storage
ollama_client: Ollama client for embeddings
settings: Application settings
"""
self.redis = redis_client
self.qdrant = qdrant_client
self.ollama = ollama_client
self.settings = settings
logger.info("Initialized VolatileCacheService")
logger.info("Initialized VolatileCacheService (Qdrant backend)")
def _build_key(self, user: str, namespace: str, key: str) -> str:
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:
"""
Build Redis key for volatile record.
Generate deterministic vector ID for namespace/key.
Pattern: {user}:volatile:{namespace}:{key_hash}
Uses hash to ensure safe key characters and consistent length.
Same namespace+key always produces same ID for upsert behavior.
"""
key_hash = hashlib.md5(key.encode()).hexdigest()[:12]
return f"{user}:{self.KEY_PREFIX}:{namespace}:{key_hash}"
def _build_pattern(self, user: str, namespace: Optional[str] = None) -> str:
"""Build pattern for key scanning."""
if namespace:
return f"{user}:{self.KEY_PREFIX}:{namespace}:*"
return f"{user}:{self.KEY_PREFIX}:*"
combined = f"{namespace}:{key}"
return hashlib.md5(combined.encode()).hexdigest()
def _get_default_ttl(self, namespace: str) -> int:
"""Get default TTL for a namespace."""
@@ -78,84 +78,116 @@ class VolatileCacheService:
except ValueError:
return self.settings.volatile_default_ttl
def _serialize_record(self, record: VolatileRecord) -> str:
"""Serialize record to JSON for storage."""
return json.dumps({
"key": record.key,
"namespace": record.namespace,
"data": record.data,
"source": record.source,
"created_at": record.created_at.isoformat(),
"updated_at": record.updated_at.isoformat(),
"ttl": record.ttl,
"refresh_schedule": record.refresh_schedule,
"user": record.user,
})
def _current_timestamp_ms(self) -> int:
"""Get current timestamp in milliseconds."""
return int(time.time() * 1000)
def _deserialize_record(self, data: str) -> VolatileRecord:
"""Deserialize record from JSON."""
obj = json.loads(data)
return VolatileRecord(
key=obj["key"],
namespace=obj["namespace"],
data=obj["data"],
source=obj.get("source"),
created_at=datetime.fromisoformat(obj["created_at"]),
updated_at=datetime.fromisoformat(obj["updated_at"]),
ttl=obj["ttl"],
refresh_schedule=obj.get("refresh_schedule"),
user=obj["user"],
)
async def get(
def _to_natural_language(
self,
user: str,
namespace: str,
key: str
) -> Optional[VolatileRecordResponse]:
key: str,
data: Dict[str, Any]
) -> str:
"""
Get a volatile record.
Convert structured data to natural language for embedding.
Args:
user: User identifier
namespace: Data namespace
key: Record key
Returns:
Record if found and not expired, None otherwise
This creates a text representation that embeds well semantically.
"""
redis_key = self._build_key(user, namespace, key)
# 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
try:
data = await self.redis.get(redis_key)
if not data:
return None
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
record = self._deserialize_record(data)
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
# Get TTL remaining
ttl_remaining = await self.redis.ttl(redis_key)
if ttl_remaining < 0:
return None
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
return VolatileRecordResponse(
key=record.key,
namespace=record.namespace,
data=record.data,
source=record.source,
created_at=record.created_at,
updated_at=record.updated_at,
ttl=record.ttl,
ttl_remaining=max(0, ttl_remaining),
refresh_schedule=record.refresh_schedule,
user=record.user,
)
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
except Exception as e:
logger.error(f"Failed to get volatile record {redis_key}: {e}")
return None
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
async def set(
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,
@@ -166,13 +198,13 @@ class VolatileCacheService:
refresh_schedule: Optional[str] = None
) -> VolatileRecordResponse:
"""
Store or update a volatile record.
Store volatile data as an embedded vector.
Args:
user: User identifier
namespace: Data namespace
key: Record key
data: Content to store
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
@@ -180,49 +212,158 @@ class VolatileCacheService:
Returns:
The stored record
"""
redis_key = self._build_key(user, namespace, key)
collection = self._collection_name(user)
# Use provided TTL or namespace default
# 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)
# Check if record exists (for created_at)
existing = await self.get(user, namespace, key)
# 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,
}
record = VolatileRecord(
# 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=existing.created_at if existing else now,
created_at=now,
updated_at=now,
ttl=effective_ttl,
ttl_remaining=effective_ttl,
refresh_schedule=refresh_schedule,
user=user,
)
try:
serialized = self._serialize_record(record)
await self.redis.setex(redis_key, effective_ttl, serialized)
async def search(
self,
user: str,
query: str,
limit: int = 5,
score_threshold: float = 0.75
) -> List[VolatileRecordResponse]:
"""
Semantic search across volatile data.
logger.debug(f"Stored volatile record {redis_key} with TTL {effective_ttl}s")
Args:
user: User identifier
query: Search query
limit: Maximum results
score_threshold: Minimum similarity score (higher = stricter)
return VolatileRecordResponse(
key=record.key,
namespace=record.namespace,
data=record.data,
source=record.source,
created_at=record.created_at,
updated_at=record.updated_at,
ttl=record.ttl,
ttl_remaining=effective_ttl,
refresh_schedule=record.refresh_schedule,
user=record.user,
)
Returns:
List of matching volatile records
"""
collection = self._collection_name(user)
except Exception as e:
logger.error(f"Failed to store volatile record {redis_key}: {e}")
raise
# 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,
@@ -231,7 +372,7 @@ class VolatileCacheService:
key: str
) -> bool:
"""
Delete a volatile record.
Delete a specific volatile record.
Args:
user: User identifier
@@ -239,51 +380,24 @@ class VolatileCacheService:
key: Record key
Returns:
True if record was deleted, False if not found
True if deleted, False if not found
"""
redis_key = self._build_key(user, namespace, key)
collection = self._collection_name(user)
try:
deleted = await self.redis.delete(redis_key)
if deleted:
logger.debug(f"Deleted volatile record {redis_key}")
return deleted > 0
except Exception as e:
logger.error(f"Failed to delete volatile record {redis_key}: {e}")
if not await self.qdrant.collection_exists(collection):
return False
async def list_namespace(
self,
user: str,
namespace: str
) -> List[str]:
"""
List all keys in a namespace.
Args:
user: User identifier
namespace: Data namespace
Returns:
List of keys (original keys, not Redis keys)
"""
pattern = self._build_pattern(user, namespace)
vector_id = self._make_vector_id(namespace, key)
try:
keys = []
async for redis_key in self.redis.scan_iter(match=pattern):
# Get the record to retrieve original key
data = await self.redis.get(redis_key)
if data:
record = self._deserialize_record(data)
keys.append(record.key)
return keys
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 list namespace {namespace}: {e}")
return []
logger.error(f"Failed to delete volatile {namespace}:{key}: {e}")
return False
async def get_scheduled(
self,
@@ -300,33 +414,49 @@ class VolatileCacheService:
Returns:
List of records with refresh_schedule set
"""
pattern = self._build_pattern(user)
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:
scheduled = []
async for redis_key in self.redis.scan_iter(match=pattern):
data = await self.redis.get(redis_key)
if data:
record = self._deserialize_record(data)
if record.refresh_schedule:
ttl_remaining = await self.redis.ttl(redis_key)
scheduled.append(VolatileRecordResponse(
key=record.key,
namespace=record.namespace,
data=record.data,
source=record.source,
created_at=record.created_at,
updated_at=record.updated_at,
ttl=record.ttl,
ttl_remaining=max(0, ttl_remaining),
refresh_schedule=record.refresh_schedule,
user=record.user,
))
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 records: {e}")
logger.error(f"Failed to get scheduled volatile records: {e}")
return []
async def get_stats(
@@ -342,92 +472,93 @@ class VolatileCacheService:
Returns:
Statistics dict
"""
pattern = self._build_pattern(user)
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:
by_namespace: Dict[str, int] = {}
total = 0
scheduled = 0
all_points = await self.qdrant.scroll_all_points(
collection_name=collection,
with_payload=True
)
async for redis_key in self.redis.scan_iter(match=pattern):
data = await self.redis.get(redis_key)
if data:
record = self._deserialize_record(data)
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[record.namespace] = by_namespace.get(record.namespace, 0) + 1
if record.refresh_schedule:
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,
"total_memory_bytes": None, # Could implement with DEBUG MEMORY
"expired_count": expired,
}
except Exception as e:
logger.error(f"Failed to get stats: {e}")
logger.error(f"Failed to get volatile stats: {e}")
return {
"total_records": 0,
"by_namespace": {},
"scheduled_count": 0,
"total_memory_bytes": None,
"expired_count": 0,
}
async def delete_namespace(
self,
user: str,
namespace: str
) -> int:
"""
Delete all records in a namespace.
Args:
user: User identifier
namespace: Data namespace
Returns:
Number of records deleted
"""
pattern = self._build_pattern(user, namespace)
try:
deleted = 0
async for redis_key in self.redis.scan_iter(match=pattern):
await self.redis.delete(redis_key)
deleted += 1
logger.info(f"Deleted {deleted} records from namespace {namespace}")
return deleted
except Exception as e:
logger.error(f"Failed to delete namespace {namespace}: {e}")
return 0
async def delete_all(
async def purge_expired(
self,
user: str
) -> int:
"""
Delete all volatile records for user.
Purge all expired volatile records for user.
Args:
user: User identifier
Returns:
Number of records deleted
Number of records purged
"""
pattern = self._build_pattern(user)
collection = self._collection_name(user)
try:
deleted = 0
async for redis_key in self.redis.scan_iter(match=pattern):
await self.redis.delete(redis_key)
deleted += 1
logger.info(f"Deleted all {deleted} volatile records for user {user}")
return deleted
except Exception as e:
logger.error(f"Failed to delete all records: {e}")
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
+317 -37
View File
@@ -1,11 +1,13 @@
"""
Tests for volatile cache router and service.
Tests for volatile cache router and service (Qdrant backend).
Tests:
- Volatile record CRUD operations
- Namespace listing and management
- Scheduled record retrieval
- TTL behavior
- TTL behavior and expiry filtering
- Semantic search
- Natural language conversion
"""
import pytest
@@ -106,6 +108,10 @@ class TestVolatileNamespaces:
"""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."""
@@ -210,18 +216,28 @@ class TestVolatileDeleteResponses:
class TestVolatileService:
"""Test VolatileCacheService functionality."""
"""Test VolatileCacheService functionality (Qdrant backend)."""
@pytest.fixture
def mock_redis(self):
"""Create mock Redis client."""
redis = AsyncMock()
redis.get = AsyncMock(return_value=None)
redis.setex = AsyncMock()
redis.delete = AsyncMock(return_value=1)
redis.ttl = AsyncMock(return_value=1500)
redis.scan_iter = MagicMock(return_value=iter([]))
return redis
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):
@@ -231,29 +247,29 @@ class TestVolatileService:
return settings
@pytest.fixture
def volatile_service(self, mock_redis, mock_settings):
def volatile_service(self, mock_qdrant, mock_ollama, mock_settings):
"""Create VolatileCacheService with mocks."""
from src.services.volatile_service import VolatileCacheService
return VolatileCacheService(
redis_client=mock_redis,
qdrant_client=mock_qdrant,
ollama_client=mock_ollama,
settings=mock_settings
)
def test_build_key(self, volatile_service):
"""Test Redis key building."""
key = volatile_service._build_key("jpmschweitzer", "weather", "rotterdam")
assert key.startswith("jpmschweitzer:volatile:weather:")
assert len(key) > 30 # Has hash suffix
def test_collection_name(self, volatile_service):
"""Test collection naming pattern."""
name = volatile_service._collection_name("jpmschweitzer")
assert name == "volatile_jpmschweitzer"
def test_build_pattern(self, volatile_service):
"""Test pattern building."""
pattern = volatile_service._build_pattern("jpmschweitzer", "weather")
assert pattern == "jpmschweitzer:volatile:weather:*"
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")
def test_build_pattern_all(self, volatile_service):
"""Test pattern building for all namespaces."""
pattern = volatile_service._build_pattern("jpmschweitzer")
assert pattern == "jpmschweitzer:volatile:*"
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."""
@@ -265,23 +281,287 @@ class TestVolatileService:
ttl = volatile_service._get_default_ttl("unknown_namespace")
assert ttl == 3600 # Falls back to settings default
@pytest.mark.asyncio
async def test_get_not_found(self, volatile_service, mock_redis):
"""Test get when record not found."""
mock_redis.get.return_value = None
result = await volatile_service.get("jpmschweitzer", "weather", "rotterdam")
assert result is None
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_delete_success(self, volatile_service, mock_redis):
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_redis.delete.return_value = 1
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_redis):
async def test_delete_not_found(self, volatile_service, mock_qdrant):
"""Test delete when record not found."""
mock_redis.delete.return_value = 0
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