"""Uploading, serving and removing chat attachments.""" from __future__ import annotations import logging from fastapi import ( APIRouter, Depends, File, Form, HTTPException, Request, Response, UploadFile, status, ) from fastapi.responses import FileResponse from lembas.api.deps import Db, RequiredUser, require_permission from lembas.db.models import Attachment, Document from lembas.services import files as files_service from lembas.services import settings_store from lembas.services.fetch import FetchError, fetch from lembas.services.library import documents as documents_service from lembas.web.templating import templates log = logging.getLogger(__name__) router = APIRouter(prefix="/api/files", tags=["files"]) def _owned(db: Db, attachment_id: str, user_id: str) -> Attachment: attachment = db.get(Attachment, attachment_id) if attachment is None or attachment.user_id != user_id: raise HTTPException(status.HTTP_404_NOT_FOUND, "That file no longer exists.") return attachment @router.post("", dependencies=[Depends(require_permission("files.upload"))]) async def upload( request: Request, db: Db, user: RequiredUser, file: UploadFile = File(...), chat_id: str = "", ) -> Response: """Accept one file and return the chip that represents it in the composer. The attachment is stored immediately but left unbound: it only joins a message when that message is sent. That is what lets a file be removed before sending, and what the orphan sweep later cleans up. """ payload = await file.read() try: attachment = files_service.store( db, user_id=user.id, chat_id=chat_id or None, payload=payload, filename=file.filename or "file", ) except files_service.FileError as exc: # 200 with an error chip rather than a 4xx: htmx swaps the response # body either way, and an error the user can read beats a silent # failure in the console. return templates.TemplateResponse( request, "chat/_attachment_error.html", {"request": request, "filename": file.filename or "file", "error": str(exc)}, ) return templates.TemplateResponse( request, "chat/_attachment_chip.html", {"request": request, "attachment": attachment}, ) @router.post("/link", dependencies=[Depends(require_permission("files.upload"))]) async def attach_link( request: Request, db: Db, user: RequiredUser, url: str = Form(""), chat_id: str = Form("") ) -> Response: """Fetch a web page and attach its text. The page is reduced to text here and stored, rather than being fetched again when the message is sent: the same rule as PDF extraction. A reply must not change because a page was edited between composing and sending. """ config = settings_store.search(db) try: page = await fetch(url, allow_private=bool(config.get("allow_private_fetch"))) except FetchError as exc: return templates.TemplateResponse( request, "chat/_attachment_error.html", {"request": request, "filename": url[:80] or "link", "error": exc.message}, ) attachment = files_service.store_text( db, user_id=user.id, chat_id=chat_id or None, filename=f"{page.title[:120] or 'page'}.txt", text=page.text, truncated=page.truncated, source_note=page.url, ) return templates.TemplateResponse( request, "chat/_attachment_chip.html", {"request": request, "attachment": attachment} ) @router.post("/from-knowledge", dependencies=[Depends(require_permission("files.upload"))]) async def attach_from_knowledge( request: Request, db: Db, user: RequiredUser, document_id: str = Form(""), chat_id: str = Form(""), ) -> Response: """Attach a library document to the message being composed. The document is **copied**, not referenced. History must not change under a conversation because a document was later edited or deleted -- the same reason a PDF's text is extracted once at upload rather than per request. """ document = documents_service.get(db, document_id, user) if document is None: return templates.TemplateResponse( request, "chat/_attachment_error.html", { "request": request, "filename": "document", "error": "That document is not available.", }, ) attachment = files_service.copy_document( db, user_id=user.id, chat_id=chat_id or None, document=document ) return templates.TemplateResponse( request, "chat/_attachment_chip.html", {"request": request, "attachment": attachment} ) @router.get("/knowledge-picker", dependencies=[Depends(require_permission("files.upload"))]) async def knowledge_picker( request: Request, db: Db, user: RequiredUser, q: str = "", chat_id: str = "" ) -> Response: """The list of documents shown by the composer's Knowledge option.""" if q.strip(): found = documents_service.search(db, user, q, limit=20) else: found = list( db.scalars( documents_service.visible(db, user) .order_by(Document.created_at.desc()) .limit(20) ) ) return templates.TemplateResponse( request, "chat/_knowledge_picker.html", # `user` is read by the template to mark documents shared by someone # else; render() would inject it, but this is a fragment. {"request": request, "documents": found, "q": q, "chat_id": chat_id, "user": user}, ) @router.delete("/{attachment_id}") async def remove(db: Db, user: RequiredUser, attachment_id: str) -> Response: """Detach a file before it has been sent.""" attachment = _owned(db, attachment_id, user.id) if attachment.message_id is not None: # Deleting it now would rewrite a conversation that has already been # sent to a model and read by the user. raise HTTPException( status.HTTP_409_CONFLICT, "That file is part of a sent message." ) files_service.delete(db, attachment) return Response(status_code=status.HTTP_200_OK) @router.get("/{attachment_id}/content") async def content(db: Db, user: RequiredUser, attachment_id: str) -> Response: """Serve an attachment back to its owner.""" attachment = _owned(db, attachment_id, user.id) path = files_service.stored_path(attachment.stored_name) if path is None: raise HTTPException(status.HTTP_404_NOT_FOUND, "That file is no longer on disk.") # inline for images so they render in the thread; attachment for everything # else so a text/html upload can never be executed in this origin. disposition = "inline" if attachment.is_image else "attachment" return FileResponse( path, media_type=attachment.media_type if attachment.is_image else "application/octet-stream", headers={ "Content-Disposition": f'{disposition}; filename="{attachment.filename}"', "Cache-Control": "private, max-age=604800", # Belt and braces: even for images, never let a browser sniff its # way to treating the bytes as something executable. "X-Content-Type-Options": "nosniff", }, ) @router.get("/{attachment_id}/text") async def extracted_text(db: Db, user: RequiredUser, attachment_id: str) -> Response: """The text a document contributed to the prompt. Worth being able to see: a PDF that extracted badly explains a strange reply, and there is otherwise no way to tell what the model was given. """ attachment = _owned(db, attachment_id, user.id) return Response( attachment.extracted_text or attachment.extraction_error, media_type="text/plain; charset=utf-8", )