shaguabijia-app-server/app/core/media.py

"""用户上传文件(头像 / 反馈截图)的本地落盘 + 对外 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_named(subdir: str, name_prefix: str, 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"{name_prefix}_{secrets.token_hex(8)}{ext}"
    (_media_dir(subdir) / fname).write_bytes(data)
    return f"{settings.MEDIA_URL_PREFIX}/{subdir}/{fname}"


def _save_image(subdir: str, user_id: int, data: bytes) -> str:
    """用户上传图片落盘(文件名带 user_id 前缀)。"""
    return _save_named(subdir, f"u{user_id}", data)


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 save_feedback_qr(data: bytes) -> str:
    """保存反馈页二维码(运营后台上传的运营素材,非用户文件),返回相对 URL(`/media/feedback_qr/<file>`)。"""
    return _save_named("feedback_qr", "qr", data)


def _sniff_video_ext(data: bytes) -> str | None:
    """按魔数判定视频类型,返回扩展名;非支持类型返回 None。

    只认 MP4 家族(ISO BMFF):`....ftyp` 在偏移 4。Android ExoPlayer 与浏览器 <video>
    都稳吃 H.264/AAC 的 mp4;放开 mkv/avi 只会让端上放不出来,不如在入口就挡掉。
    """
    if len(data) >= 12 and data[4:8] == b"ftyp":
        return ".mp4"
    return None


def save_guide_video(data: bytes) -> str:
    """保存新手引导视频(运营后台上传的运营素材),返回相对 URL(`/media/guide_video/<file>`)。

    与图片分开一套校验:体积上限走 [settings.GUIDE_VIDEO_MAX_BYTES],类型只认 MP4。
    """
    if not data:
        raise MediaError("空文件")
    limit = settings.GUIDE_VIDEO_MAX_BYTES
    if len(data) > limit:
        raise MediaError(f"视频过大(上限 {limit // (1024 * 1024)}MB)")
    if _sniff_video_ext(data) is None:
        raise MediaError("仅支持 MP4 视频(H.264 编码)")

    fname = f"guide_{secrets.token_hex(8)}.mp4"
    (_media_dir("guide_video") / fname).write_bytes(data)
    return f"{settings.MEDIA_URL_PREFIX}/guide_video/{fname}"


def save_cps_image(admin_id: int, data: bytes) -> str:
    """保存 CPS 活动落地页图,返回相对 URL(`/media/cps/<file>`)。admin_id 入文件名便于追溯。"""
    return _save_image("cps", admin_id, data)


def to_abs_media_url(rel_url: str | None) -> str | None:
    """相对 `/media/...` → 基于 CPS_REDIRECT_BASE 的绝对 URL(CPS 落地页域名,跨域可访问)。

    已是绝对 URL / 空 / base 未配(dev) 时原样返回。CPS 活动图入库前转绝对,使落地页
    (coupon 域)与 admin-web(另一域,不 serve /media)都能直接 <img src> 加载。
    """
    if not rel_url or rel_url.startswith(("http://", "https://")):
        return rel_url
    base = settings.CPS_REDIRECT_BASE.rstrip("/")
    return f"{base}{rel_url}" if base else rel_url


def _delete_managed(subdir: str, url: str | None) -> None:
    """删除本服务托管的某子目录下文件;外部 URL / 空值 / 非本子目录的不处理。"""
    prefix = f"{settings.MEDIA_URL_PREFIX}/{subdir}/"
    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(subdir) / fname).unlink(missing_ok=True)
    except OSError:
        pass


def delete_avatar(url: str | None) -> None:
    """删除本服务托管的旧头像文件;外部 URL(如微信头像)或空值不处理。"""
    _delete_managed("avatars", url)


def delete_feedback_qr(url: str | None) -> None:
    """删除本服务托管的旧反馈页二维码文件;外部 URL 或空值不处理。"""
    _delete_managed("feedback_qr", url)


def delete_guide_video(url: str | None) -> None:
    """删除本服务托管的旧新手引导视频文件;外部 URL 或空值不处理。"""
    _delete_managed("guide_video", url)