zuochenyong
ebacf01742
Co-authored-by: exinglang <exinglang@qq.com> Reviewed-on: #200 Co-authored-by: zuochenyong <zuochenyong@wonderable.ai> Co-committed-by: zuochenyong <zuochenyong@wonderable.ai>
220 lines
8.3 KiB
Python
220 lines
8.3 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 json
|
||
import os
|
||
import secrets
|
||
import subprocess
|
||
from decimal import ROUND_HALF_UP, Decimal, InvalidOperation
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
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 或 HEVC/H.265 视频及 AAC 音频;放开 mkv/avi 只会让端上放不出来,
|
||
不如在入口就挡掉。
|
||
"""
|
||
if len(data) >= 12 and data[4:8] == b"ftyp":
|
||
return ".mp4"
|
||
return None
|
||
|
||
|
||
def _probe_guide_video(path: Path) -> dict[str, Any]:
|
||
"""用 ffprobe 读取服务端可信的时长/编码信息。"""
|
||
try:
|
||
proc = subprocess.run(
|
||
[
|
||
settings.FFPROBE_BINARY,
|
||
"-v", "error",
|
||
"-show_format",
|
||
"-show_streams",
|
||
"-of", "json",
|
||
str(path),
|
||
],
|
||
capture_output=True,
|
||
text=True,
|
||
timeout=30,
|
||
check=False,
|
||
)
|
||
except FileNotFoundError as exc:
|
||
raise MediaError("服务器未安装 ffprobe，暂时无法分析视频") from exc
|
||
except (OSError, subprocess.TimeoutExpired) as exc:
|
||
raise MediaError("ffprobe 分析视频失败") from exc
|
||
if proc.returncode != 0:
|
||
raise MediaError("无法分析视频，请确认文件是有效的 MP4")
|
||
try:
|
||
payload = json.loads(proc.stdout)
|
||
streams = payload.get("streams") or []
|
||
video = next(s for s in streams if s.get("codec_type") == "video")
|
||
audio = next((s for s in streams if s.get("codec_type") == "audio"), None)
|
||
raw_duration = (
|
||
(payload.get("format") or {}).get("duration")
|
||
or video.get("duration")
|
||
)
|
||
duration_ms = int(
|
||
(Decimal(str(raw_duration)) * 1000).quantize(Decimal("1"), rounding=ROUND_HALF_UP)
|
||
)
|
||
except (StopIteration, TypeError, ValueError, InvalidOperation) as exc:
|
||
raise MediaError("视频缺少可识别的视频轨或时长") from exc
|
||
video_codec = str(video.get("codec_name") or "").lower()
|
||
audio_codec = str(audio.get("codec_name") or "").lower() if audio else None
|
||
if not 30_000 <= duration_ms <= 180_000:
|
||
raise MediaError("视频时长必须在 30～180 秒之间")
|
||
if video_codec not in ("h264", "hevc", "h265"):
|
||
raise MediaError("视频编码必须为 H.264 或 HEVC/H.265")
|
||
if audio_codec not in (None, "aac"):
|
||
raise MediaError("音频编码必须为 AAC")
|
||
return {
|
||
"duration_ms": duration_ms,
|
||
"video_codec": video_codec,
|
||
"audio_codec": audio_codec,
|
||
"analysis_status": "valid",
|
||
"analysis_error": None,
|
||
}
|
||
|
||
|
||
def save_guide_video(data: bytes) -> tuple[str, dict[str, Any]]:
|
||
"""临时落盘、ffprobe 校验后原子发布，返回 URL 与分析结果。"""
|
||
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 或 HEVC/H.265 编码)")
|
||
|
||
directory = _media_dir("guide_video")
|
||
stem = f"guide_{secrets.token_hex(8)}"
|
||
temp_path = directory / f".{stem}.upload"
|
||
final_path = directory / f"{stem}.mp4"
|
||
try:
|
||
temp_path.write_bytes(data)
|
||
analysis = _probe_guide_video(temp_path)
|
||
os.replace(temp_path, final_path)
|
||
except Exception:
|
||
temp_path.unlink(missing_ok=True)
|
||
final_path.unlink(missing_ok=True)
|
||
raise
|
||
return f"{settings.MEDIA_URL_PREFIX}/guide_video/{final_path.name}", analysis
|
||
|
||
|
||
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)
|