diff --git a/apps/api/app/api/router.py b/apps/api/app/api/router.py index 17a421028..a7c808825 100644 --- a/apps/api/app/api/router.py +++ b/apps/api/app/api/router.py @@ -2,6 +2,7 @@ from app.api.routes.asset_diagnosis import router as asset_diagnosis_router from app.api.routes.asset_libraries import router as asset_libraries_router from app.api.routes.assets import router as assets_router from app.api.routes.auth import router as auth_router +from app.api.routes.chunked_upload import router as chunked_upload_router from app.api.routes.classification_jobs import router as classification_jobs_router from app.api.routes.edit_plans import router as edit_plans_router from app.api.routes.generated_videos import router as generated_videos_router @@ -22,69 +23,74 @@ health_router.include_router(health_check_router) api_router.include_router( auth_router, - tags=["认证"], + tags=["Auth"], ) api_router.include_router( workspaces_router, - tags=["工作空间"], + tags=["Workspace"], ) api_router.include_router( projects_router, prefix="/projects", - tags=["项目管理"], + tags=["Project"], ) api_router.include_router( project_titles_router, - tags=["标题库"], + tags=["TitleLibrary"], ) api_router.include_router( edit_plans_router, ) api_router.include_router( task_center_router, - tags=["任务中心"], + tags=["TaskCenter"], ) api_router.include_router( asset_diagnosis_router, - tags=["素材诊断"], + tags=["AssetDiagnosis"], ) api_router.include_router( asset_libraries_router, prefix="/asset-libraries", - tags=["素材库管理"], + tags=["AssetLibrary"], ) api_router.include_router( assets_router, prefix="/assets", - tags=["素材资产"], + tags=["Asset"], ) api_router.include_router( ingest_jobs_router, prefix="/ingest-jobs", - tags=["导入任务"], + tags=["IngestJob"], ) api_router.include_router( classification_jobs_router, prefix="/classification-jobs", - tags=["分类任务"], + tags=["ClassificationJob"], ) api_router.include_router( upload_router, prefix="/upload", - tags=["文件上传"], + tags=["Upload"], +) +api_router.include_router( + chunked_upload_router, + prefix="/upload/chunk", + tags=["ChunkedUpload"], ) api_router.include_router( generation_tasks_router, prefix="/generation", - tags=["生成任务"], + tags=["Generation"], ) api_router.include_router( generated_videos_router, prefix="/generated-videos", - tags=["成片管理"], + tags=["GeneratedVideo"], ) api_router.include_router( project_management_router, prefix="/project-management", - tags=["项目推进管理"], + tags=["ProjectManagement"], ) diff --git a/apps/api/app/api/routes/chunked_upload.py b/apps/api/app/api/routes/chunked_upload.py new file mode 100644 index 000000000..45664d786 --- /dev/null +++ b/apps/api/app/api/routes/chunked_upload.py @@ -0,0 +1,478 @@ +""" +Chunked upload routes for large file uploads (up to 2GB). +Supports chunked upload, resume, and automatic cleanup of expired uploads. +""" + +import fcntl +import json +import logging +import os +import shutil +from datetime import datetime, timedelta, timezone +from pathlib import Path +from typing import Any +from uuid import uuid4 + +from app.auth import AuthenticatedUser, get_current_user +from app.config import get_settings +from app.core.celery_app import celery_app +from app.core.storage import OSSStorageService, get_storage_service +from app.dependencies import ( + get_asset_library_repository, + get_ingest_job_repository, + get_project_repository, + get_workspace_member_repository, +) +from app.schemas.chunked_upload import ( + ChunkedUploadCompleteRequest, + ChunkedUploadCompleteResponse, + ChunkedUploadInitRequest, + ChunkedUploadInitResponse, + ChunkedUploadStatusResponse, +) +from fastapi import APIRouter, Depends, HTTPException, UploadFile, status +from fastapi.params import File + +from packages.application import GetProjectUseCase, SubmitIngestJobCommand, SubmitIngestJobUseCase +from packages.ports.workspace_member_repository import WorkspaceMemberRepository + +router = APIRouter() +logger = logging.getLogger(__name__) + +# Configuration +DEFAULT_CHUNK_SIZE = 5 * 1024 * 1024 # 5MB +MAX_FILE_SIZE = 2 * 1024 * 1024 * 1024 # 2GB +CHUNK_EXPIRY_HOURS = 24 + +# Allowed file types (consistent with existing upload.py) +ALLOWED_MIME_TYPES = { + "image/jpeg", "image/png", "image/gif", "image/webp", + "video/mp4", "video/quicktime", "video/x-msvideo", "video/webm", + "audio/mpeg", "audio/wav", "audio/ogg", "audio/mp3", +} + +# Chunk storage root directory +CHUNK_STORAGE_ROOT = Path("/tmp/chunked_uploads") + + +def _get_chunk_dir(upload_id: str) -> Path: + """Get chunk storage directory""" + return CHUNK_STORAGE_ROOT / upload_id + + +def _get_upload_meta_path(upload_id: str) -> Path: + """Get upload metadata file path""" + return CHUNK_STORAGE_ROOT / f"{upload_id}.meta.json" + + +def _atomic_check_and_record(upload_id: str, chunk_index: int) -> bool: + """ + Atomically check if chunk is uploaded and record if not. + Uses file locking to prevent race conditions. + + Returns: + True if chunk was newly recorded, False if already exists + """ + meta_path = _get_upload_meta_path(upload_id) + CHUNK_STORAGE_ROOT.mkdir(parents=True, exist_ok=True) + + with open(meta_path, "r+", encoding="utf-8") as f: + fcntl.flock(f.fileno(), fcntl.LOCK_EX) + try: + meta = json.load(f) + if chunk_index in meta["uploaded_chunks"]: + return False + meta["uploaded_chunks"].append(chunk_index) + meta["status"] = "uploading" + f.seek(0) + json.dump(meta, f, ensure_ascii=False, indent=2) + f.truncate() + return True + finally: + fcntl.flock(f.fileno(), fcntl.LOCK_UN) + + +def _require_workspace_member( + workspace_id: str, + authenticated_user: AuthenticatedUser, + workspace_member_repository: WorkspaceMemberRepository, +) -> None: + """Verify user has workspace permission""" + member = workspace_member_repository.find_by_workspace_and_user(workspace_id, authenticated_user.user.id) + if member is None: + raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Workspace access denied") + + +def _require_project_and_library( + workspace_id: str, + project_id: str, + library_id: str, + project_repository: Any, + asset_library_repository: Any, +) -> None: + """Verify project and asset library exist""" + project = GetProjectUseCase(project_repository).execute(project_id) + if project is None or project.workspace_id != workspace_id: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Project not found") + + libraries = asset_library_repository.list_by_project(project_id) + if not any(item.id == library_id and item.workspace_id == workspace_id for item in libraries): + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Asset library not found") + + +def _load_upload_meta(upload_id: str) -> dict[str, Any]: + """Load upload metadata""" + meta_path = _get_upload_meta_path(upload_id) + if not meta_path.exists(): + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Upload not found") + + with open(meta_path, "r", encoding="utf-8") as f: + return json.load(f) + + +def _save_upload_meta(upload_id: str, meta: dict[str, Any]) -> None: + """Save upload metadata""" + meta_path = _get_upload_meta_path(upload_id) + CHUNK_STORAGE_ROOT.mkdir(parents=True, exist_ok=True) + with open(meta_path, "w", encoding="utf-8") as f: + json.dump(meta, f, ensure_ascii=False, indent=2) + + +def _validate_file_type(content: bytes, filename: str) -> str: + """Validate file type""" + try: + import magic + detected_mime = magic.from_buffer(content, mime=True) + except ImportError: + import mimetypes + detected_mime = mimetypes.guess_type(filename)[0] or "application/octet-stream" + + if detected_mime not in ALLOWED_MIME_TYPES: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail=f"Unsupported file type: {detected_mime}. Allowed types: {', '.join(sorted(ALLOWED_MIME_TYPES))}" + ) + return detected_mime + + +def _cleanup_expired_uploads() -> int: + """Cleanup expired uploads, returns number of cleaned uploads""" + if not CHUNK_STORAGE_ROOT.exists(): + return 0 + + now = datetime.now(timezone.utc) + cleaned = 0 + + for meta_file in CHUNK_STORAGE_ROOT.glob("*.meta.json"): + try: + with open(meta_file, "r", encoding="utf-8") as f: + meta = json.load(f) + + expires_at = datetime.fromisoformat(meta["expires_at"]) + if expires_at.tzinfo is None: + expires_at = expires_at.replace(tzinfo=timezone.utc) + + # Only cleanup uploads that are not actively being uploaded + if expires_at < now and meta.get("status") != "uploading": + upload_id = meta["upload_id"] + chunk_dir = _get_chunk_dir(upload_id) + if chunk_dir.exists(): + shutil.rmtree(chunk_dir) + meta_file.unlink() + cleaned += 1 + logger.info(f"Cleaned up expired upload: {upload_id}") + except Exception as e: + logger.warning(f"Failed to cleanup upload metadata {meta_file}: {e}") + + return cleaned + + +@router.post("/init", response_model=ChunkedUploadInitResponse) +async def init_chunked_upload( + request: ChunkedUploadInitRequest, + authenticated_user: AuthenticatedUser = Depends(get_current_user), + project_repository: Any = Depends(get_project_repository), + asset_library_repository: Any = Depends(get_asset_library_repository), + workspace_member_repository: WorkspaceMemberRepository = Depends(get_workspace_member_repository), +) -> ChunkedUploadInitResponse: + """Initialize chunked upload""" + settings = get_settings() + + # Validate file size + if request.file_size > MAX_FILE_SIZE: + raise HTTPException( + status_code=status.HTTP_413_REQUEST_ENTITY_TOO_LARGE, + detail=f"File exceeds maximum size ({MAX_FILE_SIZE // (1024 * 1024 * 1024)}GB)", + ) + + # Validate project exists and get workspace_id + project = GetProjectUseCase(project_repository).execute(request.project_id) + if project is None: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Project not found") + + workspace_id = project.workspace_id + + # Verify permission and asset library + _require_workspace_member(workspace_id, authenticated_user, workspace_member_repository) + _require_project_and_library( + workspace_id, + request.project_id, + request.library_id, + project_repository, + asset_library_repository, + ) + + # Calculate chunk size + chunk_size = DEFAULT_CHUNK_SIZE + expected_chunks = (request.file_size + chunk_size - 1) // chunk_size + if expected_chunks != request.total_chunks: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail=f"total_chunks mismatch. Expected {expected_chunks} for file size {request.file_size} with chunk size {chunk_size}", + ) + + # Cleanup expired uploads + _cleanup_expired_uploads() + + # Generate upload ID + upload_id = uuid4().hex + now = datetime.now(timezone.utc) + expires_at = now + timedelta(hours=CHUNK_EXPIRY_HOURS) + + # Create chunk directory + chunk_dir = _get_chunk_dir(upload_id) + chunk_dir.mkdir(parents=True, exist_ok=True) + + # Save metadata + meta = { + "upload_id": upload_id, + "filename": request.filename.replace("/", "_").replace("\\", "_"), + "file_size": request.file_size, + "total_chunks": request.total_chunks, + "uploaded_chunks": [], + "content_type": request.content_type, + "project_id": request.project_id, + "library_id": request.library_id, + "workspace_id": workspace_id, + "status": "pending", + "created_at": now.isoformat(), + "expires_at": expires_at.isoformat(), + } + _save_upload_meta(upload_id, meta) + + return ChunkedUploadInitResponse( + upload_id=upload_id, + chunk_size=chunk_size, + total_chunks=request.total_chunks, + filename=meta["filename"], + expires_at=expires_at, + ) + + +@router.post("/{upload_id}/{chunk_index}") +async def upload_chunk( + upload_id: str, + chunk_index: int, + chunk: UploadFile = File(..., description="Chunk data"), + authenticated_user: AuthenticatedUser = Depends(get_current_user), + workspace_member_repository: WorkspaceMemberRepository = Depends(get_workspace_member_repository), +) -> dict[str, Any]: + """Upload a single chunk""" + # Load metadata + meta = _load_upload_meta(upload_id) + + # Check expiry + expires_at = datetime.fromisoformat(meta["expires_at"]) + if expires_at.tzinfo is None: + expires_at = expires_at.replace(tzinfo=timezone.utc) + + if expires_at < datetime.now(timezone.utc): + raise HTTPException(status_code=status.HTTP_410_GONE, detail="Upload has expired") + + # Validate chunk index + if chunk_index < 0 or chunk_index >= meta["total_chunks"]: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail=f"Invalid chunk index. Must be between 0 and {meta['total_chunks'] - 1}", + ) + + # Verify permission + _require_workspace_member(meta["workspace_id"], authenticated_user, workspace_member_repository) + + # Atomic check and record to prevent race conditions + if not _atomic_check_and_record(upload_id, chunk_index): + return {"message": "Chunk already uploaded", "chunk_index": chunk_index} + + # Read chunk data + chunk_data = await chunk.read() + + # Validate chunk size (last chunk can be smaller than chunk_size) + expected_size = DEFAULT_CHUNK_SIZE + if chunk_index == meta["total_chunks"] - 1: + expected_size = meta["file_size"] - (chunk_index * DEFAULT_CHUNK_SIZE) + + if len(chunk_data) != expected_size: + # Rollback the recorded chunk + meta_path = _get_upload_meta_path(upload_id) + with open(meta_path, "r+", encoding="utf-8") as f: + fcntl.flock(f.fileno(), fcntl.LOCK_EX) + try: + meta = json.load(f) + if chunk_index in meta["uploaded_chunks"]: + meta["uploaded_chunks"].remove(chunk_index) + f.seek(0) + json.dump(meta, f, ensure_ascii=False, indent=2) + f.truncate() + finally: + fcntl.flock(f.fileno(), fcntl.LOCK_UN) + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail=f"Chunk size mismatch. Expected {expected_size}, got {len(chunk_data)}", + ) + + # Save chunk + chunk_path = _get_chunk_dir(upload_id) / f"chunk_{chunk_index:06d}" + with open(chunk_path, "wb") as f: + f.write(chunk_data) + + # Reload metadata for response + meta = _load_upload_meta(upload_id) + + return { + "message": "Chunk uploaded successfully", + "chunk_index": chunk_index, + "uploaded_chunks": len(meta["uploaded_chunks"]), + "total_chunks": meta["total_chunks"], + } + + +@router.get("/{upload_id}/status", response_model=ChunkedUploadStatusResponse) +async def get_upload_status( + upload_id: str, + authenticated_user: AuthenticatedUser = Depends(get_current_user), + workspace_member_repository: WorkspaceMemberRepository = Depends(get_workspace_member_repository), +) -> ChunkedUploadStatusResponse: + """Get upload status (for resume)""" + meta = _load_upload_meta(upload_id) + + # Verify permission + _require_workspace_member(meta["workspace_id"], authenticated_user, workspace_member_repository) + + return ChunkedUploadStatusResponse( + upload_id=upload_id, + filename=meta["filename"], + file_size=meta["file_size"], + total_chunks=meta["total_chunks"], + uploaded_chunks=sorted(meta["uploaded_chunks"]), + status=meta["status"], + created_at=datetime.fromisoformat(meta["created_at"]), + expires_at=datetime.fromisoformat(meta["expires_at"]), + ) + + +@router.post("/{upload_id}/complete", response_model=ChunkedUploadCompleteResponse) +async def complete_chunked_upload( + upload_id: str, + request: ChunkedUploadCompleteRequest, + authenticated_user: AuthenticatedUser = Depends(get_current_user), + project_repository: Any = Depends(get_project_repository), + asset_library_repository: Any = Depends(get_asset_library_repository), + workspace_member_repository: WorkspaceMemberRepository = Depends(get_workspace_member_repository), + ingest_job_repository: Any = Depends(get_ingest_job_repository), + storage_service: OSSStorageService = Depends(get_storage_service), +) -> ChunkedUploadCompleteResponse: + """Complete chunked upload, merge chunks""" + # Load metadata + meta = _load_upload_meta(upload_id) + + # Verify permission + _require_workspace_member(meta["workspace_id"], authenticated_user, workspace_member_repository) + + # Verify project ID and library ID + if request.project_id != meta["project_id"] or request.library_id != meta["library_id"]: + raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Project or library ID mismatch") + + # Verify all chunks are uploaded + expected_chunks = set(range(meta["total_chunks"])) + uploaded_chunks = set(meta["uploaded_chunks"]) + missing_chunks = expected_chunks - uploaded_chunks + + if missing_chunks: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail=f"Missing chunks: {sorted(missing_chunks)}. Please upload remaining chunks first.", + ) + + # Validate file type + chunk_dir = _get_chunk_dir(upload_id) + sample_chunk_path = chunk_dir / "chunk_000000" + if sample_chunk_path.exists(): + with open(sample_chunk_path, "rb") as f: + sample_data = f.read(8192) # Read first 8KB for type detection + detected_mime = _validate_file_type(sample_data, meta["filename"]) + if detected_mime not in ALLOWED_MIME_TYPES: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail=f"Unsupported file type: {detected_mime}", + ) + + # Merge chunks to temp file + temp_file_path = CHUNK_STORAGE_ROOT / f"{upload_id}_complete.tmp" + try: + with open(temp_file_path, "wb") as out_file: + for i in range(meta["total_chunks"]): + chunk_path = chunk_dir / f"chunk_{i:06d}" + with open(chunk_path, "rb") as in_file: + shutil.copyfileobj(in_file, out_file) + + # Verify file size + actual_size = temp_file_path.stat().st_size + if actual_size != meta["file_size"]: + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + detail=f"File size mismatch. Expected {meta['file_size']}, got {actual_size}", + ) + + # Upload to OSS + file_id = uuid4().hex[:8] + safe_filename = meta["filename"] + storage_key = f"uploads/{file_id}/{safe_filename}" + + file_url = storage_service.upload_file( + str(temp_file_path), + storage_key, + content_type=meta["content_type"], + ) + + # Create ingest job + use_case = SubmitIngestJobUseCase(ingest_job_repository) + job = use_case.execute( + SubmitIngestJobCommand( + workspace_id=meta["workspace_id"], + project_id=meta["project_id"], + library_id=meta["library_id"], + storage_key=storage_key, + ) + ) + celery_app.send_task("worker.ingest_asset", args=[job.id]) + + # Update metadata status + meta["status"] = "completed" + _save_upload_meta(upload_id, meta) + + return ChunkedUploadCompleteResponse( + storage_key=storage_key, + ingest_job_id=job.id, + url=file_url, + ) + + finally: + # Cleanup temp file and chunks + if temp_file_path.exists(): + temp_file_path.unlink() + if chunk_dir.exists(): + shutil.rmtree(chunk_dir) + # Delete metadata file + meta_path = _get_upload_meta_path(upload_id) + if meta_path.exists(): + meta_path.unlink() diff --git a/apps/api/app/schemas/chunked_upload.py b/apps/api/app/schemas/chunked_upload.py new file mode 100644 index 000000000..cb0d7d122 --- /dev/null +++ b/apps/api/app/schemas/chunked_upload.py @@ -0,0 +1,42 @@ +from datetime import datetime + +from pydantic import BaseModel, Field + + +class ChunkedUploadInitRequest(BaseModel): + filename: str = Field(..., min_length=1, max_length=255, description="Filename") + file_size: int = Field(..., gt=0, le=2147483648, description="File size in bytes, max 2GB") + total_chunks: int = Field(..., gt=0, description="Total number of chunks") + content_type: str = Field(default="application/octet-stream", min_length=1, max_length=100, description="Content type") + project_id: str = Field(..., min_length=1, description="Project ID") + library_id: str = Field(..., min_length=1, description="Asset library ID") + + +class ChunkedUploadInitResponse(BaseModel): + upload_id: str = Field(..., description="Upload ID") + chunk_size: int = Field(..., description="Chunk size in bytes") + total_chunks: int = Field(..., description="Total number of chunks") + filename: str = Field(..., description="Filename") + expires_at: datetime = Field(..., description="Expiration time") + + +class ChunkedUploadStatusResponse(BaseModel): + upload_id: str = Field(..., description="Upload ID") + filename: str = Field(..., description="Filename") + file_size: int = Field(..., description="File size") + total_chunks: int = Field(..., description="Total number of chunks") + uploaded_chunks: list[int] = Field(..., description="List of uploaded chunk indices") + status: str = Field(..., description="Upload status: pending/uploading/completed/failed") + created_at: datetime = Field(..., description="Creation time") + expires_at: datetime = Field(..., description="Expiration time") + + +class ChunkedUploadCompleteRequest(BaseModel): + project_id: str = Field(..., min_length=1, description="Project ID") + library_id: str = Field(..., min_length=1, description="Asset library ID") + + +class ChunkedUploadCompleteResponse(BaseModel): + storage_key: str = Field(..., description="Storage key") + ingest_job_id: str = Field(..., description="Ingest job ID") + url: str = Field(..., description="File URL")