fix(admin): 按任务白名单聚合常规金币

alembic/versions/monitoring_audit_rbac.py

@@ -1,56 +0,0 @@
-"""补齐监控审计页面权限。
-
-Revision ID: monitoring_audit_rbac
-Revises: merge_signin_boost_main
-Create Date: 2026-07-22 00:00:00.000000
-"""
-from collections.abc import Sequence
-
-import sqlalchemy as sa
-from sqlalchemy.dialects import postgresql
-
-from alembic import op
-
-revision: str = "monitoring_audit_rbac"
-down_revision: str | Sequence[str] | None = "merge_signin_boost_main"
-branch_labels: str | Sequence[str] | None = None
-depends_on: str | Sequence[str] | None = None
-
-_JSON = sa.JSON().with_variant(postgresql.JSONB(), "postgresql")
-_PAGE = "analytics-health"
-
-
-def _role_table() -> sa.TableClause:
-    return sa.table(
-        "admin_role",
-        sa.column("name", sa.String),
-        sa.column("pages", _JSON),
-    )
-
-
-def upgrade() -> None:
-    role = _role_table()
-    conn = op.get_bind()
-    pages = conn.execute(
-        sa.select(role.c.pages).where(role.c.name == "tech")
-    ).scalar_one_or_none()
-    if pages is not None and _PAGE not in pages:
-        conn.execute(
-            role.update()
-            .where(role.c.name == "tech")
-            .values(pages=[*pages, _PAGE])
-        )
-
-
-def downgrade() -> None:
-    role = _role_table()
-    conn = op.get_bind()
-    pages = conn.execute(
-        sa.select(role.c.pages).where(role.c.name == "tech")
-    ).scalar_one_or_none()
-    if pages is not None and _PAGE in pages:
-        conn.execute(
-            role.update()
-            .where(role.c.name == "tech")
-            .values(pages=[page for page in pages if page != _PAGE])
-        )

app/admin/deps.py

@@ -10,8 +10,6 @@ from fastapi import Depends, HTTPException, Request, status
 from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
 from sqlalchemy.orm import Session
 
-from app.admin.permissions import ALL_PAGE_KEYS, CUSTOM_ROLE, SUPER_ADMIN_ROLE, sanitize_pages
-from app.admin.repositories import admin_role as role_repo
 from app.admin.repositories import admin_user as admin_repo
 from app.admin.security import AdminTokenError, decode_admin_token
 from app.db.session import get_db
@@ -74,33 +72,6 @@ def require_role(*roles: str):
     return _checker
 
 
-def require_page(page: str):
-    """页面权限守卫依赖工厂。
-
-    左侧导航隐藏只是 UI，这个守卫确保直接调用 API 也必须持有对应页面权限。
-    super_admin 恒通过；custom 读个人 pages_override；其余角色读 admin_role.pages。
-    """
-    if page not in ALL_PAGE_KEYS:
-        raise ValueError(f"unknown admin page permission: {page}")
-
-    def _checker(admin: CurrentAdmin, db: AdminDb) -> AdminUser:
-        if admin.role == SUPER_ADMIN_ROLE:
-            return admin
-        pages = (
-            sanitize_pages(admin.pages_override)
-            if admin.role == CUSTOM_ROLE
-            else role_repo.effective_pages_of(db, admin.role)
-        )
-        if page not in pages:
-            raise HTTPException(
-                status_code=status.HTTP_403_FORBIDDEN,
-                detail=f"page '{page}' not allowed",
-            )
-        return admin
-
-    return _checker
-
-
 def get_client_ip(request: Request) -> str:
     """取客户端 IP(审计日志用)。生产经 nginx 反代,优先 X-Forwarded-For 第一段;否则直连 IP。
 

app/admin/permissions.py

@@ -21,6 +21,7 @@ PERMISSION_CATALOG: list[dict] = [
         {"key": "ad-revenue-report", "label": "广告收益"},
         {"key": "comparison-records", "label": "比价记录"},
         {"key": "cps", "label": "CPS收益"},
+        {"key": "device-liveness", "label": "设备存活"},
     ]},
     {"group": "奖励审核", "pages": [
         {"key": "withdraws", "label": "提现审核"},
@@ -33,14 +34,10 @@ PERMISSION_CATALOG: list[dict] = [
         {"key": "huawei-review", "label": "华为审核开关"},
         {"key": "users", "label": "用户管理"},
     ]},
-    {"group": "监控审计", "pages": [
-        {"key": "device-liveness", "label": "设备存活"},
-        {"key": "analytics-health", "label": "埋点成功率"},
-        {"key": "event-logs", "label": "埋点日志"},
-        {"key": "audit-logs", "label": "审计日志"},
-    ]},
     {"group": "其他", "pages": [
         {"key": "admins", "label": "权限管理"},
+        {"key": "event-logs", "label": "埋点日志"},
+        {"key": "audit-logs", "label": "审计日志"},
     ]},
 ]
 
@@ -61,7 +58,7 @@ BUILTIN_ROLES: list[dict] = [
         "dashboard", "ad-revenue-report", "cps", "withdraws",
     ]},
     {"name": "tech", "label": "技术", "pages": [
-        "dashboard", "device-liveness", "analytics-health", "config", "ad-revenue", "huawei-review",
+        "dashboard", "device-liveness", "config", "ad-revenue", "huawei-review",
         "event-logs", "audit-logs",
     ]},
 ]

app/admin/repositories/coupon_data.py

@@ -21,9 +21,6 @@ from app.models.user import User
 from app.repositories import ad_ecpm as crud_ecpm
 from app.repositories.coupon_state import DEFAULT_PLATFORMS, coupon_id_to_platform
 
-_SLOT_OK = ("success", "already_claimed")
-_SLOT_TRIED = ("success", "already_claimed", "failed")
-
 
 def _cn_hour(dt: datetime) -> int:
     """started_at(UTC 口径)→ 北京时间小时(0–23)。naive 当 UTC(sqlite),tz-aware 直接换算(pg)。"""
@@ -89,13 +86,7 @@ def _success_rates(rows: list) -> dict:
     }
 
 
-def _session_to_row(
+def _session_to_row(r, phone: str | None = None, nickname: str | None = None, ad_revenue_yuan: float = 0.0) -> dict:
-    r,
-    phone: str | None = None,
-    nickname: str | None = None,
-    ad_revenue_yuan: float = 0.0,
-    point_stats: dict | None = None,
-) -> dict:
     """CouponSession ORM → 明细行 dict(主表「领券数据」与「用户全部领券」抽屉共用)。"""
     return {
         "id": r.id,
@@ -113,60 +104,11 @@ 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,
         "trace_url": r.trace_url,
         "ad_revenue_yuan": ad_revenue_yuan,
     }
 
 
-def _point_scores_by_trace(db: Session, trace_ids: list[str]) -> dict[str, dict[str, int]]:
-    """聚合查询批量返回逐场点位分数，不加载逐券明细。"""
-    if not trace_ids:
-        return {}
-    succeeded = func.sum(case((CouponClaimRecord.status.in_(_SLOT_OK), 1), else_=0))
-    rows = db.execute(
-        select(
-            CouponClaimRecord.trace_id,
-            succeeded.label("succeeded"),
-            func.count().label("tried"),
-        )
-        .where(
-            CouponClaimRecord.trace_id.in_(trace_ids),
-            CouponClaimRecord.status.in_(_SLOT_TRIED),
-        )
-        .group_by(CouponClaimRecord.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
-        if trace_id is not None
-    }
-
-
-def coupon_point_details(db: Session, *, trace_id: str) -> list[dict]:
-    """按单个 trace 查询逐券结果；仅在后台用户点击分数时调用。"""
-    rows = db.execute(
-        select(
-            CouponClaimRecord.coupon_id,
-            CouponClaimRecord.coupon_name,
-            CouponClaimRecord.status,
-            CouponClaimRecord.reason,
-        )
-        .where(CouponClaimRecord.trace_id == trace_id)
-        .order_by(CouponClaimRecord.id)
-    ).all()
-    return [
-        {
-            "coupon_id": coupon_id,
-            "coupon_name": coupon_name,
-            "status": status,
-            "reason": reason,
-        }
-        for coupon_id, coupon_name, status, reason in rows
-    ]
-
-
 def _empty_result() -> dict:
     return {
         "summary": {
@@ -307,17 +249,10 @@ def coupon_data_report(
             ).all()
         }
     rev_map = crud_ecpm.revenue_yuan_by_trace(db, [r.trace_id for r in page])
-    point_stats_map = _point_scores_by_trace(db, [r.trace_id for r in page])
     items = []
     for r in page:
         phone, nickname = user_map.get(r.user_id, (None, None)) if r.user_id is not None else (None, None)
-        items.append(_session_to_row(
+        items.append(_session_to_row(r, phone, nickname, ad_revenue_yuan=rev_map.get(r.trace_id, 0.0)))
-            r,
-            phone,
-            nickname,
-            ad_revenue_yuan=rev_map.get(r.trace_id, 0.0),
-            point_stats=point_stats_map.get(r.trace_id),
-        ))
 
     return {
         "summary": summary,
@@ -341,17 +276,15 @@ def coupon_user_records(db: Session, *, user_id: int, limit: int = 100) -> dict:
     ).scalar_one()
     rev_map = crud_ecpm.revenue_yuan_by_trace(db, [r.trace_id for r in rows])
     return {
-        "items": [
+        "items": [_session_to_row(r, ad_revenue_yuan=rev_map.get(r.trace_id, 0.0)) for r in rows],
-            _session_to_row(
-                r,
-                ad_revenue_yuan=rev_map.get(r.trace_id, 0.0),
-            )
-            for r in rows
-        ],
         "total": int(total),
     }
 
 
+_SLOT_OK = ("success", "already_claimed")
+_SLOT_TRIED = ("success", "already_claimed", "failed")
+
+
 def coupon_slot_report(
     db: Session, *, date_from: str, date_to: str, app_env: str | None = None
 ) -> dict:

app/admin/routers/analytics_health.py

@@ -6,7 +6,7 @@ from typing import Annotated
 
 from fastapi import APIRouter, Depends, Query
 
-from app.admin.deps import AdminDb, require_page
+from app.admin.deps import AdminDb, get_current_admin
 from app.admin.repositories import analytics_health as repo
 from app.admin.schemas.analytics_health import (
     HealthBreakdownRow,
@@ -17,7 +17,7 @@ from app.admin.schemas.analytics_health import (
 router = APIRouter(
     prefix="/admin/api/analytics-health",
     tags=["admin-analytics-health"],
-    dependencies=[Depends(require_page("analytics-health"))],
+    dependencies=[Depends(get_current_admin)],
 )
 
 

app/admin/routers/audit.py

@@ -1,11 +1,11 @@
-"""admin 操作审计日志查询(需要 audit-logs 页面权限)。"""
+"""admin 操作审计日志查询(所有 admin 可看:谁在何时对什么做了什么)。"""
 from __future__ import annotations
 
 from typing import Annotated
 
 from fastapi import APIRouter, Depends, Query
 
-from app.admin.deps import AdminDb, require_page
+from app.admin.deps import AdminDb, get_current_admin
 from app.admin.repositories import audit_log as audit_repo
 from app.admin.schemas.admin import AdminAuditLogOut
 from app.admin.schemas.common import CursorPage
@@ -13,7 +13,7 @@ from app.admin.schemas.common import CursorPage
 router = APIRouter(
     prefix="/admin/api/audit-logs",
     tags=["admin-audit"],
-    dependencies=[Depends(require_page("audit-logs"))],
+    dependencies=[Depends(get_current_admin)],
 )
 
 

app/admin/routers/coupon_data.py

@@ -18,8 +18,6 @@ from app.admin.schemas.coupon_data import (
     CouponDataOut,
     CouponDataRow,
     CouponDataSummary,
-    CouponPointDetail,
-    CouponPointDetailsOut,
     CouponSlotRow,
     CouponSlotsOut,
     CouponUserRecordsOut,
@@ -124,22 +122,6 @@ def get_coupon_slots(
     )
 
 
-@router.get(
-    "/point-details",
-    response_model=CouponPointDetailsOut,
-    summary="按 trace 查询单次领券任务的逐券点位明细",
-)
-def get_coupon_point_details(
-    db: AdminDb,
-    trace_id: Annotated[str, Query(min_length=1, max_length=64, description="领券 trace_id")],
-) -> CouponPointDetailsOut:
-    items = coupon_data.coupon_point_details(db, trace_id=trace_id)
-    return CouponPointDetailsOut(
-        trace_id=trace_id,
-        items=[CouponPointDetail(**item) for item in items],
-    )
-
-
 @router.get(
     "/user-records",
     response_model=CouponUserRecordsOut,

app/admin/routers/device_liveness.py

@@ -2,7 +2,7 @@
 
 数据源 device_liveness 表(心跳 last_heartbeat_at + liveness_state + kill_alert_pending,
 见 app/models/device.py)。在线/掉线、掉线时长由 repo 按 HEARTBEAT_TIMEOUT_MINUTES 阈值派生。
-纯读:无写、无审计。需要 device-liveness 页面权限。
+纯读:无写、无审计。任意登录管理员可看(同大盘/设备管理,无角色门)。
 """
 from __future__ import annotations
 
@@ -10,7 +10,7 @@ from typing import Annotated
 
 from fastapi import APIRouter, Depends, Query
 
-from app.admin.deps import AdminDb, require_page
+from app.admin.deps import AdminDb, get_current_admin
 from app.admin.repositories import queries
 from app.admin.schemas.common import CursorPage
 from app.admin.schemas.device import DeviceLivenessItem, DeviceLivenessStats
@@ -18,7 +18,7 @@ from app.admin.schemas.device import DeviceLivenessItem, DeviceLivenessStats
 router = APIRouter(
     prefix="/admin/api/device-liveness",
     tags=["admin-device-liveness"],
-    dependencies=[Depends(require_page("device-liveness"))],
+    dependencies=[Depends(get_current_admin)],
 )
 
 

app/admin/routers/event_logs.py

@@ -6,7 +6,7 @@ from typing import Annotated
 
 from fastapi import APIRouter, Depends, Query
 
-from app.admin.deps import AdminDb, require_page
+from app.admin.deps import AdminDb, get_current_admin
 from app.admin.repositories import queries
 from app.admin.schemas.analytics import AnalyticsEventOut
 from app.admin.schemas.common import CursorPage
@@ -14,7 +14,7 @@ from app.admin.schemas.common import CursorPage
 router = APIRouter(
     prefix="/admin/api/event-logs",
     tags=["admin-event-logs"],
-    dependencies=[Depends(require_page("event-logs"))],
+    dependencies=[Depends(get_current_admin)],
 )
 
 

app/admin/schemas/coupon_data.py

@@ -49,15 +49,6 @@ class CouponDataHourly(BaseModel):
     avg_elapsed_ms: int | None = None
 
 
-class CouponPointDetail(BaseModel):
-    """一次领券任务中的单券点位结果。"""
-
-    coupon_id: str
-    coupon_name: str | None = None
-    status: str = Field(..., description="success / already_claimed / failed / skipped")
-    reason: str | None = None
-
-
 class CouponDataRow(BaseModel):
     """一条领券明细(一次领券任务)。"""
 
@@ -78,12 +69,6 @@ class CouponDataRow(BaseModel):
     app_env: str | None = None
     started_at: datetime = Field(..., description="发起时刻(明细「时间」列)")
     claimed_count: int | None = None
-    point_success_count: int | None = Field(
-        None, description="本次成功券点位数(success+already_claimed);无逐券埋点为空"
-    )
-    point_total_count: int | None = Field(
-        None, description="本次尝试券点位数(success+already_claimed+failed,不含 skipped);无逐券埋点为空"
-    )
     trace_url: str | None = Field(None, description="pricebot 公网 trace 链接(仅 completed 有);admin 渲染可点链接,无则显示可复制 trace_id")
     ad_revenue_yuan: float = Field(
         0.0, description="本次领券看的信息流广告预估收益(元);按 trace_id 聚合 ad_ecpm_record"
@@ -104,13 +89,6 @@ class CouponDataOut(BaseModel):
     items: list[CouponDataRow] = Field(..., description="逐条领券明细(当前页)")
 
 
-class CouponPointDetailsOut(BaseModel):
-    """单次领券任务的逐券点位结果，供点击分数时按需加载。"""
-
-    trace_id: str
-    items: list[CouponPointDetail] = Field(default_factory=list)
-
-
 class CouponUserRecordsOut(BaseModel):
     """某用户全部领券记录(点手机号抽屉用):total=该用户领券总次数,items=记录列表(UserRecordsDrawer 渲染)。"""
 

docs/api/admin/admin-audit-logs.md

@@ -1,6 +1,6 @@
 # GET /admin/api/audit-logs — 审计日志(谁改了什么,游标分页)
 
-> 所属:Admin·Audit 组（前缀 `/admin/api/audit-logs`） | 鉴权:Bearer admin_token + `audit-logs` 页面权限 | [← 返回 API 索引](../README.md)
+> 所属:Admin·Audit 组（前缀 `/admin/api/audit-logs`） | 鉴权:Bearer admin_token（角色:任意已登录 admin） | [← 返回 API 索引](../README.md)
 
 ## 入参（query）
 | 字段 | 类型 | 必填 | 默认 | 说明 |
@@ -29,8 +29,7 @@
 
 ## 错误码
 - `401` 未带 admin token / token 无效或过期 / 管理员被禁用
-- `403` 当前管理员没有 `audit-logs` 页面权限
 
 ## 说明
-- 整组(`/admin/api/audit-logs`)守卫为 `require_page("audit-logs")`，默认仅超级管理员和技术角色可查看，也可由超管给自定义角色授权。
+- 整组(`/admin/api/audit-logs`)守卫为 `get_current_admin`,任意已登录 admin 均可查看,无角色限制。
 - 审计日志只增不改不删,任何写操作经 `write_audit` 落一条。数据表见 [admin_audit_log](../database/admin_audit_log.md)。

docs/api/admin/admin-device-liveness.md

@@ -1,6 +1,6 @@
 # /admin/api/device-liveness — 设备存活监控(#80)
 
-> 所属:Admin 子应用(前缀 `/admin/api`) | 鉴权:admin + `device-liveness` 页面权限 | 表 [device_liveness](../../database/device_liveness.md) | [← 返回 API 索引](../README.md)
+> 所属:Admin 子应用(前缀 `/admin/api`) | 鉴权:admin | 表 [device_liveness](../../database/device_liveness.md) | [← 返回 API 索引](../README.md)
 
 无障碍保护存活的后台视角:哪些设备开过保护(`ever_protected`)、现在在线还是掉线(心跳超时,#107 起阈值 1 小时)、首次开启时间(`first_protected_at`)。
 
@@ -13,4 +13,3 @@
 
 ## 说明
 - 「在线」= `last_heartbeat_at` 距今 < 超时阈值;掉线召回链路(worker 置 `kill_alert_pending` → 客户端 pull)见表文档。
-- 无 `device-liveness` 页面权限时返回 `403`。

docs/api/admin/admin-event-logs.md

@@ -1,6 +1,6 @@
 # /admin/api/event-logs — 埋点日志(#83)
 
-> 所属:Admin 子应用(前缀 `/admin/api`) | 鉴权:admin + `event-logs` 页面权限 | 表 [analytics_event](../../database/analytics_event.md) | [← 返回 API 索引](../README.md)
+> 所属:Admin 子应用(前缀 `/admin/api`) | 鉴权:admin | 表 [analytics_event](../../database/analytics_event.md) | [← 返回 API 索引](../README.md)
 
 客户端埋点(`POST /api/v1/analytics/events` 批量上报)的后台检索页。
 
@@ -13,4 +13,3 @@
 ## 说明
 - 纯只读;无聚合报表(要分析导出后自己算)。
 - 时间轴用 `client_ts`(事件真实发生时刻),入库时间受客户端攒批影响。
-- 无 `event-logs` 页面权限时返回 `403`。

tests/test_admin_roles.py

@@ -44,78 +44,17 @@ def operator_token() -> str:
     return _token("r_operator", "operator")
 
 
-@pytest.fixture()
-def tech_token() -> str:
-    return _token("r_tech", "tech")
-
-
 def _auth(t: str) -> dict:
     return {"Authorization": f"Bearer {t}"}
 
 
 def test_super_pages_all_operator_limited(admin_client, super_token, operator_token) -> None:
     su = admin_client.get("/admin/api/auth/me", headers=_auth(super_token)).json()
-    assert "admins" in su["pages"] and "analytics-health" in su["pages"]  # 超管全页
+    assert "admins" in su["pages"] and "dashboard" in su["pages"]  # 超管全页
     op = admin_client.get("/admin/api/auth/me", headers=_auth(operator_token)).json()
     assert "dashboard" in op["pages"] and "admins" not in op["pages"]  # 运营看不到管理员页
 
 
-def test_monitoring_audit_catalog_and_api_permissions(
-    admin_client, super_token, operator_token, tech_token
-) -> None:
-    catalog = admin_client.get(
-        "/admin/api/roles/catalog", headers=_auth(super_token)
-    ).json()
-    monitoring = next(group for group in catalog if group["group"] == "监控审计")
-    assert [page["key"] for page in monitoring["pages"]] == [
-        "device-liveness", "analytics-health", "event-logs", "audit-logs",
-    ]
-
-    # 运营默认只能查设备存活，不能绕过导航直调技术/审计接口。
-    assert admin_client.get(
-        "/admin/api/device-liveness/stats", headers=_auth(operator_token)
-    ).status_code == 200
-    for path in (
-        "/admin/api/analytics-health/overview?date_from=2026-07-01T00:00:00Z&date_to=2026-07-02T00:00:00Z",
-        "/admin/api/event-logs",
-        "/admin/api/audit-logs",
-    ):
-        assert admin_client.get(path, headers=_auth(operator_token)).status_code == 403
-
-    # 技术角色默认拥有监控审计组全部四项权限。
-    for path in (
-        "/admin/api/device-liveness/stats",
-        "/admin/api/analytics-health/overview?date_from=2026-07-01T00:00:00Z&date_to=2026-07-02T00:00:00Z",
-        "/admin/api/event-logs",
-        "/admin/api/audit-logs",
-    ):
-        assert admin_client.get(path, headers=_auth(tech_token)).status_code == 200
-
-
-def test_custom_admin_api_permission_uses_pages_override(admin_client) -> None:
-    db = SessionLocal()
-    try:
-        admin = admin_repo.get_by_username(db, "r_monitoring_custom")
-        if admin is None:
-            admin = admin_repo.create_admin(
-                db, username="r_monitoring_custom", password="pass1234", role="custom"
-            )
-        admin.password_hash = hash_password("pass1234")
-        admin.role = "custom"
-        admin.pages_override = ["event-logs"]
-        admin.status = "active"
-        db.commit()
-    finally:
-        db.close()
-
-    token = admin_client.post(
-        "/admin/api/auth/login",
-        json={"username": "r_monitoring_custom", "password": "pass1234"},
-    ).json()["access_token"]
-    assert admin_client.get("/admin/api/event-logs", headers=_auth(token)).status_code == 200
-    assert admin_client.get("/admin/api/audit-logs", headers=_auth(token)).status_code == 403
-
-
 def test_roles_endpoints_super_only(admin_client, super_token, operator_token) -> None:
     assert admin_client.get("/admin/api/roles", headers=_auth(super_token)).status_code == 200
     assert admin_client.get("/admin/api/roles", headers=_auth(operator_token)).status_code == 403
@@ -184,7 +123,7 @@ def test_builtin_roles_labels_and_pages(admin_client, super_token) -> None:
     # 页集对齐 Prototypes/dashboard/permissions.md 的 ROLES
     assert set(roles["finance"]["pages"]) == {"dashboard", "ad-revenue-report", "cps", "withdraws"}
     assert set(roles["tech"]["pages"]) == {
-        "dashboard", "device-liveness", "analytics-health", "config", "ad-revenue", "huawei-review",
+        "dashboard", "device-liveness", "config", "ad-revenue", "huawei-review",
         "event-logs", "audit-logs",
     }
 

tests/test_coupon_point_score.py

@@ -1,161 +0,0 @@
-"""admin 领券明细逐场点位分数与按需明细。"""
-from datetime import UTC, date, datetime
-
-from fastapi.testclient import TestClient
-from sqlalchemy import delete
-
-from app.admin.main import admin_app
-from app.admin.repositories import admin_user as admin_repo
-from app.admin.repositories.coupon_data import (
-    _point_scores_by_trace,
-    coupon_data_report,
-    coupon_point_details,
-)
-from app.admin.security import create_admin_token
-from app.db.session import SessionLocal
-from app.models.coupon_state import CouponClaimRecord, CouponSession
-
-
-def test_point_scores_by_trace() -> None:
-    """已领算成功、失败算尝试、跳过不进分母。"""
-    db = SessionLocal()
-    trace = "point-score-trace"
-    try:
-        db.add_all([
-            CouponClaimRecord(
-                device_id="score-device",
-                coupon_id=f"mt-score-{status}",
-                claim_date=date(2020, 1, 2),
-                status=status,
-                coupon_name=f"测试点位-{status}",
-                reason="测试失败" if status == "failed" else None,
-                trace_id=trace,
-            )
-            for status in ("success", "already_claimed", "failed", "skipped")
-        ])
-        db.flush()
-
-        stats = _point_scores_by_trace(db, [trace])[trace]
-        assert stats["succeeded"] == 2
-        assert stats["tried"] == 3
-        details = coupon_point_details(db, trace_id=trace)
-        assert [item["status"] for item in details] == [
-            "success", "already_claimed", "failed", "skipped"
-        ]
-        assert details[2]["reason"] == "测试失败"
-    finally:
-        db.rollback()
-        db.close()
-
-
-def test_skipped_detail_does_not_create_a_score() -> None:
-    """仅有 skipped 时按需明细仍可查到，但列表没有虚假的 0/0 分数。"""
-    db = SessionLocal()
-    trace = "point-score-skipped"
-    try:
-        db.add(CouponClaimRecord(
-            device_id="score-device-skipped",
-            coupon_id="mt-score-skipped-only",
-            claim_date=date(2020, 1, 2),
-            status="skipped",
-            trace_id=trace,
-        ))
-        db.flush()
-
-        scores = _point_scores_by_trace(db, [trace, "missing-trace"])
-        assert trace not in scores
-        assert "missing-trace" not in scores
-        assert coupon_point_details(db, trace_id=trace)[0]["status"] == "skipped"
-    finally:
-        db.rollback()
-        db.close()
-
-
-def test_coupon_data_report_returns_scores_without_embedding_details() -> None:
-    """主列表只返回聚合分数，逐券记录必须走按 trace 的明细查询。"""
-    db = SessionLocal()
-    trace = "point-score-report"
-    report_date = date(2020, 1, 4)
-    try:
-        db.add(CouponSession(
-            trace_id=trace,
-            device_id="score-report-device",
-            status="completed",
-            app_env="prod",
-            platforms=["meituan-waimai"],
-            started_at=datetime(2020, 1, 4, tzinfo=UTC),
-            started_date=report_date,
-        ))
-        db.add_all([
-            CouponClaimRecord(
-                device_id="score-report-device",
-                coupon_id=f"mt-report-{status}",
-                claim_date=report_date,
-                status=status,
-                trace_id=trace,
-            )
-            for status in ("success", "failed")
-        ])
-        db.flush()
-
-        report = coupon_data_report(
-            db,
-            date_from=report_date.isoformat(),
-            date_to=report_date.isoformat(),
-            app_env="prod",
-        )
-        row = next(item for item in report["items"] if item["trace_id"] == trace)
-        assert row["point_success_count"] == 1
-        assert row["point_total_count"] == 2
-        assert "point_details" not in row
-        assert len(coupon_point_details(db, trace_id=trace)) == 2
-    finally:
-        db.rollback()
-        db.close()
-
-
-def test_coupon_point_details_endpoint() -> None:
-    """前端点击使用的接口按约定返回 trace_id 和逐券 items。"""
-    db = SessionLocal()
-    trace = "point-details-endpoint"
-    try:
-        admin = admin_repo.get_by_username(db, "point_details_admin")
-        if admin is None:
-            admin = admin_repo.create_admin(
-                db,
-                username="point_details_admin",
-                password="pass1234",
-                role="super_admin",
-            )
-        token, _expires_at = create_admin_token(admin_id=admin.id, role=admin.role)
-        db.add(CouponClaimRecord(
-            device_id="point-details-endpoint-device",
-            coupon_id="mt-point-details-endpoint",
-            coupon_name="接口测试券",
-            claim_date=date(2020, 1, 5),
-            status="failed",
-            reason="接口测试失败",
-            trace_id=trace,
-        ))
-        db.commit()
-
-        response = TestClient(admin_app).get(
-            "/admin/api/coupon-data/point-details",
-            params={"trace_id": trace},
-            headers={"Authorization": f"Bearer {token}"},
-        )
-        assert response.status_code == 200, response.text
-        assert response.json() == {
-            "trace_id": trace,
-            "items": [{
-                "coupon_id": "mt-point-details-endpoint",
-                "coupon_name": "接口测试券",
-                "status": "failed",
-                "reason": "接口测试失败",
-            }],
-        }
-    finally:
-        db.rollback()
-        db.execute(delete(CouponClaimRecord).where(CouponClaimRecord.trace_id == trace))
-        db.commit()
-        db.close()