Spaces:
Running
Running
File size: 5,066 Bytes
e9d730a |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 |
# src/implementations/document_service.py
from pathlib import Path
import shutil
import os
import uuid
from typing import List, Tuple
from fastapi import UploadFile, BackgroundTasks
from ..vectorstores.chroma_vectorstore import ChromaVectorStore
from ..utils.document_processor import DocumentProcessor
from ..models import DocumentResponse, DocumentInfo, BatchUploadResponse
from ..utils.logger import logger
class DocumentService:
def __init__(self, doc_processor: DocumentProcessor):
self.doc_processor = doc_processor
self.upload_dir = Path("temp_uploads")
self.upload_dir.mkdir(exist_ok=True)
async def process_documents(
self,
files: List[UploadFile],
vector_store: ChromaVectorStore,
background_tasks: BackgroundTasks
) -> BatchUploadResponse:
"""Process multiple document uploads"""
processed_files, failed_files = await self._handle_file_uploads(
files,
vector_store,
background_tasks
)
return BatchUploadResponse(
message=f"Processed {len(processed_files)} documents with {len(failed_files)} failures",
processed_files=processed_files,
failed_files=failed_files
)
async def _handle_file_uploads(
self,
files: List[UploadFile],
vector_store: ChromaVectorStore,
background_tasks: BackgroundTasks
) -> Tuple[List[DocumentResponse], List[dict]]:
"""Handle individual file uploads and processing"""
processed_files = []
failed_files = []
for file in files:
try:
if not self._is_supported_format(file.filename):
failed_files.append(self._create_failed_file_entry(
file.filename,
"Unsupported file format"
))
continue
document_response = await self._process_single_file(
file,
vector_store,
background_tasks
)
processed_files.append(document_response)
except Exception as e:
logger.error(f"Error processing file {file.filename}: {str(e)}")
failed_files.append(self._create_failed_file_entry(
file.filename,
str(e)
))
return processed_files, failed_files
def _is_supported_format(self, filename: str) -> bool:
"""Check if file format is supported"""
return any(filename.lower().endswith(ext)
for ext in self.doc_processor.supported_formats)
async def _process_single_file(
self,
file: UploadFile,
vector_store: ChromaVectorStore,
background_tasks: BackgroundTasks
) -> DocumentResponse:
"""Process a single file upload"""
document_id = str(uuid.uuid4())
temp_path = self.upload_dir / f"{document_id}_{file.filename}"
# Save file
with open(temp_path, "wb") as buffer:
shutil.copyfileobj(file.file, buffer)
# Add background task for processing
background_tasks.add_task(
self._process_and_store_document,
temp_path,
vector_store,
document_id
)
return DocumentResponse(
message="Document queued for processing",
document_id=document_id,
status="processing",
document_info=DocumentInfo(
original_filename=file.filename,
size=os.path.getsize(temp_path),
content_type=file.content_type
)
)
async def _process_and_store_document(
self,
file_path: Path,
vector_store: ChromaVectorStore,
document_id: str
):
"""Process document and store in vector database"""
try:
processed_doc = await self.doc_processor.process_document(file_path)
vector_store.add_documents(
documents=processed_doc['chunks'],
metadatas=[{
'document_id': document_id,
'chunk_id': i,
'source': str(file_path.name),
'metadata': processed_doc['metadata']
} for i in range(len(processed_doc['chunks']))],
ids=[f"{document_id}_chunk_{i}" for i in range(len(processed_doc['chunks']))]
)
return processed_doc
finally:
if file_path.exists():
file_path.unlink()
def _create_failed_file_entry(self, filename: str, error: str) -> dict:
"""Create a failed file entry"""
return {
"filename": filename,
"error": error
}
def cleanup(self):
"""Clean up upload directory"""
if self.upload_dir.exists() and not any(self.upload_dir.iterdir()):
self.upload_dir.rmdir() |