| """Document processing and management service""" |
|
|
| import os |
| from pathlib import Path |
| from typing import Optional, List, Dict, Any |
| from datetime import datetime |
| from uuid import uuid4 |
| import logging |
|
|
| logger = logging.getLogger(__name__) |
|
|
|
|
| class DocumentService: |
| """Service for managing documents""" |
|
|
| def __init__(self, upload_dir: str): |
| self.upload_dir = Path(upload_dir) |
| self.upload_dir.mkdir(parents=True, exist_ok=True) |
| self.documents: Dict[str, Dict[str, Any]] = {} |
|
|
| async def save_document( |
| self, |
| file_content: bytes, |
| filename: str, |
| document_type: str, |
| metadata: Optional[Dict[str, Any]] = None, |
| ) -> Dict[str, Any]: |
| """Save uploaded document""" |
| doc_id = str(uuid4()) |
| file_path = self.upload_dir / f"{doc_id}_{filename}" |
|
|
| |
| file_path.write_bytes(file_content) |
|
|
| |
| doc_info = { |
| "id": doc_id, |
| "name": filename, |
| "document_type": document_type, |
| "upload_date": datetime.now(), |
| "file_size": len(file_content), |
| "file_path": str(file_path), |
| "chunks_count": 0, |
| "metadata": metadata or {}, |
| } |
| self.documents[doc_id] = doc_info |
|
|
| logger.info(f"Document saved: {doc_id} ({filename})") |
| return doc_info |
|
|
| def get_document(self, doc_id: str) -> Optional[Dict[str, Any]]: |
| """Get document info""" |
| return self.documents.get(doc_id) |
|
|
| def list_documents(self) -> List[Dict[str, Any]]: |
| """List all documents""" |
| return list(self.documents.values()) |
|
|
| def delete_document(self, doc_id: str) -> bool: |
| """Delete document""" |
| if doc_id in self.documents: |
| doc = self.documents[doc_id] |
| file_path = Path(doc["file_path"]) |
| if file_path.exists(): |
| file_path.unlink() |
| del self.documents[doc_id] |
| logger.info(f"Document deleted: {doc_id}") |
| return True |
| return False |
|
|
| def update_chunks_count(self, doc_id: str, count: int) -> None: |
| """Update chunk count for document""" |
| if doc_id in self.documents: |
| self.documents[doc_id]["chunks_count"] = count |
|
|
