zhangxianze
4506677483
Co-authored-by: xianze <ze@192.168.0.138> Reviewed-on: #13 Co-authored-by: zhangxianze <zhangxianze@wonderable.ai> Co-committed-by: zhangxianze <zhangxianze@wonderable.ai>
82 lines
2.9 KiB
Python
82 lines
2.9 KiB
Python
"""用户上传文件(头像 / 反馈截图)的本地落盘 + 对外 URL 构建。
|
|
|
|
落盘到 `settings.MEDIA_ROOT/<子目录>/`,对外以 `settings.MEDIA_URL_PREFIX` 暴露
|
|
(main.py 用 StaticFiles 挂载)。返回**相对路径** URL(如 `/media/avatars/xxx.jpg`),
|
|
由客户端按自己的 BASE_URL 解析为绝对地址——服务端不知道客户端怎么访问到自己(dev 下
|
|
可能是 10.0.2.2 / LAN IP),所以不返回绝对 URL。
|
|
|
|
安全:不信任客户端给的 content-type / 文件名,改用魔数嗅探判定真实图片类型;
|
|
文件名服务端随机生成,杜绝路径穿越与覆盖。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import secrets
|
|
from pathlib import Path
|
|
|
|
from app.core.config import settings
|
|
|
|
|
|
class MediaError(Exception):
|
|
"""上传文件不合法(类型/大小)。调用方转 400。"""
|
|
|
|
|
|
def _media_dir(subdir: str) -> Path:
|
|
d = Path(settings.MEDIA_ROOT) / subdir
|
|
d.mkdir(parents=True, exist_ok=True)
|
|
return d
|
|
|
|
|
|
def _sniff_ext(data: bytes) -> str | None:
|
|
"""按魔数判定图片类型,返回扩展名;非支持类型返回 None。"""
|
|
if data[:3] == b"\xff\xd8\xff":
|
|
return ".jpg"
|
|
if data[:8] == b"\x89PNG\r\n\x1a\n":
|
|
return ".png"
|
|
if data[:4] == b"RIFF" and data[8:12] == b"WEBP":
|
|
return ".webp"
|
|
return None
|
|
|
|
|
|
def _save_image(subdir: str, user_id: int, data: bytes) -> str:
|
|
"""通用图片落盘:校验非空 / ≤上限 / 魔数为图片,随机文件名,返回相对 URL。"""
|
|
if not data:
|
|
raise MediaError("空文件")
|
|
if len(data) > settings.AVATAR_MAX_BYTES:
|
|
raise MediaError("图片过大(上限 5MB)")
|
|
ext = _sniff_ext(data)
|
|
if ext is None:
|
|
raise MediaError("仅支持 JPEG / PNG / WebP 图片")
|
|
|
|
fname = f"u{user_id}_{secrets.token_hex(8)}{ext}"
|
|
(_media_dir(subdir) / fname).write_bytes(data)
|
|
return f"{settings.MEDIA_URL_PREFIX}/{subdir}/{fname}"
|
|
|
|
|
|
def save_avatar(user_id: int, data: bytes) -> str:
|
|
"""保存头像,返回相对 URL(`/media/avatars/<file>`)。"""
|
|
return _save_image("avatars", user_id, data)
|
|
|
|
|
|
def save_feedback_image(user_id: int, data: bytes) -> str:
|
|
"""保存反馈截图,返回相对 URL(`/media/feedback/<file>`)。"""
|
|
return _save_image("feedback", user_id, data)
|
|
|
|
|
|
def save_report_image(user_id: int, data: bytes) -> str:
|
|
"""保存上报更低价截图,返回相对 URL(`/media/price_report/<file>`)。"""
|
|
return _save_image("price_report", user_id, data)
|
|
|
|
|
|
def delete_avatar(url: str | None) -> None:
|
|
"""删除本服务托管的旧头像文件;外部 URL(如微信头像)或空值不处理。"""
|
|
prefix = f"{settings.MEDIA_URL_PREFIX}/avatars/"
|
|
if not url or not url.startswith(prefix):
|
|
return
|
|
fname = url[len(prefix):]
|
|
if not fname or "/" in fname or "\\" in fname or ".." in fname:
|
|
return # 防路径穿越
|
|
try:
|
|
(_media_dir("avatars") / fname).unlink(missing_ok=True)
|
|
except OSError:
|
|
pass
|