"""本地文件存储(开发环境):头像等小文件与上传视频存 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)