diff --git a/alembic/versions/cps_activity_image.py b/alembic/versions/cps_activity_image.py
new file mode 100644
index 0000000..e677950
--- /dev/null
+++ b/alembic/versions/cps_activity_image.py
@@ -0,0 +1,38 @@
+"""cps activity 落地页图: cps_activity.image_url + 回填存量淘宝活动
+
+新建淘宝活动时运营提供落地页主视觉图(上传或选已有),落地页按活动展示对应图。
+存量淘宝活动回填为此前写死的那张图,无缝衔接。
+
+Revision ID: cps_activity_image
+Revises: cps_v2_platforms
+Create Date: 2026-06-17 12:00:00.000000
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+from app.core.config import settings
+
+
+revision: str = "cps_activity_image"
+down_revision: Union[str, Sequence[str], None] = "cps_v2_platforms"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+ op.add_column("cps_activity", sa.Column("image_url", sa.String(512), nullable=True))
+ # 回填存量淘宝活动 = 此前写死的那张图(绝对 URL,基于落地页域名 CPS_REDIRECT_BASE;
+ # dev 未配则相对路径)。新活动由运营自传/选,不回填。
+ base = settings.CPS_REDIRECT_BASE.rstrip("/")
+ default_img = f"{base}/media/taobao_landing.jpg" if base else "/media/taobao_landing.jpg"
+ op.execute(
+ sa.text(
+ "UPDATE cps_activity SET image_url = :img WHERE platform = 'taobao'"
+ ).bindparams(img=default_img)
+ )
+
+
+def downgrade() -> None:
+ op.drop_column("cps_activity", "image_url")
diff --git a/app/admin/repositories/cps.py b/app/admin/repositories/cps.py
index 6733de9..ee39955 100644
--- a/app/admin/repositories/cps.py
+++ b/app/admin/repositories/cps.py
@@ -137,14 +137,29 @@ def list_activities(
return offset_paginate(db, stmt, (desc(CpsActivity.id),), limit=limit, cursor=cursor)
+def list_activity_images(db: Session) -> list[str]:
+ """已用过的活动落地页图(distinct 非空 image_url),供新建活动复用选择。"""
+ rows = (
+ db.execute(
+ select(CpsActivity.image_url)
+ .where(CpsActivity.image_url.is_not(None))
+ .distinct()
+ )
+ .scalars()
+ .all()
+ )
+ return [r for r in rows if r]
+
+
def create_activity(
db: Session, *, name: str, platform: str = "meituan", act_id: str | None = None,
product_view_sign: str | None = None, payload: str | None = None,
- remark: str | None = None, commit: bool = True,
+ image_url: str | None = None, remark: str | None = None, commit: bool = True,
) -> CpsActivity:
activity = CpsActivity(
name=name, platform=platform, act_id=act_id,
- product_view_sign=product_view_sign, payload=payload, remark=remark,
+ product_view_sign=product_view_sign, payload=payload,
+ image_url=image_url, remark=remark,
)
db.add(activity)
if commit:
diff --git a/app/admin/routers/cps.py b/app/admin/routers/cps.py
index 5eae8fe..71b7078 100644
--- a/app/admin/routers/cps.py
+++ b/app/admin/routers/cps.py
@@ -10,7 +10,7 @@ import time
from datetime import datetime, timedelta, timezone
from typing import Annotated
-from fastapi import APIRouter, Depends, HTTPException, Query, Request
+from fastapi import APIRouter, Depends, File, HTTPException, Query, Request, UploadFile
from app.admin.audit import write_audit
from app.admin.deps import AdminDb, get_client_ip, get_current_admin, require_role
@@ -30,6 +30,7 @@ from app.admin.schemas.cps import (
CpsReferralLinksRequest,
CpsStatsOut,
)
+from app.core import media
from app.core.config import settings
from app.integrations import meituan
from app.integrations.meituan import MeituanCpsError
@@ -154,10 +155,12 @@ def create_activity(
if not body.payload:
label = "淘口令" if body.platform == "taobao" else "推广链接"
raise HTTPException(status_code=400, detail=f"{body.platform} 活动需填{label}")
+ if body.platform == "taobao" and not body.image_url:
+ raise HTTPException(status_code=400, detail="淘宝活动需提供落地页图(上传或选已有)")
activity = cps_repo.create_activity(
db, name=body.name, platform=body.platform, act_id=body.act_id,
product_view_sign=body.product_view_sign, payload=body.payload,
- remark=body.remark, commit=False,
+ image_url=media.to_abs_media_url(body.image_url), remark=body.remark, commit=False,
)
write_audit(
db, admin, action="cps.activity.create", target_type="cps_activity", target_id=activity.id,
@@ -169,6 +172,24 @@ def create_activity(
return CpsActivityOut.model_validate(activity)
+@router.post("/upload-image", summary="上传活动落地页图(返回绝对 URL)")
+async def upload_activity_image(
+ admin: Annotated[AdminUser, Depends(require_role("operator"))],
+ file: UploadFile = File(..., description="落地页图(jpeg/png/webp,≤5MB)"),
+) -> dict:
+ data = await file.read()
+ try:
+ url = media.save_cps_image(admin.id, data)
+ except media.MediaError as e:
+ raise HTTPException(status_code=400, detail=str(e)) from e
+ return {"url": media.to_abs_media_url(url)}
+
+
+@router.get("/activity-images", summary="已有活动落地页图(供新建复用选择)")
+def activity_images(db: AdminDb) -> dict:
+ return {"images": cps_repo.list_activity_images(db)}
+
+
# ───────────── 生成链接(批量,按平台分支) ─────────────
def _gen_one_link(db, group: CpsGroup, activity: CpsActivity):
"""给一个活动生成一条落地页 link。美团:转链拿短链(需群 sid);淘宝/京东:用 payload。"""
diff --git a/app/admin/schemas/cps.py b/app/admin/schemas/cps.py
index 5ca7dbb..4e57c6c 100644
--- a/app/admin/schemas/cps.py
+++ b/app/admin/schemas/cps.py
@@ -52,6 +52,7 @@ class CpsActivityOut(BaseModel):
act_id: str | None = None
product_view_sign: str | None = None
payload: str | None = None # 淘宝淘口令 / 京东链接
+ image_url: str | None = None # 淘宝落地页图(绝对 URL)
status: str
remark: str | None = None
created_at: datetime
@@ -65,6 +66,8 @@ class CpsActivityCreate(BaseModel):
product_view_sign: str | None = Field(default=None, max_length=128)
# 淘宝:整段淘口令文本 / 京东:推广链接
payload: str | None = Field(default=None, max_length=4096)
+ # 淘宝落地页图(绝对/相对 URL;淘宝必填,由 router 校验)
+ image_url: str | None = Field(default=None, max_length=512)
remark: str | None = Field(default=None, max_length=256)
diff --git a/app/api/v1/cps_redirect.py b/app/api/v1/cps_redirect.py
index 0c604dd..9be15e6 100644
--- a/app/api/v1/cps_redirect.py
+++ b/app/api/v1/cps_redirect.py
@@ -6,13 +6,16 @@
"""
from __future__ import annotations
+import html
import json
from fastapi import APIRouter, Depends, Request
from fastapi.responses import HTMLResponse, RedirectResponse
from sqlalchemy.orm import Session
+from app.core import media
from app.db.session import get_db
+from app.models.cps_activity import CpsActivity
from app.models.cps_link import CpsLink
from app.repositories import cps_link as cps_link_repo
@@ -21,6 +24,9 @@ router = APIRouter(tags=["cps-redirect"])
# code 不存在/失效时的兜底落地
_FALLBACK_URL = "https://www.meituan.com/"
+# 淘宝活动未设图时的兜底主视觉(存量已回填,基本只在老 link/异常时触发)
+_DEFAULT_TAOBAO_IMAGE = "/media/taobao_landing.jpg"
+
def _client_ip(request: Request) -> str | None:
xff = request.headers.get("x-forwarded-for")
@@ -46,7 +52,9 @@ def cps_landing(code: str, request: Request, db: Session = Depends(get_db)):
return RedirectResponse(_FALLBACK_URL, status_code=302)
_record(db, link, request, "visit")
if link.platform == "taobao":
- return HTMLResponse(_taobao_landing_html(link.target_url))
+ activity = db.get(CpsActivity, link.activity_id)
+ image_url = (activity.image_url if activity else None) or _DEFAULT_TAOBAO_IMAGE
+ return HTMLResponse(_taobao_landing_html(link.target_url, image_url))
# 美团短链 / 京东链接:直接 302 跳
return RedirectResponse(link.target_url, status_code=302)
@@ -59,9 +67,18 @@ def cps_copy(code: str, request: Request, db: Session = Depends(get_db)) -> dict
return {"ok": True}
-def _taobao_landing_html(token: str) -> str:
- """淘宝落地页 H5(主视觉为设计图 /media/taobao_landing.jpg,复制淘口令按钮在 75% 屏高)。token 安全嵌入 JS。"""
- return _TAOBAO_HTML.replace("__TOKEN_JS__", json.dumps(token))
+def _taobao_landing_html(token: str, image_url: str) -> str:
+ """淘宝落地页 H5(主视觉图按活动传入,复制淘口令按钮在 75% 屏高)。
+
+ image_url 转绝对(落地页 coupon 域,相对也可达,绝对更稳)后经 html.escape 嵌入
+ (防属性注入);token 经 json.dumps 安全嵌入 JS。
+ """
+ safe_img = html.escape(media.to_abs_media_url(image_url) or image_url, quote=True)
+ return (
+ _TAOBAO_HTML
+ .replace("__IMAGE_URL__", safe_img)
+ .replace("__TOKEN_JS__", json.dumps(token))
+ )
_TAOBAO_HTML = """
@@ -82,7 +99,7 @@ body{font-family:-apple-system,"PingFang SC",sans-serif;background:#fff0ef;color
+