132 lines
3.5 KiB
Python
132 lines
3.5 KiB
Python
"""角色头像文件服务。
|
||
|
||
职责：
|
||
- 上传头像文件到本地文件系统
|
||
- 提供头像文件流
|
||
- 清理头像文件
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from pathlib import Path
|
||
|
||
from fastapi.responses import FileResponse
|
||
|
||
from app.core.config import settings
|
||
from app.core.errors import AppError
|
||
from app.services.characters import get_owned_character
|
||
|
||
_ALLOWED_EXTENSIONS = frozenset({".jpg", ".jpeg", ".png", ".webp"})
|
||
_MAX_FILE_SIZE = 5 * 1024 * 1024 # 5MB
|
||
|
||
|
||
def _avatar_path(character_id: str) -> Path:
|
||
"""获取角色头像文件路径（无扩展名，需遍历查找实际文件）。"""
|
||
return settings.resolved_avatar_dir / character_id
|
||
|
||
|
||
def _find_avatar_file(character_id: str) -> Path | None:
|
||
"""查找角色已有的头像文件（任一格式）。"""
|
||
base = _avatar_path(character_id)
|
||
for ext in _ALLOWED_EXTENSIONS:
|
||
f = base.with_suffix(ext)
|
||
if f.exists():
|
||
return f
|
||
return None
|
||
|
||
|
||
def _allowed(content_type: str) -> bool:
|
||
return content_type in {
|
||
"image/jpeg",
|
||
"image/png",
|
||
"image/webp",
|
||
}
|
||
|
||
|
||
async def upload_avatar(
|
||
character_id: str,
|
||
content: bytes,
|
||
content_type: str,
|
||
owner_scope: str = settings.default_owner_scope,
|
||
) -> str:
|
||
"""上传角色头像。
|
||
|
||
Args:
|
||
character_id: 角色 ID
|
||
content: 图片文件二进制内容
|
||
content_type: MIME 类型
|
||
owner_scope: 归属范围
|
||
|
||
Returns:
|
||
头像访问的相对 URL 路径
|
||
|
||
Raises:
|
||
AppError: 文件类型不允许、大小超限、角色不存在
|
||
"""
|
||
from app.db.session import AsyncSessionLocal
|
||
|
||
# 校验角色存在性
|
||
async with AsyncSessionLocal() as session:
|
||
await get_owned_character(session, character_id, owner_scope=owner_scope)
|
||
|
||
if not _allowed(content_type):
|
||
raise AppError(
|
||
"invalid_input",
|
||
f"不支持的文件类型 '{content_type}'，仅支持 JPG/PNG/WebP。",
|
||
)
|
||
if len(content) > _MAX_FILE_SIZE:
|
||
raise AppError(
|
||
"invalid_input",
|
||
"头像文件过大，最大允许 5MB。",
|
||
)
|
||
|
||
# 扩展名映射
|
||
ext_map = {
|
||
"image/jpeg": ".jpg",
|
||
"image/png": ".png",
|
||
"image/webp": ".webp",
|
||
}
|
||
ext = ext_map[content_type]
|
||
|
||
# 删除旧头像
|
||
old = _find_avatar_file(character_id)
|
||
if old:
|
||
old.unlink(missing_ok=True)
|
||
|
||
# 写入新文件
|
||
dest = _avatar_path(character_id).with_suffix(ext)
|
||
dest.parent.mkdir(parents=True, exist_ok=True)
|
||
dest.write_bytes(content)
|
||
|
||
return f"/api/v1/characters/{character_id}/avatar"
|
||
|
||
|
||
async def get_avatar_response(character_id: str) -> FileResponse:
|
||
"""获取角色头像文件响应。
|
||
|
||
Args:
|
||
character_id: 角色 ID
|
||
|
||
Returns:
|
||
FileResponse 或 None（无头像时抛出 not_found）
|
||
"""
|
||
avatar_file = _find_avatar_file(character_id)
|
||
if avatar_file is None:
|
||
raise AppError("not_found", "该角色未设置头像。")
|
||
# 推测 media_type
|
||
ext = avatar_file.suffix.lower()
|
||
media_type = {
|
||
".jpg": "image/jpeg",
|
||
".jpeg": "image/jpeg",
|
||
".png": "image/png",
|
||
".webp": "image/webp",
|
||
}.get(ext, "application/octet-stream")
|
||
return FileResponse(str(avatar_file), media_type=media_type)
|
||
|
||
|
||
async def delete_avatar(character_id: str) -> None:
|
||
"""删除角色头像文件。"""
|
||
avatar_file = _find_avatar_file(character_id)
|
||
if avatar_file:
|
||
avatar_file.unlink(missing_ok=True)
|