"""用户上传文件(头像 / 反馈截图)的本地落盘 + 对外 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/`)。""" return _save_image("avatars", user_id, data) def save_feedback_image(user_id: int, data: bytes) -> str: """保存反馈截图,返回相对 URL(`/media/feedback/`)。""" return _save_image("feedback", 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