meijiaka-zy/python-api/app/api/v1/upload.py

"""
文件上传 API
============

提供通用文件上传功能，直接上传到七牛云对象存储。
"""

import io
import logging
import uuid
from pathlib import Path

from fastapi import APIRouter, Depends, File, HTTPException, UploadFile
from pydantic import BaseModel, Field

from app.api.deps import get_current_user
from app.config import get_settings
from app.models.user import User
from app.schemas.common import ApiResponse, success_response
from app.services.qiniu_service import get_qiniu_service
from app.utils.file_validation import check_upload_file

router = APIRouter(prefix="/upload", tags=["Upload"])

logger = logging.getLogger(__name__)
settings = get_settings()



class UploadResponse(BaseModel):
    """上传响应"""

    url: str = Field(..., description="七牛云文件 URL")
    key: str = Field(..., description="七牛云文件 key")
    size: int = Field(..., description="文件大小（字节）")


@router.post("/video", response_model=ApiResponse[UploadResponse])
async def upload_video(
    file: UploadFile = File(..., description="视频文件"),
    current_user: User = Depends(get_current_user),
):
    """
    上传视频到七牛云

    支持格式：mp4, mov, avi, webm
    返回七牛云永久访问 URL。
    """
    try:
        # 验证文件格式
        allowed_types = {
            "video/mp4",
            "video/quicktime",
            "video/x-msvideo",
            "video/webm",
        }
        content_type = file.content_type or ""

        # 如果 content_type 为空，尝试从文件名推断
        if not content_type:
            ext = Path(file.filename or "").suffix.lower()
            ext_to_mime = {
                ".mp4": "video/mp4",
                ".mov": "video/quicktime",
                ".avi": "video/x-msvideo",
                ".webm": "video/webm",
            }
            content_type = ext_to_mime.get(ext, "")

        if content_type not in allowed_types:
            raise HTTPException(
                status_code=400,
                detail=f"不支持的文件格式: {content_type}，请上传 mp4/mov/avi/webm 视频",
            )

        # 读取文件内容
        content = await file.read()
        if not content:
            raise HTTPException(status_code=400, detail="文件内容为空")

        # 校验大小和魔数
        check_upload_file(
            content,
            settings.UPLOAD_MAX_VIDEO_SIZE,
            content_type,
            "视频",
        )

        # 生成唯一文件名
        ext = Path(file.filename or "video.mp4").suffix or ".mp4"
        unique_name = f"{uuid.uuid4().hex[:16]}{ext}"

        # 上传到七牛云
        qiniu = get_qiniu_service()
        bucket, domain = qiniu._get_bucket_and_domain("video")
        key = qiniu.generate_key("video", unique_name)
        stream = io.BytesIO(content)
        result = await qiniu.upload_stream_async(
            stream=stream,
            key=key,
            mime_type=content_type or "video/mp4",
            bucket=bucket,
            domain=domain,
        )

        url = result.get("url")
        key = result.get("key")

        if not url:
            raise HTTPException(status_code=500, detail="上传到七牛云失败：未返回 URL")

        logger.info(f"[Upload] 视频上传成功: {url[:80]}..., size={len(content)}")

        return success_response(
            data=UploadResponse(
                url=url,
                key=key or unique_name,
                size=len(content),
            )
        )

    except HTTPException:
        raise
    except Exception as e:
        logger.error(f"[Upload] 视频上传失败: {e}")
        raise HTTPException(status_code=500, detail=f"上传失败: {e}")




@router.post("/audio", response_model=ApiResponse[UploadResponse])
async def upload_audio(
    file: UploadFile = File(..., description="音频文件"),
    current_user: User = Depends(get_current_user),
):
    """
    上传音频到七牛云

    支持格式：mp3, wav, aac, m4a, ogg, flac
    """
    try:
        allowed_types = {
            "audio/mpeg",
            "audio/mp3",
            "audio/wav",
            "audio/x-wav",
            "audio/aac",
            "audio/mp4",
            "audio/ogg",
            "audio/flac",
            "audio/x-flac",
        }
        content_type = file.content_type or ""

        if not content_type:
            ext = Path(file.filename or "").suffix.lower()
            ext_to_mime = {
                ".mp3": "audio/mpeg",
                ".wav": "audio/wav",
                ".aac": "audio/aac",
                ".m4a": "audio/mp4",
                ".ogg": "audio/ogg",
                ".flac": "audio/flac",
            }
            content_type = ext_to_mime.get(ext, "")

        if content_type not in allowed_types:
            raise HTTPException(
                status_code=400,
                detail=f"不支持的音频格式: {content_type}，请上传 mp3/wav/aac/m4a/ogg/flac",
            )

        content = await file.read()
        if not content:
            raise HTTPException(status_code=400, detail="文件内容为空")

        # 校验大小和魔数
        check_upload_file(
            content,
            settings.UPLOAD_MAX_AUDIO_SIZE,
            content_type,
            "音频",
        )

        ext = Path(file.filename or "audio.mp3").suffix or ".mp3"
        unique_name = f"{uuid.uuid4().hex[:16]}{ext}"

        qiniu = get_qiniu_service()
        # 复用视频 bucket（或根据配置使用音频 bucket）
        bucket, domain = qiniu._get_bucket_and_domain("video")
        key = qiniu.generate_key("audio", unique_name)
        stream = io.BytesIO(content)
        result = await qiniu.upload_stream_async(
            stream=stream,
            key=key,
            mime_type=content_type or "audio/mpeg",
            bucket=bucket,
            domain=domain,
        )

        url = result.get("url")
        key = result.get("key")

        if not url:
            raise HTTPException(status_code=500, detail="上传到七牛云失败：未返回 URL")

        logger.info(f"[Upload] 音频上传成功: {url[:80]}..., size={len(content)}")

        return success_response(
            data=UploadResponse(
                url=url,
                key=key or unique_name,
                size=len(content),
            )
        )

    except HTTPException:
        raise
    except Exception as e:
        logger.error(f"[Upload] 音频上传失败: {e}")
        raise HTTPException(status_code=500, detail=f"上传失败: {e}")