m.dabbagh/text_processor
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix

Browse Source
This commit is contained in:
m.dabbagh 2026-01-19 15:42:46 +03:30
parent e783d92eca
commit 0084ae6bc0
1 changed files with 10 additions and 27 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

37
src/adapters/incoming/api_routes.py
View File

@ -65,24 +65,6 @@ def _get_service() -> ITextProcessor:
return get_processor_service() return get_processor_service()
def _to_domain_strategy(request_strategy) -> ChunkingStrategy:
"""
Convert API request strategy to domain model.
Args:
request_strategy: API request strategy schema
Returns:
ChunkingStrategy: Domain strategy model
"""
return ChunkingStrategy(
strategy_name=request_strategy.strategy_name,
chunk_size=request_strategy.chunk_size,
overlap_size=request_strategy.overlap_size,
respect_boundaries=request_strategy.respect_boundaries,
)
def _to_document_response(document) -> DocumentResponse: def _to_document_response(document) -> DocumentResponse:
""" """
Convert domain document to API response. Convert domain document to API response.
@ -221,14 +203,14 @@ async def process_file(
# Pull service from bootstrap # Pull service from bootstrap
service: ITextProcessor = _get_service() service: ITextProcessor = _get_service()
# Create temporary file with appropriate suffix # Create temporary directory and file with original filename
suffix = Path(file.filename).suffix if file.filename else ".tmp" temp_dir = tempfile.mkdtemp()
temp_file = tempfile.NamedTemporaryFile(delete=False, suffix=suffix) original_filename = file.filename if file.filename else "uploaded_file.tmp"
temp_file_path = Path(temp_file.name) temp_file_path = Path(temp_dir) / original_filename
# Copy uploaded file to temporary location # Copy uploaded file to temporary location
logger.info(f"Processing uploaded file: {file.filename}") logger.info(f"Processing uploaded file: {file.filename}")
with temp_file: with open(temp_file_path, 'wb') as temp_file:
shutil.copyfileobj(file.file, temp_file) shutil.copyfileobj(file.file, temp_file)
# Create chunking strategy # Create chunking strategy
@ -261,13 +243,14 @@ async def process_file(
detail=f"Internal server error: {str(e)}", detail=f"Internal server error: {str(e)}",
) )
finally: finally:
# Clean up temporary file # Clean up temporary file and directory
if temp_file_path and temp_file_path.exists(): if temp_file_path and temp_file_path.exists():
try: try:
temp_file_path.unlink() temp_dir = temp_file_path.parent
logger.debug(f"Cleaned up temporary file: {temp_file_path}") shutil.rmtree(temp_dir)
logger.debug(f"Cleaned up temporary directory: {temp_dir}")
except Exception as e: except Exception as e:
logger.warning(f"Failed to delete temporary file {temp_file_path}: {str(e)}") logger.warning(f"Failed to delete temporary directory: {str(e)}")
@router.get( @router.get(