修复 DeepSeek V4 Flash TOKEN 成本高估 (#214)

## 改动
- 为 deepseek-v4-flash 配置 DashScope 华北 2 官方单价：输入 ¥1 / 输出 ¥2（每百万 Token）
- 定向重算历史上误用 3/15 兜底价冻结的成本和价格快照
- 保留已有人工单价及配置生效时间，避免影响历史缺失成本回填
- 增加迁移升级、降级和原配置保留测试

## 验证
- 相关测试：17 passed
- ruff check：通过

---------

Co-authored-by: guke <guke@wonderable.ai>
Co-authored-by: linkeyu <798648091@qq.com>
Co-authored-by: unknown <798648091@qq.com>
Reviewed-on: #214
Co-authored-by: linkeyu <linkeyu@wonderable.ai>
Co-committed-by: linkeyu <linkeyu@wonderable.ai>

app/admin/repositories/coupon_data.py

@@ -166,6 +166,18 @@ def _session_to_row(
     point_stats: dict | None = None,
 ) -> dict:
     """CouponSession ORM → 明细行 dict(主表「领券数据」与「用户全部领券」抽屉共用)。"""
+    # 中途退出可能发生在第一张券产生终态之前，此时没有逐券事件。
+    # 明确返回 0/0，让前端区分「退出前无单券结果」与其它状态的埋点缺失。
+    if point_stats is not None:
+        point_success_count = point_stats["succeeded"]
+        point_total_count = point_stats["tried"]
+    elif r.status == "abandoned":
+        point_success_count = 0
+        point_total_count = 0
+    else:
+        point_success_count = None
+        point_total_count = None
+    point_event_count = point_stats["events"] if point_stats is not None else 0
     return {
         "id": r.id,
         "trace_id": r.trace_id,
@@ -182,8 +194,9 @@ def _session_to_row(
         "app_env": r.app_env,
         "started_at": r.started_at,
         "claimed_count": r.claimed_count,
-        "point_success_count": point_stats["succeeded"] if point_stats else None,
-        "point_total_count": point_stats["tried"] if point_stats else None,
+        "point_success_count": point_success_count,
+        "point_total_count": point_total_count,
+        "point_event_count": point_event_count,
         "trace_url": r.trace_url,
         "ad_revenue_yuan": ad_revenue_yuan,
     }
@@ -194,21 +207,24 @@ def _point_scores_by_trace(db: Session, trace_ids: list[str]) -> dict[str, dict[
     if not trace_ids:
         return {}
     succeeded = func.sum(case((CouponClaimEvent.status.in_(_SLOT_OK), 1), else_=0))
+    tried = func.sum(case((CouponClaimEvent.status.in_(_SLOT_TRIED), 1), else_=0))
     rows = db.execute(
         select(
             CouponClaimEvent.trace_id,
             succeeded.label("succeeded"),
-            func.count().label("tried"),
-        )
-        .where(
-            CouponClaimEvent.trace_id.in_(trace_ids),
-            CouponClaimEvent.status.in_(_SLOT_TRIED),
+            tried.label("tried"),
+            func.count().label("events"),
         )
+        .where(CouponClaimEvent.trace_id.in_(trace_ids))
         .group_by(CouponClaimEvent.trace_id)
     ).all()
     return {
-        trace_id: {"succeeded": int(success_count or 0), "tried": int(tried or 0)}
-        for trace_id, success_count, tried in rows
+        trace_id: {
+            "succeeded": int(success_count or 0),
+            "tried": int(tried_count or 0),
+            "events": int(event_count or 0),
+        }
+        for trace_id, success_count, tried_count, event_count in rows
         if trace_id is not None
     }
 
@@ -400,12 +416,15 @@ def coupon_user_records(db: Session, *, user_id: int, limit: int = 100) -> dict:
     total = db.execute(
         select(func.count()).select_from(CouponSession).where(CouponSession.user_id == user_id)
     ).scalar_one()
-    rev_map = crud_ecpm.revenue_yuan_by_trace(db, [r.trace_id for r in rows])
+    trace_ids = [r.trace_id for r in rows]
+    rev_map = crud_ecpm.revenue_yuan_by_trace(db, trace_ids)
+    point_stats_map = _point_scores_by_trace(db, trace_ids)
     return {
         "items": [
             _session_to_row(
                 r,
                 ad_revenue_yuan=rev_map.get(r.trace_id, 0.0),
+                point_stats=point_stats_map.get(r.trace_id),
             )
             for r in rows
         ],

app/admin/repositories/stats.py

@@ -566,6 +566,10 @@ def dashboard_overview(
         )
     ).all()
     coupon_started = len(period_coupon_sessions)
+    coupon_abandoned = sum(s.status == "abandoned" for s in period_coupon_sessions)
+    # 用户主动中途退出不代表领券流程失败，不进入整场成功率样本。
+    # started / failed 仍留在分母：前者是尚未形成终态的流失，后者是实际执行失败。
+    coupon_success_denominator = coupon_started - coupon_abandoned
     coupon_completed_elapsed = sorted(
         s.elapsed_ms
         for s in period_coupon_sessions
@@ -731,9 +735,13 @@ def dashboard_overview(
             },
             "coupon": {
                 "started": coupon_started,
+                "abandoned": coupon_abandoned,
+                "success_denominator": coupon_success_denominator,
                 "all_success": coupon_all_success,
                 "success_rate": (
-                    round(coupon_all_success / coupon_started, 4) if coupon_started else None
+                    round(coupon_all_success / coupon_success_denominator, 4)
+                    if coupon_success_denominator
+                    else None
                 ),
                 "point_success": coupon_point_success,
                 "points_per_session": coupon_points_per_session,

app/admin/schemas/coupon_data.py

@@ -79,10 +79,16 @@ class CouponDataRow(BaseModel):
     started_at: datetime = Field(..., description="发起时刻(明细「时间」列)")
     claimed_count: int | None = None
     point_success_count: int | None = Field(
-        None, description="本次成功单券数(success+already_claimed);无逐券事件为空"
+        None,
+        description="本次成功单券数(success+already_claimed);中途退出且无逐券结果为0,其它无事件为空",
     )
     point_total_count: int | None = Field(
-        None, description="本次尝试单券数(success+already_claimed+failed,不含 skipped);无逐券事件为空"
+        None,
+        description="本次尝试单券数(success+already_claimed+failed,不含 skipped);中途退出且无逐券结果为0,其它无事件为空",
+    )
+    point_event_count: int = Field(
+        0,
+        description="本次全部逐券事件数(含 skipped);用于区分无有效计分事件与完全无事件",
     )
     trace_url: str | None = Field(None, description="pricebot 公网 trace 链接(仅 completed 有);admin 渲染可点链接,无则显示可复制 trace_id")
     ad_revenue_yuan: float = Field(

app/admin/schemas/dashboard.py

@@ -70,6 +70,9 @@ class DashboardPeriodCoupon(BaseModel):
     成功口径 success+already_claimed(与「我的」页累计领券一致)。"""
 
     started: int = 0
+    # 用户主动中途退出，不计入整场成功率分母。
+    abandoned: int = 0
+    success_denominator: int = 0
     # 全部领成功的次数:completed 且当日该设备全部点位成功
     all_success: int = 0
     success_rate: float | None = None

app/api/v1/feedback.py

@@ -54,11 +54,13 @@ def _app_status(db_status: str) -> str:
 
 
 def _record_out(fb) -> FeedbackRecordOut:
+    images = fb.images or []
     return FeedbackRecordOut(
         id=fb.id,
         content=fb.content,
         scene=getattr(fb, "scene", None),
-        images=fb.images or [],
+        images=images,
+        image_thumbnails=[media.feedback_thumbnail_url(url) for url in images],
         status=_app_status(fb.status),
         reject_reason=getattr(fb, "reject_reason", None),
         reward_coins=getattr(fb, "reward_coins", None),
@@ -84,7 +86,7 @@ async def submit_feedback(
     device_model: str = Form(default=""),
     rom_name: str = Form(default=""),
     android_version: str = Form(default=""),
-    images: list[UploadFile] = File(default=[]),
+    images: list[UploadFile] = File(default=[]),  # noqa: B008 - FastAPI dependency declaration
 ) -> FeedbackOut:
     content = content.strip()
     contact = contact.strip()

app/core/media.py

@@ -11,6 +11,7 @@
 from __future__ import annotations
 
 import json
+import logging
 import os
 import secrets
 import subprocess
@@ -18,8 +19,17 @@ from decimal import ROUND_HALF_UP, Decimal, InvalidOperation
 from pathlib import Path
 from typing import Any
 
+from PIL import Image, ImageOps
+
 from app.core.config import settings
 
+logger = logging.getLogger("shagua.media")
+
+_FEEDBACK_DIR = "feedback"
+_FEEDBACK_THUMB_DIR = "feedback_thumbs"
+_FEEDBACK_THUMB_MAX_PX = 256
+_FEEDBACK_THUMB_QUALITY = 78
+
 
 class MediaError(Exception):
     """上传文件不合法(类型/大小)。调用方转 400。"""
@@ -68,8 +78,86 @@ def save_avatar(user_id: int, data: bytes) -> str:
 
 
 def save_feedback_image(user_id: int, data: bytes) -> str:
-    """保存反馈截图,返回相对 URL(`/media/feedback/<file>`)。"""
-    return _save_image("feedback", user_id, data)
+    """保存反馈截图并预生成历史页缩略图，返回原图相对 URL。"""
+    url = _save_image(_FEEDBACK_DIR, user_id, data)
+    # 缩略图失败不影响反馈受理；读取缩略图 URL 时会按需重试并回退原图。
+    ensure_feedback_thumbnail(url)
+    return url
+
+
+def feedback_thumbnail_url(image_url: str) -> str:
+    """把反馈原图 URL 映射成确定的缩略图 URL，不在 records 接口内做图片解码。
+
+    上传文件名由服务端生成且不会覆盖；旧数据在客户端真正请求可见图片时按需补图。
+    """
+    paths = _feedback_thumbnail_paths(image_url)
+    return paths[2] if paths is not None else image_url
+
+
+def _feedback_thumbnail_paths(image_url: str) -> tuple[Path, Path, str] | None:
+    prefix = f"{settings.MEDIA_URL_PREFIX}/{_FEEDBACK_DIR}/"
+    if not image_url.startswith(prefix):
+        return None
+
+    filename = image_url.removeprefix(prefix)
+    # 只接受当前目录下的单个文件名，避免数据库脏数据造成路径穿越。
+    if not filename or Path(filename).name != filename:
+        return None
+
+    source = _media_dir(_FEEDBACK_DIR) / filename
+    thumb_name = f"{Path(filename).stem}.jpg"
+    thumb = _media_dir(_FEEDBACK_THUMB_DIR) / thumb_name
+    thumb_url = f"{settings.MEDIA_URL_PREFIX}/{_FEEDBACK_THUMB_DIR}/{thumb_name}"
+    return source, thumb, thumb_url
+
+
+def ensure_feedback_thumbnail(image_url: str) -> Path | None:
+    """确保缩略图存在并返回文件；生成失败时回退原图，供动态缩略图路由使用。"""
+    paths = _feedback_thumbnail_paths(image_url)
+    if paths is None:
+        return None
+    source, thumb, _ = paths
+    if thumb.is_file():
+        return thumb
+    if not source.is_file():
+        return None
+
+    temp = thumb.with_name(f".{thumb.name}.{secrets.token_hex(4)}.tmp")
+    try:
+        with Image.open(source) as opened:
+            image = ImageOps.exif_transpose(opened)
+            image.thumbnail(
+                (_FEEDBACK_THUMB_MAX_PX, _FEEDBACK_THUMB_MAX_PX),
+                Image.Resampling.LANCZOS,
+            )
+            if image.mode != "RGB":
+                image = image.convert("RGB")
+            image.save(
+                temp,
+                format="JPEG",
+                quality=_FEEDBACK_THUMB_QUALITY,
+                optimize=True,
+            )
+        os.replace(temp, thumb)
+        return thumb
+    except (Image.DecompressionBombError, OSError, ValueError):
+        logger.warning("生成反馈缩略图失败: %s", source, exc_info=True)
+        return source
+    finally:
+        temp.unlink(missing_ok=True)
+
+
+def feedback_thumbnail_file(filename: str) -> Path | None:
+    """由缩略图文件名找到原反馈图并按需生成，非法/不存在返回 None。"""
+    if not filename or Path(filename).name != filename or Path(filename).suffix.lower() != ".jpg":
+        return None
+    stem = Path(filename).stem
+    for ext in (".jpg", ".png", ".webp"):
+        original = _media_dir(_FEEDBACK_DIR) / f"{stem}{ext}"
+        if original.is_file():
+            original_url = f"{settings.MEDIA_URL_PREFIX}/{_FEEDBACK_DIR}/{original.name}"
+            return ensure_feedback_thumbnail(original_url)
+    return None
 
 
 def save_report_image(user_id: int, data: bytes) -> str:

app/main.py

@@ -10,7 +10,7 @@ from collections.abc import AsyncIterator
 from contextlib import asynccontextmanager
 from pathlib import Path
 
-from fastapi import FastAPI
+from fastapi import FastAPI, HTTPException
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.responses import FileResponse
 from fastapi.staticfiles import StaticFiles
@@ -44,6 +44,7 @@ from app.api.v1.tasks import router as tasks_router
 from app.api.v1.user import router as user_router
 from app.api.v1.wallet import router as wallet_router
 from app.api.v1.wxpay import router as wxpay_router
+from app.core import media
 from app.core.config import settings
 from app.core.cps_reconcile_worker import (
     start_cps_reconcile_worker,
@@ -82,6 +83,19 @@ setup_logging(debug=settings.APP_DEBUG)
 logger = logging.getLogger("shagua.main")
 
 
+class FeedbackMediaStaticFiles(StaticFiles):
+    """反馈原图/缩略图文件名不可变，可长期缓存，避免列表反复回源。"""
+
+    async def get_response(self, path: str, scope):
+        response = await super().get_response(path, scope)
+        media_path = path.replace("\\", "/").lstrip("/")
+        if response.status_code == 200 and media_path.startswith(
+            ("feedback/", "feedback_thumbs/")
+        ):
+            response.headers["Cache-Control"] = "public, max-age=31536000, immutable"
+        return response
+
+
 @asynccontextmanager
 async def lifespan(_: FastAPI) -> AsyncIterator[None]:
     # 提示而非强制建表:生产用 alembic upgrade head,本地 dev 也建议先跑一次 migration。
@@ -212,8 +226,24 @@ def download_apk() -> FileResponse:
     )
 
 
+@app.get(
+    f"{settings.MEDIA_URL_PREFIX}/feedback_thumbs/{{filename}}",
+    tags=["feedback"],
+    include_in_schema=False,
+)
+def feedback_thumbnail(filename: str) -> FileResponse:
+    """旧反馈图按首次可见请求补缩略图；新图上传时已预生成。"""
+    path = media.feedback_thumbnail_file(filename)
+    if path is None:
+        raise HTTPException(status_code=404, detail="图片不存在")
+    return FileResponse(
+        path,
+        headers={"Cache-Control": "public, max-age=31536000, immutable"},
+    )
+
+
 app.mount(
     settings.MEDIA_URL_PREFIX,
-    StaticFiles(directory=str(_media_root)),
+    FeedbackMediaStaticFiles(directory=str(_media_root)),
     name="media",
 )

app/schemas/feedback.py

@@ -33,6 +33,8 @@ class FeedbackRecordOut(BaseModel):
     # 比价反馈的问题场景(找错商品/优惠不对…);普通反馈为 None
     scene: str | None = None
     images: list[str] = Field(default_factory=list)
+    # 与 images 下标一一对应；生成失败时该项回退原图 URL，兼容历史数据。
+    image_thumbnails: list[str] = Field(default_factory=list)
     status: str
     reject_reason: str | None = None
     reward_coins: int | None = None

pyproject.toml

@@ -35,6 +35,9 @@ dependencies = [
     # multipart form (FastAPI 表单上传依赖)
     "python-multipart>=0.0.9",
 
+    # 用户反馈截图缩略图，避免 App 历史页为 48dp 小图下载数 MB 原图
+    "pillow>=11.0.0",
+
     # admin 后台账号密码 hash(用户侧是手机号+验证码登录,不需要密码;admin 才用)
     "bcrypt>=4.0.0",
 

tests/test_admin_read.py

@@ -1,7 +1,7 @@
 """Admin M2 读接口测试:大盘聚合 + 用户/流水/提现/反馈列表 + 鉴权拦截。"""
 from __future__ import annotations
 
-from datetime import UTC, datetime
+from datetime import UTC, date, datetime
 
 import pytest
 from fastapi.testclient import TestClient
@@ -12,6 +12,7 @@ from app.admin.repositories import admin_user as admin_repo
 from app.admin.repositories import queries
 from app.db.session import SessionLocal, engine
 from app.models.comparison import ComparisonRecord
+from app.models.coupon_state import CouponClaimRecord, CouponSession
 from app.models.feedback import Feedback
 from app.models.invite import InviteRelation
 from app.models.savings import SavingsRecord
@@ -120,6 +121,86 @@ def test_dashboard_period_comparison_is_aggregated_by_backend(
     assert comparison["token_cost_total_yuan"] == pytest.approx(1.0)
 
 
+def test_dashboard_coupon_success_rate_excludes_abandoned_sessions(
+    admin_client: TestClient, admin_token: str
+) -> None:
+    started_date = date(2038, 1, 16)
+    started_at = datetime(2038, 1, 16, 8, tzinfo=UTC)
+    sessions = [
+        ("coupon-rate-completed-1", "coupon-rate-device-1", "completed"),
+        ("coupon-rate-completed-2", "coupon-rate-device-2", "completed"),
+        ("coupon-rate-failed", "coupon-rate-device-3", "failed"),
+        ("coupon-rate-abandoned", "coupon-rate-device-4", "abandoned"),
+    ]
+    db = SessionLocal()
+    try:
+        for trace_id, device_id, status in sessions:
+            db.add(
+                CouponSession(
+                    trace_id=trace_id,
+                    device_id=device_id,
+                    status=status,
+                    app_env="prod",
+                    platforms=["meituan-waimai"],
+                    started_at=started_at,
+                    started_date=started_date,
+                )
+            )
+        for index, device_id in enumerate(("coupon-rate-device-1", "coupon-rate-device-2")):
+            db.add(
+                CouponClaimRecord(
+                    device_id=device_id,
+                    coupon_id=f"mt_dashboard_rate_{index}",
+                    claim_date=started_date,
+                    status="success",
+                    app_env="prod",
+                )
+            )
+        db.commit()
+    finally:
+        db.close()
+
+    response = admin_client.get(
+        "/admin/api/stats/overview",
+        params={"date_from": "2038-01-16", "date_to": "2038-01-16"},
+        headers=_auth(admin_token),
+    )
+    assert response.status_code == 200, response.text
+    coupon = response.json()["period"]["coupon"]
+    assert coupon["started"] == 4
+    assert coupon["abandoned"] == 1
+    assert coupon["success_denominator"] == 3
+    assert coupon["all_success"] == 2
+    assert coupon["success_rate"] == pytest.approx(2 / 3, abs=0.0001)
+
+    db = SessionLocal()
+    try:
+        db.add(
+            CouponSession(
+                trace_id="coupon-rate-only-abandoned",
+                device_id="coupon-rate-device-only-abandoned",
+                status="abandoned",
+                app_env="prod",
+                platforms=["meituan-waimai"],
+                started_at=datetime(2038, 1, 17, 8, tzinfo=UTC),
+                started_date=date(2038, 1, 17),
+            )
+        )
+        db.commit()
+    finally:
+        db.close()
+
+    empty_denominator_response = admin_client.get(
+        "/admin/api/stats/overview",
+        params={"date_from": "2038-01-17", "date_to": "2038-01-17"},
+        headers=_auth(admin_token),
+    )
+    assert empty_denominator_response.status_code == 200
+    only_abandoned = empty_denominator_response.json()["period"]["coupon"]
+    assert only_abandoned["success_denominator"] == 0
+    assert only_abandoned["success_rate"] is None
+
+
 def test_user_list_and_detail(admin_client: TestClient, admin_token: str) -> None:
     uid = _seed_user_with_data("13800000002")
     r = admin_client.get("/admin/api/users", headers=_auth(admin_token))

tests/test_coupon_point_score.py

@@ -10,6 +10,7 @@ from app.admin.repositories.coupon_data import (
     _point_scores_by_trace,
     coupon_data_report,
     coupon_point_details,
+    coupon_user_records,
 )
 from app.admin.security import create_admin_token
 from app.db.session import SessionLocal
@@ -38,6 +39,7 @@ def test_point_scores_by_trace() -> None:
         stats = _point_scores_by_trace(db, [trace])[trace]
         assert stats["succeeded"] == 2
         assert stats["tried"] == 3
+        assert stats["events"] == 4
         details = coupon_point_details(db, trace_id=trace)
         assert [item["status"] for item in details] == [
             "success", "already_claimed", "failed", "skipped"
@@ -48,8 +50,8 @@ def test_point_scores_by_trace() -> None:
         db.close()
 
 
-def test_skipped_detail_does_not_create_a_score() -> None:
-    """仅有 skipped 时按需明细仍可查到，但列表没有虚假的 0/0 分数。"""
+def test_skipped_detail_is_distinguished_from_no_events() -> None:
+    """仅有 skipped 时分数仍为0/0，但保留事件数供前端开放明细。"""
     db = SessionLocal()
     trace = "point-score-skipped"
     try:
@@ -63,7 +65,7 @@ def test_skipped_detail_does_not_create_a_score() -> None:
         db.flush()
 
         scores = _point_scores_by_trace(db, [trace, "missing-trace"])
-        assert trace not in scores
+        assert scores[trace] == {"succeeded": 0, "tried": 0, "events": 1}
         assert "missing-trace" not in scores
         assert coupon_point_details(db, trace_id=trace)[0]["status"] == "skipped"
     finally:
@@ -114,6 +116,112 @@ def test_coupon_data_report_returns_scores_without_embedding_details() -> None:
         db.close()
 
 
+def test_coupon_data_report_marks_abandoned_without_point_results() -> None:
+    """中途退出且没有逐券终态时返回0/0，其他状态缺埋点仍保持为空。"""
+    db = SessionLocal()
+    report_date = date(2020, 1, 6)
+    user_id = 910006
+    try:
+        db.add_all([
+            CouponSession(
+                trace_id="point-score-abandoned-without-result",
+                device_id="score-abandoned-device",
+                user_id=user_id,
+                status="abandoned",
+                app_env="prod",
+                platforms=["meituan-waimai"],
+                started_at=datetime(2020, 1, 6, tzinfo=UTC),
+                started_date=report_date,
+            ),
+            CouponSession(
+                trace_id="point-score-completed-without-result",
+                device_id="score-completed-device",
+                user_id=user_id,
+                status="completed",
+                app_env="prod",
+                platforms=["meituan-waimai"],
+                started_at=datetime(2020, 1, 6, 1, tzinfo=UTC),
+                started_date=report_date,
+            ),
+            CouponSession(
+                trace_id="point-score-abandoned-with-result",
+                device_id="score-abandoned-result-device",
+                user_id=user_id,
+                status="abandoned",
+                app_env="prod",
+                platforms=["meituan-waimai"],
+                started_at=datetime(2020, 1, 6, 2, tzinfo=UTC),
+                started_date=report_date,
+            ),
+            CouponSession(
+                trace_id="point-score-abandoned-skipped-only",
+                device_id="score-abandoned-skipped-device",
+                user_id=user_id,
+                status="abandoned",
+                app_env="prod",
+                platforms=["meituan-waimai"],
+                started_at=datetime(2020, 1, 6, 3, tzinfo=UTC),
+                started_date=report_date,
+            ),
+        ])
+        db.add_all([
+            CouponClaimEvent(
+                trace_id="point-score-abandoned-with-result",
+                device_id="score-abandoned-result-device",
+                coupon_id=f"mt-abandoned-{status}",
+                claim_date=report_date,
+                status=status,
+            )
+            for status in ("success", "failed")
+        ])
+        db.add(CouponClaimEvent(
+            trace_id="point-score-abandoned-skipped-only",
+            device_id="score-abandoned-skipped-device",
+            coupon_id="mt-abandoned-skipped",
+            claim_date=report_date,
+            status="skipped",
+        ))
+        db.flush()
+
+        report = coupon_data_report(
+            db,
+            date_from=report_date.isoformat(),
+            date_to=report_date.isoformat(),
+            app_env="prod",
+        )
+        rows = {item["trace_id"]: item for item in report["items"]}
+        abandoned = rows["point-score-abandoned-without-result"]
+        assert abandoned["point_success_count"] == 0
+        assert abandoned["point_total_count"] == 0
+        assert abandoned["point_event_count"] == 0
+
+        abandoned_with_result = rows["point-score-abandoned-with-result"]
+        assert abandoned_with_result["point_success_count"] == 1
+        assert abandoned_with_result["point_total_count"] == 2
+        assert abandoned_with_result["point_event_count"] == 2
+
+        abandoned_skipped = rows["point-score-abandoned-skipped-only"]
+        assert abandoned_skipped["point_success_count"] == 0
+        assert abandoned_skipped["point_total_count"] == 0
+        assert abandoned_skipped["point_event_count"] == 1
+
+        completed = rows["point-score-completed-without-result"]
+        assert completed["point_success_count"] is None
+        assert completed["point_total_count"] is None
+
+        user_rows = {
+            item["trace_id"]: item
+            for item in coupon_user_records(db, user_id=user_id)["items"]
+        }
+        assert user_rows["point-score-abandoned-without-result"]["point_total_count"] == 0
+        assert user_rows["point-score-abandoned-with-result"]["point_total_count"] == 2
+        assert user_rows["point-score-abandoned-skipped-only"]["point_event_count"] == 1
+        assert user_rows["point-score-completed-without-result"]["point_total_count"] is None
+    finally:
+        db.rollback()
+        db.close()
+
+
 def test_coupon_point_details_endpoint() -> None:
     """前端点击使用的接口按约定返回 trace_id 和逐券 items。"""
     db = SessionLocal()