marco
d55f47fc59
- model: cps_activity.image_url + 迁移 cps_activity_image 回填存量淘宝活动为现有图 - admin: 新建淘宝活动必传图(上传 /upload-image 或选已有 /activity-images),存绝对 URL - 落地页 cps_redirect: 按 link→activity 取图展示,缺图兜底默认图,img src 经 html.escape - media: save_cps_image + to_abs_media_url(基于 CPS_REDIRECT_BASE,跨域 admin-web 可显示) Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
99 lines
3.7 KiB
Python
99 lines
3.7 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 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_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
|