from pathlib import Path from fastapi import APIRouter, Depends, UploadFile, File, Form, HTTPException from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy import select from typing import List from app.core.security import get_current_user_id from app.db.database import get_db, DocumentRecord, User from app.schemas.schemas import BatchIngestResponse, DocumentOut from app.services.ingestion_service import ( create_document_record, get_conversation_documents, get_all_documents, get_document, schedule_ingestion, validate_medical_document, ) from app.db import chat_db from app.services.qdrant_service import get_qdrant_service router = APIRouter(prefix="/documents", tags=["documents"]) ALLOWED_EXT = {"pdf", "docx", "txt"} def _check_ext(filename: str) -> None: ext = filename.rsplit(".", 1)[-1].lower() if "." in filename else "" if ext not in ALLOWED_EXT: raise HTTPException(status_code=400, detail=f"Unsupported file type. Allowed: {', '.join(ALLOWED_EXT)}") async def _require_admin(user_id: str, db: AsyncSession) -> User: result = await db.execute(select(User).where(User.id == user_id)) user = result.scalar_one_or_none() if not user or user.role != "admin": raise HTTPException(status_code=403, detail="Admin access required") return user @router.post("/upload", response_model=BatchIngestResponse, status_code=202) async def upload_document( files: List[UploadFile] = File(...), conversation_id: str = Form(...), title: str | None = Form(default=None), source: str | None = Form(default=None), user_id: str = Depends(get_current_user_id), db: AsyncSession = Depends(get_db), ): """Queue one or more medical documents for ingestion.""" conv = await chat_db.get_conversation(user_id, conversation_id) if not conv: raise HTTPException(status_code=404, detail="Conversation not found") pending_documents: List[tuple[str, bytes, str]] = [] for file in files: filename = file.filename or "unknown" _check_ext(filename) raw_bytes = await file.read() try: validate_medical_document(filename, raw_bytes) except ValueError as exc: raise HTTPException(status_code=400, detail=f"{filename}: {exc}") from exc doc_title = title.strip() if title and len(files) == 1 else Path(filename).stem.replace("-", " ").replace("_", " ") pending_documents.append((filename, raw_bytes, doc_title)) document_ids: List[str] = [] for filename, raw_bytes, doc_title in pending_documents: doc_id = await create_document_record( filename=filename, title=doc_title, source=source, user_id=user_id, conversation_id=conversation_id, db=db, ) schedule_ingestion( doc_id=doc_id, filename=filename, title=doc_title, source=source, raw_bytes=raw_bytes, ) document_ids.append(doc_id) noun = "document" if len(document_ids) == 1 else "documents" return BatchIngestResponse( document_ids=document_ids, message=f"Queued {len(document_ids)} {noun} for ingestion", ) @router.get("", response_model=List[DocumentOut]) async def list_documents( conversation_id: str | None = None, user_id: str = Depends(get_current_user_id), db: AsyncSession = Depends(get_db), ): if conversation_id: conv = await chat_db.get_conversation(user_id, conversation_id) if not conv: raise HTTPException(status_code=404, detail="Conversation not found") docs = await get_conversation_documents(conversation_id, db) else: docs = await get_all_documents(db) return docs @router.get("/{doc_id}", response_model=DocumentOut) async def get_document_detail( doc_id: str, user_id: str = Depends(get_current_user_id), db: AsyncSession = Depends(get_db), ): doc = await get_document(doc_id, db) if not doc: raise HTTPException(status_code=404, detail="Document not found") return doc @router.delete("/conversation/{conversation_id}", status_code=204) async def clear_conversation_documents( conversation_id: str, user_id: str = Depends(get_current_user_id), db: AsyncSession = Depends(get_db), ): conv = await chat_db.get_conversation(user_id, conversation_id) if not conv: raise HTTPException(status_code=404, detail="Conversation not found") docs = await get_conversation_documents(conversation_id, db) qdrant = get_qdrant_service() for doc in docs: await qdrant.delete_by_doc_id(doc.id) await db.delete(doc) await db.commit() @router.delete("/{doc_id}", status_code=204) async def delete_document( doc_id: str, user_id: str = Depends(get_current_user_id), db: AsyncSession = Depends(get_db), ): await _require_admin(user_id, db) doc = await get_document(doc_id, db) if not doc: raise HTTPException(status_code=404, detail="Document not found") qdrant = get_qdrant_service() await qdrant.delete_by_doc_id(doc_id) await db.delete(doc) await db.commit()