小鱼开发
d18e705a99
- provider: 增加 need_contour/contour_color/contour_size/need_crop_background 参数 - service: 默认 scene=human，human/product 场景自动启用白色描边 + 裁剪背景 - adapter: 透传新参数到 provider - API: scene 默认值改为 human - 前端: removeBackground 默认 scene 改为 human
238 lines
7.8 KiB
Python
238 lines
7.8 KiB
Python
"""
|
||
图片处理 API
|
||
============
|
||
|
||
提供图片上传（七牛云）和 AI 抠图（火山引擎 MediaKit）功能。
|
||
"""
|
||
|
||
import io
|
||
import logging
|
||
import uuid
|
||
from pathlib import Path
|
||
|
||
import httpx
|
||
from fastapi import APIRouter, Depends, File, HTTPException, Request, 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.services.volcengine_mediakit_service import VolcengineMediakitService
|
||
from app.utils.file_validation import check_upload_file
|
||
|
||
router = APIRouter(tags=["Image"])
|
||
|
||
logger = logging.getLogger(__name__)
|
||
settings = get_settings()
|
||
|
||
|
||
# ── Dependencies ──
|
||
|
||
async def get_mediakit_service(request: Request) -> VolcengineMediakitService:
|
||
"""FastAPI Depends：从 app.state 获取全局 VolcengineMediakitService 实例。"""
|
||
service = getattr(request.app.state, "volcengine_mediakit_service", None)
|
||
if service is None:
|
||
raise HTTPException(
|
||
status_code=503,
|
||
detail="MediaKit 服务未初始化，请检查配置",
|
||
)
|
||
return service
|
||
|
||
|
||
# ── Schemas ──
|
||
|
||
class ImageUploadResponse(BaseModel):
|
||
"""图片上传响应"""
|
||
|
||
url: str = Field(..., description="七牛云图片 URL")
|
||
key: str = Field(..., description="七牛云文件 key")
|
||
size: int = Field(..., description="文件大小（字节）")
|
||
|
||
|
||
class RemoveBackgroundResponse(BaseModel):
|
||
"""抠图响应"""
|
||
|
||
url: str = Field(..., description="抠图结果图片 URL")
|
||
|
||
|
||
class RemoveBackgroundRequest(BaseModel):
|
||
"""抠图请求"""
|
||
|
||
image_url: str = Field(..., description="原始图片 URL")
|
||
scene: str = Field(default="human", description="场景类型：general（通用）、human（人物，默认白色描边）或 product（商品）")
|
||
|
||
|
||
# ── Endpoints ──
|
||
|
||
@router.post("/upload/image", response_model=ApiResponse[ImageUploadResponse])
|
||
async def upload_image(
|
||
file: UploadFile = File(..., description="图片文件"),
|
||
current_user: User = Depends(get_current_user),
|
||
) -> ApiResponse[ImageUploadResponse]:
|
||
"""
|
||
上传图片到七牛云
|
||
|
||
支持格式：jpg, jpeg, png, gif, webp
|
||
返回七牛云永久访问 URL。
|
||
"""
|
||
try:
|
||
allowed_types = {
|
||
"image/jpeg",
|
||
"image/jpg",
|
||
"image/png",
|
||
"image/gif",
|
||
"image/webp",
|
||
}
|
||
content_type = file.content_type or ""
|
||
|
||
# 如果 content_type 为空，尝试从文件名推断
|
||
if not content_type:
|
||
ext = Path(file.filename or "").suffix.lower()
|
||
ext_to_mime = {
|
||
".jpg": "image/jpeg",
|
||
".jpeg": "image/jpeg",
|
||
".png": "image/png",
|
||
".gif": "image/gif",
|
||
".webp": "image/webp",
|
||
}
|
||
content_type = ext_to_mime.get(ext, "")
|
||
|
||
if content_type not in allowed_types:
|
||
raise HTTPException(
|
||
status_code=400,
|
||
detail=f"不支持的图片格式: {content_type}，请上传 jpg/png/gif/webp 图片",
|
||
)
|
||
|
||
# 读取文件内容
|
||
content = await file.read()
|
||
if not content:
|
||
raise HTTPException(status_code=400, detail="文件内容为空")
|
||
|
||
# 校验大小和魔数
|
||
check_upload_file(
|
||
content,
|
||
settings.UPLOAD_MAX_IMAGE_SIZE,
|
||
content_type,
|
||
"图片",
|
||
)
|
||
|
||
# 生成唯一文件名
|
||
ext = Path(file.filename or "image.jpg").suffix or ".jpg"
|
||
unique_name = f"{uuid.uuid4().hex[:16]}{ext}"
|
||
|
||
# 上传到七牛云
|
||
qiniu = get_qiniu_service()
|
||
bucket, domain = qiniu._get_bucket_and_domain("image")
|
||
file_key = qiniu.generate_key("image", unique_name)
|
||
stream = io.BytesIO(content)
|
||
result = await qiniu.upload_stream_async(
|
||
stream=stream,
|
||
key=file_key,
|
||
mime_type=content_type or "image/jpeg",
|
||
bucket=bucket,
|
||
domain=domain,
|
||
)
|
||
|
||
url = result.get("url")
|
||
returned_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=ImageUploadResponse(
|
||
url=url,
|
||
key=returned_key or file_key,
|
||
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("/image/remove-background", response_model=ApiResponse[RemoveBackgroundResponse])
|
||
async def remove_background(
|
||
req: RemoveBackgroundRequest,
|
||
current_user: User = Depends(get_current_user),
|
||
mediakit_service: VolcengineMediakitService = Depends(get_mediakit_service),
|
||
) -> ApiResponse[RemoveBackgroundResponse]:
|
||
"""
|
||
AI 抠图（火山引擎 MediaKit）
|
||
|
||
移除图片背景，返回透明背景图片 URL。
|
||
"""
|
||
try:
|
||
logger.info(
|
||
f"[RemoveBackground] 开始抠图: image_url={req.image_url[:80]}..., scene={req.scene}"
|
||
)
|
||
result = await mediakit_service.remove_background(
|
||
image_url=req.image_url,
|
||
scene=req.scene,
|
||
)
|
||
|
||
if not result.image_url:
|
||
logger.error(
|
||
f"[RemoveBackground] 抠图返回空 URL: raw={result.raw}"
|
||
)
|
||
raise HTTPException(status_code=500, detail="抠图失败：未返回结果图片 URL")
|
||
|
||
logger.info(f"[RemoveBackground] 抠图成功: {result.image_url[:80]}...")
|
||
|
||
# 下载抠图结果并转存到七牛云（避免前端 CORS 问题）
|
||
try:
|
||
async with httpx.AsyncClient(timeout=60.0) as client:
|
||
img_resp = await client.get(result.image_url, follow_redirects=True)
|
||
img_resp.raise_for_status()
|
||
img_content = img_resp.content
|
||
|
||
if not img_content:
|
||
raise HTTPException(status_code=500, detail="抠图结果下载失败：内容为空")
|
||
|
||
# 上传到七牛云 image bucket
|
||
qiniu = get_qiniu_service()
|
||
bucket, domain = qiniu._get_bucket_and_domain("image")
|
||
unique_name = f"{uuid.uuid4().hex[:16]}.png"
|
||
file_key = qiniu.generate_key("image", unique_name)
|
||
stream = io.BytesIO(img_content)
|
||
upload_result = await qiniu.upload_stream_async(
|
||
stream=stream,
|
||
key=file_key,
|
||
mime_type="image/png",
|
||
bucket=bucket,
|
||
domain=domain,
|
||
)
|
||
|
||
qiniu_url = upload_result.get("url")
|
||
if not qiniu_url:
|
||
raise HTTPException(status_code=500, detail="抠图结果转存到七牛云失败")
|
||
|
||
logger.info(f"[RemoveBackground] 结果已转存七牛云: {qiniu_url[:80]}...")
|
||
|
||
return success_response(
|
||
data=RemoveBackgroundResponse(url=qiniu_url),
|
||
message="抠图成功",
|
||
)
|
||
except HTTPException:
|
||
raise
|
||
except Exception as e:
|
||
logger.error(f"[RemoveBackground] 结果转存失败: {e}")
|
||
raise HTTPException(status_code=500, detail=f"抠图结果转存失败: {e}")
|
||
|
||
except HTTPException:
|
||
raise
|
||
except ValueError as e:
|
||
raise HTTPException(status_code=400, detail=str(e))
|
||
except Exception as e:
|
||
logger.error(
|
||
f"[RemoveBackground] 抠图失败: image_url={req.image_url[:80]}..., error={e}"
|
||
)
|
||
raise HTTPException(status_code=500, detail=f"抠图失败: {e}")
|