Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
9e7d8394f3 | ||
|
|
983a934b85 | ||
|
|
6d5760c297 | ||
|
|
867de65354 |
+1
-1
@@ -7,7 +7,7 @@ QDRANT_PORT=6333
|
||||
OLLAMA_URL=http://192.168.86.149:11434
|
||||
SEARXNG_URL=http://192.168.86.149:8080
|
||||
REDIS_HOST=192.168.86.149
|
||||
PAPERLESS_URL=http://192.168.86.149:8000
|
||||
PAPERLESS_URL=http://192.168.86.149:8091
|
||||
|
||||
OLLAMA_MODEL=mistral-nemo-large:latest
|
||||
OLLAMA_EMBEDDING_MODEL=nomic-embed-text
|
||||
|
||||
@@ -5,6 +5,28 @@ 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.8] - 2025-12-25
|
||||
|
||||
### Added
|
||||
|
||||
- **Paperless Orphan Cleanup** - `POST /maintenance/cleanup/paperless` endpoint
|
||||
- Detects documents deleted from Paperless but still indexed in Library Desk
|
||||
- Removes orphaned vectors and graph nodes
|
||||
- Supports `dry_run=true` for preview mode
|
||||
|
||||
## [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
|
||||
|
||||
+1
-1
@@ -1,6 +1,6 @@
|
||||
[project]
|
||||
name = "library-desk"
|
||||
version = "1.4.6"
|
||||
version = "1.4.8"
|
||||
description = "Coordination service for The Library system - HybridRAG queries, document ingestion, entity extraction, and knowledge consolidation"
|
||||
readme = "README.md"
|
||||
requires-python = ">=3.12"
|
||||
|
||||
+18
-13
@@ -86,22 +86,27 @@ class DocumentUploadResponse(BaseModel):
|
||||
|
||||
|
||||
class PaperlessWebhookPayload(BaseModel):
|
||||
"""Payload from Paperless-ngx webhook (include_document=true format)."""
|
||||
id: int = Field(..., description="Paperless document ID")
|
||||
"""
|
||||
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")
|
||||
content: Optional[str] = Field(None, description="Document text content")
|
||||
correspondent: Optional[int] = Field(None, description="Correspondent ID")
|
||||
document_type: Optional[int] = Field(None, description="Document type ID")
|
||||
tags: List[int] = Field(default_factory=list, description="Tag IDs")
|
||||
created: Optional[str] = Field(None, description="Created date")
|
||||
modified: Optional[str] = Field(None, description="Modified datetime")
|
||||
added: Optional[str] = Field(None, description="Added datetime")
|
||||
original_file_name: Optional[str] = Field(None, description="Original filename")
|
||||
owner: Optional[int] = Field(None, description="Owner user ID")
|
||||
custom_fields: List[Dict[str, Any]] = Field(default_factory=list, description="Custom fields")
|
||||
|
||||
class Config:
|
||||
extra = "ignore" # Ignore extra fields from Paperless
|
||||
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):
|
||||
|
||||
+133
-9
@@ -7,7 +7,7 @@ Event-driven integration with Paperless-ngx:
|
||||
- Search across indexed documents
|
||||
"""
|
||||
|
||||
from fastapi import APIRouter, HTTPException, Depends, Query, UploadFile, File
|
||||
from fastapi import APIRouter, HTTPException, Depends, Query, UploadFile, File, Request
|
||||
from typing import Optional
|
||||
import logging
|
||||
import time
|
||||
@@ -64,12 +64,138 @@ async def receive_webhook(
|
||||
"""
|
||||
from src.services.document_sync_service import DocumentSyncService
|
||||
|
||||
logger.info(f"Webhook received: document_id={payload.id}, title={payload.title}")
|
||||
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=payload.id,
|
||||
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"
|
||||
@@ -86,23 +212,21 @@ async def receive_webhook(
|
||||
)
|
||||
|
||||
result = await sync_service.index_document(
|
||||
document_id=payload.id,
|
||||
document_id=doc_id,
|
||||
user=user,
|
||||
content=payload.content, # Use content from webhook payload
|
||||
title=payload.title,
|
||||
)
|
||||
|
||||
return WebhookResponse(
|
||||
document_id=payload.id,
|
||||
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 {payload.id}: {e}", exc_info=True)
|
||||
logger.error(f"Webhook-simple failed for document {doc_id}: {e}", exc_info=True)
|
||||
return WebhookResponse(
|
||||
document_id=payload.id,
|
||||
document_id=doc_id,
|
||||
status="error",
|
||||
indexed=False,
|
||||
message=str(e)
|
||||
|
||||
+113
-1
@@ -19,7 +19,7 @@ from src.services.graph_service import GraphService
|
||||
from src.services.volatile_service import VolatileCacheService
|
||||
from src.core.dependencies import (
|
||||
VectorServiceDep, GraphServiceDep, WikiJSDep, RedisDep,
|
||||
QdrantDep, OllamaDep, verify_api_key
|
||||
QdrantDep, OllamaDep, PaperlessDep, verify_api_key
|
||||
)
|
||||
from src.config import get_settings
|
||||
from src.core.multi_tenancy import DEFAULT_USER
|
||||
@@ -232,6 +232,18 @@ class TestDataCleanupResponse(BaseModel):
|
||||
duration_ms: float
|
||||
|
||||
|
||||
class PaperlessCleanupResponse(BaseModel):
|
||||
"""Response from Paperless orphan cleanup operation."""
|
||||
success: bool
|
||||
dry_run: bool
|
||||
paperless_ids_checked: int = Field(description="Total Paperless IDs found in indexes")
|
||||
orphans_found: int = Field(description="Documents deleted from Paperless but still indexed")
|
||||
orphan_ids: List[int] = Field(default_factory=list, description="Paperless IDs that are orphans")
|
||||
vector_chunks_deleted: int = Field(description="Vector chunks removed")
|
||||
graph_nodes_deleted: int = Field(description="Graph Document nodes removed")
|
||||
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 = [
|
||||
@@ -675,6 +687,106 @@ async def cleanup_test_data(
|
||||
raise HTTPException(status_code=500, detail=str(e))
|
||||
|
||||
|
||||
@router.post("/cleanup/paperless", response_model=PaperlessCleanupResponse)
|
||||
async def cleanup_paperless_orphans(
|
||||
user: str = Query(..., description="User identifier"),
|
||||
dry_run: bool = Query(default=True, description="Preview only, don't delete"),
|
||||
vector_service: VectorServiceDep = None,
|
||||
graph_service: GraphServiceDep = None,
|
||||
paperless: PaperlessDep = None,
|
||||
api_key: str = Depends(verify_api_key)
|
||||
):
|
||||
"""
|
||||
Find and clean up Paperless document orphans.
|
||||
|
||||
Detects documents that were indexed in Library Desk but have since been
|
||||
deleted from Paperless-ngx. Removes orphaned vectors and graph nodes.
|
||||
|
||||
**Use dry_run=true (default) to preview what would be deleted.**
|
||||
|
||||
**Scheduler Integration:**
|
||||
```json
|
||||
{
|
||||
"task_name": "paperless_orphan_cleanup",
|
||||
"schedule": "0 5 * * *",
|
||||
"endpoint": "POST /maintenance/cleanup/paperless?user=jpmschweitzer&dry_run=false",
|
||||
"description": "Daily cleanup of orphaned Paperless documents"
|
||||
}
|
||||
```
|
||||
"""
|
||||
start_time = time.time()
|
||||
|
||||
try:
|
||||
settings = get_settings()
|
||||
if not settings.paperless_token:
|
||||
raise HTTPException(status_code=503, detail="Paperless not configured")
|
||||
|
||||
# Get all document chunks from vectors with doc_type="document"
|
||||
chunk_refs = await vector_service.get_all_chunk_references(user)
|
||||
doc_chunks = [ref for ref in chunk_refs if ref.get("doc_type") == "document"]
|
||||
|
||||
# Extract unique paperless_ids
|
||||
paperless_ids = list(set(
|
||||
ref.get("paperless_id") for ref in doc_chunks
|
||||
if ref.get("paperless_id")
|
||||
))
|
||||
|
||||
logger.info(f"Found {len(paperless_ids)} unique Paperless IDs in indexes")
|
||||
|
||||
# Check each against Paperless API
|
||||
orphan_ids = []
|
||||
for pid in paperless_ids:
|
||||
try:
|
||||
doc = await paperless.get_document(pid)
|
||||
if doc is None:
|
||||
orphan_ids.append(pid)
|
||||
except Exception as e:
|
||||
# Document not found or API error - treat as orphan
|
||||
logger.debug(f"Paperless document {pid} not found: {e}")
|
||||
orphan_ids.append(pid)
|
||||
|
||||
logger.info(f"Found {len(orphan_ids)} orphaned Paperless documents")
|
||||
|
||||
# Delete orphans if not dry run
|
||||
vectors_deleted = 0
|
||||
graph_deleted = 0
|
||||
|
||||
if not dry_run and orphan_ids:
|
||||
for pid in orphan_ids:
|
||||
try:
|
||||
# Delete vector chunks for this paperless_id
|
||||
chunks_removed = await vector_service.delete_paperless_document_chunks(pid, user)
|
||||
vectors_deleted += chunks_removed
|
||||
|
||||
# Delete graph node for this paperless_id
|
||||
graph_removed = await graph_service.delete_paperless_document(pid, user)
|
||||
graph_deleted += graph_removed
|
||||
|
||||
logger.info(f"Cleaned up orphaned Paperless document {pid}: {chunks_removed} chunks, {graph_removed} nodes")
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to cleanup Paperless document {pid}: {e}")
|
||||
|
||||
duration_ms = (time.time() - start_time) * 1000
|
||||
|
||||
return PaperlessCleanupResponse(
|
||||
success=True,
|
||||
dry_run=dry_run,
|
||||
paperless_ids_checked=len(paperless_ids),
|
||||
orphans_found=len(orphan_ids),
|
||||
orphan_ids=orphan_ids,
|
||||
vector_chunks_deleted=vectors_deleted,
|
||||
graph_nodes_deleted=graph_deleted,
|
||||
duration_ms=duration_ms
|
||||
)
|
||||
|
||||
except HTTPException:
|
||||
raise
|
||||
except Exception as e:
|
||||
logger.error(f"Paperless orphan 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"),
|
||||
|
||||
@@ -281,10 +281,13 @@ class DocumentSyncService:
|
||||
"""Mark document as indexed in Paperless custom field."""
|
||||
# Try to update library_indexed custom field if it exists
|
||||
try:
|
||||
await self.paperless.update_document(
|
||||
document_id=document_id,
|
||||
custom_fields=[{"field": "library_indexed", "value": True}]
|
||||
)
|
||||
# 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
|
||||
|
||||
@@ -1306,6 +1306,48 @@ Feel free to expand it with more details!
|
||||
logger.error(f"Failed to delete document {document_id} from graph: {e}", exc_info=True)
|
||||
return 0
|
||||
|
||||
async def delete_paperless_document(
|
||||
self,
|
||||
paperless_id: int,
|
||||
user: str
|
||||
) -> int:
|
||||
"""
|
||||
Delete a Paperless document node and all its relationships.
|
||||
|
||||
Args:
|
||||
paperless_id: Paperless-ngx document ID
|
||||
user: User identifier
|
||||
|
||||
Returns:
|
||||
Number of nodes deleted (1 if successful, 0 if not found)
|
||||
"""
|
||||
user_doc_label = get_neo4j_user_label(user)
|
||||
|
||||
delete_query = f"""
|
||||
MATCH (d:{user_doc_label}:Document {{paperless_id: $paperless_id}})
|
||||
DETACH DELETE d
|
||||
RETURN count(d) as deleted_count
|
||||
"""
|
||||
|
||||
try:
|
||||
result = await self.neo4j.execute_query(
|
||||
delete_query,
|
||||
{"paperless_id": paperless_id}
|
||||
)
|
||||
|
||||
deleted_count = result[0]["deleted_count"] if result else 0
|
||||
|
||||
if deleted_count > 0:
|
||||
logger.info(f"Deleted Document node for Paperless document {paperless_id}")
|
||||
else:
|
||||
logger.debug(f"No Document node found for Paperless document {paperless_id}")
|
||||
|
||||
return deleted_count
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to delete Paperless document {paperless_id} from graph: {e}", exc_info=True)
|
||||
return 0
|
||||
|
||||
async def delete_collection_node(
|
||||
self,
|
||||
collection_id: str,
|
||||
|
||||
@@ -389,6 +389,39 @@ class VectorService:
|
||||
logger.error(f"Failed to delete chunks for document {document_id}: {e}", exc_info=True)
|
||||
return 0
|
||||
|
||||
async def delete_paperless_document_chunks(
|
||||
self,
|
||||
paperless_id: int,
|
||||
user: str
|
||||
) -> int:
|
||||
"""
|
||||
Delete all chunks for a Paperless document.
|
||||
|
||||
Args:
|
||||
paperless_id: Paperless-ngx document ID
|
||||
user: User identifier
|
||||
|
||||
Returns:
|
||||
Number of chunks deleted
|
||||
"""
|
||||
collection_name = get_qdrant_collection_name(user)
|
||||
|
||||
try:
|
||||
deleted_count = await self.qdrant.delete_by_filter(
|
||||
collection_name=collection_name,
|
||||
filter_conditions={
|
||||
"doc_type": "document",
|
||||
"paperless_id": paperless_id
|
||||
}
|
||||
)
|
||||
|
||||
logger.info(f"Deleted chunks for Paperless document {paperless_id}")
|
||||
return deleted_count
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to delete chunks for Paperless document {paperless_id}: {e}", exc_info=True)
|
||||
return 0
|
||||
|
||||
async def delete_collection_chunks(
|
||||
self,
|
||||
collection_id: str,
|
||||
|
||||
Reference in New Issue
Block a user