Merge remote-tracking branch 'origin/main' into feat-huaweipermissiontoggle

feat(huawei-review): 华为审核开关(admin 可切 + 客户端下发 + 审计)
华为应用市场审核要求新手引导的「快速设置」权限步必须可被用户关闭,平时又要保住权限开启率,故做成后台可切的两态开关,送审期间切开、过审后收回。 - app_config 新增 huawei_review 行(default / review),空库与脏值一律回退 default = 上线至今的现状,宁可不给退出按钮也不误放开 - admin: GET/PATCH /admin/api/huawei-review,权限 operator/tech,切换写审计 - 客户端: GET /api/v1/platform/huawei-review 不鉴权(引导页在登录前就展示), 下发 onboarding_closable;机型 gate 由客户端做,故此处不判 ROM Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-22 16:39:10 +08:00 · 2026-07-20 20:53:04 +08:00
39 changed files with 102 additions and 2082 deletions
@@ -113,13 +113,6 @@ JD_UNION_APP_SECRET=
 JD_UNION_SITE_ID=
 JD_UNION_AUTH_KEY=

-# 美团 + 京东订单每天北京时间 05:00 自动对账；按更新时间回拉近 3 天，重叠防漏单并刷新状态。
-# 手动对账按钮不受该开关影响。通常保持开启；临时停自动任务时设为 false。
-CPS_AUTO_RECONCILE_ENABLED=true
-CPS_AUTO_RECONCILE_RUN_HOUR=5
-CPS_AUTO_RECONCILE_LOOKBACK_DAYS=3
-CPS_AUTO_RECONCILE_CHECK_INTERVAL_SEC=60
-
 # ===== Pricebot 上游 (领券/比价业务透传目标) =====
 # 客户端调本服务的 /api/v1/coupon/step 等,我们透传到 pricebot-backend。
 # 本地开发用 localhost:8000。生产部署改成内网地址(如 http://pricebot.internal:8000)。
@@ -1,26 +0,0 @@
-"""merge notification/comparison_user_idx/monitoring_audit heads
-
-Revision ID: 8e04cc13a211
-Revises: comparison_user_created_idx, monitoring_audit_rbac, notification_table
-Create Date: 2026-07-23 15:37:26.967540
-
-"""
-from typing import Sequence, Union
-
-from alembic import op
-import sqlalchemy as sa
-
-
-# revision identifiers, used by Alembic.
-revision: str = '8e04cc13a211'
-down_revision: Union[str, Sequence[str], None] = ('comparison_user_created_idx', 'monitoring_audit_rbac', 'notification_table')
-branch_labels: Union[str, Sequence[str], None] = None
-depends_on: Union[str, Sequence[str], None] = None
-
-
-def upgrade() -> None:
-    pass
-
-
-def downgrade() -> None:
-    pass
@@ -1,52 +0,0 @@
-"""add composite index (user_id, created_at, id) on comparison_record
-
-C 端「我的比价记录」列表(GET /api/v1/compare/records)是
-`WHERE user_id=? ORDER BY created_at DESC, id DESC LIMIT n` —— 原来只有单列 user_id 索引,
-过滤完还要把该用户的**全部**记录取出来排序才能拿前 n 条,重度用户随记录数线性变慢。
-
-本复合索引的反向扫恰好等于 (created_at DESC, id DESC),规划器直接取前 n 条、免排序。
-列序 (user_id, created_at, id) 与查询一一对应,不要调整。
-
-Revision ID: comparison_user_created_idx
-Revises: merge_active_phone
-Create Date: 2026-07-21
-"""
-
-from __future__ import annotations
-
-from alembic import op
-
-revision = "comparison_user_created_idx"
-down_revision = "merge_active_phone"
-branch_labels = None
-depends_on = None
-
-INDEX_NAME = "ix_comparison_user_created"
-COLUMNS = ["user_id", "created_at", "id"]
-
-
-def upgrade() -> None:
-    bind = op.get_bind()
-    if bind.dialect.name == "postgresql":
-        # 线上 comparison_record 已有数据量,普通 CREATE INDEX 持表写锁会阻塞比价 harvest 写入;
-        # 用 CONCURRENTLY 不锁表(须脱离事务,autocommit_block 切到自动提交)。
-        # 同 comparison_status_created_idx 的做法。
-        with op.get_context().autocommit_block():
-            op.create_index(
-                INDEX_NAME, "comparison_record", COLUMNS,
-                unique=False, postgresql_concurrently=True,
-            )
-    else:
-        op.create_index(INDEX_NAME, "comparison_record", COLUMNS, unique=False)
-
-
-def downgrade() -> None:
-    bind = op.get_bind()
-    if bind.dialect.name == "postgresql":
-        with op.get_context().autocommit_block():
-            op.drop_index(
-                INDEX_NAME, table_name="comparison_record",
-                postgresql_concurrently=True,
-            )
-    else:
-        op.drop_index(INDEX_NAME, table_name="comparison_record")
@@ -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])
-        )
@@ -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。

@@ -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",
    ]},
 ]
@@ -81,10 +81,6 @@ def _date_range(date_from: str, date_to: str) -> list[str]:
 # ad_feed_reward_record,由 audit 内部按 ad_type 区分(feed 含历史 NULL,draw 仅 ad_type=="draw")。
 _AUDIT_SCENES = {"reward_video", "feed", "draw"}

-# 激励视频未满足有效播放条件时不计客户端预估收益。客户端仍会在 onAdShow
-# 上报 eCPM，随后才在关闭时补报以下终态，因此必须在展示/发奖合并后修正收益。
-_ZERO_REVENUE_REWARD_VIDEO_STATUSES = frozenset({"closed_early", "too_short"})
-

 # 发奖复算明细字段(展开下钻看「金币怎么算出来的」)——从 audit 行原样取这些 key。
 _REWARD_DETAIL_KEYS = (
@@ -206,11 +202,6 @@ def ad_revenue_report(
                "matched": bool(rwd["matched"]),
                "reward_detail": _reward_detail(rwd),
            })
-            if (
-                rec.ad_type == "reward_video"
-                and rwd["status"] in _ZERO_REVENUE_REWARD_VIDEO_STATUSES
-            ):
-                ev["revenue_yuan"] = 0.0
        else:
            # 纯展示(信息流逐条展示、激励视频缺发奖记录):不计对账,matched=True。
            ev.update({
@@ -1,7 +1,7 @@
 """admin「领券数据」看板聚合:发起/完成数、耗时均值与分位、按天/小时趋势、逐条明细。

-数据源 coupon_session(一次领券一行,客户端 /api/v1/coupon/session 两段上报)。生产 PostgreSQL
-使用 percentile_cont 聚合耗时分位；SQLite 本地/测试环境回退读取耗时单列计算。
+数据源 coupon_session(一次领券一行,客户端 /api/v1/coupon/session 两段上报)。量级不大,全量拉
+区间数据后 Python 聚合(分位 SQLite 无 percentile,统一 Python 算,PG 上也一致)。
 - 发起数 = 区间内全部 session(含 started/completed/failed/abandoned),= 流失统计的基数。
 - 完成数 / 耗时均值 / 分位 = 仅 status==completed 子集(成功跑完才有可比的"领券耗时")。
 - summary/daily/hourly/total 在全量上算,不受分页;items 为排序后当前页。
@@ -45,75 +45,6 @@ def _percentile(sorted_vals: list[int], q: float) -> int | None:
    return round(sorted_vals[lo] * (1 - frac) + sorted_vals[hi] * frac)


-def _round_duration_ms(value) -> int | None:
-    """将数据库聚合结果按既有 Python round 口径转为整数毫秒。"""
-    if value is None:
-        return None
-    return int(round(value))
-
-
-def _coupon_summary_aggregate_stmt(conditions: list):
-    """PostgreSQL 汇总卡聚合语句；计数、均值与四个分位一次返回。"""
-    completed = CouponSession.status == "completed"
-    completed_elapsed = completed & CouponSession.elapsed_ms.is_not(None)
-    return select(
-        func.count(CouponSession.id),
-        func.sum(case((completed, 1), else_=0)),
-        func.avg(CouponSession.elapsed_ms).filter(completed_elapsed),
-        *(
-            func.percentile_cont(q)
-            .within_group(CouponSession.elapsed_ms)
-            .filter(completed_elapsed)
-            for q in (0.05, 0.5, 0.95, 0.99)
-        ),
-    ).where(*conditions)
-
-
-def _coupon_summary_aggregates(db: Session, conditions: list) -> dict:
-    """汇总卡基础指标；生产 PG 全部在数据库内完成，SQLite 仅作测试回退。"""
-    if db.bind is not None and db.bind.dialect.name == "postgresql":
-        row = db.execute(_coupon_summary_aggregate_stmt(conditions)).one()
-        return {
-            "started_count": int(row[0] or 0),
-            "completed_count": int(row[1] or 0),
-            "avg_elapsed_ms": _round_duration_ms(row[2]),
-            "p5_ms": _round_duration_ms(row[3]),
-            "p50_ms": _round_duration_ms(row[4]),
-            "p95_ms": _round_duration_ms(row[5]),
-            "p99_ms": _round_duration_ms(row[6]),
-        }
-
-    counts = db.execute(
-        select(
-            func.count(CouponSession.id),
-            func.sum(case((CouponSession.status == "completed", 1), else_=0)),
-        ).where(*conditions)
-    ).one()
-    # SQLite 没有 percentile_cont；本地/测试只回退读取耗时单列，不加载完整记录。
-    completed_elapsed = list(
-        db.execute(
-            select(CouponSession.elapsed_ms)
-            .where(
-                *conditions,
-                CouponSession.status == "completed",
-                CouponSession.elapsed_ms.is_not(None),
-            )
-            .order_by(CouponSession.elapsed_ms)
-        ).scalars()
-    )
-    return {
-        "started_count": int(counts[0] or 0),
-        "completed_count": int(counts[1] or 0),
-        "avg_elapsed_ms": _round_duration_ms(
-            sum(completed_elapsed) / len(completed_elapsed)
-        ) if completed_elapsed else None,
-        "p5_ms": _percentile(completed_elapsed, 5),
-        "p50_ms": _percentile(completed_elapsed, 50),
-        "p95_ms": _percentile(completed_elapsed, 95),
-        "p99_ms": _percentile(completed_elapsed, 99),
-    }
-
-
 def _avg(vals: list[int]) -> int | None:
    return round(sum(vals) / len(vals)) if vals else None

@@ -284,22 +215,30 @@ def coupon_data_report(
        if not user_ids:
            return _empty_result()

-    conditions = [
+    stmt = select(CouponSession).where(
        CouponSession.started_date >= d_from,
        CouponSession.started_date <= d_to,
-    ]
+    )
    if app_env is not None:
-        conditions.append(CouponSession.app_env == app_env)
+        stmt = stmt.where(CouponSession.app_env == app_env)
    if statuses:
-        conditions.append(CouponSession.status.in_(statuses))
+        stmt = stmt.where(CouponSession.status.in_(statuses))
    if user_ids is not None:
-        conditions.append(CouponSession.user_id.in_(user_ids))
-    stmt = select(CouponSession).where(*conditions)
+        stmt = stmt.where(CouponSession.user_id.in_(user_ids))
    rows = list(db.execute(stmt).scalars())

    # ── 汇总卡 ──
+    completed_elapsed = sorted(
+        r.elapsed_ms for r in rows if r.status == "completed" and r.elapsed_ms is not None
+    )
    summary = {
-        **_coupon_summary_aggregates(db, conditions),
+        "started_count": len(rows),
+        "completed_count": sum(1 for r in rows if r.status == "completed"),
+        "avg_elapsed_ms": _avg(completed_elapsed),
+        "p5_ms": _percentile(completed_elapsed, 5),
+        "p50_ms": _percentile(completed_elapsed, 50),
+        "p95_ms": _percentile(completed_elapsed, 95),
+        "p99_ms": _percentile(completed_elapsed, 99),
        **_success_rates(rows),
    }

@@ -384,7 +323,7 @@ def coupon_data_report(
        "summary": summary,
        "daily": daily,
        "hourly": hourly,
-        "total": summary["started_count"],
+        "total": len(rows),
        "items": items,
    }

@@ -5,7 +5,6 @@
 """
 from __future__ import annotations

-import logging
 from datetime import datetime, timedelta, timezone
 from decimal import Decimal, InvalidOperation
 from typing import Any
@@ -16,14 +15,12 @@ from sqlalchemy.orm import Session

 from app.admin.repositories.queries import _as_utc, offset_paginate
 from app.integrations import jd_union, meituan
+from app.repositories import cps_link as cps_link_repo
 from app.models.cps_activity import CpsActivity
 from app.models.cps_group import CpsGroup
 from app.models.cps_link import CpsClick, CpsLink
 from app.models.cps_order import CpsOrder
 from app.models.cps_wx_user import CpsWxUser
-from app.repositories import cps_link as cps_link_repo
-
-logger = logging.getLogger("shagua.cps_reconcile")

 # 美团订单状态:取消(4)/风控(5)不计佣金;结算(6)为佣金真正到账
 _INVALID_STATUS = {"4", "5"}
@@ -381,35 +378,18 @@ def effective_commission_cents(order: CpsOrder) -> int:
 def reconcile_orders(
    db: Session, *, start_time: int, end_time: int,
    query_time_type: int = 1, sid: str | None = None, max_pages: int = 200,
-    audit_context: dict[str, Any] | None = None,
 ) -> dict:
    """调美团 query_order 分页拉单 → 按 order_id upsert。返回 {fetched, inserted, updated, pages}。

    订单状态会随时间变(付款→完成→结算/退款),重复拉同一单则更新。max_pages 防异常死循环。
    """
-    fetched = inserted = updated = pages = api_requests = 0
+    fetched = inserted = updated = pages = 0
    page = 1
    while page <= max_pages:
-        api_requests += 1
-        try:
-            resp = meituan.query_order(
-                sid=sid, start_time=start_time, end_time=end_time,
-                query_time_type=query_time_type, page=page, limit=100,
-            )
-        except Exception:  # noqa: BLE001 - 记录失败页后保持原异常类型继续抛出
-            if audit_context is not None:
-                logger.exception(
-                    "CPS reconcile upstream request failed platform=meituan page=%s",
-                    page,
-                    extra={
-                        **audit_context,
-                        "event": "cps_reconcile.request_failed",
-                        "platform": "meituan",
-                        "failed_page": page,
-                        "api_request_number": api_requests,
-                    },
-                )
-            raise
+        resp = meituan.query_order(
+            sid=sid, start_time=start_time, end_time=end_time,
+            query_time_type=query_time_type, page=page, limit=100,
+        )
        rows = ((resp.get("data") or {}).get("dataList")) or []
        if not rows:
            break
@@ -439,58 +419,30 @@ def reconcile_orders(
            break
        page += 1
    db.commit()
-    return {
-        "fetched": fetched,
-        "inserted": inserted,
-        "updated": updated,
-        "pages": pages,
-        "api_requests": api_requests,
-    }
+    return {"fetched": fetched, "inserted": inserted, "updated": updated, "pages": pages}


 def reconcile_jd_orders(
    db: Session, *, start_time: datetime, end_time: datetime,
    query_time_type: int = 3, max_pages: int = 100,
-    audit_context: dict[str, Any] | None = None,
 ) -> dict:
    """调京东 order.row.query 拉单 → 按订单行 upsert。

    京东单次查询窗口最多 1 小时,这里按北京自然时间切窗并逐页拉取。
    """
-    fetched = inserted = updated = pages = api_requests = windows = 0
+    fetched = inserted = updated = pages = 0
    cur = start_time
    while cur < end_time:
        win_end = min(cur + timedelta(hours=1), end_time)
-        windows += 1
        page = 1
        while page <= max_pages:
-            api_requests += 1
-            try:
-                resp = jd_union.query_order_rows(
-                    start_time=cur,
-                    end_time=win_end,
-                    query_time_type=query_time_type,
-                    page_index=page,
-                    page_size=200,
-                )
-            except Exception:  # noqa: BLE001 - 记录失败窗口后保持原异常类型继续抛出
-                if audit_context is not None:
-                    logger.exception(
-                        "CPS reconcile upstream request failed platform=jd window=%s..%s page=%s",
-                        cur.isoformat(),
-                        win_end.isoformat(),
-                        page,
-                        extra={
-                            **audit_context,
-                            "event": "cps_reconcile.request_failed",
-                            "platform": "jd",
-                            "failed_window_start": cur.isoformat(),
-                            "failed_window_end": win_end.isoformat(),
-                            "failed_page": page,
-                            "api_request_number": api_requests,
-                        },
-                    )
-                raise
+            resp = jd_union.query_order_rows(
+                start_time=cur,
+                end_time=win_end,
+                query_time_type=query_time_type,
+                page_index=page,
+                page_size=200,
+            )
            rows = resp.get("rows") or []
            has_more = bool(resp.get("has_more"))
            if not rows:
@@ -517,14 +469,7 @@ def reconcile_jd_orders(
            page += 1
        cur = win_end
    db.commit()
-    return {
-        "fetched": fetched,
-        "inserted": inserted,
-        "updated": updated,
-        "pages": pages,
-        "api_requests": api_requests,
-        "windows": windows,
-    }
+    return {"fetched": fetched, "inserted": inserted, "updated": updated, "pages": pages}


 def list_orders(
@@ -6,7 +6,6 @@
 from __future__ import annotations

 from datetime import date, datetime, time, timedelta, timezone
-from decimal import ROUND_HALF_UP, Decimal
 from zoneinfo import ZoneInfo

 from sqlalchemy import Select, asc, case, desc, func, or_, select
@@ -298,58 +297,6 @@ def _comparison_percentile(sorted_values: list[int], q: float) -> int | None:
    return int(value + 0.5)


-def _round_duration_ms(value) -> int | None:
-    """将数据库聚合结果按既有口径四舍五入为整数毫秒。"""
-    if value is None:
-        return None
-    return int(Decimal(str(value)).quantize(Decimal("1"), rounding=ROUND_HALF_UP))
-
-
-def _comparison_duration_aggregate_stmt(conditions: list, status: str, quantiles: tuple[float, ...]):
-    """PostgreSQL 耗时聚合语句；每种状态只返回一行。"""
-    return select(
-        func.avg(ComparisonRecord.total_ms),
-        *(
-            func.percentile_cont(q).within_group(ComparisonRecord.total_ms)
-            for q in quantiles
-        ),
-    ).where(
-        *conditions,
-        ComparisonRecord.status == status,
-        ComparisonRecord.total_ms.is_not(None),
-    )
-
-
-def _comparison_duration_aggregates(
-    db: Session,
-    *,
-    conditions: list,
-    status: str,
-    quantiles: tuple[float, ...],
-) -> list[int | None]:
-    """返回平均值和各分位数；生产 PG 在数据库内聚合，SQLite 仅作测试回退。"""
-    if db.bind is not None and db.bind.dialect.name == "postgresql":
-        row = db.execute(
-            _comparison_duration_aggregate_stmt(conditions, status, quantiles)
-        ).one()
-        return [_round_duration_ms(value) for value in row]
-
-    # SQLite 没有 percentile_cont；本地/测试只回退读取耗时单列，不加载完整记录。
-    values = list(
-        db.execute(
-            select(ComparisonRecord.total_ms)
-            .where(
-                *conditions,
-                ComparisonRecord.status == status,
-                ComparisonRecord.total_ms.is_not(None),
-            )
-            .order_by(ComparisonRecord.total_ms)
-        ).scalars()
-    )
-    average = _round_duration_ms(sum(values) / len(values)) if values else None
-    return [average, *(_comparison_percentile(values, q) for q in quantiles)]
-
-
 def comparison_records_summary(
    db: Session,
    *,
@@ -385,18 +332,20 @@ def comparison_records_summary(
    success = int(row[2] or 0)
    lower_price = int(row[4] or 0)
    cancelled = int(row[5] or 0)
-    success_duration_stats = _comparison_duration_aggregates(
-        db,
-        conditions=conditions,
-        status="success",
-        quantiles=(0.05, 0.5, 0.95, 0.99),
-    )
-    cancelled_duration_stats = _comparison_duration_aggregates(
-        db,
-        conditions=conditions,
-        status="cancelled",
-        quantiles=(0.05, 0.5, 0.95),
-    )
+    success_durations = sorted(db.execute(
+        select(ComparisonRecord.total_ms).where(
+            *conditions,
+            ComparisonRecord.status == "success",
+            ComparisonRecord.total_ms.is_not(None),
+        )
+    ).scalars().all())
+    cancelled_durations = sorted(db.execute(
+        select(ComparisonRecord.total_ms).where(
+            *conditions,
+            ComparisonRecord.status == "cancelled",
+            ComparisonRecord.total_ms.is_not(None),
+        )
+    ).scalars().all())
    success_rate_denominator = started - cancelled
    return {
        "started": started,
@@ -405,16 +354,19 @@ def comparison_records_summary(
        "success_rate": success / success_rate_denominator if success_rate_denominator else None,
        "avg_token_cost": float(row[3]) if row[3] is not None else None,
        "lower_price_rate": lower_price / success if success else None,
-        "avg_duration_ms": success_duration_stats[0],
-        "p5_duration_ms": success_duration_stats[1],
-        "p50_duration_ms": success_duration_stats[2],
-        "p95_duration_ms": success_duration_stats[3],
-        "p99_duration_ms": success_duration_stats[4],
+        "avg_duration_ms": (
+            int(sum(success_durations) / len(success_durations) + 0.5)
+            if success_durations else None
+        ),
+        "p5_duration_ms": _comparison_percentile(success_durations, 0.05),
+        "p50_duration_ms": _comparison_percentile(success_durations, 0.5),
+        "p95_duration_ms": _comparison_percentile(success_durations, 0.95),
+        "p99_duration_ms": _comparison_percentile(success_durations, 0.99),
        "cancelled": cancelled,
        "cancelled_rate": cancelled / started if started else None,
-        "cancelled_p5_ms": cancelled_duration_stats[1],
-        "cancelled_p50_ms": cancelled_duration_stats[2],
-        "cancelled_p95_ms": cancelled_duration_stats[3],
+        "cancelled_p5_ms": _comparison_percentile(cancelled_durations, 0.05),
+        "cancelled_p50_ms": _comparison_percentile(cancelled_durations, 0.5),
+        "cancelled_p95_ms": _comparison_percentile(cancelled_durations, 0.95),
    }


@@ -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)],
 )


@@ -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)],
 )


@@ -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)],
 )


@@ -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)],
 )


@@ -26,10 +26,7 @@ class AdRevenueRecord(BaseModel):

    record_id: int
    created_at: datetime
-    status: str = Field(
-        ...,
-        description="granted / capped / ecpm_missing / closed_early / too_short",
-    )
+    status: str = Field(..., description="granted / capped / ecpm_missing")
    ecpm: str | None = Field(None, description="本次采用的 eCPM 原始值(分/千次展示)")
    ecpm_factor: float | None = Field(None, description="因子1(eCPM 档);非 granted 为空")
    units: int = Field(..., description="折算份数:激励视频恒 1;信息流 = 满 10 秒份数")
@@ -47,7 +44,7 @@ class AdRevenueDaily(BaseModel):

    date: str = Field(..., description="北京时间 YYYY-MM-DD")
    impressions: int = Field(..., description="当天展示条数合计")
-    revenue_yuan: float = Field(..., description="当天客户端有效预估收益合计(元;eCPM 折算)")
+    revenue_yuan: float = Field(..., description="当天客户端预估收益合计(元;eCPM 折算)")
    pangle_revenue_yuan: float | None = Field(
        None, description="当天穿山甲后台预估收益(元;GroMore revenue);非全量视图/无数据为空"
    )
@@ -96,10 +93,7 @@ class AdRevenueRow(BaseModel):
    has_impression: bool = Field(..., description="是否有广告展示(信息流逐条展示=True,纯发奖行=False)")
    impressions: int = Field(..., description="本行展示条数:有展示=1 / 纯发奖=0(供日汇总、趋势图复用)")
    ecpm: str | None = Field(None, description="eCPM 原始值(分/千次);展示行取展示值,纯发奖行取发奖采用值")
-    revenue_yuan: float = Field(
-        ...,
-        description="本次有效展示预估收益(元)= eCPM元 ÷ 1000;纯发奖、激励视频提前关闭/时长不足=0",
-    )
+    revenue_yuan: float = Field(..., description="本次展示预估收益(元)= eCPM元 ÷ 1000;纯发奖行=0")
    row_revenue_yuan: float | None = Field(
        None,
        description="主表逐行展示用的预估收益(元):一次比价/领券聚合行=该次发奖广告 eCPM 折算之和;"
@@ -156,7 +150,7 @@ class AdRevenueReportOut(BaseModel):
    total: int = Field(..., description="广告事件总数(全量,不受分页影响;= 当前筛选下的分页总条数)")
    truncated: bool = Field(..., description="当前页之后是否还有更多事件(len(events) > offset + limit)")
    total_impressions: int = Field(..., description="全量展示条数合计")
-    total_revenue_yuan: float = Field(..., description="全量客户端有效预估收益合计(元;eCPM 折算)")
+    total_revenue_yuan: float = Field(..., description="全量客户端预估收益合计(元;eCPM 折算)")
    total_pangle_revenue_yuan: float | None = Field(
        None,
        description="全量穿山甲后台预估收益合计(元;GroMore revenue)。穿山甲无用户/类型/场景维度,"
@@ -115,22 +115,13 @@ def list_records(
    db: DbSession,
    limit: int = Query(20, ge=1, le=100),
    cursor: int | None = Query(None, description="上一页末条 id"),
-    ordered: bool | None = Query(
-        None,
-        description="true=只看「已下单」(店名命中本人真实下单)的记录;不传=全部",
-    ),
-    keyword: str | None = Query(
-        None,
-        max_length=64,
-        description="按店名 / 菜名模糊搜索,忽略大小写;空白串等同不传",
-    ),
    include_trace: bool = Query(
        False,
        description="客户端开了本机 agent 调试模式时带 true,放行本人记录的 trace_url",
    ),
 ) -> ComparisonRecordPage:
    items, next_cursor = crud_compare.list_records(
-        db, user.id, limit=limit, cursor=cursor, ordered=ordered, keyword=keyword
+        db, user.id, limit=limit, cursor=cursor
    )
    outs = [ComparisonRecordOut.model_validate(it) for it in items]
    # 权限闸:未开 debug_trace_enabled 的用户不下发 trace_url(列表页「复制调试链接」靠它)。
@@ -192,7 +192,7 @@ def withdraw_info(
        wechat_bound=bool(u and u.wechat_openid),
        wechat_nickname=u.wechat_nickname if u else None,
        wechat_avatar_url=u.wechat_avatar_url if u else None,
-        transfer_auth_enabled=bool(auth and auth.state == "active" and auth.authorization_id),
+        transfer_auth_enabled=bool(auth and auth.state == "active"),
        tiers=[WithdrawTierOut(**t) for t in crud_wallet.withdraw_tier_states(db, user.id, source)],
    )

@@ -335,8 +335,7 @@ def open_transfer_auth(user: CurrentUser, db: DbSession) -> TransferAuthResultOu
 def transfer_auth_status(user: CurrentUser, db: DbSession) -> TransferAuthStatusOut:
    auth = crud_wallet.sync_transfer_auth(db, user.id)
    state = auth.state if auth else "none"
-    enabled = bool(auth and auth.state == "active" and auth.authorization_id)
-    return TransferAuthStatusOut(state=state, enabled=enabled)
+    return TransferAuthStatusOut(state=state, enabled=(state == "active"))


@router.post(
@@ -188,13 +188,6 @@ class Settings(BaseSettings):
        """京东联盟订单查询凭证齐全。"""
        return bool(self.JD_UNION_APP_KEY and self.JD_UNION_APP_SECRET)

-    # 美团 + 京东 CPS 订单自动对账：进程内 worker 每天北京时间 05:00 后跑一轮。
-    # 按更新时间回拉近 N 天（重叠窗口防漏单并刷新状态），order_id 幂等更新；手动接口不受影响。
-    CPS_AUTO_RECONCILE_ENABLED: bool = True
-    CPS_AUTO_RECONCILE_RUN_HOUR: int = 5
-    CPS_AUTO_RECONCILE_LOOKBACK_DAYS: int = 3
-    CPS_AUTO_RECONCILE_CHECK_INTERVAL_SEC: int = 60
-
    # ===== 微信服务号(网页授权) =====
    # CPS 落地页在微信内拿用户 openid(base 静默)/昵称头像(userinfo),做用户级群统计。
    # ⚠️ 区别于 WECHAT_APP_ID(那是 App 移动应用,用于微信支付);这是【已认证服务号】。
@@ -1,428 +0,0 @@
-"""美团、京东 CPS 订单每日自动对账任务。
-
-每天北京时间 `CPS_AUTO_RECONCILE_RUN_HOUR`（默认 05:00）后执行一次，按更新时间
-回拉最近若干天订单并复用 admin CPS 仓储层的幂等 upsert。手动对账接口保持独立、不受影响。
-"""
-from __future__ import annotations
-
-import asyncio
-import contextlib
-import logging
-import os
-import socket
-import time
-from collections.abc import Callable, Iterator
-from datetime import date, datetime, timedelta
-from pathlib import Path
-from uuid import uuid4
-
-from app.admin.repositories import cps as cps_repo
-from app.core.config import settings
-from app.core.rewards import CN_TZ
-from app.db.session import SessionLocal, engine
-
-logger = logging.getLogger("shagua.cps_reconcile")
-_LOCK_PATH = Path(__file__).resolve().parents[2] / "data" / "cps_reconcile.lock"
-
-
-def _cn_now() -> datetime:
-    return datetime.now(CN_TZ)
-
-
-def _new_run_id(now: datetime) -> str:
-    return f"{now:%Y%m%d-%H%M%S}-{uuid4().hex[:8]}"
-
-
-def _next_run_at(now: datetime, run_hour: int) -> datetime:
-    scheduled = now.replace(hour=run_hour, minute=0, second=0, microsecond=0)
-    return scheduled if now < scheduled else scheduled + timedelta(days=1)
-
-
-def _trigger_for_run(worker_started_at: datetime, now: datetime, run_hour: int) -> str:
-    if worker_started_at.date() == now.date() and worker_started_at.hour >= run_hour:
-        return "startup_catchup"
-    return "scheduled"
-
-
-def _touch_lock() -> None:
-    with contextlib.suppress(FileNotFoundError):
-        os.utime(_LOCK_PATH, None)
-
-
-@contextlib.contextmanager
-def _single_instance_lock(stale_after_sec: int) -> Iterator[bool]:
-    """同机多进程保护：同一时间只允许一个 CPS 自动对账 worker 运行。"""
-    _LOCK_PATH.parent.mkdir(parents=True, exist_ok=True)
-    fd: int | None = None
-    try:
-        try:
-            fd = os.open(str(_LOCK_PATH), os.O_CREAT | os.O_EXCL | os.O_WRONLY)
-        except FileExistsError:
-            try:
-                age = time.time() - _LOCK_PATH.stat().st_mtime
-            except FileNotFoundError:
-                age = stale_after_sec + 1
-            if age > stale_after_sec:
-                with contextlib.suppress(FileNotFoundError):
-                    _LOCK_PATH.unlink()
-                try:
-                    fd = os.open(str(_LOCK_PATH), os.O_CREAT | os.O_EXCL | os.O_WRONLY)
-                except FileExistsError:
-                    fd = None
-
-        if fd is None:
-            yield False
-            return
-
-        os.write(fd, f"pid={os.getpid()} started_at={int(time.time())}\n".encode("ascii"))
-        yield True
-    finally:
-        if fd is not None:
-            os.close(fd)
-            with contextlib.suppress(FileNotFoundError):
-                _LOCK_PATH.unlink()
-
-
-def _empty_result() -> dict:
-    return {
-        "fetched": 0,
-        "inserted": 0,
-        "updated": 0,
-        "pages": 0,
-        "api_requests": 0,
-    }
-
-
-def _platform_log_fields(platform: str, result: dict) -> dict:
-    fields = {
-        "platform": platform,
-        "platform_status": result["status"],
-        "duration_ms": result["duration_ms"],
-    }
-    for key in ("fetched", "inserted", "updated", "pages", "api_requests", "windows"):
-        if key in result:
-            fields[key] = result[key]
-    return fields
-
-
-def _run_platform(
-    *,
-    platform: str,
-    query_time_type: int,
-    common: dict,
-    reconcile: Callable[[], dict],
-) -> tuple[dict, str | None]:
-    started = time.perf_counter()
-    logger.info(
-        "CPS auto reconcile platform started run_id=%s platform=%s",
-        common["run_id"],
-        platform,
-        extra={
-            **common,
-            "event": "cps_reconcile.platform_started",
-            "platform": platform,
-            "query_time_type": query_time_type,
-        },
-    )
-    try:
-        raw_result = reconcile()
-    except Exception as exc:  # noqa: BLE001 - 单平台失败不能阻断另一平台
-        result = {
-            "status": "failed",
-            "duration_ms": round((time.perf_counter() - started) * 1000, 3),
-            "error_type": type(exc).__name__,
-            "error_summary": str(exc)[:500],
-        }
-        logger.exception(
-            "CPS auto reconcile platform failed run_id=%s platform=%s error_type=%s",
-            common["run_id"],
-            platform,
-            type(exc).__name__,
-            extra={
-                **common,
-                "event": "cps_reconcile.platform_failed",
-                **_platform_log_fields(platform, result),
-                "error_type": type(exc).__name__,
-                "error_summary": str(exc)[:500],
-            },
-        )
-        return result, str(exc)
-
-    result = {
-        **raw_result,
-        "status": "success",
-        "duration_ms": round((time.perf_counter() - started) * 1000, 3),
-    }
-    logger.info(
-        "CPS auto reconcile platform completed run_id=%s platform=%s fetched=%s inserted=%s updated=%s",
-        common["run_id"],
-        platform,
-        result.get("fetched", 0),
-        result.get("inserted", 0),
-        result.get("updated", 0),
-        extra={
-            **common,
-            "event": "cps_reconcile.platform_completed",
-            **_platform_log_fields(platform, result),
-        },
-    )
-    return result, None
-
-
-def _skip_platform(platform: str, common: dict) -> dict:
-    result = {
-        **_empty_result(),
-        "status": "skipped",
-        "skipped": "not_configured",
-        "duration_ms": 0.0,
-    }
-    logger.warning(
-        "CPS auto reconcile platform skipped run_id=%s platform=%s reason=not_configured",
-        common["run_id"],
-        platform,
-        extra={
-            **common,
-            "event": "cps_reconcile.platform_skipped",
-            **_platform_log_fields(platform, result),
-            "skip_reason": "not_configured",
-        },
-    )
-    return result
-
-
-def _reconcile_once(
-    now: datetime | None = None,
-    *,
-    run_id: str | None = None,
-    trigger: str = "scheduled",
-) -> dict:
-    """独立拉取美团和京东；单平台异常只记日志，不影响另一平台。"""
-    end = (now or _cn_now()).astimezone(CN_TZ)
-    run_id = run_id or _new_run_id(end)
-    lookback_days = max(1, int(settings.CPS_AUTO_RECONCILE_LOOKBACK_DAYS))
-    start = end - timedelta(days=lookback_days)
-    task_started = time.perf_counter()
-    common = {
-        "run_id": run_id,
-        "trigger": trigger,
-        "scheduled_date": end.date().isoformat(),
-        "app_env": settings.APP_ENV,
-        "db_dialect": engine.dialect.name,
-        "hostname": socket.gethostname(),
-        "pid": os.getpid(),
-        "window_start": start.isoformat(),
-        "window_end": end.isoformat(),
-        "lookback_days": lookback_days,
-    }
-    result = {
-        "run_id": run_id,
-        "trigger": trigger,
-        "window_start": start.isoformat(),
-        "window_end": end.isoformat(),
-        "meituan": None,
-        "jd": None,
-        "errors": {},
-    }
-    logger.info(
-        "CPS auto reconcile started run_id=%s trigger=%s window=%s..%s",
-        run_id,
-        trigger,
-        result["window_start"],
-        result["window_end"],
-        extra={**common, "event": "cps_reconcile.started"},
-    )
-
-    if settings.mt_cps_configured:
-        def reconcile_meituan() -> dict:
-            with SessionLocal() as db:
-                return cps_repo.reconcile_orders(
-                    db,
-                    start_time=int(start.timestamp()),
-                    end_time=int(end.timestamp()),
-                    query_time_type=2,
-                    audit_context=common,
-                )
-
-        result["meituan"], error = _run_platform(
-            platform="meituan",
-            query_time_type=2,
-            common=common,
-            reconcile=reconcile_meituan,
-        )
-        if error is not None:
-            result["errors"]["meituan"] = error
-    else:
-        result["meituan"] = _skip_platform("meituan", common)
-
-    _touch_lock()
-    if settings.jd_union_configured:
-        def reconcile_jd() -> dict:
-            with SessionLocal() as db:
-                return cps_repo.reconcile_jd_orders(
-                    db,
-                    start_time=start,
-                    end_time=end,
-                    query_time_type=3,
-                    audit_context=common,
-                )
-
-        result["jd"], error = _run_platform(
-            platform="jd",
-            query_time_type=3,
-            common=common,
-            reconcile=reconcile_jd,
-        )
-        if error is not None:
-            result["errors"]["jd"] = error
-    else:
-        result["jd"] = _skip_platform("jd", common)
-
-    successes = sum(
-        platform_result.get("status") == "success"
-        for platform_result in (result["meituan"], result["jd"])
-    )
-    if result["errors"]:
-        task_status = "partial_success" if successes else "failed"
-    else:
-        task_status = "success"
-    completed_at = _cn_now()
-    duration_ms = round((time.perf_counter() - task_started) * 1000, 3)
-    next_run_at = _next_run_at(
-        completed_at,
-        min(max(int(settings.CPS_AUTO_RECONCILE_RUN_HOUR), 0), 23),
-    ).isoformat()
-    result.update({
-        "status": task_status,
-        "duration_ms": duration_ms,
-        "manual_retry_required": bool(result["errors"]),
-        "next_run_at": next_run_at,
-    })
-    completion_fields = {
-        **common,
-        "event": "cps_reconcile.completed",
-        "task_status": task_status,
-        "duration_ms": duration_ms,
-        "manual_retry_required": result["manual_retry_required"],
-        "failed_platforms": sorted(result["errors"]),
-        "next_run_at": next_run_at,
-        "meituan_result": result["meituan"],
-        "jd_result": result["jd"],
-    }
-    log_method = logger.info if task_status == "success" else logger.warning
-    log_method(
-        "CPS auto reconcile completed run_id=%s status=%s duration_ms=%s failed_platforms=%s next_run_at=%s",
-        run_id,
-        task_status,
-        duration_ms,
-        ",".join(sorted(result["errors"])) or "-",
-        next_run_at,
-        extra=completion_fields,
-    )
-
-    return result
-
-
-def _should_run(last_run: date | None, now: datetime, run_hour: int) -> bool:
-    return last_run != now.date() and now.hour >= run_hour
-
-
-async def _run_loop() -> None:
-    interval = max(30, int(settings.CPS_AUTO_RECONCILE_CHECK_INTERVAL_SEC))
-    run_hour = min(max(int(settings.CPS_AUTO_RECONCILE_RUN_HOUR), 0), 23)
-    lock_stale_after = max(interval * 3, 1800)
-    with _single_instance_lock(lock_stale_after) as lock_acquired:
-        if not lock_acquired:
-            logger.warning(
-                "CPS auto reconcile worker skipped: another worker owns lock",
-                extra={
-                    "event": "cps_reconcile.worker_skipped",
-                    "skip_reason": "lock_not_acquired",
-                    "pid": os.getpid(),
-                },
-            )
-            return
-        await _run_locked_loop(interval, run_hour)
-
-
-async def _run_locked_loop(interval: int, run_hour: int) -> None:
-    worker_started_at = _cn_now()
-    logger.info(
-        "CPS auto reconcile worker started run_hour=%s interval=%ss lookback_days=%s",
-        run_hour,
-        interval,
-        settings.CPS_AUTO_RECONCILE_LOOKBACK_DAYS,
-        extra={
-            "event": "cps_reconcile.worker_started",
-            "pid": os.getpid(),
-            "hostname": socket.gethostname(),
-            "app_env": settings.APP_ENV,
-            "db_dialect": engine.dialect.name,
-            "run_hour": run_hour,
-            "interval_sec": interval,
-            "lookback_days": settings.CPS_AUTO_RECONCILE_LOOKBACK_DAYS,
-            "next_run_at": _next_run_at(worker_started_at, run_hour).isoformat(),
-        },
-    )
-    last_run: date | None = None
-    try:
-        while True:
-            run_id: str | None = None
-            try:
-                _touch_lock()
-                now = _cn_now()
-                if _should_run(last_run, now, run_hour):
-                    run_id = _new_run_id(now)
-                    trigger = _trigger_for_run(worker_started_at, now, run_hour)
-                    await asyncio.to_thread(
-                        _reconcile_once,
-                        now,
-                        run_id=run_id,
-                        trigger=trigger,
-                    )
-                    last_run = now.date()
-            except Exception:  # noqa: BLE001 - 后台任务不能因单次异常退出
-                logger.exception(
-                    "CPS auto reconcile unexpected worker error",
-                    extra={
-                        "event": "cps_reconcile.unexpected_failed",
-                        "run_id": run_id or "unassigned",
-                        "pid": os.getpid(),
-                    },
-                )
-            await asyncio.sleep(interval)
-    except asyncio.CancelledError:
-        logger.info(
-            "CPS auto reconcile worker stopped",
-            extra={"event": "cps_reconcile.worker_stopped", "pid": os.getpid()},
-        )
-        raise
-
-
-def start_cps_reconcile_worker() -> asyncio.Task | None:
-    if not settings.CPS_AUTO_RECONCILE_ENABLED:
-        logger.info(
-            "CPS auto reconcile disabled",
-            extra={
-                "event": "cps_reconcile.worker_skipped",
-                "skip_reason": "disabled",
-            },
-        )
-        return None
-    if not settings.mt_cps_configured and not settings.jd_union_configured:
-        logger.warning(
-            "CPS auto reconcile not started: Meituan and JD credentials are missing",
-            extra={
-                "event": "cps_reconcile.worker_skipped",
-                "skip_reason": "all_platform_credentials_missing",
-            },
-        )
-        return None
-    return asyncio.create_task(_run_loop(), name="cps-auto-reconcile")
-
-
-async def stop_cps_reconcile_worker(task: asyncio.Task | None) -> None:
-    if task is None:
-        return
-    task.cancel()
-    with contextlib.suppress(asyncio.CancelledError):
-        await task
@@ -22,13 +22,13 @@ from __future__ import annotations
 import json
 import logging
 import os
-import shutil
 import sys
 from contextvars import ContextVar
 from datetime import datetime
 from logging.handlers import RotatingFileHandler
 from pathlib import Path

+
 # 请求级 trace_id:入口(如 compare.py 透传壳)set 之后, 本请求上下文(含 run_in_threadpool
 # 拷贝出去的线程)内所有日志自动带上。默认空串 = 非请求上下文(启动/后台 worker)。
 trace_id_ctx: ContextVar[str] = ContextVar("trace_id", default="")
@@ -91,55 +91,6 @@ class TextFormatter(logging.Formatter):
        return f"{base} trace={tid}" if tid else base


-class SafeRotatingFileHandler(RotatingFileHandler):
-    """Windows 下不会被外部句柄卡死的 RotatingFileHandler。
-
-    stdlib 轮转靠 rename 活动文件(app-server.log → .1);Windows 只要有别的句柄(IDE 索引、
-    app.admin.main 第二进程、残留 --reload worker、杀软扫描)开着它, rename 就 WinError 32,
-    轮转永久卡死——文件停在 maxBytes、之后每条日志被丢。这里 Windows 改用 copytruncate:把活动
-    文件拷进备份、再通过自己的句柄原地清空, 从不 rename 活动文件, 故外部句柄开着也能转。
-    POSIX(生产 Linux)rename 打开中的文件本就合法, 保留 stdlib 的原子轮转不变。
-
-    代价:copytruncate 在“拷贝→清空”极窄窗口内并发写可能丢几行(仅跨进程;同进程 emit 有
-    handler 锁串行, 无此问题)。对本地开发日志可接受。
-    """
-
-    def doRollover(self) -> None:
-        if os.name != "nt":
-            super().doRollover()
-            return
-        if self.stream is None:
-            self.stream = self._open()
-        else:
-            self.stream.flush()
-        try:
-            self._copytruncate_backups()
-        except OSError:
-            # 备份腾挪是尽力而为:任一备份被占用也绝不能挡住下面的清空, 否则活动文件继续涨、
-            # 轮转又卡死——那就白改了。
-            pass
-        # 通过自己独占的句柄原地清空:不涉及 rename, 外部只读句柄不受影响。
-        self.stream.seek(0)
-        self.stream.truncate()
-        self.stream.flush()
-
-    def _copytruncate_backups(self) -> None:
-        """把 .N-1→.N 逐级腾挪, 再把活动文件拷到 .1(不动活动文件本身)。"""
-        if self.backupCount <= 0:
-            return
-        for i in range(self.backupCount - 1, 0, -1):
-            sfn = self.rotation_filename(f"{self.baseFilename}.{i}")
-            dfn = self.rotation_filename(f"{self.baseFilename}.{i + 1}")
-            if os.path.exists(sfn):
-                if os.path.exists(dfn):
-                    os.remove(dfn)
-                os.replace(sfn, dfn)
-        dfn = self.rotation_filename(f"{self.baseFilename}.1")
-        if os.path.exists(dfn):
-            os.remove(dfn)
-        shutil.copyfile(self.baseFilename, dfn)
-
-
 _CONFIGURED = False


@@ -175,7 +126,7 @@ def setup_logging(debug: bool = False) -> None:
        Path(os.getenv("LOG_DIR", "logs")) / "app-server.log"
    )
    Path(log_file).parent.mkdir(parents=True, exist_ok=True)
-    file_handler = SafeRotatingFileHandler(
+    file_handler = RotatingFileHandler(
        log_file, maxBytes=10 * 1024 * 1024, backupCount=5, encoding="utf-8",
    )
    file_handler.setFormatter(JsonFormatter(service))
@@ -43,10 +43,6 @@ 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.config import settings
-from app.core.cps_reconcile_worker import (
-    start_cps_reconcile_worker,
-    stop_cps_reconcile_worker,
-)
 from app.core.daily_exchange_worker import (
    start_daily_exchange_worker,
    stop_daily_exchange_worker,
@@ -93,7 +89,6 @@ async def lifespan(_: FastAPI) -> AsyncIterator[None]:
    except Exception:  # noqa: BLE001
        logger.exception("reverse_geocoder 预热失败(城市反查将在首个请求时懒加载)")
    reconcile_task = start_withdraw_reconcile_worker()
-    cps_reconcile_task = start_cps_reconcile_worker()
    heartbeat_task = start_heartbeat_monitor()
    daily_exchange_task = start_daily_exchange_worker()
    observe_task = start_observe_worker()
@@ -103,7 +98,6 @@ async def lifespan(_: FastAPI) -> AsyncIterator[None]:
    finally:
        await stop_heartbeat_monitor(heartbeat_task)
        await stop_withdraw_reconcile_worker(reconcile_task)
-        await stop_cps_reconcile_worker(cps_reconcile_task)
        await stop_daily_exchange_worker(daily_exchange_task)
        await stop_observe_worker(observe_task)
        await stop_inactivity_reset_worker(inactivity_task)
@@ -45,10 +45,6 @@ class ComparisonRecord(Base):
        # 首页轮播 / 省钱战绩聚合都按 status='success' 过滤 + created_at 近期排序;
        # 复合索引避免随数据量增大退化成全表扫(单列 created_at 索引不含 status)。
        Index("ix_comparison_status_created", "status", "created_at"),
-        # C 端「我的比价记录」列表:WHERE user_id=? ORDER BY created_at DESC, id DESC LIMIT n。
-        # 单列 user_id 索引只能过滤,排序仍要把该用户全部记录取出来排一遍;这条复合索引的**反向扫**
-        # 恰好等于 (created_at DESC, id DESC),PG 直接取前 n 条、免排序。列序不能动。
-        Index("ix_comparison_user_created", "user_id", "created_at", "id"),
    )

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
@@ -7,8 +7,8 @@ from __future__ import annotations

 from datetime import datetime

-from sqlalchemy import func, or_, select
-from sqlalchemy.orm import Session, defer
+from sqlalchemy import func, select
+from sqlalchemy.orm import Session

 from app.core.rewards import CN_TZ
 from app.models.ad_feed_reward import AdFeedRewardRecord
@@ -375,48 +375,19 @@ def harvest_abort(
    return rec


-def _ordered_shop_name_select(user_id: int):
-    """该用户「真实下单」(source='compare')覆盖到的店名 select,给「已下单」筛选当子查询。
-
-    口径与 [_ordered_shop_names] 完全一致,只是时机不同:那边是**拿到本页之后**按 candidates
-    反查打标;这边是**分页之前**就要过滤,拿不到 candidates,只能整段下推成子查询。
-    没有先捞成集合再展开 IN (...) 字面量 —— 重度用户下单过的店名可能上千,展开会撞 SQLite
-    的绑定变量上限,而且又变回了那个「随下单量线性变慢」的老写法。
-    """
-    return select(SavingsRecord.shop_name).where(
-        SavingsRecord.user_id == user_id,
-        SavingsRecord.source == "compare",
-        SavingsRecord.shop_name.is_not(None),
-    )
-
-
-def _like_escape(kw: str) -> str:
-    """转义 LIKE 通配符(百分号 / 下划线 / 反斜杠),让用户输入只按字面量匹配(配合 escape 参数)。
-
-    不转义的话搜一个「%」就等于把整表拉回来。
-    """
-    return kw.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
-
-
-def _ordered_shop_names(db: Session, user_id: int, candidates: set[str]) -> set[str]:
-    """[candidates] 里哪些店名被该用户「真实下单」(source='compare')覆盖过,用来打「已下单」。
+def _ordered_shop_names(db: Session, user_id: int) -> set[str]:
+    """该用户「真实下单」(source='compare')覆盖到的店名集合,用来给比价记录打「已下单」。

    只认 compare(归因命中后真实上报),demo 演示数据不算。下单上报不带 trace_id,
    只能按店名对齐——两边店名同源(都来自比价意图识别阶段的门店名 query),精确相等即视为同店。
    语义=店级:同一家店比价过多次,这些记录会一并标「已下单」。
-
-    ⚠️ 只查**本页出现过的店名**(candidates ≤ limit 条),不再把该用户全部下单店名捞回内存:
-    老写法随下单量线性增长,重度用户几千行全读一遍只为跟 50 条记录取交集。空集合直接返回
-    (避免 IN () 非法)。
    """
-    if not candidates:
-        return set()
    rows = db.execute(
        select(SavingsRecord.shop_name).where(
            SavingsRecord.user_id == user_id,
            SavingsRecord.source == "compare",
-            SavingsRecord.shop_name.in_(candidates),
-        ).distinct()
+            SavingsRecord.shop_name.is_not(None),
+        )
    ).scalars().all()
    return {s for s in rows if s}

@@ -444,60 +415,17 @@ def _ad_coins_by_trace(db: Session, user_id: int, trace_ids: list[str]) -> dict[
    return {tid: int(coin) for tid, coin in rows if tid}


-# 列表出参(ComparisonRecordOut)根本不读、但 select(ORM) 默认会一并捞回来的重型 JSON 列:
-#   - raw_payload:done.params 上报体全量,**每条记录都有**(harvest 与 POST 两条写路径都落)。
-#     单条几 KB~几十 KB,一页 50 条就是稳定几百 KB~几 MB 的白读 + 白反序列化。
-#   - llm_calls:每次 LLM 调用的 input_messages + output 全文。只有走老客户端 POST /compare/record
-#     的记录才有(_backfill_llm_calls 回填;harvest 路径不落),但有的时候单条就能到 MB 级 —— 一页里
-#     混进几条这种记录,整个请求就被它们拖住。
-#   - llm_price_snapshot:逐模型单价快照,同样只在回填时落。
-# 三列全部读出来再被 pydantic 丢掉,是「比价记录/全部记录」页慢的主要来源。
-# ⚠️ defer 的列一旦在别处被读到会触发**逐行**懒加载(N+1);列表这条链路(ComparisonRecordOut
-# 不声明这三个字段 → 不会 getattr 到)是安全的。详情接口 get_record 不 defer,raw_payload 照常返回。
-_LIST_DEFERRED = (
-    ComparisonRecord.raw_payload,
-    ComparisonRecord.llm_calls,
-    ComparisonRecord.llm_price_snapshot,
-)
-
-
 def list_records(
    db: Session,
    user_id: int,
    *,
    limit: int = 20,
    cursor: int | None = None,
-    ordered: bool | None = None,
-    keyword: str | None = None,
 ) -> tuple[list[ComparisonRecord], int | None]:
    """比价记录分页(按创建时间倒序、id 兜底,游标式)。附「已下单」店级标记 + 「看广告赚的金币」(瞬态,不写库)。"""
-    stmt = (
-        select(ComparisonRecord)
-        .where(ComparisonRecord.user_id == user_id)
-        .options(*(defer(col) for col in _LIST_DEFERRED))
-    )
+    stmt = select(ComparisonRecord).where(ComparisonRecord.user_id == user_id)
    if cursor is not None:
        stmt = stmt.where(ComparisonRecord.id < cursor)
-    # 「已下单」tab 与搜索框的过滤都下推到这里,不能留给客户端对整页结果 filter ——
-    # 分页之后一页里可能一条都不命中,列表看着就是空的/卡住的,得翻很多页才蹦出一条。
-    if ordered:
-        stmt = stmt.where(
-            ComparisonRecord.store_name.in_(_ordered_shop_name_select(user_id))
-        )
-    kw = (keyword or "").strip()
-    if kw:
-        # product_names 是写路径从 items[].name 派生的普通文本列(items 本身是 JSON,SQLite 下
-        # 中文被 ensure_ascii 转义,没法直接 LIKE)—— 搜「菜名」靠的就是它。
-        # ilike:PG 原生 ILIKE,SQLite 渲染成 lower() LIKE lower(),两边都忽略大小写。
-        pattern = f"%{_like_escape(kw)}%"
-        stmt = stmt.where(
-            or_(
-                ComparisonRecord.store_name.ilike(pattern, escape="\\"),
-                ComparisonRecord.product_names.ilike(pattern, escape="\\"),
-            )
-        )
-    # 排序与 ix_comparison_user_created(user_id, created_at, id)对齐 —— DESC/DESC 正好是该索引的
-    # 反向扫,PG 免排序直接取前 limit 条。改排序方向前先想清楚索引还吃不吃得上。
    stmt = stmt.order_by(ComparisonRecord.created_at.desc(), ComparisonRecord.id.desc()).limit(limit)

    items = list(db.execute(stmt).scalars().all())
@@ -505,10 +433,7 @@ def list_records(

    # 「已下单」标记:本页记录的 store_name 若落在该用户真实下单的店名集合里即 True。
    # ordered / ad_coins_earned 均非 ORM 列,仅挂实例上供 ComparisonRecordOut(from_attributes) 读出,不持久化。
-    page_shops = {it.store_name for it in items if it.store_name}
-    # ordered=True 时上面已按同一口径(_ordered_shop_name_select)筛过,本页必然全是已下单,
-    # 省掉这次反查;其余情况照旧按本页店名反查 savings。
-    ordered_shops = page_shops if ordered else _ordered_shop_names(db, user_id, page_shops)
+    ordered_shops = _ordered_shop_names(db, user_id)
    # 「本次比价看广告赚的金币」:按本页 trace_id 一次性聚合(同 ordered 范式)。
    ad_coins = _ad_coins_by_trace(db, user_id, [it.trace_id for it in items])
    for it in items:
@@ -25,18 +25,6 @@ server {
    # (纯文字反馈体积小、不受影响 → 呈现为「时好时坏」)。根治仍需客户端上传前压缩。
    client_max_body_size 32m;

-    # JSON 响应压缩。nginx 默认 gzip off,且就算 on 了 gzip_types 也只含 text/html、
-    # gzip_proxied 默认 off(反代来的响应一律不压)—— 三个默认值凑一起 = 我们所有接口都在裸奔。
-    # 比价记录列表这种一次 50 条、字段名 + 中文店名/菜名高度重复的 JSON,gzip 压缩比稳定在 8~10 倍
-    # (几百 KB → 几十 KB),弱网下省的就是首屏那几秒。
-    # 只压 JSON:APK 直链(/media/shaguabijia.apk)、图片本身已是压缩格式,再压纯浪费 CPU。
-    gzip              on;
-    gzip_proxied      any;      # 反代响应也压(默认 off = 对我们这套反代等于没开)
-    gzip_types        application/json;
-    gzip_min_length   1024;     # 小响应压了反而更大(gzip 头开销),不值当
-    gzip_comp_level   5;        # 5 是体积/CPU 的常用折中点,再往上收益递减
-    gzip_vary         on;       # 给 CDN/中间缓存正确按 Accept-Encoding 分桶
-
    location / {
        proxy_pass http://127.0.0.1:8770;
        proxy_http_version 1.1;
@@ -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)。
@@ -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`。
@@ -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`。
@@ -10,12 +10,6 @@
 |---|---|---|---|---|
 | `limit` | int | ❌ | 20 | 1–100 |
 | `cursor` | int | ❌ | null | 上一页末条 `id`,首页不传 |
-| `ordered` | bool | ❌ | null | `true`=只出「已下单」(店名命中本人真实下单)的记录;不传=不筛 |
-| `keyword` | string | ❌ | null | 按店名 / 菜名模糊搜索,忽略大小写,≤64 字符;纯空白等同不传 |
-| `include_trace` | bool | ❌ | false | 客户端开了本机 agent 调试模式时带 `true`,放行**本人**记录的 `trace_url` |
-
-`ordered` / `keyword` 都在服务端过滤后再分页,客户端不要拿一页结果自己 filter ——
-分页之后一页里可能一条都不命中,列表会看着像空的。

 ## 出参
 响应 `200`：`{ items: ComparisonRecordOut[], next_cursor: int|null }`（分页见 [索引#游标分页约定](./README.md#游标分页约定)）
@@ -44,9 +38,6 @@
 | `items` | object[] | 下单菜品 `{name, qty, specs?}` |
 | `comparison_results` | object[] | 逐平台对比（price 单位元，已按 rank 升序） |
 | `skipped_dish_names` | string[] | 被跳过的菜名 |
-| `ordered` | bool | 「已下单」店级标记:店名命中本人 `source='compare'` 的下单记录即 `true`。**瞬态字段,不在表里**,每次查询现算 |
-| `ad_coins_earned` | int | 本次比价看信息流广告实发的金币(按 `trace_id` 聚合)。同为瞬态字段 |
-| `trace_url` | string \| null | pricebot 调试链接。未开 `debug_trace_enabled` 且未带 `include_trace=true` 时为 `null` |
 | `created_at` | datetime | 时间 |

 ## 错误
@@ -1,277 +0,0 @@
-"""一次性 mock:造几条不同状态的用户反馈,供运营后台「反馈工单」页联调验收。
-
-覆盖:
-  - 三个状态 tab(待审核 pending / 已采纳 adopted / 未采纳 rejected),重点铺「待审核」;
-  - 两种反馈类型 source(普通反馈 profile /「我的」页入口、比价反馈 comparison / 比价结果页入口),
-    比价反馈带「问题场景」scene(找错商品/优惠不对/比价太慢…);
-  - 提交端环境快照(app_version / device_model / rom_name / android_version)——新端反馈才有,
-    另留 1~2 条历史反馈(env 全 NULL、contact 有值)测「旧数据」展示;
-  - 截图:在 data/media/feedback/ 生成真实可加载的纯色 PNG(手写字节,无需 Pillow),
-    让审核抽屉的图能真加载出来(与 seed_mock_price_reports 同法)。
-  - 已采纳条带 reward_coins + admin_reply + review_note;未采纳条带 reject_reason + admin_reply。
-
-幂等:每次运行先按固定 mock 手机号清掉旧 mock 用户/反馈 + 删 mock 截图再重建。仅清理用 --clean-only。
-
-  .venv\\Scripts\\python.exe scripts\\seed_mock_feedback.py
-  .venv\\Scripts\\python.exe scripts\\seed_mock_feedback.py --clean-only
-
-看图:前端「反馈工单」页(http://localhost:3001 → 反馈)。图经 NEXT_PUBLIC_MEDIA_BASE
-(本地 = http://localhost:8770)由 App 后端 /media 加载——改过 .env.local 后需重启 next dev,
-且 App 后端(:8770)要在跑。
-"""
-from __future__ import annotations
-
-import argparse
-import struct
-import sys
-import zlib
-from datetime import UTC, datetime, timedelta
-from pathlib import Path
-
-from sqlalchemy import delete, select
-
-from app.core.config import settings
-from app.db.session import SessionLocal
-from app.models.feedback import Feedback
-from app.models.user import User
-from app.repositories.user import _gen_unique_username
-
-# Windows GBK 控制台下也能正常打印中文/¥(避免 UnicodeEncodeError)
-if hasattr(sys.stdout, "reconfigure"):
-    sys.stdout.reconfigure(encoding="utf-8")
-
-# 固定 mock 手机号:脚本只动这些号,便于幂等重建 / 清理。
-# 刻意与 seed_mock_withdraws / seed_mock_price_reports 的号段错开,互不干扰。
-MOCK_PHONES = [
-    "13255550001",
-    "13255550002",
-    "13255550003",
-    "13255550004",
-    "13255550005",
-]
-
-_FEEDBACK_DIR = Path(settings.MEDIA_ROOT) / "feedback"
-_MOCK_IMG_GLOB = "mock_fb_*.png"  # 本脚本生成的图前缀,清理时按此删
-
-
-def _naive_utc_now() -> datetime:
-    """与 func.now() 在 SQLite 的口径一致:naive UTC。反馈 created_at 走 server_default=func.now(),
-    这里显式造数据也用 naive UTC,和真实反馈行同源,前端展示口径一致。"""
-    return datetime.now(UTC).replace(tzinfo=None)
-
-
-def _solid_png(width: int, height: int, rgb: tuple[int, int, int]) -> bytes:
-    """生成一张纯色 PNG(truecolor RGB)的字节,无需 Pillow。颜色块即可肉眼判断「图加载出来了」。"""
-    def _chunk(typ: bytes, data: bytes) -> bytes:
-        body = typ + data
-        return struct.pack(">I", len(data)) + body + struct.pack(">I", zlib.crc32(body) & 0xFFFFFFFF)
-
-    ihdr = struct.pack(">IIBBBBB", width, height, 8, 2, 0, 0, 0)  # 8bit/通道, color type 2 = RGB
-    row = b"\x00" + bytes(rgb) * width  # 每行前缀 filter byte 0
-    idat = zlib.compress(row * height, 9)
-    return b"\x89PNG\r\n\x1a\n" + _chunk(b"IHDR", ihdr) + _chunk(b"IDAT", idat) + _chunk(b"IEND", b"")
-
-
-def _write_mock_image(name: str, rgb: tuple[int, int, int]) -> str:
-    """写一张 mock 截图到 media/feedback/,返回其相对 URL(/media/feedback/<name>)。"""
-    _FEEDBACK_DIR.mkdir(parents=True, exist_ok=True)
-    (_FEEDBACK_DIR / name).write_bytes(_solid_png(320, 320, rgb))
-    return f"{settings.MEDIA_URL_PREFIX}/feedback/{name}"
-
-
-def clean(db) -> int:
-    uids = list(db.execute(select(User.id).where(User.phone.in_(MOCK_PHONES))).scalars())
-    if uids:
-        db.execute(delete(Feedback).where(Feedback.user_id.in_(uids)))
-        db.execute(delete(User).where(User.id.in_(uids)))
-        db.commit()
-    # 删 mock 截图文件
-    if _FEEDBACK_DIR.exists():
-        for f in _FEEDBACK_DIR.glob(_MOCK_IMG_GLOB):
-            f.unlink(missing_ok=True)
-    return len(uids)
-
-
-def seed(db) -> list[Feedback]:
-    now = _naive_utc_now()
-
-    def ago(**kw) -> datetime:
-        return now - timedelta(**kw)
-
-    # 1) 建 5 个 mock 用户(U5 昵称留空测 "-" 展示)
-    users_spec = [
-        ("13255550001", "反馈小达人"),
-        ("13255550002", "比价挑刺王"),
-        ("13255550003", "热心用户阿明"),
-        ("13255550004", "老用户张姐"),
-        ("13255550005", None),
-    ]
-    users: dict[str, User] = {}
-    for phone, nickname in users_spec:
-        u = User(
-            phone=phone,
-            username=_gen_unique_username(db),  # 列 NOT NULL + unique,仿真实注册生成
-            nickname=nickname,
-            register_channel="sms",
-            status="active",
-            created_at=ago(days=15),
-            last_login_at=ago(hours=1),
-        )
-        db.add(u)
-        users[phone] = u
-    db.flush()  # 拿 user.id
-
-    # 2) 生成 mock 截图(不同颜色块,便于肉眼区分「都加载出来了」)
-    palette = [
-        (24, 144, 255),   # 蓝
-        (82, 196, 26),    # 绿
-        (250, 173, 20),   # 橙
-        (245, 34, 45),    # 红
-    ]
-    imgs = [_write_mock_image(f"mock_fb_{i}.png", palette[i]) for i in range(len(palette))]
-
-    # 3) 造反馈记录
-    def fb(
-        phone: str,
-        content: str,
-        *,
-        source: str = "profile",
-        scene: str | None = None,
-        images: list[str] | None = None,
-        contact: str = "",
-        status: str = "pending",
-        reject_reason: str | None = None,
-        reward_coins: int | None = None,
-        review_note: str | None = None,
-        admin_reply: str | None = None,
-        app_version: str | None = None,
-        device_model: str | None = None,
-        rom_name: str | None = None,
-        android_version: str | None = None,
-        created: datetime,
-    ) -> Feedback:
-        return Feedback(
-            user_id=users[phone].id,
-            content=content,
-            contact=contact,  # 列 NOT NULL:新端存空串,历史数据有值
-            source=source,
-            scene=scene,
-            images=images,
-            status=status,
-            reject_reason=reject_reason,
-            reward_coins=reward_coins,
-            review_note=review_note,
-            admin_reply=admin_reply,
-            app_version=app_version,
-            device_model=device_model,
-            rom_name=rom_name,
-            android_version=android_version,
-            reviewed_at=(created + timedelta(hours=2)) if status != "pending" else None,
-            created_at=created,
-        )
-
-    feedbacks = [
-        # ===== 待审核 pending(默认 tab,重点铺量)=====
-        # 普通反馈 · 新端(带环境快照)· 无图
-        fb("13255550001", "签到金币到账有时候会延迟一两分钟,能不能做成实时到账?",
-           source="profile",
-           app_version="2.3.1", device_model="PJA110", rom_name="ColorOS", android_version="14",
-           created=ago(minutes=6)),
-        # 比价反馈 · scene=优惠不对 · 新端 · 2 图
-        fb("13255550002", "这家店京东外卖的到手价比你们算出来的最低价还低,截图为证,麻烦核实。",
-           source="comparison", scene="优惠不对", images=[imgs[0], imgs[1]],
-           app_version="2.3.1", device_model="M2012K11AC", rom_name="MIUI", android_version="13",
-           created=ago(minutes=22)),
-        # 比价反馈 · scene=找错商品 · 新端 · 无图
-        fb("13255550002", "比价结果里的商品跟我搜的不是同一个规格,数量对不上。",
-           source="comparison", scene="找错商品",
-           app_version="2.3.0", device_model="V2309A", rom_name="OriginOS", android_version="14",
-           created=ago(hours=1)),
-        # 普通反馈 · 新端 · 1 图(表扬 + 小问题)
-        fb("13255550003", "提现秒到账,好评!顺手反馈个小 bug:金币记录页偶尔白屏,要退出去重进。",
-           source="profile", images=[imgs[2]],
-           app_version="2.3.1", device_model="23078RKD5C", rom_name="MIUI", android_version="14",
-           created=ago(hours=3)),
-        # 比价反馈 · scene=比价太慢 · 历史数据(env 全 NULL、contact 有值)
-        fb("13255550004", "比价转圈太久了,经常要等十几秒才出结果,体验不太好。",
-           source="comparison", scene="比价太慢", contact="微信 zhangjie_66",
-           created=ago(days=1, hours=2)),
-        # 普通反馈 · 历史数据(env 全 NULL、contact 有值)· 无昵称用户
-        fb("13255550005", "希望能增加支付宝提现,微信零钱用不太习惯。",
-           source="profile", contact="QQ 100200300",
-           created=ago(days=1, hours=8)),
-
-        # ===== 已采纳 adopted(发金币 + 回复)=====
-        fb("13255550003", "建议在比价结果页加个「一键复制口令」,分享给家人更方便。",
-           source="profile", images=[imgs[3]],
-           status="adopted", reward_coins=2000,
-           review_note="有效产品建议,已排期到 2.4.0", admin_reply="感谢反馈!该功能已在规划中,金币奖励已发放~",
-           app_version="2.2.8", device_model="PJA110", rom_name="ColorOS", android_version="13",
-           created=ago(days=2)),
-
-        # ===== 未采纳 rejected(带原因 + 回复)=====
-        fb("13255550002", "你们算的价格不准,我看到的更便宜。",
-           source="comparison", scene="价格不准",
-           status="rejected", reject_reason="截图价格为限时活动价且已过期,不满足「长期可复现更低价」条件,暂不采纳。",
-           admin_reply="感谢参与,本次未通过,欢迎继续上报有效更低价~",
-           app_version="2.3.0", device_model="M2012K11AC", rom_name="MIUI", android_version="13",
-           created=ago(days=3)),
-    ]
-    db.add_all(feedbacks)
-    db.commit()
-    for f in feedbacks:
-        db.refresh(f)
-    return feedbacks
-
-
-def main() -> None:
-    parser = argparse.ArgumentParser(description="造用户反馈 mock 数据(运营后台反馈工单页联调用)")
-    parser.add_argument("--clean-only", action="store_true", help="只清理 mock 数据,不重建")
-    args = parser.parse_args()
-
-    db = SessionLocal()
-    try:
-        removed = clean(db)
-        if removed:
-            print(f"🧹 已清理旧 mock:{removed} 个用户及其反馈 + mock 截图")
-        if args.clean_only:
-            print("✅ 仅清理,已完成。")
-            return
-
-        feedbacks = seed(db)
-
-        status_label = {"pending": "待审核", "adopted": "已采纳", "rejected": "未采纳"}
-        source_label = {"profile": "普通反馈", "comparison": "比价反馈"}
-        by_status: dict[str, list[Feedback]] = {}
-        for f in feedbacks:
-            by_status.setdefault(f.status, []).append(f)
-
-        print(f"\n✅ 已生成 {len(feedbacks)} 条反馈(截图落 {_FEEDBACK_DIR}),分布:")
-        for st in ("pending", "adopted", "rejected"):
-            lst = by_status.get(st, [])
-            print(f"   {status_label[st]:<4} {len(lst)} 条")
-
-        print("\n   明细(#id | 状态 | 类型/场景 | 图 | 内容):")
-        uid2phone = dict(
-            db.execute(select(User.id, User.phone).where(User.phone.in_(MOCK_PHONES))).all()
-        )
-        for f in feedbacks:
-            src = source_label.get(f.source, f.source)
-            scene = f"·{f.scene}" if f.scene else ""
-            nimg = len(f.images or [])
-            snippet = f.content[:20] + ("…" if len(f.content) > 20 else "")
-            print(
-                f"   #{f.id} [{status_label.get(f.status, f.status)}] "
-                f"{src}{scene}  {nimg}图  {uid2phone.get(f.user_id, '?')}  {snippet}"
-            )
-        print(
-            "\n👉 打开 http://localhost:3001 → 反馈 查看(默认「待审核」tab)。"
-            "\n   图加载不出来时排查:① 是否重启过 next dev(读 .env.local 的 NEXT_PUBLIC_MEDIA_BASE)"
-            "  ② App 后端(:8770)是否在跑(它托管 /media)。"
-        )
-    finally:
-        db.close()
-
-
-if __name__ == "__main__":
-    main()
@@ -30,7 +30,6 @@ from app.core.config import settings
 from app.db.session import SessionLocal
 from app.models.price_report import PriceReport
 from app.models.user import User
-from app.repositories.user import _gen_unique_username

 if hasattr(sys.stdout, "reconfigure"):
    sys.stdout.reconfigure(encoding="utf-8")  # Windows 控制台输出中文/¥
@@ -85,11 +84,10 @@ def seed(db) -> list[PriceReport]:
    # 1) 建 3 个 mock 用户
    users: dict[str, User] = {}
    for i, (phone, nickname) in enumerate(
-        zip(MOCK_PHONES, ["省钱小王", "比价老李", "薅羊毛阿珍"], strict=True)
+        zip(MOCK_PHONES, ["省钱小王", "比价老李", "薅羊毛阿珍"])
    ):
        u = User(
            phone=phone,
-            username=_gen_unique_username(db),  # 列 NOT NULL + unique,仿真实注册生成
            nickname=nickname,
            register_channel="sms",
            status="active",
@@ -21,7 +21,7 @@ import argparse
 import sys
 import uuid
 from collections import defaultdict
-from datetime import UTC, datetime, timedelta
+from datetime import datetime, timedelta, timezone

 from sqlalchemy import delete, select

@@ -31,7 +31,6 @@ from app.models.ad_feed_reward import AdFeedRewardRecord
 from app.models.ad_reward import AdRewardRecord
 from app.models.user import User
 from app.models.wallet import CashTransaction, CoinAccount, CoinTransaction, WithdrawOrder
-from app.repositories.user import _gen_unique_username

 # Windows GBK 控制台下也能正常打印中文(避免 UnicodeEncodeError)
 if hasattr(sys.stdout, "reconfigure"):
@@ -52,7 +51,7 @@ REMARK = {"exchange_in": "金币兑入", "withdraw": "提现扣款", "withdraw_r

 def _naive_utc_now() -> datetime:
    """与 func.now() 在 SQLite 的口径一致:naive UTC。前端按 UTC 解析再转北京时间。"""
-    return datetime.now(UTC).replace(tzinfo=None)
+    return datetime.now(timezone.utc).replace(tzinfo=None)


 def _mock_transfer_no() -> str:
@@ -97,7 +96,6 @@ def seed(db) -> list[WithdrawOrder]:
    for phone, nickname in users_spec:
        u = User(
            phone=phone,
-            username=_gen_unique_username(db),  # 列 NOT NULL + unique,仿真实注册生成
            nickname=nickname,
            register_channel="sms",
            status="active",
@@ -9,11 +9,9 @@ from app.admin.repositories import ad_revenue
 from app.db.session import SessionLocal
 from app.models.ad_ecpm import AdEcpmRecord
 from app.models.ad_pangle_revenue import AdPangleDailyRevenue
-from app.models.ad_reward import AdRewardRecord
 from app.models.user import User

 REPORT_DATE = "2040-02-03"
-PLAYBACK_DATE = "2040-02-04"


 def test_business_scope_filters_client_and_pangle_by_env_and_code(monkeypatch) -> None:
@@ -129,82 +127,3 @@ def test_business_scope_filters_client_and_pangle_by_env_and_code(monkeypatch) -
        db.execute(delete(User).where(User.phone == "18800009991"))
        db.commit()
        db.close()
-
-
-def test_reward_video_incomplete_playback_has_zero_revenue() -> None:
-    db = SessionLocal()
-    phone = "18800009992"
-    sessions = {
-        "closed_early": "rv-zero-closed",
-        "too_short": "rv-zero-short",
-        "capped": "rv-keep-capped",
-        "granted": "rv-keep-granted",
-    }
-    try:
-        user = User(phone=phone, username="29999999992", register_channel="sms")
-        db.add(user)
-        db.flush()
-
-        for index, (status, session_id) in enumerate(sessions.items(), start=1):
-            created_at = datetime(2040, 2, 4, index, tzinfo=UTC)
-            db.add(AdEcpmRecord(
-                user_id=user.id,
-                ad_type="reward_video",
-                ad_session_id=session_id,
-                app_env="prod",
-                our_code_id="prod-reward",
-                ecpm_raw="10000",
-                report_date=PLAYBACK_DATE,
-                created_at=created_at,
-            ))
-            db.add(AdRewardRecord(
-                trans_id=f"{session_id}-trans",
-                user_id=user.id,
-                coin=0,
-                status=status,
-                reward_scene="reward_video",
-                ad_session_id=session_id,
-                app_env="prod",
-                our_code_id="prod-reward",
-                ecpm_raw="10000",
-                reward_date=PLAYBACK_DATE,
-                created_at=created_at,
-            ))
-        db.commit()
-
-        result = ad_revenue.ad_revenue_report(
-            db,
-            date_from=PLAYBACK_DATE,
-            date_to=PLAYBACK_DATE,
-            user_id=user.id,
-            ad_type="reward_video",
-            app_env="prod",
-            revenue_scope="all",
-            granularity="hour",
-        )
-
-        revenue_by_status = {row["status"]: row["revenue_yuan"] for row in result["items"]}
-        assert revenue_by_status == {
-            "closed_early": 0.0,
-            "too_short": 0.0,
-            "capped": 0.1,
-            "granted": 0.1,
-        }
-        assert result["total_impressions"] == 4
-        assert result["total_revenue_yuan"] == 0.2
-        assert len(result["daily"]) == 1
-        assert result["daily"][0]["date"] == PLAYBACK_DATE
-        assert result["daily"][0]["impressions"] == 4
-        assert result["daily"][0]["revenue_yuan"] == 0.2
-        assert sum(row["revenue_yuan"] for row in result["hourly"]) == 0.2
-        assert result["type_stats"]["reward_video"] == {
-            "impressions": 4,
-            "revenue_yuan": 0.2,
-        }
-    finally:
-        db.rollback()
-        db.execute(delete(AdRewardRecord).where(AdRewardRecord.reward_date == PLAYBACK_DATE))
-        db.execute(delete(AdEcpmRecord).where(AdEcpmRecord.report_date == PLAYBACK_DATE))
-        db.execute(delete(User).where(User.phone == phone))
-        db.commit()
-        db.close()
@@ -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",
    }

@@ -222,210 +222,6 @@ def test_stats_compare_count_and_saved(client) -> None:
    assert s2["compare_count"] == 2  # 仍 2(failed 不计)


-def test_records_ordered_flag(client) -> None:
-    """「已下单」店级标记:店名命中该用户 source='compare' 的下单记录才 True。
-
-    覆盖 list_records 只按**本页店名**反查 savings 的写法(原来是把该用户全部下单店名捞回内存
-    再取交集,随下单量线性变慢)——两种写法结果必须一致,故这里按店名逐条断言。
-    """
-    token = _login(client, "13800002010")
-
-    # 两条比价记录:一条海底捞(稍后会有对应下单),一条没下过单的店
-    client.post("/api/v1/compare/record", json=_food_payload("ord-1"), headers=_auth(token))
-    other = _food_payload("ord-2")
-    other["store_name"] = "没下过单的店"
-    client.post("/api/v1/compare/record", json=other, headers=_auth(token))
-
-    # 下单前:两条都不该带「已下单」
-    items = client.get("/api/v1/compare/records", headers=_auth(token)).json()["items"]
-    assert {it["store_name"]: it["ordered"] for it in items} == {
-        "海底捞(朝阳店)": False,
-        "没下过单的店": False,
-    }
-
-    # 对海底捞真实下单一笔(order/report 写 source='compare' 的 savings_record)
-    r = client.post(
-        "/api/v1/order/report",
-        json={
-            "client_event_id": "evt-ordered-flag",
-            "platform": "美团",
-            "platform_package": "com.sankuai.meituan",
-            "pay_channel": "wechat",
-            "compared_price_cents": 12350,
-            "paid_amount_cents": 12350,
-            "shop_name": "海底捞(朝阳店)",
-            "original_price_cents": 12850,
-        },
-        headers=_auth(token),
-    )
-    assert r.status_code == 200, r.text
-
-    # 下单后:只有同店名那条翻成 True,另一条不受影响
-    items = client.get("/api/v1/compare/records", headers=_auth(token)).json()["items"]
-    assert {it["store_name"]: it["ordered"] for it in items} == {
-        "海底捞(朝阳店)": True,
-        "没下过单的店": False,
-    }
-
-    # 别人的下单不该影响本人标记(_ordered_shop_names 按 user_id 过滤)
-    token_b = _login(client, "13800002011")
-    client.post("/api/v1/compare/record", json=_food_payload("ord-b"), headers=_auth(token_b))
-    items_b = client.get("/api/v1/compare/records", headers=_auth(token_b)).json()["items"]
-    assert [it["ordered"] for it in items_b] == [False]
-
-
-def test_records_list_omits_raw_payload(client) -> None:
-    """列表出参不含 raw_payload(仓库层 defer 掉了重型 JSON 列);详情接口照常返回。
-
-    defer 的列一旦被 ORM 实例读到会触发逐行懒加载(N+1),而列表 schema 本就不该带 raw_payload
-    —— 这条同时守住「列表不泄露上报体全量」和「没人不小心把它加回出参」。
-    """
-    token = _login(client, "13800002012")
-    rid = client.post(
-        "/api/v1/compare/record", json=_food_payload("no-raw"), headers=_auth(token)
-    ).json()["id"]
-
-    items = client.get("/api/v1/compare/records", headers=_auth(token)).json()["items"]
-    assert len(items) == 1
-    assert "raw_payload" not in items[0]
-    # 概要字段照常齐全(defer 没误伤列表要用的列)
-    assert items[0]["store_name"] == "海底捞(朝阳店)"
-    assert items[0]["best_platform_id"] == "meituan"
-    assert items[0]["comparison_results"] and items[0]["items"]
-
-    # 详情不 defer:raw_payload 全量还在
-    d = client.get(f"/api/v1/compare/records/{rid}", headers=_auth(token)).json()
-    assert d["raw_payload"]["trace_id"] == "no-raw"
-
-
-def test_records_ordered_filter(client) -> None:
-    """ordered=true 只出「已下单」的记录,且过滤结果自身能翻页。
-
-    这个筛选必须在服务端做:客户端早先是对「已经拉回来的那一页」做 filter,分页之后一页里
-    很可能一条已下单都没有 ——「已下单」tab 就会看着像空的,得手动翻很多页才蹦出一条。
-    """
-    token = _login(client, "13800002013")
-
-    # 3 条「下过单的店」+ 2 条没下过单的店,交错写入,确保过滤不是靠顺序碰巧对上
-    for i in range(3):
-        p = _food_payload(f"of-ordered-{i}")
-        p["store_name"] = "下过单的店"
-        client.post("/api/v1/compare/record", json=p, headers=_auth(token))
-        if i < 2:
-            q = _food_payload(f"of-plain-{i}")
-            q["store_name"] = "没下过单的店"
-            client.post("/api/v1/compare/record", json=q, headers=_auth(token))
-
-    client.post(
-        "/api/v1/order/report",
-        json={
-            "client_event_id": "evt-ordered-filter",
-            "platform": "美团",
-            "platform_package": "com.sankuai.meituan",
-            "pay_channel": "wechat",
-            "compared_price_cents": 12350,
-            "paid_amount_cents": 12350,
-            "shop_name": "下过单的店",
-            "original_price_cents": 12850,
-        },
-        headers=_auth(token),
-    )
-
-    # 不传 ordered:5 条全出(「全部记录」tab 口径不变)
-    assert len(client.get("/api/v1/compare/records", headers=_auth(token)).json()["items"]) == 5
-
-    # ordered=true:只出那 3 条,且每条都自带 ordered=True
-    page = client.get("/api/v1/compare/records?ordered=true", headers=_auth(token)).json()
-    assert [it["store_name"] for it in page["items"]] == ["下过单的店"] * 3
-    assert all(it["ordered"] for it in page["items"])
-    assert page["next_cursor"] is None
-
-    # 游标只在「已下单」集合内走 —— 不会把没下单的记录算进一页的 limit 里
-    p1 = client.get(
-        "/api/v1/compare/records?ordered=true&limit=2", headers=_auth(token)
-    ).json()
-    assert len(p1["items"]) == 2
-    assert p1["next_cursor"] is not None
-    p2 = client.get(
-        f"/api/v1/compare/records?ordered=true&limit=2&cursor={p1['next_cursor']}",
-        headers=_auth(token),
-    ).json()
-    assert [it["store_name"] for it in p2["items"]] == ["下过单的店"]
-    # 两页不重叠,合起来正好 3 条
-    assert len({it["id"] for it in p1["items"] + p2["items"]}) == 3
-
-    # 别人的下单不该让本人记录进「已下单」
-    token_b = _login(client, "13800002014")
-    pb = _food_payload("of-b")
-    pb["store_name"] = "下过单的店"
-    client.post("/api/v1/compare/record", json=pb, headers=_auth(token_b))
-    assert client.get(
-        "/api/v1/compare/records?ordered=true", headers=_auth(token_b)
-    ).json()["items"] == []
-
-
-def test_records_keyword_search(client) -> None:
-    """keyword 按店名 / 菜名模糊搜(忽略大小写),LIKE 通配符只当字面量;搜索结果也能翻页。
-
-    菜名走写路径派生的 product_names 文本列 —— items 是 JSON,SQLite 下中文被 ensure_ascii
-    转义,直接 LIKE 搜不到。
-    """
-    token = _login(client, "13800002015")
-
-    b = _food_payload("kw-b")
-    b["store_name"] = "Pizza Hut"
-    b["items"] = [{"name": "榴莲比萨", "qty": 1}]
-    client.post("/api/v1/compare/record", json=b, headers=_auth(token))
-
-    c = _food_payload("kw-c")
-    c["store_name"] = "100%纯牛肉汉堡"
-    c["items"] = [{"name": "双层牛肉堡", "qty": 1}]
-    client.post("/api/v1/compare/record", json=c, headers=_auth(token))
-
-    # 默认 payload 的店名是「海底捞(朝阳店)」
-    client.post("/api/v1/compare/record", json=_food_payload("kw-a"), headers=_auth(token))
-
-    def _search(kw: str, **extra) -> list[str]:
-        r = client.get(
-            "/api/v1/compare/records",
-            params={"keyword": kw, **extra},
-            headers=_auth(token),
-        )
-        assert r.status_code == 200, r.text
-        return [it["store_name"] for it in r.json()["items"]]
-
-    assert _search("海底捞") == ["海底捞(朝阳店)"]          # 店名命中
-    assert _search("榴莲") == ["Pizza Hut"]                  # 菜名命中(product_names)
-    assert _search("pizza") == ["Pizza Hut"]                 # 忽略大小写
-    assert _search("PIZZA") == ["Pizza Hut"]
-    assert _search("不存在的店") == []                        # 没命中就是空
-    # 通配符只当普通字符:搜 % 不该把整表拉回来,搜 _ 也不该匹配任意单字符
-    assert _search("%") == ["100%纯牛肉汉堡"]
-    assert _search("_") == []
-    # 纯空白等同不传 → 不过滤
-    assert len(_search("   ")) == 3
-
-    # 搜索结果自身可翻页
-    for i in range(3):
-        p = _food_payload(f"kw-page-{i}")
-        p["store_name"] = "连锁烤鱼店"
-        client.post("/api/v1/compare/record", json=p, headers=_auth(token))
-    p1 = client.get(
-        "/api/v1/compare/records",
-        params={"keyword": "烤鱼", "limit": 2},
-        headers=_auth(token),
-    ).json()
-    assert len(p1["items"]) == 2
-    assert p1["next_cursor"] is not None
-    p2 = client.get(
-        "/api/v1/compare/records",
-        params={"keyword": "烤鱼", "limit": 2, "cursor": p1["next_cursor"]},
-        headers=_auth(token),
-    ).json()
-    assert [it["store_name"] for it in p2["items"]] == ["连锁烤鱼店"]
-    assert len({it["id"] for it in p1["items"] + p2["items"]}) == 3
-
-
 def test_requires_auth(client) -> None:
    """不带 token 统一 401。"""
    assert client.post("/api/v1/compare/record", json={"trace_id": "t"}).status_code == 401
@@ -4,28 +4,12 @@ from __future__ import annotations
 from datetime import UTC, date, datetime

 import pytest
-from sqlalchemy.dialects import postgresql

 from app.admin.repositories import queries
 from app.db.session import SessionLocal
 from app.models.comparison import ComparisonRecord


-def test_postgresql_duration_summary_uses_ordered_set_aggregates() -> None:
-    stmt = queries._comparison_duration_aggregate_stmt(
-        [], "success", (0.05, 0.5, 0.95, 0.99)
-    )
-    sql = str(
-        stmt.compile(
-            dialect=postgresql.dialect(),
-            compile_kwargs={"literal_binds": True},
-        )
-    )
-
-    assert sql.count("percentile_cont") == 4
-    assert "comparison_record.status = 'success'" in sql
-
-
 def test_summary_uses_only_success_durations_and_filters_beijing_date() -> None:
    db = SessionLocal()
    try:
@@ -7,12 +7,8 @@ from __future__ import annotations
 from datetime import UTC, date, datetime

 from sqlalchemy import delete, func, select
-from sqlalchemy.dialects import postgresql

-from app.admin.repositories.coupon_data import (
-    _coupon_summary_aggregate_stmt,
-    coupon_data_report,
-)
+from app.admin.repositories.coupon_data import coupon_data_report
 from app.db.session import SessionLocal
 from app.models.coupon_state import CouponSession
 from app.repositories.coupon_state import (
@@ -23,12 +19,7 @@ from app.repositories.coupon_state import (


 def _agg_session(
-    trace: str,
-    platforms,
-    platform_success,
-    *,
-    status: str = "completed",
-    elapsed_ms: int | None = None,
+    trace: str, platforms, platform_success, *, status: str = "completed"
 ) -> CouponSession:
    """构造一条聚合测试用 session(started_date 固定 2020-01-02、app_env=prod,不 commit)。"""
    return CouponSession(
@@ -40,53 +31,9 @@ def _agg_session(
        platform_success=platform_success,
        started_at=datetime(2020, 1, 2, tzinfo=UTC),
        started_date=date(2020, 1, 2),
-        elapsed_ms=elapsed_ms,
    )


-def test_postgresql_coupon_summary_uses_ordered_set_aggregates() -> None:
-    sql = str(
-        _coupon_summary_aggregate_stmt([]).compile(
-            dialect=postgresql.dialect(),
-            compile_kwargs={"literal_binds": True},
-        )
-    )
-
-    assert sql.count("percentile_cont") == 4
-    assert "FILTER (WHERE coupon_session.status = 'completed'" in sql
-
-
-def test_coupon_duration_summary_uses_only_completed_rows() -> None:
-    db = SessionLocal()
-    try:
-        db.add_all([
-            _agg_session("duration-a", [], [], elapsed_ms=1000),
-            _agg_session("duration-b", [], [], elapsed_ms=3000),
-            _agg_session(
-                "duration-failed", [], [], status="failed", elapsed_ms=100_000
-            ),
-        ])
-        db.flush()
-
-        summary = coupon_data_report(
-            db,
-            date_from="2020-01-02",
-            date_to="2020-01-02",
-            app_env="prod",
-        )["summary"]
-
-        assert summary["started_count"] == 3
-        assert summary["completed_count"] == 2
-        assert summary["avg_elapsed_ms"] == 2000
-        assert summary["p5_ms"] == 1100
-        assert summary["p50_ms"] == 2000
-        assert summary["p95_ms"] == 2900
-        assert summary["p99_ms"] == 2980
-    finally:
-        db.rollback()
-        db.close()
-
-
 def _make_session(db, trace_id: str, **kw) -> CouponSession:
    row = CouponSession(
        trace_id=trace_id,
@@ -1,197 +0,0 @@
-"""美团、京东 CPS 每日自动对账 worker。"""
-from __future__ import annotations
-
-import logging
-from datetime import date, datetime, timedelta
-
-import pytest
-
-from app.core import cps_reconcile_worker as worker
-from app.core.config import settings
-from app.core.rewards import CN_TZ
-from app.db.session import SessionLocal
-from app.integrations.meituan import MeituanCpsError
-
-
-def _configure_platforms(monkeypatch) -> None:
-    monkeypatch.setattr(settings, "MT_CPS_APP_KEY", "mt-key")
-    monkeypatch.setattr(settings, "MT_CPS_APP_SECRET", "mt-secret")
-    monkeypatch.setattr(settings, "JD_UNION_APP_KEY", "jd-key")
-    monkeypatch.setattr(settings, "JD_UNION_APP_SECRET", "jd-secret")
-    monkeypatch.setattr(settings, "CPS_AUTO_RECONCILE_LOOKBACK_DAYS", 3)
-
-
-def test_reconcile_once_pulls_meituan_and_jd_by_update_time(monkeypatch) -> None:
-    _configure_platforms(monkeypatch)
-    calls: dict[str, dict] = {}
-
-    def fake_meituan(db, **kwargs):
-        calls["meituan"] = kwargs
-        return {
-            "fetched": 2,
-            "inserted": 1,
-            "updated": 1,
-            "pages": 1,
-            "api_requests": 1,
-        }
-
-    def fake_jd(db, **kwargs):
-        calls["jd"] = kwargs
-        return {
-            "fetched": 3,
-            "inserted": 2,
-            "updated": 1,
-            "pages": 2,
-            "api_requests": 72,
-            "windows": 72,
-        }
-
-    monkeypatch.setattr(worker.cps_repo, "reconcile_orders", fake_meituan)
-    monkeypatch.setattr(worker.cps_repo, "reconcile_jd_orders", fake_jd)
-    monkeypatch.setattr(worker, "_touch_lock", lambda: None)
-    now = datetime(2026, 7, 22, 5, 0, tzinfo=CN_TZ)
-
-    result = worker._reconcile_once(now)
-
-    assert result["errors"] == {}
-    assert result["meituan"]["fetched"] == 2
-    assert result["jd"]["fetched"] == 3
-    assert calls["meituan"]["query_time_type"] == 2
-    assert calls["jd"]["query_time_type"] == 3
-    assert calls["meituan"]["end_time"] == int(now.timestamp())
-    assert calls["meituan"]["start_time"] == int((now - timedelta(days=3)).timestamp())
-    assert calls["jd"]["start_time"] == now - timedelta(days=3)
-    assert calls["jd"]["end_time"] == now
-    assert calls["meituan"]["audit_context"]["run_id"] == result["run_id"]
-    assert calls["jd"]["audit_context"]["run_id"] == result["run_id"]
-    assert result["status"] == "success"
-    assert result["manual_retry_required"] is False
-
-
-def test_meituan_failure_does_not_block_jd(monkeypatch) -> None:
-    _configure_platforms(monkeypatch)
-    jd_called = False
-
-    def fail_meituan(db, **kwargs):
-        raise MeituanCpsError("temporary failure")
-
-    def fake_jd(db, **kwargs):
-        nonlocal jd_called
-        jd_called = True
-        return {"fetched": 1, "inserted": 1, "updated": 0, "pages": 1}
-
-    monkeypatch.setattr(worker.cps_repo, "reconcile_orders", fail_meituan)
-    monkeypatch.setattr(worker.cps_repo, "reconcile_jd_orders", fake_jd)
-    monkeypatch.setattr(worker, "_touch_lock", lambda: None)
-
-    result = worker._reconcile_once(datetime(2026, 7, 22, 5, 0, tzinfo=CN_TZ))
-
-    assert result["errors"]["meituan"] == "temporary failure"
-    assert jd_called is True
-    assert result["jd"]["fetched"] == 1
-    assert result["meituan"]["status"] == "failed"
-    assert result["status"] == "partial_success"
-    assert result["manual_retry_required"] is True
-
-
-def test_should_run_once_after_five_am() -> None:
-    before = datetime(2026, 7, 22, 4, 59, tzinfo=CN_TZ)
-    at_five = datetime(2026, 7, 22, 5, 0, tzinfo=CN_TZ)
-
-    assert worker._should_run(None, before, 5) is False
-    assert worker._should_run(None, at_five, 5) is True
-    assert worker._should_run(date(2026, 7, 22), at_five, 5) is False
-
-
-def test_trigger_and_next_run_are_beijing_time() -> None:
-    before_five = datetime(2026, 7, 22, 4, 30, tzinfo=CN_TZ)
-    after_five = datetime(2026, 7, 22, 6, 0, tzinfo=CN_TZ)
-
-    assert worker._trigger_for_run(before_five, after_five, 5) == "scheduled"
-    assert worker._trigger_for_run(after_five, after_five, 5) == "startup_catchup"
-    assert worker._next_run_at(before_five, 5) == datetime(2026, 7, 22, 5, 0, tzinfo=CN_TZ)
-    assert worker._next_run_at(after_five, 5) == datetime(2026, 7, 23, 5, 0, tzinfo=CN_TZ)
-
-
-def test_reconcile_emits_structured_audit_logs(monkeypatch, caplog) -> None:
-    _configure_platforms(monkeypatch)
-    monkeypatch.setattr(
-        worker.cps_repo,
-        "reconcile_orders",
-        lambda db, **kwargs: {
-            "fetched": 2,
-            "inserted": 1,
-            "updated": 1,
-            "pages": 1,
-            "api_requests": 1,
-        },
-    )
-    monkeypatch.setattr(
-        worker.cps_repo,
-        "reconcile_jd_orders",
-        lambda db, **kwargs: {
-            "fetched": 3,
-            "inserted": 2,
-            "updated": 1,
-            "pages": 1,
-            "api_requests": 72,
-            "windows": 72,
-        },
-    )
-    monkeypatch.setattr(worker, "_touch_lock", lambda: None)
-    monkeypatch.setattr(
-        worker,
-        "_cn_now",
-        lambda: datetime(2026, 7, 22, 5, 1, tzinfo=CN_TZ),
-    )
-
-    with caplog.at_level(logging.INFO, logger=worker.logger.name):
-        result = worker._reconcile_once(
-            datetime(2026, 7, 22, 5, 0, tzinfo=CN_TZ),
-            run_id="audit-run-1",
-            trigger="scheduled",
-        )
-
-    records = {record.event: record for record in caplog.records if hasattr(record, "event")}
-    assert records["cps_reconcile.started"].run_id == "audit-run-1"
-    assert records["cps_reconcile.started"].lookback_days == 3
-    assert records["cps_reconcile.started"].scheduled_date == "2026-07-22"
-    assert records["cps_reconcile.started"].hostname
-    assert records["cps_reconcile.completed"].task_status == "success"
-    assert records["cps_reconcile.completed"].manual_retry_required is False
-    assert records["cps_reconcile.completed"].meituan_result["fetched"] == 2
-    assert records["cps_reconcile.completed"].jd_result["api_requests"] == 72
-    assert result["next_run_at"] == "2026-07-23T05:00:00+08:00"
-
-
-def test_jd_request_failure_logs_exact_window(monkeypatch, caplog) -> None:
-    start = datetime(2026, 7, 20, 5, 0, tzinfo=CN_TZ)
-
-    def fail_query(**kwargs):
-        raise RuntimeError("upstream timeout")
-
-    monkeypatch.setattr(worker.cps_repo.jd_union, "query_order_rows", fail_query)
-    with SessionLocal() as db, caplog.at_level(logging.ERROR, logger=worker.logger.name):
-        with pytest.raises(RuntimeError, match="upstream timeout"):
-            worker.cps_repo.reconcile_jd_orders(
-                db,
-                start_time=start,
-                end_time=start + timedelta(hours=2),
-                audit_context={"run_id": "audit-run-2", "trigger": "scheduled"},
-            )
-
-    record = next(
-        item
-        for item in caplog.records
-        if getattr(item, "event", "") == "cps_reconcile.request_failed"
-    )
-    assert record.run_id == "audit-run-2"
-    assert record.platform == "jd"
-    assert record.failed_window_start == "2026-07-20T05:00:00+08:00"
-    assert record.failed_window_end == "2026-07-20T06:00:00+08:00"
-    assert record.failed_page == 1
-
-
-def test_start_worker_respects_auto_switch(monkeypatch) -> None:
-    monkeypatch.setattr(settings, "CPS_AUTO_RECONCILE_ENABLED", False)
-    assert worker.start_cps_reconcile_worker() is None
@@ -1,73 +0,0 @@
-"""SafeRotatingFileHandler:Windows 轮转不被外部句柄卡死(WinError 32)。
-
-stdlib RotatingFileHandler 靠 rename 活动文件轮转;Windows 上只要有第二个句柄(admin
-第二进程、残留 --reload worker、IDE 索引、杀软)开着它, rename 就 WinError 32、轮转永久
-卡死。Safe 版在 Windows 改走 copytruncate(拷贝→就地清空, 从不 rename 活动文件)。
-
-这些用例用 monkeypatch 把 os.name 强制成 "nt", 使 copytruncate 分支在任何 OS 的 CI 上都被
-覆盖;在真实 Windows 上则天然命中。
-"""
-from __future__ import annotations
-
-import logging
-from pathlib import Path
-
-from app.core.logging import SafeRotatingFileHandler
-
-
-def _emit(handler: logging.Handler, msg: str) -> None:
-    handler.emit(logging.LogRecord("t", logging.INFO, __file__, 0, msg, (), None))
-
-
-def test_rollover_survives_second_open_handle(tmp_path: Path, monkeypatch) -> None:
-    """第二个句柄开着活动文件时轮转:不抛异常, 且确实转了(生成 .1、活动文件就地清空)。"""
-    monkeypatch.setattr("app.core.logging.os.name", "nt")
-
-    log_file = tmp_path / "app-server.log"
-    # maxBytes 放大, 避免 emit 期间自动轮转干扰;本用例手动触发 doRollover。
-    handler = SafeRotatingFileHandler(
-        str(log_file), maxBytes=10**9, backupCount=3, encoding="utf-8",
-    )
-    handler.setFormatter(logging.Formatter("%(message)s"))
-    try:
-        for i in range(20):
-            _emit(handler, f"line-{i:03d}")
-        handler.flush()
-        before = log_file.stat().st_size
-        assert before > 0
-
-        # 正是 Windows 上 rename 失败的条件:另一个句柄开着活动文件。
-        with open(log_file, "a", encoding="utf-8"):
-            handler.doRollover()  # 不应抛 PermissionError / WinError 32
-
-        backup = tmp_path / "app-server.log.1"
-        assert backup.exists()
-        assert backup.stat().st_size == before   # 轮转前内容完整进了备份
-        assert log_file.stat().st_size == 0      # 活动文件就地清空(不是 rename)
-
-        # 句柄没被 rename 破坏, 仍能继续写。
-        _emit(handler, "after-rollover")
-        handler.flush()
-        assert log_file.stat().st_size > 0
-    finally:
-        handler.close()
-
-
-def test_backups_shift_and_capped(tmp_path: Path, monkeypatch) -> None:
-    """多次轮转:.1/.2 逐级腾挪, 超过 backupCount 的丢弃(不出现 .3)。"""
-    monkeypatch.setattr("app.core.logging.os.name", "nt")
-
-    log_file = tmp_path / "app-server.log"
-    handler = SafeRotatingFileHandler(
-        str(log_file), maxBytes=10**9, backupCount=2, encoding="utf-8",
-    )
-    handler.setFormatter(logging.Formatter("%(message)s"))
-    try:
-        for _ in range(4):
-            _emit(handler, "x" * 50)
-            handler.doRollover()
-        assert (tmp_path / "app-server.log.1").exists()
-        assert (tmp_path / "app-server.log.2").exists()
-        assert not (tmp_path / "app-server.log.3").exists()
-    finally:
-        handler.close()
@@ -9,7 +9,7 @@ from sqlalchemy import select

 from app.db.session import SessionLocal
 from app.models.user import User
-from app.models.wallet import CoinAccount, WithdrawOrder, WechatTransferAuthorization
+from app.models.wallet import CoinAccount, WithdrawOrder
 from app.repositories import wallet as crud_wallet


@@ -426,62 +426,3 @@ def test_withdraw_reject_refunds(client, monkeypatch) -> None:
    r = client.get("/api/v1/wallet/withdraw/status", params={"out_bill_no": bill}, headers=_auth(token))
    assert r.json()["status"] == "rejected"
    assert r.json()["fail_reason"] == "测试拒绝"
-
-
-# ===== §fix 免确认授权 enabled 判定收严:active+authorization_id 非空才算已授权 =====
-
-def _seed_transfer_auth(phone: str, state: str, authorization_id: str | None) -> None:
-    """直接在库里写/改该用户的免确认授权记录(绕过微信,用于判定测试)。"""
-    db = SessionLocal()
-    try:
-        user = db.execute(select(User).where(User.phone == phone)).scalar_one()
-        auth = db.get(WechatTransferAuthorization, user.id)
-        if auth is None:
-            auth = WechatTransferAuthorization(
-                user_id=user.id, openid=user.wechat_openid or "openid_test_abc",
-                out_authorization_no=f"oan_{user.id}",
-            )
-            db.add(auth)
-        auth.state = state
-        auth.authorization_id = authorization_id
-        db.commit()
-    finally:
-        db.close()
-
-
-def test_withdraw_info_auth_enabled_requires_authorization_id(client, monkeypatch) -> None:
-    _patch_userinfo(monkeypatch)
-    token = _login(client, "13800002051")
-    # 触发建号 + 绑定微信(withdraw-info 读 openid)
-    client.get("/api/v1/wallet/withdraw-info", headers=_auth(token))
-    client.post("/api/v1/wallet/bind-wechat", json={"code": "c1"}, headers=_auth(token))
-
-    # active 但无 authorization_id → 视为未授权
-    _seed_transfer_auth("13800002051", "active", None)
-    r = client.get("/api/v1/wallet/withdraw-info", headers=_auth(token))
-    assert r.status_code == 200, r.text
-    assert r.json()["transfer_auth_enabled"] is False
-
-    # active 且有 authorization_id → 已授权
-    _seed_transfer_auth("13800002051", "active", "wx_auth_123")
-    r = client.get("/api/v1/wallet/withdraw-info", headers=_auth(token))
-    assert r.status_code == 200, r.text
-    assert r.json()["transfer_auth_enabled"] is True
-
-
-def test_transfer_auth_status_requires_authorization_id(client, monkeypatch) -> None:
-    _patch_userinfo(monkeypatch)
-    token = _login(client, "13800002052")
-    client.get("/api/v1/wallet/withdraw-info", headers=_auth(token))
-    client.post("/api/v1/wallet/bind-wechat", json={"code": "c2"}, headers=_auth(token))
-
-    _seed_transfer_auth("13800002052", "active", None)
-    r = client.get("/api/v1/wallet/transfer-auth/status", headers=_auth(token))
-    assert r.status_code == 200, r.text
-    assert r.json()["state"] == "active"
-    assert r.json()["enabled"] is False
-
-    _seed_transfer_auth("13800002052", "active", "wx_auth_456")
-    r = client.get("/api/v1/wallet/transfer-auth/status", headers=_auth(token))
-    assert r.status_code == 200, r.text
-    assert r.json()["enabled"] is True