"""Files: upload, download, list, preview, and polymorphic entity attachment.""" import os import uuid import shutil from pathlib import Path from fastapi import APIRouter, Request, Form, Depends, UploadFile, File as FastAPIFile from fastapi.templating import Jinja2Templates from fastapi.responses import RedirectResponse, FileResponse from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy import text from typing import Optional from core.database import get_db from core.base_repository import BaseRepository from core.sidebar import get_sidebar_data router = APIRouter(prefix="/files", tags=["files"]) templates = Jinja2Templates(directory="templates") FILE_STORAGE_PATH = os.getenv("FILE_STORAGE_PATH", "/opt/lifeos/files/dev") # Ensure storage dir exists Path(FILE_STORAGE_PATH).mkdir(parents=True, exist_ok=True) # MIME types that can be previewed inline PREVIEWABLE = { "image/jpeg", "image/png", "image/gif", "image/webp", "image/svg+xml", "application/pdf", "text/plain", "text/html", "text/csv", } @router.get("/") async def list_files( request: Request, context_type: Optional[str] = None, context_id: Optional[str] = None, db: AsyncSession = Depends(get_db), ): sidebar = await get_sidebar_data(db) if context_type and context_id: # Files attached to a specific entity result = await db.execute(text(""" SELECT f.*, fm.context_type, fm.context_id FROM files f JOIN file_mappings fm ON fm.file_id = f.id WHERE f.is_deleted = false AND fm.context_type = :ct AND fm.context_id = :cid ORDER BY f.created_at DESC """), {"ct": context_type, "cid": context_id}) else: # All files result = await db.execute(text(""" SELECT f.* FROM files f WHERE f.is_deleted = false ORDER BY f.created_at DESC LIMIT 100 """)) items = [dict(r._mapping) for r in result] return templates.TemplateResponse("files.html", { "request": request, "sidebar": sidebar, "items": items, "context_type": context_type or "", "context_id": context_id or "", "page_title": "Files", "active_nav": "files", }) @router.get("/upload") async def upload_form( request: Request, context_type: Optional[str] = None, context_id: Optional[str] = None, db: AsyncSession = Depends(get_db), ): sidebar = await get_sidebar_data(db) return templates.TemplateResponse("file_upload.html", { "request": request, "sidebar": sidebar, "context_type": context_type or "", "context_id": context_id or "", "page_title": "Upload File", "active_nav": "files", }) @router.post("/upload") async def upload_file( request: Request, file: UploadFile = FastAPIFile(...), description: Optional[str] = Form(None), tags: Optional[str] = Form(None), context_type: Optional[str] = Form(None), context_id: Optional[str] = Form(None), db: AsyncSession = Depends(get_db), ): # Generate storage filename file_uuid = str(uuid.uuid4()) original = file.filename or "unknown" safe_name = original.replace("/", "_").replace("\\", "_") storage_name = f"{file_uuid}_{safe_name}" storage_path = os.path.join(FILE_STORAGE_PATH, storage_name) # Save to disk with open(storage_path, "wb") as f: content = await file.read() f.write(content) size_bytes = len(content) # Insert file record repo = BaseRepository("files", db) data = { "filename": storage_name, "original_filename": original, "storage_path": storage_path, "mime_type": file.content_type, "size_bytes": size_bytes, "description": description, } if tags and tags.strip(): data["tags"] = [t.strip() for t in tags.split(",") if t.strip()] new_file = await repo.create(data) # Create file mapping if context provided if context_type and context_type.strip() and context_id and context_id.strip(): await db.execute(text(""" INSERT INTO file_mappings (file_id, context_type, context_id) VALUES (:fid, :ct, :cid) ON CONFLICT DO NOTHING """), {"fid": new_file["id"], "ct": context_type, "cid": context_id}) # Redirect back to context or file list if context_type and context_id: return RedirectResponse( url=f"/files?context_type={context_type}&context_id={context_id}", status_code=303, ) return RedirectResponse(url="/files", status_code=303) @router.get("/{file_id}/download") async def download_file(file_id: str, db: AsyncSession = Depends(get_db)): repo = BaseRepository("files", db) item = await repo.get(file_id) if not item or not os.path.exists(item["storage_path"]): return RedirectResponse(url="/files", status_code=303) return FileResponse( path=item["storage_path"], filename=item["original_filename"], media_type=item.get("mime_type") or "application/octet-stream", ) @router.get("/{file_id}/preview") async def preview_file(file_id: str, request: Request, db: AsyncSession = Depends(get_db)): """Inline preview for images and PDFs.""" repo = BaseRepository("files", db) sidebar = await get_sidebar_data(db) item = await repo.get(file_id) if not item: return RedirectResponse(url="/files", status_code=303) can_preview = item.get("mime_type", "") in PREVIEWABLE return templates.TemplateResponse("file_preview.html", { "request": request, "sidebar": sidebar, "item": item, "can_preview": can_preview, "page_title": item["original_filename"], "active_nav": "files", }) @router.get("/{file_id}/serve") async def serve_file(file_id: str, db: AsyncSession = Depends(get_db)): """Serve file inline (for img src, iframe, etc).""" repo = BaseRepository("files", db) item = await repo.get(file_id) if not item or not os.path.exists(item["storage_path"]): return RedirectResponse(url="/files", status_code=303) return FileResponse( path=item["storage_path"], media_type=item.get("mime_type") or "application/octet-stream", ) @router.post("/{file_id}/delete") async def delete_file(file_id: str, request: Request, db: AsyncSession = Depends(get_db)): repo = BaseRepository("files", db) await repo.soft_delete(file_id) referer = request.headers.get("referer", "/files") return RedirectResponse(url=referer, status_code=303)