"""Document upload and management endpoints.""" from uuid import UUID, uuid4 import structlog from fastapi import APIRouter, File, HTTPException, UploadFile from fastapi.responses import JSONResponse from sqlalchemy import select from src.api.dependencies import ApiKeyDep, AsyncSessionDep from src.api.tenant_context import TenantContextDep, set_rls_tenant from src.api.schemas.documents import ( BatchUploadResponse, DocumentDeletionResponse, DocumentProgress, DocumentStatus, DocumentStatusResponse, DocumentUploadResponse, ValidationErrorDetail, ) from src.api.schemas.symbols import ( SymbolDeleteResponse, SymbolEntry, SymbolListResponse, SymbolUploadRequest, SymbolUploadResponse, ) from src.core.config import settings from src.db.models import Document from src.services.batch_validator import BatchValidationError, validate_batch from src.services.document_service import create_document from src.services.file_storage import FileStorageError, save_uploaded_file from src.services.file_validator import FileValidationError, validate_uploaded_file from src.services.progress_tracker import get_progress_tracker from src.services.symbol_service import symbol_service from src.workers.tasks import process_document, re_resolve_symbols from src.workers.extraction_v2_tasks import extract_document_v2 logger = structlog.get_logger(__name__) router = APIRouter(prefix="/documents", tags=["Documents"]) # Convert MB to bytes for file size validation MAX_UPLOAD_SIZE_BYTES = settings.max_upload_size_mb * 1024 * 1024 @router.post( "", status_code=202, response_model=DocumentUploadResponse, responses={ 202: { "description": "Document accepted for processing", "model": DocumentUploadResponse, }, 400: { "description": "File validation failed", "content": { "application/json": { "examples": { "invalid_extension": { "summary": "Invalid file extension", "value": { "error": { "code": "INVALID_EXTENSION", "message": "File extension must be .xlsx, .xls, .xlsm, .docx, .doc, .docm, .pdf, .pptx, .ppt, or .pptm", } }, }, "invalid_format": { "summary": "Invalid file format (magic bytes mismatch)", "value": { "error": { "code": "INVALID_FORMAT", "message": "File signature does not match supported format", } }, }, "corrupted_file": { "summary": "Corrupted document file", "value": { "error": { "code": "CORRUPTED_FILE", "message": "File cannot be opened or is corrupted", } }, }, "empty_file": { "summary": "Empty document file", "value": { "error": { "code": "EMPTY_FILE", "message": "Document file is empty or has no content", } }, }, } } }, }, 413: { "description": "File size exceeds maximum allowed", "content": { "application/json": { "example": { "error": { "code": "FILE_TOO_LARGE", "message": f"File size exceeds maximum allowed size of {settings.max_upload_size_mb}MB", } } } }, }, 500: { "description": "Internal server error during file upload", "content": { "application/json": { "example": { "error": { "code": "UPLOAD_FAILED", "message": "Failed to save uploaded file", } } } }, }, }, ) async def upload_document( api_key: ApiKeyDep, session: AsyncSessionDep, ctx: TenantContextDep, file: UploadFile = File(..., description="Document file to upload (.xlsx, .xls, .xlsm, .docx, .doc, .docm, .pdf, .pptx, .ppt, or .pptm)"), ) -> DocumentUploadResponse: """Upload a document for processing. Accepts document files (Excel, Word, PDF, PowerPoint) up to 50MB. The file undergoes comprehensive validation (extension, magic bytes, corruption checks) before being saved to storage. A database record is created with status "pending", and an async processing task is queued for background execution. Supported formats: - Excel: .xlsx, .xls, .xlsm - Word: .docx, .doc, .docm - PDF: .pdf - PowerPoint: .pptx, .ppt, .pptm Validation checks (fail-fast order): 1. File extension must be one of the supported formats 2. File size ≤ 50MB 3. Magic bytes match the declared format 4. File is not corrupted and can be processed Args: api_key: Authenticated API key (injected by dependency). session: Async database session (injected by dependency). file: The uploaded Excel file. Returns: DocumentUploadResponse with document_id, filename, status, and message. Raises: HTTPException: 400 if validation fails (INVALID_EXTENSION, INVALID_FORMAT, CORRUPTED_FILE, EMPTY_FILE), 413 if file size exceeds limit, 500 if upload fails. """ logger.info( "document_upload_started", filename=file.filename, content_type=file.content_type, api_key_id=str(api_key.id), ) # Read file to check size and validate content # Note: This loads the entire file into memory for validation # For production, consider streaming validation or nginx client_max_body_size file_content = await file.read() file_size = len(file_content) # Validate file size (fast check first, before deeper validation) if file_size > MAX_UPLOAD_SIZE_BYTES: logger.warning( "file_too_large", filename=file.filename, file_size=file_size, max_size=MAX_UPLOAD_SIZE_BYTES, api_key_id=str(api_key.id), ) raise HTTPException( status_code=413, detail={ "error": { "code": "FILE_TOO_LARGE", "message": f"File size exceeds maximum allowed size of {settings.max_upload_size_mb}MB", } }, ) # Validate file format, corruption, and content # This uses the already-loaded file_content bytes (no additional I/O) try: validate_uploaded_file(file.filename, file_content, MAX_UPLOAD_SIZE_BYTES) except FileValidationError as e: logger.warning( "file_validation_failed", filename=file.filename, error_code=e.code, error_message=e.message, api_key_id=str(api_key.id), ) raise HTTPException( status_code=400, detail={ "error": { "code": e.code, "message": e.message, } }, ) from e # Reset file pointer for subsequent operations await file.seek(0) try: # Create document record first to get document_id document = await create_document( filename=file.filename, file_path="", # Will be updated after file save file_size=file_size, api_key_id=api_key.id, session=session, tenant_id=ctx.tenant_id, node_id=ctx.node_id, ) # Save file to storage file_path = await save_uploaded_file(document.id, file) # Update document with file path document.file_path = file_path session.add(document) await session.commit() await session.refresh(document) # Queue Celery task for async processing using V2 pipeline # V2 automatically falls back to V1 if conversion fails task = extract_document_v2.delay(str(document.id), file_path) logger.info( "document_upload_completed", document_id=str(document.id), filename=file.filename, file_size=file_size, celery_task_id=task.id, api_key_id=str(api_key.id), pipeline_version="v2_with_v1_fallback", ) return DocumentUploadResponse( document_id=document.id, filename=document.filename, status=document.status, message="Document uploaded successfully and queued for processing", ) except FileStorageError as e: logger.error( "file_storage_error", filename=file.filename, error=str(e), api_key_id=str(api_key.id), ) raise HTTPException( status_code=500, detail={ "error": { "code": "UPLOAD_FAILED", "message": "Failed to save uploaded file", } }, ) from e except Exception as e: logger.error( "document_upload_error", filename=file.filename, error=str(e), api_key_id=str(api_key.id), ) raise HTTPException( status_code=500, detail={ "error": { "code": "INTERNAL_ERROR", "message": "An unexpected error occurred during upload", } }, ) from e @router.post( "/batch", status_code=207, response_model=BatchUploadResponse, responses={ 202: { "description": "All documents accepted for processing", "model": BatchUploadResponse, }, 207: { "description": "Multi-Status - Partial success (some files valid, some invalid)", "model": BatchUploadResponse, }, 400: { "description": "Batch validation failed (too many files or batch too large)", "content": { "application/json": { "examples": { "too_many_files": { "summary": "Too many files in batch", "value": { "error": { "code": "TOO_MANY_FILES", "message": "Batch contains 21 files, maximum allowed is 20", } }, }, "batch_too_large": { "summary": "Total batch size exceeds limit", "value": { "error": { "code": "BATCH_TOO_LARGE", "message": "Total batch size 550.00MB exceeds maximum allowed size of 500MB", } }, }, } } }, }, 500: { "description": "Internal server error during batch upload", "content": { "application/json": { "example": { "error": { "code": "INTERNAL_ERROR", "message": "An unexpected error occurred during batch upload", } } } }, }, }, ) async def upload_documents_batch( api_key: ApiKeyDep, session: AsyncSessionDep, ctx: TenantContextDep, files: list[UploadFile] = File(..., description="List of document files to upload (.xlsx, .xls, .xlsm, .docx, .doc, .docm, .pdf, .pptx, .ppt, or .pptm)"), ) -> JSONResponse: """Upload multiple documents in a single batch request. Accepts up to 20 document files (Excel, Word, PDF, PowerPoint) with a maximum total batch size of 500MB. Files are validated and processed individually - valid files are accepted even if some files in the batch fail validation (partial success). Supported formats: - Excel: .xlsx, .xls, .xlsm - Word: .docx, .doc, .docm - PDF: .pdf - PowerPoint: .pptx, .ppt, .pptm Batch-level validation (fails entire batch): 1. File count ≤ 20 files 2. Total batch size ≤ 500MB Per-file validation (partial success allowed): 3. File extension must be one of the supported formats 4. File size ≤ 50MB (individual file limit) 5. Magic bytes match the declared format 6. File is not corrupted and can be processed Args: api_key: Authenticated API key (injected by dependency). session: Async database session (injected by dependency). files: List of uploaded Excel files. Returns: BatchUploadResponse with: - 202 if all files valid - 207 Multi-Status if partial success (some files valid, some invalid) - batch_id for tracking - documents array with accepted files - errors array with failed files and error details Raises: HTTPException: 400 if batch validation fails (too many files, too large), 500 if unexpected server error. """ batch_id = uuid4() logger.info( "batch_upload_started", batch_id=str(batch_id), file_count=len(files), api_key_id=str(api_key.id), ) try: # Batch-level validation (fail entire batch if violated) total_batch_size = await validate_batch(files) logger.info( "batch_validation_passed", batch_id=str(batch_id), file_count=len(files), total_size_bytes=total_batch_size, ) except BatchValidationError as e: logger.warning( "batch_validation_failed", batch_id=str(batch_id), error_code=e.code, error_message=e.message, file_count=len(files), api_key_id=str(api_key.id), ) raise HTTPException( status_code=400, detail={ "error": { "code": e.code, "message": e.message, } }, ) from e # Per-file processing (partial success allowed) successful_docs: list[DocumentStatus] = [] errors: list[ValidationErrorDetail] = [] for file in files: try: # Read file content for validation file_content = await file.read() file_size = len(file_content) # Validate individual file using existing validator # Note: Size check happens in validate_uploaded_file validate_uploaded_file(file.filename, file_content, MAX_UPLOAD_SIZE_BYTES) # Reset file pointer for save operation await file.seek(0) # Create document record document = await create_document( filename=file.filename, file_path="", # Will be updated after file save file_size=file_size, api_key_id=api_key.id, session=session, tenant_id=ctx.tenant_id, node_id=ctx.node_id, ) # Save file to storage file_path = await save_uploaded_file(document.id, file) # Update document with file path document.file_path = file_path session.add(document) await session.commit() await session.refresh(document) # Queue Celery task for async processing using V2 pipeline # V2 automatically falls back to V1 if conversion fails task = extract_document_v2.delay(str(document.id), file_path) logger.info( "batch_document_accepted", batch_id=str(batch_id), document_id=str(document.id), filename=file.filename, file_size=file_size, celery_task_id=task.id, pipeline_version="v2_with_v1_fallback", ) successful_docs.append( DocumentStatus( document_id=document.id, filename=document.filename, status="accepted", ) ) except FileValidationError as e: # Individual file validation failed - add to errors, continue batch logger.warning( "batch_file_validation_failed", batch_id=str(batch_id), filename=file.filename, error_code=e.code, error_message=e.message, ) errors.append( ValidationErrorDetail( filename=file.filename, code=e.code, message=e.message, ) ) continue except FileStorageError as e: # File storage failed - add to errors, continue batch logger.error( "batch_file_storage_error", batch_id=str(batch_id), filename=file.filename, error=str(e), ) errors.append( ValidationErrorDetail( filename=file.filename, code="UPLOAD_FAILED", message="Failed to save uploaded file", ) ) continue except Exception as e: # Unexpected error for this file - add to errors, continue batch logger.error( "batch_file_processing_error", batch_id=str(batch_id), filename=file.filename, error=str(e), ) errors.append( ValidationErrorDetail( filename=file.filename, code="INTERNAL_ERROR", message="An unexpected error occurred processing this file", ) ) continue # Generate summary message success_count = len(successful_docs) error_count = len(errors) if error_count == 0: message = f"Batch accepted: {success_count} documents queued for processing" status_code = 202 elif success_count == 0: message = f"Batch rejected: All {error_count} files failed validation" status_code = 207 else: message = f"Batch accepted: {success_count} documents queued for processing, {error_count} failed validation" status_code = 207 logger.info( "batch_upload_completed", batch_id=str(batch_id), success_count=success_count, error_count=error_count, status_code=status_code, api_key_id=str(api_key.id), ) response_data = BatchUploadResponse( batch_id=batch_id, documents=successful_docs, errors=errors, message=message, ) return JSONResponse( status_code=status_code, content=response_data.model_dump(mode="json"), ) @router.get( "/{document_id}", status_code=200, response_model=DocumentStatusResponse, responses={ 200: { "description": "Document status retrieved successfully", "model": DocumentStatusResponse, }, 404: { "description": "Document not found or not accessible", "content": { "application/json": { "example": { "error": { "code": "DOCUMENT_NOT_FOUND", "message": "Document not found or you do not have access to it", } } } }, }, }, ) async def get_document_status( document_id: UUID, api_key: ApiKeyDep, session: AsyncSessionDep, ctx: TenantContextDep, ) -> DocumentStatusResponse: """Get the current status of a document. Retrieves the processing status of a document with conditional fields based on the current status: - pending: Basic info only - processing: Includes progress information (stage, sheets_processed, sheets_total) - completed: Includes metadata (sheet_count, record_count, etc.) - failed: Includes error_message Args: document_id: UUID of the document to query. api_key: Authenticated API key (injected by dependency). session: Async database session (injected by dependency). Returns: DocumentStatusResponse with status-specific fields. Raises: HTTPException: 404 if document not found or not owned by requesting API key. """ logger.info( "document_status_check_started", document_id=str(document_id), api_key_id=str(api_key.id), ) # Set RLS session variable (defense-in-depth with application-level WHERE) await set_rls_tenant(session, ctx.tenant_id) # Query document by ID, scoped by tenant + node result = await session.execute( select(Document) .where(Document.id == document_id) .where(Document.tenant_id == ctx.tenant_id) .where(Document.node_id == ctx.node_id) ) document = result.scalar_one_or_none() # Check existence and authorization (return 404 for both to avoid leaking existence) if not document or document.api_key_id != api_key.id: logger.warning( "document_not_found_or_unauthorized", document_id=str(document_id), api_key_id=str(api_key.id), document_exists=document is not None, ) raise HTTPException( status_code=404, detail={ "error": { "code": "DOCUMENT_NOT_FOUND", "message": "Document not found or you do not have access to it", } }, ) # Build base response response = DocumentStatusResponse( document_id=document.id, filename=document.filename, status=document.status, created_at=document.created_at, ) # Add status-specific fields if document.status == "processing": # Get progress from Redis progress_tracker = get_progress_tracker() progress_data = await progress_tracker.get_progress(document.id) if progress_data: response.progress = DocumentProgress( stage=progress_data["stage"], sheets_processed=progress_data["sheets_processed"], sheets_total=progress_data["sheets_total"], ) logger.debug( "progress_data_included", document_id=str(document_id), stage=progress_data["stage"], ) else: # Processing but no progress data yet (worker may not have started) logger.debug( "progress_data_not_available", document_id=str(document_id), ) elif document.status == "completed": # Include completion metadata response.processed_at = document.processed_at response.sheet_count = document.sheet_count response.record_count = document.record_count # Note: symbols_found and cross_references will be added in Epic 3 # For now, these fields remain None elif document.status == "failed": # Include error message response.error_message = document.error_message logger.info( "document_status_check_completed", document_id=str(document_id), status=document.status, api_key_id=str(api_key.id), ) return response @router.delete( "/{document_id}", status_code=200, response_model=DocumentDeletionResponse, responses={ 200: { "description": "Document deleted successfully", "model": DocumentDeletionResponse, }, 404: { "description": "Document not found or not accessible", "content": { "application/json": { "example": { "error": { "code": "DOCUMENT_NOT_FOUND", "message": "Document not found or you do not have access to it", } } } }, }, 409: { "description": "Cannot delete document while processing", "content": { "application/json": { "example": { "error": { "code": "DOCUMENT_PROCESSING", "message": "Cannot delete document while processing", } } } }, }, }, ) async def delete_document( document_id: UUID, api_key: ApiKeyDep, session: AsyncSessionDep, ctx: TenantContextDep, ) -> DocumentDeletionResponse: """Delete a document and all associated data. Removes the document from the database, deletes the physical file from storage, and removes all associated data including extracted records, symbol dictionaries, cross-references, and vector embeddings. Deletion cascade removes: - File from storage - documents record from database - All extracted_records for this document - All symbol_dictionaries for this document - All cross_references from this document - All vectors in Milvus for this document Args: document_id: UUID of the document to delete. api_key: Authenticated API key (injected by dependency). session: Async database session (injected by dependency). Returns: DocumentDeletionResponse with deletion confirmation. Raises: HTTPException: 404 if document not found or not owned by requesting API key, 409 if document is currently being processed. """ logger.info( "document_deletion_started", document_id=str(document_id), api_key_id=str(api_key.id), tenant_id=str(ctx.tenant_id), ) # Set RLS session variable (defense-in-depth with application-level WHERE) await set_rls_tenant(session, ctx.tenant_id) # Query document by ID, scoped by tenant + node (defense-in-depth with RLS) result = await session.execute( select(Document) .where(Document.id == document_id) .where(Document.tenant_id == ctx.tenant_id) .where(Document.node_id == ctx.node_id) ) document = result.scalar_one_or_none() # Check existence and authorization (return 404 for both to avoid leaking existence) if not document or document.api_key_id != api_key.id: logger.warning( "document_deletion_unauthorized", document_id=str(document_id), api_key_id=str(api_key.id), document_exists=document is not None, ) raise HTTPException( status_code=404, detail={ "error": { "code": "DOCUMENT_NOT_FOUND", "message": "Document not found or you do not have access to it", } }, ) # Check if document is processing (409 Conflict) if document.status == "processing": logger.warning( "document_deletion_blocked", document_id=str(document_id), status=document.status, api_key_id=str(api_key.id), ) raise HTTPException( status_code=409, detail={ "error": { "code": "DOCUMENT_PROCESSING", "message": "Cannot delete document while processing", } }, ) # Import here to avoid circular dependency from src.services.document_deletion_service import DocumentDeletionService import aiofiles.os # Delete extracted records and vectors atomically deletion_service = DocumentDeletionService(session) deletion_counts = await deletion_service.delete_document(document_id=document.id) # Delete physical file from storage file_deleted = False if document.file_path and await aiofiles.os.path.exists(document.file_path): try: await aiofiles.os.remove(document.file_path) file_deleted = True logger.info("file_deleted", file_path=document.file_path) except Exception as e: logger.warning( "file_deletion_failed", file_path=document.file_path, error=str(e), ) # Delete document record from database (CASCADE will handle remaining relations) await session.delete(document) await session.commit() deletion_summary = { "records_deleted": deletion_counts["records"], "vectors_deleted": deletion_counts["vectors"], "file_deleted": file_deleted, } # Log deletion for audit trail logger.info( "document_deleted", document_id=str(document_id), filename=document.filename, api_key_id=str(api_key.id), deletion_summary=deletion_summary, ) return DocumentDeletionResponse( document_id=document_id, status="deleted", message="Document and all associated data removed", ) @router.get("") async def list_documents(api_key: ApiKeyDep, ctx: TenantContextDep) -> dict: """List all documents. Requires valid API key authentication. Returns: List of documents. Note: Placeholder endpoint. Full implementation in Epic 2. """ return { "documents": [], "message": "Document endpoints coming in Epic 2", "authenticated_as": api_key.name or str(api_key.id), } # ============================================================================= # Symbol Upload Endpoints # ============================================================================= @router.post( "/{document_id}/symbols", status_code=200, response_model=SymbolUploadResponse, responses={ 200: { "description": "Custom symbols uploaded successfully", "model": SymbolUploadResponse, }, 400: { "description": "Validation error in request body", "content": { "application/json": { "examples": { "empty_symbols": { "summary": "Empty symbols array", "value": { "error": { "code": "VALIDATION_ERROR", "message": "Symbols array must contain at least 1 item", } }, }, "duplicate_symbol": { "summary": "Duplicate symbol in request", "value": { "error": { "code": "VALIDATION_ERROR", "message": "Duplicate symbol '★' in request", } }, }, } } }, }, 404: { "description": "Document not found or not accessible", "content": { "application/json": { "example": { "error": { "code": "DOCUMENT_NOT_FOUND", "message": "Document not found or you do not have access to it", } } } }, }, }, ) async def upload_custom_symbols( document_id: UUID, request: SymbolUploadRequest, api_key: ApiKeyDep, session: AsyncSessionDep, ctx: TenantContextDep, ) -> SymbolUploadResponse: """Upload custom symbol definitions for a document. Custom symbols are used to define mappings from symbols (e.g., ○, ×, △, ★) to their human-readable meanings. These custom symbols take precedence over auto-detected symbols when resolving symbol values in extracted records. If the document has already been processed (status is "completed"), symbol re-resolution is automatically triggered to apply the new symbols. Args: document_id: UUID of the document to add symbols to. request: Request body containing list of symbol definitions. api_key: Authenticated API key (injected by dependency). session: Async database session (injected by dependency). Returns: SymbolUploadResponse with upload statistics. Raises: HTTPException: 400 if validation fails, 404 if document not found or not owned by API key. """ logger.info( "custom_symbol_upload_started", document_id=str(document_id), symbol_count=len(request.symbols), api_key_id=str(api_key.id), ) # Verify document exists and belongs to API key + tenant scope document = await symbol_service.get_document(document_id, session) if not document or document.api_key_id != api_key.id or document.tenant_id != ctx.tenant_id or document.node_id != ctx.node_id: logger.warning( "document_not_found_or_unauthorized_for_symbols", document_id=str(document_id), api_key_id=str(api_key.id), document_exists=document is not None, ) raise HTTPException( status_code=404, detail={ "error": { "code": "DOCUMENT_NOT_FOUND", "message": "Document not found or you do not have access to it", } }, ) # Convert request symbols to dicts for service symbols_data = [ {"symbol": s.symbol, "meaning": s.meaning} for s in request.symbols ] # Upload symbols via service result = await symbol_service.upload_custom_symbols( document_id=document_id, symbols=symbols_data, session=session, ) # Check if re-resolution is needed needs_reprocessing = await symbol_service.check_document_needs_reprocessing( document_id, session ) if needs_reprocessing: # Queue Celery task for re-resolution task = re_resolve_symbols.delay(str(document_id)) result.reprocessing_triggered = True logger.info( "symbol_re_resolution_triggered", document_id=str(document_id), celery_task_id=task.id, ) message = "Custom symbols added. Re-resolution triggered for processed document." else: message = "Custom symbols will be applied to this document" logger.info( "custom_symbol_upload_completed", document_id=str(document_id), symbols_added=result.symbols_added, symbols_updated=result.symbols_updated, reprocessing_triggered=result.reprocessing_triggered, api_key_id=str(api_key.id), ) return SymbolUploadResponse( document_id=document_id, symbols_added=result.symbols_added, symbols_updated=result.symbols_updated, message=message, reprocessing_triggered=result.reprocessing_triggered, ) @router.delete( "/{document_id}/symbols/{symbol}", status_code=200, response_model=SymbolDeleteResponse, responses={ 200: { "description": "Symbol deleted successfully", "model": SymbolDeleteResponse, }, 404: { "description": "Document or symbol not found", "content": { "application/json": { "examples": { "document_not_found": { "summary": "Document not found", "value": { "error": { "code": "DOCUMENT_NOT_FOUND", "message": "Document not found or you do not have access to it", } }, }, "symbol_not_found": { "summary": "Symbol not found", "value": { "error": { "code": "SYMBOL_NOT_FOUND", "message": "Custom symbol '★' not found for this document", } }, }, } } }, }, }, ) async def delete_custom_symbol( document_id: UUID, symbol: str, api_key: ApiKeyDep, session: AsyncSessionDep, ctx: TenantContextDep, ) -> SymbolDeleteResponse: """Delete a custom symbol from a document. Only custom symbols (uploaded via API) can be deleted. Auto-detected symbols cannot be deleted via this endpoint. If the document has already been processed (status is "completed"), symbol re-resolution is automatically triggered to update records. Args: document_id: UUID of the document. symbol: The symbol string to delete (URL-encoded if special chars). api_key: Authenticated API key (injected by dependency). session: Async database session (injected by dependency). Returns: SymbolDeleteResponse with deletion status. Raises: HTTPException: 404 if document or symbol not found. """ logger.info( "custom_symbol_deletion_started", document_id=str(document_id), symbol=symbol, api_key_id=str(api_key.id), ) # Verify document exists and belongs to API key + tenant scope document = await symbol_service.get_document(document_id, session) if not document or document.api_key_id != api_key.id or document.tenant_id != ctx.tenant_id or document.node_id != ctx.node_id: logger.warning( "document_not_found_or_unauthorized_for_symbol_deletion", document_id=str(document_id), api_key_id=str(api_key.id), document_exists=document is not None, ) raise HTTPException( status_code=404, detail={ "error": { "code": "DOCUMENT_NOT_FOUND", "message": "Document not found or you do not have access to it", } }, ) # Delete symbol via service result = await symbol_service.delete_custom_symbol( document_id=document_id, symbol=symbol, session=session, ) if not result.deleted: logger.warning( "custom_symbol_not_found_for_deletion", document_id=str(document_id), symbol=symbol, ) raise HTTPException( status_code=404, detail={ "error": { "code": "SYMBOL_NOT_FOUND", "message": f"Custom symbol '{symbol}' not found for this document", } }, ) # Check if re-resolution is needed needs_reprocessing = await symbol_service.check_document_needs_reprocessing( document_id, session ) if needs_reprocessing: # Queue Celery task for re-resolution task = re_resolve_symbols.delay(str(document_id)) result.reprocessing_triggered = True logger.info( "symbol_re_resolution_triggered_after_deletion", document_id=str(document_id), celery_task_id=task.id, ) message = "Symbol deleted. Re-resolution triggered for processed document." else: message = "Symbol deleted successfully" logger.info( "custom_symbol_deletion_completed", document_id=str(document_id), symbol=symbol, reprocessing_triggered=result.reprocessing_triggered, api_key_id=str(api_key.id), ) return SymbolDeleteResponse( document_id=document_id, symbol=symbol, message=message, reprocessing_triggered=result.reprocessing_triggered, ) @router.get( "/{document_id}/symbols", status_code=200, response_model=SymbolListResponse, responses={ 200: { "description": "Symbols retrieved successfully", "model": SymbolListResponse, }, 404: { "description": "Document not found or not accessible", "content": { "application/json": { "example": { "error": { "code": "DOCUMENT_NOT_FOUND", "message": "Document not found or you do not have access to it", } } } }, }, }, ) async def list_document_symbols( document_id: UUID, api_key: ApiKeyDep, session: AsyncSessionDep, ctx: TenantContextDep, ) -> SymbolListResponse: """List all symbols for a document. Returns both auto-detected and custom-uploaded symbols. Args: document_id: UUID of the document. api_key: Authenticated API key (injected by dependency). session: Async database session (injected by dependency). Returns: SymbolListResponse with all symbols for the document. Raises: HTTPException: 404 if document not found or not owned by API key. """ logger.info( "list_document_symbols_started", document_id=str(document_id), api_key_id=str(api_key.id), ) # Verify document exists and belongs to API key + tenant scope document = await symbol_service.get_document(document_id, session) if not document or document.api_key_id != api_key.id or document.tenant_id != ctx.tenant_id or document.node_id != ctx.node_id: logger.warning( "document_not_found_or_unauthorized_for_symbol_list", document_id=str(document_id), api_key_id=str(api_key.id), document_exists=document is not None, ) raise HTTPException( status_code=404, detail={ "error": { "code": "DOCUMENT_NOT_FOUND", "message": "Document not found or you do not have access to it", } }, ) # Get all symbols via service symbols_data = await symbol_service.get_document_symbols(document_id, session) # Count custom vs auto-detected custom_count = sum(1 for s in symbols_data if s.get("is_custom", False)) auto_detected_count = len(symbols_data) - custom_count # Convert to response format symbols = [ SymbolEntry(symbol=s["symbol"], meaning=s["meaning"]) for s in symbols_data ] logger.info( "list_document_symbols_completed", document_id=str(document_id), total_symbols=len(symbols), custom_count=custom_count, auto_detected_count=auto_detected_count, api_key_id=str(api_key.id), ) return SymbolListResponse( document_id=document_id, symbols=symbols, total=len(symbols), custom_count=custom_count, auto_detected_count=auto_detected_count, )