Xiaoxia AI
199 lines
7.9 KiB
Python
199 lines
7.9 KiB
Python
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.upload import (
|
|
DirectUploadCompleteRequest,
|
|
DirectUploadCompleteResponse,
|
|
DirectUploadPrepareRequest,
|
|
DirectUploadPrepareResponse,
|
|
UploadAssetResponse,
|
|
)
|
|
from fastapi import APIRouter, Depends, File, Form, HTTPException, UploadFile, status
|
|
|
|
from packages.application import GetProjectUseCase, SubmitIngestJobCommand, SubmitIngestJobUseCase
|
|
from packages.ports.workspace_member_repository import WorkspaceMemberRepository
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
def _require_workspace_member(
|
|
workspace_id: str,
|
|
authenticated_user: AuthenticatedUser,
|
|
workspace_member_repository: WorkspaceMemberRepository,
|
|
) -> None:
|
|
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:
|
|
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 _submit_ingest_job(
|
|
workspace_id: str,
|
|
project_id: str,
|
|
library_id: str,
|
|
storage_key: str,
|
|
ingest_job_repository: Any,
|
|
) -> Any:
|
|
use_case = SubmitIngestJobUseCase(ingest_job_repository)
|
|
job = use_case.execute(
|
|
SubmitIngestJobCommand(
|
|
workspace_id=workspace_id,
|
|
project_id=project_id,
|
|
library_id=library_id,
|
|
storage_key=storage_key,
|
|
)
|
|
)
|
|
celery_app.send_task("worker.ingest_asset", args=[job.id])
|
|
return job
|
|
|
|
|
|
@router.post("/direct/prepare", response_model=DirectUploadPrepareResponse)
|
|
async def prepare_direct_upload(
|
|
request: DirectUploadPrepareRequest,
|
|
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),
|
|
storage_service: OSSStorageService = Depends(get_storage_service),
|
|
) -> DirectUploadPrepareResponse:
|
|
"""创建浏览器直传 OSS 的短期表单签名。"""
|
|
settings = get_settings()
|
|
max_size_bytes = settings.OSS_DIRECT_UPLOAD_MAX_MB * 1024 * 1024
|
|
if request.file_size > max_size_bytes:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_413_REQUEST_ENTITY_TOO_LARGE,
|
|
detail=f"File exceeds upload limit ({settings.OSS_DIRECT_UPLOAD_MAX_MB}MB)",
|
|
)
|
|
|
|
_require_workspace_member(request.workspace_id, authenticated_user, workspace_member_repository)
|
|
_require_project_and_library(
|
|
request.workspace_id,
|
|
request.project_id,
|
|
request.library_id,
|
|
project_repository,
|
|
asset_library_repository,
|
|
)
|
|
|
|
file_id = uuid4().hex[:8]
|
|
safe_filename = request.filename.replace("/", "_").replace("\\", "_")
|
|
storage_key = f"uploads/{file_id}/{safe_filename}"
|
|
try:
|
|
payload = storage_service.create_direct_upload_post(
|
|
storage_key=storage_key,
|
|
content_type=request.content_type or "application/octet-stream",
|
|
max_size_bytes=max_size_bytes,
|
|
expires_seconds=settings.OSS_DIRECT_UPLOAD_EXPIRE_SECONDS,
|
|
)
|
|
except RuntimeError as error:
|
|
raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail=str(error)) from error
|
|
|
|
return DirectUploadPrepareResponse(
|
|
upload_url=str(payload["url"]),
|
|
method=str(payload["method"]),
|
|
storage_key=str(payload["storage_key"]),
|
|
expires_at=str(payload["expires_at"]),
|
|
fields={str(key): str(value) for key, value in dict(payload["fields"]).items()},
|
|
max_size_bytes=max_size_bytes,
|
|
)
|
|
|
|
|
|
@router.post("/direct/complete", response_model=DirectUploadCompleteResponse)
|
|
async def complete_direct_upload(
|
|
request: DirectUploadCompleteRequest,
|
|
authenticated_user: AuthenticatedUser = Depends(get_current_user),
|
|
ingest_job_repository: Any = Depends(get_ingest_job_repository),
|
|
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),
|
|
storage_service: OSSStorageService = Depends(get_storage_service),
|
|
) -> DirectUploadCompleteResponse:
|
|
"""确认浏览器直传完成并创建导入任务。"""
|
|
_require_workspace_member(request.workspace_id, authenticated_user, workspace_member_repository)
|
|
_require_project_and_library(
|
|
request.workspace_id,
|
|
request.project_id,
|
|
request.library_id,
|
|
project_repository,
|
|
asset_library_repository,
|
|
)
|
|
normalized_key = storage_service._normalize_storage_key(request.storage_key)
|
|
if not normalized_key.startswith("uploads/"):
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Invalid upload key")
|
|
if not storage_service.file_exists(normalized_key):
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Uploaded file not found")
|
|
|
|
job = _submit_ingest_job(
|
|
workspace_id=request.workspace_id,
|
|
project_id=request.project_id,
|
|
library_id=request.library_id,
|
|
storage_key=normalized_key,
|
|
ingest_job_repository=ingest_job_repository,
|
|
)
|
|
return DirectUploadCompleteResponse(storage_key=normalized_key, ingest_job_id=job.id)
|
|
|
|
|
|
@router.post("", response_model=UploadAssetResponse)
|
|
async def upload_asset(
|
|
file: UploadFile = File(..., description="要上传的文件（视频、音频、图片等）"),
|
|
workspace_id: str = Form(..., description="工作空间 ID"),
|
|
project_id: str = Form(..., description="项目 ID"),
|
|
library_id: str = Form(..., description="素材库 ID"),
|
|
authenticated_user: AuthenticatedUser = Depends(get_current_user),
|
|
ingest_job_repository: Any = Depends(get_ingest_job_repository),
|
|
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),
|
|
storage_service: OSSStorageService = Depends(get_storage_service),
|
|
) -> UploadAssetResponse:
|
|
"""上传素材文件并触发导入流水线。"""
|
|
_require_workspace_member(workspace_id, authenticated_user, workspace_member_repository)
|
|
_require_project_and_library(workspace_id, project_id, library_id, project_repository, asset_library_repository)
|
|
|
|
file_id = uuid4().hex[:8]
|
|
storage_key = f"uploads/{file_id}/{file.filename}"
|
|
|
|
file_url = storage_service.upload_file(
|
|
file.file,
|
|
storage_key,
|
|
content_type=file.content_type or "application/octet-stream",
|
|
)
|
|
|
|
job = _submit_ingest_job(
|
|
workspace_id=workspace_id,
|
|
project_id=project_id,
|
|
library_id=library_id,
|
|
storage_key=storage_key,
|
|
ingest_job_repository=ingest_job_repository,
|
|
)
|
|
|
|
return UploadAssetResponse(
|
|
storage_key=storage_key,
|
|
ingest_job_id=job.id,
|
|
url=file_url,
|
|
)
|