69 lines
2.6 KiB
Python
69 lines
2.6 KiB
Python
"""本地文件存储（开发环境）：头像等小文件与上传视频存 server/uploads，经 /static 访问。
|
|
|
|
生产切换 OSS 时仅需替换本模块实现，调用方不变。
|
|
"""
|
|
import time
|
|
|
|
from fastapi import UploadFile
|
|
|
|
from app.config import BASE_DIR
|
|
from app.core import errors
|
|
from app.core.response import BizException
|
|
|
|
UPLOAD_DIR = BASE_DIR / "uploads"
|
|
MAX_AVATAR_SIZE = 5 * 1024 * 1024 # 5MB
|
|
|
|
# 允许上传的视频扩展名
|
|
VIDEO_EXTS = ("mp4", "mov", "m4v", "avi", "mkv", "webm")
|
|
# 上传视频大小上限（与 config.yaml upload.max_size_mb 对齐）
|
|
MAX_VIDEO_SIZE = 500 * 1024 * 1024 # 500MB
|
|
|
|
|
|
def video_ext_of(filename: str | None) -> str:
|
|
return (filename or "").rsplit(".", 1)[-1].lower() if filename and "." in filename else ""
|
|
|
|
|
|
async def save_avatar(file: UploadFile, user_id: int) -> str:
|
|
"""保存头像，返回可访问的 URL 路径"""
|
|
ext = (file.filename or "").rsplit(".", 1)[-1].lower()
|
|
if ext not in ("jpg", "jpeg", "png"):
|
|
raise BizException(errors.PARAM_ERROR, "仅支持 jpg/png 图片")
|
|
data = await file.read()
|
|
if not data:
|
|
raise BizException(errors.PARAM_ERROR, "文件内容为空")
|
|
if len(data) > MAX_AVATAR_SIZE:
|
|
raise BizException(errors.PARAM_ERROR, "图片不能超过 5MB")
|
|
(UPLOAD_DIR / "avatars").mkdir(parents=True, exist_ok=True)
|
|
name = f"{user_id}_{int(time.time())}.{ext}"
|
|
(UPLOAD_DIR / "avatars" / name).write_bytes(data)
|
|
return f"/static/avatars/{name}"
|
|
|
|
|
|
def validate_object_key(object_key: str) -> str:
|
|
"""校验上传凭证下发的 object_key 格式，防路径穿越；返回扩展名"""
|
|
if (
|
|
not object_key.startswith("upload/tmp/")
|
|
or ".." in object_key
|
|
or "\\" in object_key
|
|
or len(object_key) > 300
|
|
):
|
|
raise BizException(errors.PARAM_ERROR, "object_key 无效")
|
|
ext = video_ext_of(object_key)
|
|
if ext not in VIDEO_EXTS:
|
|
raise BizException(errors.PARAM_ERROR, "object_key 无效")
|
|
return ext
|
|
|
|
|
|
async def save_video(file: UploadFile, object_key: str) -> tuple[str, int]:
|
|
"""按 object_key 保存上传的视频文件，返回 (可访问 URL 路径, 文件字节数)"""
|
|
validate_object_key(object_key)
|
|
data = await file.read()
|
|
if not data:
|
|
raise BizException(errors.PARAM_ERROR, "文件内容为空")
|
|
if len(data) > MAX_VIDEO_SIZE:
|
|
raise BizException(errors.PARAM_ERROR, f"视频不能超过 {MAX_VIDEO_SIZE // 1024 // 1024}MB")
|
|
path = UPLOAD_DIR / object_key
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_bytes(data)
|
|
return f"/static/{object_key}", len(data)
|