149 lines
5.0 KiB
Python
149 lines
5.0 KiB
Python
"""Upload router for handling file uploads."""
|
|
|
|
import logging
|
|
|
|
from fastapi import APIRouter, File, HTTPException, UploadFile
|
|
from pydantic import BaseModel
|
|
|
|
from deerflow.config.paths import get_paths
|
|
from deerflow.sandbox.sandbox_provider import get_sandbox_provider
|
|
from deerflow.uploads.manager import (
|
|
PathTraversalError,
|
|
delete_file_safe,
|
|
enrich_file_listing,
|
|
ensure_uploads_dir,
|
|
get_uploads_dir,
|
|
list_files_in_dir,
|
|
normalize_filename,
|
|
upload_artifact_url,
|
|
upload_virtual_path,
|
|
)
|
|
from deerflow.utils.file_conversion import CONVERTIBLE_EXTENSIONS, convert_file_to_markdown
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
router = APIRouter(prefix="/api/threads/{thread_id}/uploads", tags=["uploads"])
|
|
|
|
|
|
class UploadResponse(BaseModel):
|
|
"""Response model for file upload."""
|
|
|
|
success: bool
|
|
files: list[dict[str, str]]
|
|
message: str
|
|
|
|
|
|
|
|
|
|
@router.post("", response_model=UploadResponse)
|
|
async def upload_files(
|
|
thread_id: str,
|
|
files: list[UploadFile] = File(...),
|
|
) -> UploadResponse:
|
|
"""Upload multiple files to a thread's uploads directory."""
|
|
if not files:
|
|
raise HTTPException(status_code=400, detail="No files provided")
|
|
|
|
try:
|
|
uploads_dir = ensure_uploads_dir(thread_id)
|
|
except ValueError as e:
|
|
raise HTTPException(status_code=400, detail=str(e))
|
|
sandbox_uploads = get_paths().sandbox_uploads_dir(thread_id)
|
|
uploaded_files = []
|
|
|
|
sandbox_provider = get_sandbox_provider()
|
|
sandbox_id = sandbox_provider.acquire(thread_id)
|
|
sandbox = sandbox_provider.get(sandbox_id)
|
|
|
|
for file in files:
|
|
if not file.filename:
|
|
continue
|
|
|
|
try:
|
|
safe_filename = normalize_filename(file.filename)
|
|
except ValueError:
|
|
logger.warning(f"Skipping file with unsafe filename: {file.filename!r}")
|
|
continue
|
|
|
|
try:
|
|
content = await file.read()
|
|
file_path = uploads_dir / safe_filename
|
|
file_path.write_bytes(content)
|
|
|
|
virtual_path = upload_virtual_path(safe_filename)
|
|
|
|
if sandbox_id != "local":
|
|
sandbox.update_file(virtual_path, content)
|
|
|
|
file_info = {
|
|
"filename": safe_filename,
|
|
"size": str(len(content)),
|
|
"path": str(sandbox_uploads / safe_filename),
|
|
"virtual_path": virtual_path,
|
|
"artifact_url": upload_artifact_url(thread_id, safe_filename),
|
|
}
|
|
|
|
logger.info(f"Saved file: {safe_filename} ({len(content)} bytes) to {file_info['path']}")
|
|
|
|
file_ext = file_path.suffix.lower()
|
|
if file_ext in CONVERTIBLE_EXTENSIONS:
|
|
md_path = await convert_file_to_markdown(file_path)
|
|
if md_path:
|
|
md_virtual_path = upload_virtual_path(md_path.name)
|
|
|
|
if sandbox_id != "local":
|
|
sandbox.update_file(md_virtual_path, md_path.read_bytes())
|
|
|
|
file_info["markdown_file"] = md_path.name
|
|
file_info["markdown_path"] = str(sandbox_uploads / md_path.name)
|
|
file_info["markdown_virtual_path"] = md_virtual_path
|
|
file_info["markdown_artifact_url"] = upload_artifact_url(thread_id, md_path.name)
|
|
|
|
uploaded_files.append(file_info)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to upload {file.filename}: {e}")
|
|
raise HTTPException(status_code=500, detail=f"Failed to upload {file.filename}: {str(e)}")
|
|
|
|
return UploadResponse(
|
|
success=True,
|
|
files=uploaded_files,
|
|
message=f"Successfully uploaded {len(uploaded_files)} file(s)",
|
|
)
|
|
|
|
|
|
@router.get("/list", response_model=dict)
|
|
async def list_uploaded_files(thread_id: str) -> dict:
|
|
"""List all files in a thread's uploads directory."""
|
|
try:
|
|
uploads_dir = get_uploads_dir(thread_id)
|
|
except ValueError as e:
|
|
raise HTTPException(status_code=400, detail=str(e))
|
|
result = list_files_in_dir(uploads_dir)
|
|
enrich_file_listing(result, thread_id)
|
|
|
|
# Gateway additionally includes the sandbox-relative path.
|
|
sandbox_uploads = get_paths().sandbox_uploads_dir(thread_id)
|
|
for f in result["files"]:
|
|
f["path"] = str(sandbox_uploads / f["filename"])
|
|
|
|
return result
|
|
|
|
|
|
@router.delete("/{filename}")
|
|
async def delete_uploaded_file(thread_id: str, filename: str) -> dict:
|
|
"""Delete a file from a thread's uploads directory."""
|
|
try:
|
|
uploads_dir = get_uploads_dir(thread_id)
|
|
except ValueError as e:
|
|
raise HTTPException(status_code=400, detail=str(e))
|
|
try:
|
|
return delete_file_safe(uploads_dir, filename, convertible_extensions=CONVERTIBLE_EXTENSIONS)
|
|
except FileNotFoundError:
|
|
raise HTTPException(status_code=404, detail=f"File not found: {filename}")
|
|
except PathTraversalError:
|
|
raise HTTPException(status_code=400, detail="Invalid path")
|
|
except Exception as e:
|
|
logger.error(f"Failed to delete {filename}: {e}")
|
|
raise HTTPException(status_code=500, detail=f"Failed to delete {filename}: {str(e)}")
|