from pathlib import Path from fastapi import APIRouter, HTTPException, Request from fastapi.responses import FileResponse, Response from .. import database as db, config router = APIRouter(prefix="/api/files", tags=["files"]) DATA = config.DATA_DIR.resolve() def _safe_path(path_str: str, campaign_id: str | None = None) -> Path: root = (config.campaign_dir(campaign_id) if campaign_id else DATA).resolve() resolved = (root / path_str).resolve() if not str(resolved).startswith(str(root)): raise HTTPException(400, "Path traversal denied") return resolved def _auto_name(parent: Path, name: str) -> Path: p = parent / name if not p.exists(): return p stem = Path(name).stem suffix = Path(name).suffix n = 2 while (parent / f"{stem}_copy_{n}{suffix}").exists(): n += 1 return parent / f"{stem}_copy_{n}{suffix}" def _enrich(entries, conn): for e in entries: if e["type"] == "file": sid = e["name"].split("_")[0] e["session_id"] = sid row = conn.execute("SELECT name, status FROM sessions WHERE id = ?", (sid,)).fetchone() e["session_name"] = row["name"] if row else "(orphan)" e["session_status"] = row["status"] if row else "unknown" @router.get("/browse") def browse(path: str = "", campaign_id: str = None): clean = path.strip("/") entries = [] conn = db.get_conn() base = config.campaign_dir(campaign_id) if campaign_id else DATA if not clean: for name in sorted(["audio", "transcriptions", "notes"]): entries.append({"name": name, "type": "dir", "path": name}) return {"entries": entries, "current_path": "", "parent_path": None} if clean == "notes": if campaign_id: rows = conn.execute( """SELECT n.session_id, n.dm_notes, n.player_recap, n.generated_at FROM notes n JOIN sessions s ON n.session_id = s.id WHERE s.campaign_id = ? ORDER BY n.generated_at DESC""", (campaign_id,), ).fetchall() else: rows = conn.execute( "SELECT session_id, dm_notes, player_recap, generated_at FROM notes ORDER BY generated_at DESC" ).fetchall() for r in rows: for kind, text in (("player_recap", r["player_recap"]), ("dm_notes", r["dm_notes"])): if not text: continue entries.append({ "name": f"{r['session_id']}_{kind}.txt", "type": "file", "path": f"notes/{r['session_id']}_{kind}.txt", "notes_kind": kind, "size": len(text), "modified_at": int(r["generated_at"] or 0), "preview": text[:120], }) _enrich(entries, conn) return {"entries": entries, "current_path": "notes", "parent_path": ""} resolved = (base / clean).resolve() if not str(resolved).startswith(str(base.resolve())): raise HTTPException(400, "Path traversal denied") if not resolved.exists() or not resolved.is_dir(): raise HTTPException(404, "Directory not found") for f in sorted(resolved.iterdir(), key=lambda p: (not p.is_dir(), p.name.lower())): if f.name.startswith("tmp_"): continue rel = str(f.relative_to(base)) entry = {"name": f.name, "type": "dir" if f.is_dir() else "file", "path": rel} if not f.is_dir(): entry["size"] = f.stat().st_size entry["modified_at"] = int(f.stat().st_mtime) entries.append(entry) _enrich([e for e in entries if e["type"] == "file"], conn) parts = clean.split("/") parent = "/".join(parts[:-1]) if len(parts) > 1 else "" return {"entries": entries, "current_path": clean, "parent_path": parent} _MIME_MAP: dict[str, str] = { ".wav": "audio/wav", ".mp3": "audio/mpeg", ".mp4": "video/mp4", ".m4a": "audio/mp4", ".ogg": "audio/ogg", ".flac": "audio/flac", ".json": "application/json", ".txt": "text/plain", } @router.get("/view") def view_file(path: str, campaign_id: str = None): if path.startswith("notes/"): parts = Path(path).stem.split("_") session_id = parts[0] kind = parts[1] if len(parts) > 1 else None conn = db.get_conn() row = conn.execute("SELECT dm_notes, player_recap FROM notes WHERE session_id = ?", (session_id,)).fetchone() if not row: raise HTTPException(404, "Notes not found") text = row[kind] if kind in ("dm_notes", "player_recap") else (row["player_recap"] or row["dm_notes"] or "") return Response(content=text, media_type="text/plain; charset=utf-8", headers={"Content-Disposition": "inline"}) file_path = _safe_path(path, campaign_id) if not file_path.exists(): raise HTTPException(404, "File not found") mime = _MIME_MAP.get(file_path.suffix.lower(), "application/octet-stream") return FileResponse(file_path, media_type=mime, headers={"Content-Disposition": "inline"}) @router.get("/download") def download_file(path: str, campaign_id: str = None): file_path = _safe_path(path, campaign_id) if not file_path.exists(): raise HTTPException(404, "File not found") return FileResponse(file_path, filename=file_path.name) @router.post("/upload") async def upload_file(request: Request): form = await request.form() file_field = form.get("file") if not file_field or not hasattr(file_field, "filename") or not file_field.filename: raise HTTPException(400, "No file provided") dest_dir_str = form.get("dir", "audio") campaign_id = form.get("campaign_id") dest_dir = _safe_path(dest_dir_str, campaign_id) dest_dir.mkdir(parents=True, exist_ok=True) dest = _auto_name(dest_dir, file_field.filename) with open(dest, "wb") as f: content = await file_field.read() f.write(content) root = config.campaign_dir(campaign_id) if campaign_id else DATA return {"path": str(dest.relative_to(root))} @router.delete("") def delete_files(body: dict): paths = body.get("paths", []) campaign_id = body.get("campaign_id") deleted = [] errors = [] for p in paths: try: fp = _safe_path(p, campaign_id) if fp.exists(): fp.unlink() deleted.append(p) elif p.startswith("notes/"): parts = Path(p).stem.split("_") session_id = parts[0] conn = db.get_conn() kind = parts[1] if len(parts) > 1 else None if kind == "dm_notes": conn.execute("UPDATE notes SET dm_notes = NULL WHERE session_id = ?", (session_id,)) elif kind == "player_recap": conn.execute("UPDATE notes SET player_recap = NULL WHERE session_id = ?", (session_id,)) else: conn.execute("DELETE FROM notes WHERE session_id = ?", (session_id,)) conn.commit() deleted.append(p) else: errors.append({"path": p, "error": "not found"}) except Exception as e: errors.append({"path": p, "error": str(e)}) return {"deleted": deleted, "errors": errors}