处理冲突

# Conflicts:
#	app/admin/repositories/ad_revenue.py
#	app/admin/repositories/stats.py

alembic/versions/11c44afbea58_analytics_selfstat_tables.py

@@ -1,72 +0,0 @@
-"""analytics_selfstat tables
-
-Revision ID: 11c44afbea58
-Revises: admin_user_plain_password
-Create Date: 2026-07-08 16:32:49.351817
-
-"""
-from collections.abc import Sequence
-
-import sqlalchemy as sa
-
-from alembic import op
-
-
-# revision identifiers, used by Alembic.
-revision: str = '11c44afbea58'
-down_revision: str | Sequence[str] | None = 'admin_user_plain_password'
-branch_labels: str | Sequence[str] | None = None
-depends_on: str | Sequence[str] | None = None
-
-
-def upgrade() -> None:
-    op.create_table(
-        'analytics_selfstat',
-        sa.Column('id', sa.Integer(), autoincrement=True, nullable=False),
-        sa.Column('device_id', sa.String(length=64), nullable=False),
-        sa.Column('epoch_id', sa.String(length=64), nullable=False),
-        sa.Column('app_ver', sa.String(length=32), nullable=True),
-        sa.Column('oem', sa.String(length=32), nullable=True),
-        sa.Column('os', sa.String(length=32), nullable=True),
-        sa.Column('batches_attempted', sa.BigInteger(), nullable=False, server_default='0'),
-        sa.Column('batches_ok', sa.BigInteger(), nullable=False, server_default='0'),
-        sa.Column('batches_fail', sa.BigInteger(), nullable=False, server_default='0'),
-        sa.Column('retries', sa.BigInteger(), nullable=False, server_default='0'),
-        sa.Column('queue_depth', sa.Integer(), nullable=False, server_default='0'),
-        sa.Column('sent_at', sa.BigInteger(), nullable=True),
-        sa.Column('created_at', sa.DateTime(timezone=True),
-                  server_default=sa.text('(CURRENT_TIMESTAMP)'), nullable=False),
-        sa.PrimaryKeyConstraint('id'),
-    )
-    with op.batch_alter_table('analytics_selfstat', schema=None) as batch_op:
-        batch_op.create_index(batch_op.f('ix_analytics_selfstat_created_at'), ['created_at'], unique=False)
-        batch_op.create_index(batch_op.f('ix_analytics_selfstat_device_id'), ['device_id'], unique=False)
-        batch_op.create_index(batch_op.f('ix_analytics_selfstat_epoch_id'), ['epoch_id'], unique=False)
-
-    op.create_table(
-        'analytics_selfstat_event',
-        sa.Column('id', sa.Integer(), autoincrement=True, nullable=False),
-        sa.Column('snapshot_id', sa.Integer(), nullable=False),
-        sa.Column('event', sa.String(length=64), nullable=False),
-        sa.Column('attempted', sa.BigInteger(), nullable=False, server_default='0'),
-        sa.Column('drop_capture', sa.BigInteger(), nullable=False, server_default='0'),
-        sa.Column('delivered', sa.BigInteger(), nullable=False, server_default='0'),
-        sa.Column('drop_undelivered', sa.BigInteger(), nullable=False, server_default='0'),
-        sa.ForeignKeyConstraint(['snapshot_id'], ['analytics_selfstat.id'], ),
-        sa.PrimaryKeyConstraint('id'),
-    )
-    with op.batch_alter_table('analytics_selfstat_event', schema=None) as batch_op:
-        batch_op.create_index(batch_op.f('ix_analytics_selfstat_event_event'), ['event'], unique=False)
-        batch_op.create_index(batch_op.f('ix_analytics_selfstat_event_snapshot_id'), ['snapshot_id'], unique=False)
-
-
-def downgrade() -> None:
-    with op.batch_alter_table('analytics_selfstat_event', schema=None) as batch_op:
-        batch_op.drop_index(batch_op.f('ix_analytics_selfstat_event_snapshot_id'))
-        batch_op.drop_index(batch_op.f('ix_analytics_selfstat_event_event'))
-    op.drop_table('analytics_selfstat_event')
-    with op.batch_alter_table('analytics_selfstat', schema=None) as batch_op:
-        batch_op.drop_index(batch_op.f('ix_analytics_selfstat_epoch_id'))
-        batch_op.drop_index(batch_op.f('ix_analytics_selfstat_device_id'))
-        batch_op.drop_index(batch_op.f('ix_analytics_selfstat_created_at'))
-    op.drop_table('analytics_selfstat')

alembic/versions/admin_user_pages_override.py

@@ -1,35 +0,0 @@
-"""admin_user 加 pages_override 列(「自定义」权限:按人存专属可见页 key 列表)
-
-权限管理页新增「自定义」角色:选它时该成员的可见页不跟随任何共享角色,而由逐页勾选决定,
-存这个人专属的一份页 key 列表。仅 role == "custom" 时有效;普通角色为 None(可见页跟随角色)。
-PG 用 JSONB,SQLite 退化为通用 JSON(同 admin_audit_log.detail / comparison_record.raw_payload)。
-
-Revision ID: admin_user_pages_override
-Revises: admin_user_plain_password
-Create Date: 2026-07-08 00:00:00.000000
-"""
-
-from collections.abc import Sequence
-
-import sqlalchemy as sa
-from sqlalchemy.dialects.postgresql import JSONB
-
-from alembic import op
-
-revision: str = "admin_user_pages_override"
-down_revision: str | Sequence[str] | None = "admin_user_plain_password"
-branch_labels: str | Sequence[str] | None = None
-depends_on: str | Sequence[str] | None = None
-
-# 与 app/models/admin.py 的 _JSON 一致:PG JSONB / 其它 JSON
-_JSON = sa.JSON().with_variant(JSONB(), "postgresql")
-
-
-def upgrade() -> None:
-    with op.batch_alter_table("admin_user", schema=None) as batch_op:
-        batch_op.add_column(sa.Column("pages_override", _JSON, nullable=True))
-
-
-def downgrade() -> None:
-    with op.batch_alter_table("admin_user", schema=None) as batch_op:
-        batch_op.drop_column("pages_override")

app/admin/main.py

@@ -26,7 +26,6 @@ from app.admin.routers.cps import router as cps_router
 from app.admin.routers.dashboard import router as dashboard_router
 from app.admin.routers.device_liveness import router as device_liveness_router
 from app.admin.routers.ops_stat_config import router as ops_stat_config_router
-from app.admin.routers.analytics_health import router as analytics_health_router
 from app.admin.routers.event_logs import router as event_logs_router
 from app.admin.routers.feedback import router as feedback_router
 from app.admin.routers.feedback_qr import router as feedback_qr_router
@@ -98,7 +97,6 @@ admin_app.include_router(withdraw_router)
 admin_app.include_router(price_report_router)
 admin_app.include_router(feedback_router)
 admin_app.include_router(event_logs_router)
-admin_app.include_router(analytics_health_router)
 admin_app.include_router(feedback_qr_router)
 admin_app.include_router(admins_router)
 admin_app.include_router(roles_router)

app/admin/permissions.py

@@ -9,9 +9,6 @@ super_admin 为内建全权角色,恒可见全部页(effective_pages 特判)。
 from __future__ import annotations
 
 SUPER_ADMIN_ROLE = "super_admin"
-# 「自定义」哨兵角色:不是 admin_role 表里的行,而是标记「这个人的可见页由 pages_override 决定」。
-# admin_user.role == CUSTOM_ROLE 时,有效可见页取 admin_user.pages_override(见 auth._admin_out_with_pages)。
-CUSTOM_ROLE = "custom"
 
 # 分组镜像前端导航(app/(main)/layout.tsx 的 NAV_GROUPS);key = 路由一级
 PERMISSION_CATALOG: list[dict] = [

app/admin/repositories/admin_user.py

@@ -26,17 +26,14 @@ def create_admin(
     password: str,
     role: str = "operator",
     plain_password: str | None = None,
-    pages_override: list[str] | None = None,
 ) -> AdminUser:
     """建管理员。plain_password 非空则额外留存明文(后台 UI 建的账号传,供权限管理页复看);
-    脚本/起后台建账号不传(留 None → 前端不显示密码)。
-    pages_override:role=="custom" 时传专属可见页 key 列表;普通角色为 None。"""
+    脚本/起后台建账号不传(留 None → 前端不显示密码)。"""
     admin = AdminUser(
         username=username,
         password_hash=hash_password(password),
         role=role,
         plain_password=plain_password,
-        pages_override=pages_override,
     )
     db.add(admin)
     db.commit()

app/admin/repositories/analytics_health.py

@@ -1,146 +0,0 @@
-"""埋点健康度聚合(埋点成功率 / 上报成功率)。
-
-只存原始累计快照,查询时在 Python 侧差分聚合(admin 低频、量级小,跨 PG/SQLite 无方言坑;
-与 cps.py / coupon_data.py 同款约定)。差分按 (device_id, epoch_id, event) 分区、created_at
-升序,相邻做差、负值夹 0;每增量按其快照 created_at 归入北京天桶。
-"""
-from __future__ import annotations
-
-from collections import defaultdict
-from datetime import UTC, datetime
-
-from sqlalchemy import func, select
-from sqlalchemy.orm import Session
-
-from app.core import rewards
-from app.models.analytics_selfstat import AnalyticsSelfStat as H
-from app.models.analytics_selfstat import AnalyticsSelfStatEvent as E
-
-_COUNTS = ("attempted", "drop_capture", "delivered", "drop_undelivered")
-
-
-def diff_snapshots(rows: list[dict]) -> list[dict]:
-    """累计快照行 → 每快照增量行(纯逻辑)。
-
-    rows 每行含 device_id/epoch_id/event/created_at/app_ver/oem/os + 四个累计计数。
-    返回每行含 dims + created_at + 四个增量 d_*(分区首行增量=累计值;负值夹 0)。
-    """
-    parts: dict[tuple, list[dict]] = defaultdict(list)
-    for r in rows:
-        parts[(r["device_id"], r["epoch_id"], r["event"])].append(r)
-
-    out: list[dict] = []
-    for group in parts.values():
-        group.sort(key=lambda r: (r["created_at"], r.get("id", 0)))
-        prev = {k: 0 for k in _COUNTS}
-        for r in group:
-            deltas = {f"d_{k}": max(0, int(r[k]) - prev[k]) for k in _COUNTS}
-            out.append({
-                "device_id": r["device_id"], "epoch_id": r["epoch_id"], "event": r["event"],
-                "created_at": r["created_at"], "app_ver": r["app_ver"],
-                "oem": r["oem"], "os": r["os"], **deltas,
-            })
-            prev = {k: int(r[k]) for k in _COUNTS}
-    return out
-
-
-def _cn_day(dt: datetime) -> str:
-    """created_at(UTC 口径)→ 北京日期字符串 YYYY-MM-DD。naive 当 UTC,tz-aware 直接换算。"""
-    if dt.tzinfo is None:
-        dt = dt.replace(tzinfo=UTC)
-    return dt.astimezone(rewards.CN_TZ).date().isoformat()
-
-
-def _rates(sums: dict) -> dict:
-    """由四个增量和派生两段率(分母 0 → None)。"""
-    persisted_denom = sums["attempted"]
-    report_denom = sums["delivered"] + sums["drop_undelivered"]
-    return {
-        **sums,
-        "track_success_rate": (
-            (sums["attempted"] - sums["drop_capture"]) / persisted_denom
-            if persisted_denom else None
-        ),
-        "report_success_rate": (
-            sums["delivered"] / report_denom if report_denom else None
-        ),
-    }
-
-
-def _sum_deltas(deltas: list[dict]) -> dict:
-    return {k: sum(d[f"d_{k}"] for d in deltas) for k in _COUNTS}
-
-
-def _fetch_rows(db: Session, date_from: datetime, date_to: datetime) -> list[dict]:
-    """取 [from, to) 区间行 + 每分区在 from 左侧的最后一条基线行(供第一条区间增量做差)。"""
-    cols = (
-        H.id, H.device_id, H.epoch_id, E.event, H.created_at,
-        H.app_ver, H.oem, H.os,
-        E.attempted, E.drop_capture, E.delivered, E.drop_undelivered,
-    )
-    in_range = db.execute(
-        select(*cols).join(E, E.snapshot_id == H.id)
-        .where(H.created_at >= date_from, H.created_at < date_to)
-    ).mappings().all()
-
-    # 注:基线子查询无下界扫 from 左侧全量(spec §7 已接受的取舍;量级变大再上物化 rollup)。
-    # 用 max(id) 而非 max(created_at) 选"最新一条":id 严格单调,避免 SQLite 秒级时间戳撞车时选歧义。
-    sub = (
-        select(H.device_id, H.epoch_id, E.event, func.max(H.id).label("max_id"))
-        .join(E, E.snapshot_id == H.id)
-        .where(H.created_at < date_from)
-        .group_by(H.device_id, H.epoch_id, E.event)
-        .subquery()
-    )
-    baseline = db.execute(
-        select(*cols).join(E, E.snapshot_id == H.id).join(
-            sub, sub.c.max_id == H.id
-        )
-    ).mappings().all()
-
-    return [dict(r) for r in list(baseline) + list(in_range)]
-
-
-def _in_range_deltas(db: Session, date_from: datetime, date_to: datetime) -> list[dict]:
-    """差分后只保留 created_at ∈ [from, to) 的增量(基线行被差分用后丢弃)。
-
-    Python 侧过滤需对齐 tz 口径:SQLite 返回 naive UTC,PG 返回 aware UTC。
-    统一转成 naive UTC 再比较,兼容两种后端。
-    """
-    def _to_naive_utc(dt: datetime) -> datetime:
-        if dt.tzinfo is not None:
-            return dt.astimezone(UTC).replace(tzinfo=None)
-        return dt
-
-    from_naive = _to_naive_utc(date_from)
-    to_naive = _to_naive_utc(date_to)
-    deltas = diff_snapshots(_fetch_rows(db, date_from, date_to))
-    return [d for d in deltas if from_naive <= _to_naive_utc(d["created_at"]) < to_naive]
-
-
-def overview(db: Session, date_from: datetime, date_to: datetime) -> dict:
-    deltas = _in_range_deltas(db, date_from, date_to)
-    return _rates(_sum_deltas(deltas))
-
-
-def trend(db: Session, date_from: datetime, date_to: datetime) -> list[dict]:
-    deltas = _in_range_deltas(db, date_from, date_to)
-    by_day: dict[str, list[dict]] = defaultdict(list)
-    for d in deltas:
-        by_day[_cn_day(d["created_at"])].append(d)
-    return [
-        {"day": day, **_rates(_sum_deltas(items))}
-        for day, items in sorted(by_day.items())
-    ]
-
-
-def breakdown(db: Session, date_from: datetime, date_to: datetime, dim: str) -> list[dict]:
-    if dim not in ("event", "app_ver", "oem"):
-        raise ValueError(f"invalid dim: {dim!r}")
-    deltas = _in_range_deltas(db, date_from, date_to)
-    by_key: dict[str, list[dict]] = defaultdict(list)
-    for d in deltas:
-        by_key[d.get(dim) or "(unknown)"].append(d)
-    rows = [{"key": key, **_rates(_sum_deltas(items))} for key, items in by_key.items()]
-    rows.sort(key=lambda r: (r["report_success_rate"] is None, r["report_success_rate"] or 0.0))
-    return rows

app/admin/repositories/queries.py

@@ -25,13 +25,7 @@ from app.models.feedback import Feedback
 from app.models.onboarding import OnboardingCompletion
 from app.models.price_report import PriceReport
 from app.models.user import User
-from app.models.wallet import (
-    CashTransaction,
-    CoinAccount,
-    CoinTransaction,
-    InviteCashTransaction,
-    WithdrawOrder,
-)
+from app.models.wallet import CashTransaction, CoinAccount, CoinTransaction, WithdrawOrder
 
 # 「最近活跃」计入的行为事件(与大盘 DAU/留存活跃口径一致:开始比价 + 开始领券)
 _ACTIVE_EVENTS = (COMPARE_START_EVENT, COUPON_START_EVENT)
@@ -852,19 +846,26 @@ def withdraw_risk_flags(
     return flags, score
 
 
-def _check_withdraw_ledger_side(
-    orders: list[WithdrawOrder], txns: list, *, withdraw_biz: str, refund_biz: str
-) -> dict:
-    """对某一本账(普通现金 / 邀请奖励金)做提现单 ↔ 流水的交叉校验。
+def withdraw_ledger_check(db: Session) -> dict:
+    cash_balance_total = int(
+        db.execute(select(func.coalesce(func.sum(CoinAccount.cash_balance_cents), 0))).scalar_one()
+    )
+    cash_txn_total = int(
+        db.execute(select(func.coalesce(func.sum(CashTransaction.amount_cents), 0))).scalar_one()
+    )
 
-    orders 已按 source 过滤到本账;txns 是本账流水表里 withdraw_biz/refund_biz 两类流水。
-    规则:每单发起应有一条扣款流水(ref_id=out_bill_no);失败/拒绝单应有且仅一条退款流水;
-    非退款终态不应出现退款流水。四个计数全为 0 即本账自洽。
-    """
-    withdraw_refs = {txn.ref_id for txn in txns if txn.biz_type == withdraw_biz}
+    orders = list(db.execute(select(WithdrawOrder)).scalars().all())
+    cash_txns = list(
+        db.execute(
+            select(CashTransaction).where(
+                CashTransaction.biz_type.in_(("withdraw", "withdraw_refund"))
+            )
+        ).scalars().all()
+    )
+    withdraw_refs = {txn.ref_id for txn in cash_txns if txn.biz_type == "withdraw"}
     refund_counts: dict[str, int] = {}
-    for txn in txns:
-        if txn.biz_type == refund_biz and txn.ref_id:
+    for txn in cash_txns:
+        if txn.biz_type == "withdraw_refund" and txn.ref_id:
             refund_counts[txn.ref_id] = refund_counts.get(txn.ref_id, 0) + 1
 
     missing_withdraw = 0
@@ -879,95 +880,24 @@ def _check_withdraw_ledger_side(
         if has_refund and order.status not in {"failed", "rejected"}:
             refund_on_non_terminal += 1
 
-    return {
-        "missing_withdraw": missing_withdraw,
-        "missing_refund": missing_refund,
-        "duplicate_refund": sum(1 for count in refund_counts.values() if count > 1),
-        "refund_on_non_terminal": refund_on_non_terminal,
-    }
-
-
-def withdraw_ledger_check(db: Session) -> dict:
-    """现金账本校验:两本物理隔离的账各自对账(产品红线:coin_cash / invite_cash 不串)。
-
-    普通现金:CoinAccount.cash_balance_cents ↔ cash_transaction(withdraw/withdraw_refund);
-    邀请奖励金:CoinAccount.invite_cash_balance_cents ↔ invite_cash_transaction
-              (invite_withdraw/invite_withdraw_refund)。
-    提现单按 source 分流到对应账核对——邀请提现的流水写在 invite_cash_transaction 表,
-    绝不能拿去和普通现金流水比(否则每笔邀请提现单都会被误报「缺扣款/缺退款流水」)。
-    分流口径与 create_withdraw 一致:仅 source==invite_cash 走邀请账,其余(含历史空值)归普通现金。
-    """
-    orders = list(db.execute(select(WithdrawOrder)).scalars().all())
-    coin_orders = [o for o in orders if o.source != "invite_cash"]
-    invite_orders = [o for o in orders if o.source == "invite_cash"]
-
-    # —— 普通现金账(coin_cash) ——
-    cash_balance_total = int(
-        db.execute(select(func.coalesce(func.sum(CoinAccount.cash_balance_cents), 0))).scalar_one()
-    )
-    cash_txn_total = int(
-        db.execute(select(func.coalesce(func.sum(CashTransaction.amount_cents), 0))).scalar_one()
-    )
-    cash_txns = list(
-        db.execute(
-            select(CashTransaction).where(
-                CashTransaction.biz_type.in_(("withdraw", "withdraw_refund"))
-            )
-        ).scalars().all()
-    )
-    coin = _check_withdraw_ledger_side(
-        coin_orders, cash_txns, withdraw_biz="withdraw", refund_biz="withdraw_refund"
-    )
-    cash_diff = cash_balance_total - cash_txn_total
-
-    # —— 邀请奖励金账(invite_cash,独立账户 + 独立流水表) ——
-    invite_balance_total = int(
-        db.execute(
-            select(func.coalesce(func.sum(CoinAccount.invite_cash_balance_cents), 0))
-        ).scalar_one()
-    )
-    invite_txn_total = int(
-        db.execute(
-            select(func.coalesce(func.sum(InviteCashTransaction.amount_cents), 0))
-        ).scalar_one()
-    )
-    invite_txns = list(
-        db.execute(
-            select(InviteCashTransaction).where(
-                InviteCashTransaction.biz_type.in_(("invite_withdraw", "invite_withdraw_refund"))
-            )
-        ).scalars().all()
-    )
-    invite = _check_withdraw_ledger_side(
-        invite_orders, invite_txns,
-        withdraw_biz="invite_withdraw", refund_biz="invite_withdraw_refund",
-    )
-    invite_diff = invite_balance_total - invite_txn_total
-
+    duplicate_refund = sum(1 for count in refund_counts.values() if count > 1)
+    diff = cash_balance_total - cash_txn_total
     ok = (
-        cash_diff == 0
-        and invite_diff == 0
-        and all(v == 0 for v in coin.values())
-        and all(v == 0 for v in invite.values())
+        diff == 0
+        and missing_withdraw == 0
+        and missing_refund == 0
+        and duplicate_refund == 0
+        and refund_on_non_terminal == 0
     )
     return {
         "ok": ok,
-        # 普通现金账(coin_cash:金币兑换的现金)
         "cash_balance_total_cents": cash_balance_total,
         "cash_transaction_total_cents": cash_txn_total,
-        "balance_diff_cents": cash_diff,
-        "missing_withdraw_txn_count": coin["missing_withdraw"],
-        "missing_refund_txn_count": coin["missing_refund"],
-        "duplicate_refund_txn_count": coin["duplicate_refund"],
-        "refund_txn_on_non_terminal_count": coin["refund_on_non_terminal"],
-        # 邀请奖励金账(invite_cash:与普通现金物理隔离,各自对账)
-        "invite_cash_balance_total_cents": invite_balance_total,
-        "invite_cash_transaction_total_cents": invite_txn_total,
-        "invite_balance_diff_cents": invite_diff,
-        "invite_missing_withdraw_txn_count": invite["missing_withdraw"],
-        "invite_missing_refund_txn_count": invite["missing_refund"],
-        "invite_duplicate_refund_txn_count": invite["duplicate_refund"],
-        "invite_refund_txn_on_non_terminal_count": invite["refund_on_non_terminal"],
+        "balance_diff_cents": diff,
+        "missing_withdraw_txn_count": missing_withdraw,
+        "missing_refund_txn_count": missing_refund,
+        "duplicate_refund_txn_count": duplicate_refund,
+        "refund_txn_on_non_terminal_count": refund_on_non_terminal,
     }
 
 

app/admin/repositories/stats.py

@@ -30,17 +30,11 @@ from app.models.user import User
 from app.models.wallet import CoinTransaction, WithdrawOrder
 
 _BEIJING = timezone(timedelta(hours=8))
-REWARD_VIDEO_BIZ_TYPES = ("reward_video", "ad_reward")
-# 领券/比价奖励金币的真实来源是信息流广告发奖(ad_feed_reward_record,按 feed_scene 分场景);
-# coin_transaction 里只有扁平的 feed_ad_reward、biz_type 不分 coupon/comparison,故这俩桶历史从未
-# 被写入,仅留作未来兜底,实际金额在下方按 feed_scene 汇总 ad_feed_reward_record 得出。reward_video/
-# ad_reward 是激励视频,单独成桶、不再混进领券奖励(历史误并会把激励视频金币双计进领券)。
-COUPON_REWARD_BIZ_TYPES = ("coupon", "coupon_reward")
+COUPON_REWARD_BIZ_TYPES = ("reward_video", "ad_reward", "coupon", "coupon_reward")
 COMPARISON_REWARD_BIZ_TYPES = ("comparison", "compare_reward", "comparison_reward")
 EXCLUDED_REWARD_BIZ_TYPES = ("invite_inviter", "invite_invitee", "admin_grant")
 UNCLASSIFIED_FEED_BIZ_TYPES = ("feed_ad_reward",)
 REGULAR_TASK_EXCLUDED_BIZ_TYPES = (
-    *REWARD_VIDEO_BIZ_TYPES,
     *COUPON_REWARD_BIZ_TYPES,
     *COMPARISON_REWARD_BIZ_TYPES,
     *EXCLUDED_REWARD_BIZ_TYPES,
@@ -368,7 +362,7 @@ def dashboard_overview(
     period_reward_video_coin_total = _sum(
         CoinTransaction.amount,
         *period_coin_conds,
-        CoinTransaction.biz_type.in_(REWARD_VIDEO_BIZ_TYPES),
+        CoinTransaction.biz_type.in_(("reward_video", "ad_reward")),
     )
     period_feed_ad_coin_total = _sum(
         CoinTransaction.amount,
@@ -390,29 +384,15 @@ def dashboard_overview(
         *period_coin_conds,
         CoinTransaction.biz_type.like("task_%"),
     )
-    # 领券/比价奖励金币 = biz_type 桶(历史空,兜底)+ 该场景信息流广告实发金币
-    # (ad_feed_reward_record.feed_scene,granted;reward_date 是北京日期串,与 period 同自然日窗口)。
     period_coupon_reward_coin_total = _sum(
         CoinTransaction.amount,
         *period_coin_conds,
         CoinTransaction.biz_type.in_(COUPON_REWARD_BIZ_TYPES),
-    ) + _sum(
-        AdFeedRewardRecord.coin,
-        AdFeedRewardRecord.status == "granted",
-        AdFeedRewardRecord.feed_scene == "coupon",
-        AdFeedRewardRecord.reward_date >= period_from.isoformat(),
-        AdFeedRewardRecord.reward_date <= period_to.isoformat(),
     )
     period_comparison_reward_coin_total = _sum(
         CoinTransaction.amount,
         *period_coin_conds,
         CoinTransaction.biz_type.in_(COMPARISON_REWARD_BIZ_TYPES),
-    ) + _sum(
-        AdFeedRewardRecord.coin,
-        AdFeedRewardRecord.status == "granted",
-        AdFeedRewardRecord.feed_scene == "comparison",
-        AdFeedRewardRecord.reward_date >= period_from.isoformat(),
-        AdFeedRewardRecord.reward_date <= period_to.isoformat(),
     )
     period_regular_task_coin_total = _sum(
         CoinTransaction.amount,
@@ -563,7 +543,7 @@ def dashboard_overview(
             "reward_video_coin_total": _sum(
                 CoinTransaction.amount,
                 CoinTransaction.amount > 0,
-                CoinTransaction.biz_type.in_(REWARD_VIDEO_BIZ_TYPES),
+                CoinTransaction.biz_type.in_(("reward_video", "ad_reward")),
             ),
             "reward_video_watch_count": _count(
                 AdRewardRecord,

app/admin/routers/admins.py

@@ -5,7 +5,7 @@ from fastapi import APIRouter, Depends, HTTPException, Request
 
 from app.admin.audit import write_audit
 from app.admin.deps import AdminDb, CurrentAdmin, get_client_ip, require_role
-from app.admin.permissions import CUSTOM_ROLE, SUPER_ADMIN_ROLE, sanitize_pages
+from app.admin.permissions import SUPER_ADMIN_ROLE
 from app.admin.repositories import admin_role as role_repo
 from app.admin.repositories import admin_user as admin_repo
 from app.admin.schemas.admin import AdminCreateRequest, AdminUpdateRequest
@@ -26,22 +26,14 @@ def _active_super_count(db: AdminDb) -> int:
 
 
 def _validate_role(db: AdminDb, role: str) -> None:
-    """角色必须是 super_admin / custom(自定义) / admin_role 表里已存在的角色,否则 400。"""
-    if role in (SUPER_ADMIN_ROLE, CUSTOM_ROLE):
+    """角色必须是 super_admin 或 admin_role 表里已存在的角色,否则 400。"""
+    if role == SUPER_ADMIN_ROLE:
         return
     role_repo.ensure_builtin_roles(db)  # 空表(测试/全新库)兜底播种,再校验
     if role_repo.get_role(db, role) is None:
         raise HTTPException(status_code=400, detail=f"角色不存在: {role}")
 
 
-def _clean_override(role: str, pages_override: list[str] | None) -> list[str] | None:
-    """按最终角色算出该存的 pages_override:custom → 勾选集(过滤非法 key,可空列表);
-    非 custom → None(切回普通角色即清空自定义页)。"""
-    if role == CUSTOM_ROLE:
-        return sanitize_pages(pages_override)
-    return None
-
-
 @router.get("", response_model=list[AdminOut], summary="管理员列表(含明文密码,super_admin 专属)")
 def list_admins(db: AdminDb) -> list[AdminOut]:
     out: list[AdminOut] = []
@@ -59,16 +51,13 @@ def create_admin(
     if admin_repo.get_by_username(db, body.username) is not None:
         raise HTTPException(status_code=409, detail="用户名已存在")
     _validate_role(db, body.role)
-    override = _clean_override(body.role, body.pages_override)
     new = admin_repo.create_admin(
         db, username=body.username, password=body.password, role=body.role,
         plain_password=body.password,  # UI 建的账号留存明文,供权限管理页复看
-        pages_override=override,
     )
     write_audit(
         db, admin, action="admin.create", target_type="admin", target_id=new.id,
-        detail={"username": new.username, "role": new.role, "pages_override": override},
-        ip=get_client_ip(request), commit=True,
+        detail={"username": new.username, "role": new.role}, ip=get_client_ip(request), commit=True,
     )
     return AdminOut.model_validate(new)
 
@@ -100,8 +89,7 @@ def update_admin(
         _validate_role(db, body.role)
 
     changes: dict = {}
-    role_changed = body.role is not None and body.role != target.role
-    if role_changed:
+    if body.role is not None and body.role != target.role:
         changes["role"] = {"before": target.role, "after": body.role}
         target.role = body.role
     if body.status is not None and body.status != target.status:
@@ -111,17 +99,6 @@ def update_admin(
         changes["password"] = "reset"
         target.password_hash = hash_password(body.password)
         target.plain_password = body.password  # 同步留存明文,权限管理页复看保持一致
-
-    # 自定义可见页:按「最终角色」(target.role,已应用完角色变更)决定该存什么。
-    # - 切到/维持 custom 且传了 pages_override → 用勾选集;切到 custom 没传 → 清成空列表。
-    # - 切回普通角色 → 清空 override(_clean_override 返回 None)。
-    # - 角色没变、只传 pages_override(编辑现有 custom 用户的勾选)→ 也更新。
-    if role_changed or body.pages_override is not None:
-        new_override = _clean_override(target.role, body.pages_override)
-        if new_override != target.pages_override:
-            changes["pages_override"] = {"before": target.pages_override, "after": new_override}
-            target.pages_override = new_override
-
     if not changes:
         raise HTTPException(status_code=400, detail="无任何变更字段")
     db.commit()

app/admin/routers/analytics_health.py

@@ -1,49 +0,0 @@
-"""admin 埋点健康度:埋点成功率 / 上报成功率 总览 + 趋势 + 下钻(只读)。"""
-from __future__ import annotations
-
-from datetime import datetime
-from typing import Annotated
-
-from fastapi import APIRouter, Depends, Query
-
-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,
-    HealthMetrics,
-    HealthTrendPoint,
-)
-
-router = APIRouter(
-    prefix="/admin/api/analytics-health",
-    tags=["admin-analytics-health"],
-    dependencies=[Depends(get_current_admin)],
-)
-
-
-@router.get("/overview", response_model=HealthMetrics, summary="两段成功率总览")
-def overview(
-    db: AdminDb,
-    date_from: Annotated[datetime, Query()],
-    date_to: Annotated[datetime, Query()],
-) -> HealthMetrics:
-    return HealthMetrics(**repo.overview(db, date_from, date_to))
-
-
-@router.get("/trend", response_model=list[HealthTrendPoint], summary="按北京天趋势")
-def trend(
-    db: AdminDb,
-    date_from: Annotated[datetime, Query()],
-    date_to: Annotated[datetime, Query()],
-) -> list[HealthTrendPoint]:
-    return [HealthTrendPoint(**p) for p in repo.trend(db, date_from, date_to)]
-
-
-@router.get("/breakdown", response_model=list[HealthBreakdownRow], summary="按维度下钻")
-def breakdown(
-    db: AdminDb,
-    date_from: Annotated[datetime, Query()],
-    date_to: Annotated[datetime, Query()],
-    dim: Annotated[str, Query(pattern="^(event|app_ver|oem)$")] = "event",
-) -> list[HealthBreakdownRow]:
-    return [HealthBreakdownRow(**r) for r in repo.breakdown(db, date_from, date_to, dim)]

app/admin/routers/auth.py

@@ -6,7 +6,6 @@ import logging
 from fastapi import APIRouter, Depends, HTTPException
 
 from app.admin.deps import AdminDb, CurrentAdmin
-from app.admin.permissions import CUSTOM_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.schemas.auth import AdminLoginRequest, AdminLoginResponse, AdminOut
@@ -20,13 +19,9 @@ router = APIRouter(prefix="/admin/api/auth", tags=["admin-auth"])
 
 
 def _admin_out_with_pages(admin, db: AdminDb) -> AdminOut:  # noqa: ANN001
-    """AdminOut + 有效可见页(前端左侧导航按此过滤)。
-    role=="custom" → 用这个人的 pages_override(按人自定义,过滤悬空 key);其余走角色解析。"""
+    """AdminOut + 当前角色有效可见页(前端左侧导航按此过滤)。"""
     out = AdminOut.model_validate(admin)
-    if admin.role == CUSTOM_ROLE:
-        out.pages = sanitize_pages(admin.pages_override)
-    else:
-        out.pages = role_repo.effective_pages_of(db, admin.role)
+    out.pages = role_repo.effective_pages_of(db, admin.role)
     return out
 
 

app/admin/routers/ops_marquee_seed.py

@@ -19,8 +19,6 @@ from app.admin.schemas.ops_marquee_seed import (
     OpsMarqueeSeedCreate,
     OpsMarqueeSeedOut,
     OpsMarqueeSeedUpdate,
-    OpsRealRecordItem,
-    OpsRealRecordsOut,
     OpsSavingsFeedPreviewOut,
 )
 from app.models.admin import AdminUser
@@ -59,31 +57,9 @@ def list_seeds(db: AdminDb) -> list[OpsMarqueeSeedOut]:
 def preview_feed(
     db: AdminDb,
     limit: Annotated[int, Query(ge=1, le=30)] = 8,
-    mode: Annotated[str | None, Query(description="mixed/real/seed;不传=当前持久化模式")] = None,
 ) -> OpsSavingsFeedPreviewOut:
-    """返回客户端实际会看到的 feed(真实记录会插队、种子随机抽取/金额随机/名字合成),供运营对效果。
-
-    mode 显式指定则预览该模式(**不改持久化配置**,供前端切换开关时实时预览);不传则用当前持久化模式。
-    """
-    if mode is not None and mode not in ops_marquee.FEED_MODES:
-        raise HTTPException(status_code=400, detail="mode 需为 mixed / real / seed")
-    return OpsSavingsFeedPreviewOut(items=ops_marquee.get_feed(db, limit=limit, mode=mode))
-
-
-@router.get("/real-records", response_model=OpsRealRecordsOut, summary="分页浏览当前模式下可展示的记录(审核用)")
-def list_real_records(
-    db: AdminDb,
-    mode: Annotated[str | None, Query(description="mixed/real/seed;不传=当前持久化模式")] = None,
-    offset: Annotated[int, Query(ge=0)] = 0,
-    limit: Annotated[int, Query(ge=1, le=50)] = 8,
-) -> OpsRealRecordsOut:
-    """分页列出**当前模式**下可在 app 轮播展示的全部记录(**不去重**):只真实=真实记录;只种子=各启用
-    种子按生成逻辑各出一行;混播=真实+种子。与 app 同口径洗牌+去连簇,固定种子→翻页稳定、能翻遍全部。
-    item.user_id=0 表示种子行。"""
-    if mode is not None and mode not in ops_marquee.FEED_MODES:
-        raise HTTPException(status_code=400, detail="mode 需为 mixed / real / seed")
-    items, total = ops_marquee.list_real_records(db, mode=mode, offset=offset, limit=limit)
-    return OpsRealRecordsOut(items=[OpsRealRecordItem(**it) for it in items], total=total)
+    """返回客户端实际会看到的 feed(真实记录会插队、种子随机抽取/金额随机/名字合成),供运营对效果。"""
+    return OpsSavingsFeedPreviewOut(items=ops_marquee.get_feed(db, limit=limit))
 
 
 # 注:/mode 两个端点须在 /{seed_id} 之前注册,否则 PATCH /mode 会被 /{seed_id} 抢先按 id 解析。

app/admin/schemas/admin.py

@@ -13,18 +13,14 @@ class AdminCreateRequest(BaseModel):
     username: str = Field(..., min_length=3, max_length=64)
     password: str = Field(..., min_length=8, max_length=72)  # bcrypt ≤72 字节
     role: str = Field("operator", min_length=1, max_length=32)
-    # 仅当 role == "custom":这个人专属可见页 key 列表(逐页勾选结果)。其余角色不传/忽略。
-    pages_override: list[str] | None = None
 
 
 class AdminUpdateRequest(BaseModel):
-    """改角色 / 启用禁用 / 重置密码 / 自定义可见页,字段都可选(只改传了的)。"""
+    """改角色 / 启用禁用 / 重置密码,字段都可选(只改传了的)。"""
 
     role: str | None = Field(None, min_length=1, max_length=32)
     status: Literal["active", "disabled"] | None = None
     password: str | None = Field(None, min_length=8, max_length=72)
-    # 改成/更新「自定义」可见页;role 切回普通角色时后端会清空 override(见路由)。
-    pages_override: list[str] | None = None
 
 
 class AdminAuditLogOut(BaseModel):

app/admin/schemas/analytics_health.py

@@ -1,21 +0,0 @@
-"""埋点健康度 admin 响应 schema。"""
-from __future__ import annotations
-
-from pydantic import BaseModel
-
-
-class HealthMetrics(BaseModel):
-    attempted: int
-    drop_capture: int
-    delivered: int
-    drop_undelivered: int
-    track_success_rate: float | None
-    report_success_rate: float | None
-
-
-class HealthTrendPoint(HealthMetrics):
-    day: str
-
-
-class HealthBreakdownRow(HealthMetrics):
-    key: str

app/admin/schemas/auth.py

@@ -21,11 +21,8 @@ class AdminOut(BaseModel):
     created_at: datetime
     last_login_at: datetime | None = None
     # 该管理员当前角色的有效可见页(= 左侧导航项 key);仅登录 / /me 填充,列表接口默认空。
-    # 前端据此过滤左侧导航(super_admin = 全部页;role=="custom" = pages_override)。见 permissions.py。
+    # 前端据此过滤左侧导航(super_admin = 全部页)。见 app/admin/permissions.py。
     pages: list[str] = []
-    # 「自定义」可见页原始勾选集(role=="custom" 时非空);列表接口下发,供权限管理页编辑回填勾选。
-    # 普通角色为 None。from_attributes 自动从 ORM 列取。
-    pages_override: list[str] | None = None
     # 明文登录密码:仅「管理员账号列表」(super_admin 专属路由)填充,供权限管理页编辑时复看;
     # 无留存(脚本建的超管 / 旧账号)为 None → 前端不显示。登录 / /me 不下发(保持 None)。
     password: str | None = None

app/admin/schemas/ops_marquee_seed.py

@@ -62,16 +62,3 @@ class OpsSavingsFeedPreviewItem(BaseModel):
 class OpsSavingsFeedPreviewOut(BaseModel):
     """运营预览:实际混播出来的 feed(真实记录会插队,与客户端一致)。"""
     items: list[OpsSavingsFeedPreviewItem]
-
-
-class OpsRealRecordItem(BaseModel):
-    masked_user: str          # 脱敏后展示名(与 app 一致)
-    saved_amount_cents: int   # 节省金额(分)
-    created_at: str           # 比价记录时间(YYYY-MM-DD HH:MM)
-    user_id: int              # 真实 user_id(供运营核对,不下发客户端)
-
-
-class OpsRealRecordsOut(BaseModel):
-    """分页浏览「全部可展示的真实记录」(success+省>0,不去重、稳定顺序)。"""
-    items: list[OpsRealRecordItem]
-    total: int                # 满足条件的真实记录总数(算页数用)

app/admin/schemas/wallet.py

@@ -130,7 +130,6 @@ class WithdrawBulkResult(BaseModel):
 
 class WithdrawLedgerCheckOut(BaseModel):
     ok: bool
-    # 普通现金账(coin_cash:金币兑换的现金)
     cash_balance_total_cents: int
     cash_transaction_total_cents: int
     balance_diff_cents: int
@@ -138,14 +137,6 @@ class WithdrawLedgerCheckOut(BaseModel):
     missing_refund_txn_count: int
     duplicate_refund_txn_count: int
     refund_txn_on_non_terminal_count: int
-    # 邀请奖励金账(invite_cash:与普通现金物理隔离,各自对账)。默认 0 向后兼容。
-    invite_cash_balance_total_cents: int = 0
-    invite_cash_transaction_total_cents: int = 0
-    invite_balance_diff_cents: int = 0
-    invite_missing_withdraw_txn_count: int = 0
-    invite_missing_refund_txn_count: int = 0
-    invite_duplicate_refund_txn_count: int = 0
-    invite_refund_txn_on_non_terminal_count: int = 0
 
 
 class WxpayHealthCheckOut(BaseModel):

app/api/v1/analytics.py

@@ -6,18 +6,13 @@ POST /api/v1/analytics/events — 批量接收新手引导(及后续)埋点,appe
 """
 from __future__ import annotations
 
-import logging
-
-from fastapi import APIRouter, HTTPException, Request
+from fastapi import APIRouter, Request
 
 from app.api.deps import DbSession
 from app.repositories import analytics as analytics_repo
-from app.repositories import analytics_selfstat as selfstat_repo
 from app.schemas.analytics import AnalyticsBatchIn, AnalyticsIngestOut
-from app.schemas.analytics_selfstat import SelfStatBatchIn, SelfStatIngestOut
 
 router = APIRouter(prefix="/api/v1/analytics", tags=["analytics"])
-logger = logging.getLogger("shagua.analytics")
 
 
 def _client_ip(request: Request) -> str:
@@ -34,13 +29,3 @@ def ingest_events(
 ) -> AnalyticsIngestOut:
     n = analytics_repo.record_batch(db, batch, client_ip=_client_ip(request))
     return AnalyticsIngestOut(received=n)
-
-
-@router.post("/selfstat", response_model=SelfStatIngestOut, summary="上报自报计数快照")
-def ingest_selfstat(batch: SelfStatBatchIn, db: DbSession) -> SelfStatIngestOut:
-    try:
-        snap_id = selfstat_repo.record_selfstat(db, batch)
-    except Exception:  # noqa: BLE001 — 计数链路要稳,落库失败不裸奔 500,记日志回明确错误
-        logger.exception("selfstat ingest failed device=%s epoch=%s", batch.device_id, batch.epoch_id)
-        raise HTTPException(status_code=503, detail="selfstat ingest failed") from None
-    return SelfStatIngestOut(snapshot_id=snap_id)

app/models/__init__.py

@@ -7,10 +7,6 @@ from app.models.ad_watch_log import AdWatchLog  # noqa: F401
 from app.models.admin import AdminAuditLog, AdminUser  # noqa: F401
 from app.models.admin_role import AdminRole  # noqa: F401
 from app.models.analytics_event import AnalyticsEvent  # noqa: F401
-from app.models.analytics_selfstat import (  # noqa: F401
-    AnalyticsSelfStat,
-    AnalyticsSelfStatEvent,
-)
 from app.models.app_config import AppConfig  # noqa: F401
 from app.models.comparison import ComparisonRecord  # noqa: F401
 from app.models.cps_activity import CpsActivity  # noqa: F401

app/models/admin.py

@@ -28,11 +28,8 @@ class AdminUser(Base):
     # 明文登录密码:仅「后台 UI 创建/重置」的管理员留存,供超管在权限管理页复看转交。
     # 脚本/起后台时建的超管账号不写(为 None → 前端「不显示密码」)。⚠️ 内部工具便利取舍,见 create/list。
     plain_password: Mapped[str | None] = mapped_column(String(128), nullable=True)
-    # super_admin(全权+管账号)/ finance(钱:提现+金币)/ operator(用户+反馈+大盘)/ custom(按人自定义)
+    # super_admin(全权+管账号)/ finance(钱:提现+金币)/ operator(用户+反馈+大盘)
     role: Mapped[str] = mapped_column(String(20), nullable=False, default="operator")
-    # 「自定义」权限:仅当 role == "custom" 时有效,存这个人专属的可见页 key 列表(不共享给他人)。
-    # 非 custom 用户为 None → 可见页跟随角色。登录/`/me` 下发有效页时,非空即优先用它(见 auth._admin_out_with_pages)。
-    pages_override: Mapped[list | None] = mapped_column(_JSON, nullable=True)
     # active / disabled
     status: Mapped[str] = mapped_column(String(20), nullable=False, default="active")
 

app/models/analytics_selfstat.py

@@ -1,59 +0,0 @@
-"""埋点/上报成功率自报计数快照表(append-only)。
-
-客户端周期上报「自 epoch 起算的累计计数」;服务端只存原始快照,查询时在 Python 侧差分聚合
-(见 app/admin/repositories/analytics_health.py)。与既有 analytics_event 表完全独立。
-
-- analytics_selfstat        :一快照一行(快照头 + 设备维度 + 设备级诊断量)
-- analytics_selfstat_event  :一 event 一行(四类累计计数),外键指向快照头
-"""
-from __future__ import annotations
-
-from datetime import datetime
-
-from sqlalchemy import BigInteger, DateTime, ForeignKey, Integer, String, func
-from sqlalchemy.orm import Mapped, mapped_column
-
-from app.db.base import Base
-
-
-class AnalyticsSelfStat(Base):
-    __tablename__ = "analytics_selfstat"
-
-    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
-    device_id: Mapped[str] = mapped_column(String(64), index=True, nullable=False)
-    epoch_id: Mapped[str] = mapped_column(String(64), index=True, nullable=False)
-    # 设备维度(每设备固定,下钻用)
-    app_ver: Mapped[str | None] = mapped_column(String(32), nullable=True)
-    oem: Mapped[str | None] = mapped_column(String(32), nullable=True)
-    os: Mapped[str | None] = mapped_column(String(32), nullable=True)
-    # 设备级诊断量(累计;queue_depth 是瞬时 gauge)
-    batches_attempted: Mapped[int] = mapped_column(BigInteger, nullable=False, default=0)
-    batches_ok: Mapped[int] = mapped_column(BigInteger, nullable=False, default=0)
-    batches_fail: Mapped[int] = mapped_column(BigInteger, nullable=False, default=0)
-    retries: Mapped[int] = mapped_column(BigInteger, nullable=False, default=0)
-    queue_depth: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
-    sent_at: Mapped[int | None] = mapped_column(BigInteger, nullable=True)  # 端上报时刻 epoch ms
-    # 服务端接收时间(权威,用于时间分桶与分区排序)
-    created_at: Mapped[datetime] = mapped_column(
-        DateTime(timezone=True), server_default=func.now(), index=True, nullable=False
-    )
-
-    def __repr__(self) -> str:  # pragma: no cover
-        return f"<AnalyticsSelfStat id={self.id} device={self.device_id} epoch={self.epoch_id}>"
-
-
-class AnalyticsSelfStatEvent(Base):
-    __tablename__ = "analytics_selfstat_event"
-
-    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
-    snapshot_id: Mapped[int] = mapped_column(
-        ForeignKey("analytics_selfstat.id"), index=True, nullable=False
-    )
-    event: Mapped[str] = mapped_column(String(64), index=True, nullable=False)
-    attempted: Mapped[int] = mapped_column(BigInteger, nullable=False, default=0)
-    drop_capture: Mapped[int] = mapped_column(BigInteger, nullable=False, default=0)
-    delivered: Mapped[int] = mapped_column(BigInteger, nullable=False, default=0)
-    drop_undelivered: Mapped[int] = mapped_column(BigInteger, nullable=False, default=0)
-
-    def __repr__(self) -> str:  # pragma: no cover
-        return f"<AnalyticsSelfStatEvent snap={self.snapshot_id} event={self.event}>"

app/repositories/ad_feed_reward.py

@@ -176,19 +176,10 @@ def grant_feed_reward(
         )
         return _commit_record(db, rec, client_event_id)
 
-    # 按点位场景拆流水文案(2026-07):比价等候期看的广告 vs 领券时看的广告,在收益明细里分开显示。
-    # feed_scene=comparison→比价奖励 / coupon→领券奖励;其它(welfare/空/旧端不带)维持通用「信息流广告奖励」。
-    # 客户端按此 biz_type 直显固定文案(见 CoinHistoryViewModel.coinTitle),故 remark 只作后台留痕/兜底。
-    if feed_scene == "comparison":
-        reward_biz, reward_remark = "feed_ad_reward_comparison", "比价奖励"
-    elif feed_scene == "coupon":
-        reward_biz, reward_remark = "feed_ad_reward_coupon", "领券奖励"
-    else:
-        reward_biz, reward_remark = "feed_ad_reward", "信息流广告奖励"
     crud_wallet.grant_coins(
         db, user_id, coin,
-        biz_type=reward_biz, ref_id=client_event_id,
-        remark=reward_remark,
+        biz_type="feed_ad_reward", ref_id=client_event_id,
+        remark="信息流广告奖励",
     )
     rec = AdFeedRewardRecord(
         client_event_id=client_event_id,

app/repositories/analytics_selfstat.py

@@ -1,38 +0,0 @@
-"""自报计数快照落库。一次事务:插 1 条快照头 + N 条 event 行,返回快照 id。"""
-from __future__ import annotations
-
-from sqlalchemy.orm import Session
-
-from app.models.analytics_selfstat import AnalyticsSelfStat, AnalyticsSelfStatEvent
-from app.schemas.analytics_selfstat import SelfStatBatchIn
-
-
-def record_selfstat(db: Session, batch: SelfStatBatchIn) -> int:
-    snap = AnalyticsSelfStat(
-        device_id=batch.device_id,
-        epoch_id=batch.epoch_id,
-        app_ver=batch.app_ver,
-        oem=batch.oem,
-        os=batch.os,
-        batches_attempted=batch.batches_attempted,
-        batches_ok=batch.batches_ok,
-        batches_fail=batch.batches_fail,
-        retries=batch.retries,
-        queue_depth=batch.queue_depth,
-        sent_at=batch.sent_at,
-    )
-    db.add(snap)
-    db.flush()  # 拿到 snap.id
-    db.add_all([
-        AnalyticsSelfStatEvent(
-            snapshot_id=snap.id,
-            event=e.event,
-            attempted=e.attempted,
-            drop_capture=e.drop_capture,
-            delivered=e.delivered,
-            drop_undelivered=e.drop_undelivered,
-        )
-        for e in batch.events
-    ])
-    db.commit()
-    return snap.id

app/repositories/ops_marquee.py

@@ -30,7 +30,6 @@ from app.models.comparison import ComparisonRecord
 from app.models.ops_marquee_seed import OpsMarqueeSeed
 from app.models.user import User
 from app.repositories import app_config
-from app.repositories.user import is_default_nickname
 
 # 首页轮播数据源模式(存 app_config.marquee_feed_mode):
 #   mixed=真实优先+种子补位+合成兜底(默认,原行为);real=只真实(不足则少/空);seed=只种子+合成兜底。
@@ -141,12 +140,9 @@ def _synth_masked_name(rng: random.Random) -> str:
 
 def _mask_real(nickname: str | None, user_id: int) -> str:
     """真实用户脱敏(对齐 PRD「用户标识打码规则」):设过昵称→昵称脱敏(中英文皆可);
-    没昵称→「用户」+5星+id 后 2 位(用户*****08),按 user_id 稳定、刷新不变脸。
-
-    创建时自动分配的默认昵称(「用户」+9 位随机,见 user.is_default_nickname)不算用户主动设的昵称,
-    按「无昵称」处理走 id 规则(产品决策 2026-07:默认昵称归入「没昵称」档)。"""
+    没昵称→「用户」+5星+id 后 2 位(用户*****08),按 user_id 稳定、刷新不变脸。"""
     nick = (nickname or "").strip()
-    if nick and not is_default_nickname(nick):
+    if nick:
         return _mask_nickname(nick)
     return _mask_anon(user_id)
 
@@ -217,47 +213,34 @@ def _recent_real_rows(db: Session) -> list[tuple[int, int, str | None]]:
     return out
 
 
-def _shuffle_declustered(rows: list, rng: random.Random | None = None) -> list:
-    """洗牌 + 「去连簇」:先洗牌,再贪心重排让相邻两条尽量不是同一 user_id(元素 [0] 即 user_id)。
-    rng=None → 用全局 _rng(feed 每次新随机);传入 rng(如固定种子 Random)→ 排列确定(admin 稳定分页)。
-    减少同一用户连续出现;只有少数几个用户时 best-effort。"""
-    rng = rng or _rng
-    pool = list(rows)
-    rng.shuffle(pool)
-    result: list = []
-    while pool:
-        prev_uid = result[-1][0] if result else None
-        # 优先挑与上一条不同 user 的;挑不到(只剩同 user)才取第一个
-        idx = next((i for i, r in enumerate(pool) if r[0] != prev_uid), 0)
-        result.append(pool.pop(idx))
-    return result
-
-
-def get_feed(db: Session, limit: int = 8, mode: str | None = None) -> list[dict]:
+def get_feed(db: Session, limit: int = 8) -> list[dict]:
     """返回最多 limit 条 {masked_user, saved_amount_cents, time(HH:MM:SS 北京)}。
 
-    mode:显式传入(admin 预览指定模式)则用它、**不改持久化配置**;不传(客户端 /savings-feed)读
-    持久化的 marquee_feed_mode;非法值一律回退到持久化模式。
-    真实条:success 且 0 < saved ≤ 上限,**不按 user 去重**(打乱 + 去连簇:相邻尽量不同用户、减少单人连刷)后取前 limit。
+    真实条:success 且 0 < saved ≤ 上限,按 user 去重(同一用户只取最新一条,避免单人刷屏)。
     不足用启用的种子补齐——**公平随机抽取** need 个(而非固定取前 N),让所有种子都有机会露出;
     种子用户名留空则随机合成(避开撞名),金额取**长尾随机**(小额居多、偶尔大额,更像真实分布)。
     真实 + 种子仍不满 limit → 内置合成条**补满**,保证轮播既不空也不稀疏。
     展示时间统一「刷新」成相对现在的最近时刻(从 now 往前**随机抖动**递减),保证轮播永远像刚发生、
     节奏自然不机械(真实用户/金额不变,只换展示时间——避免旧测试数据 / 低谷期记录显示成过时时间)。
     """
-    # 预览可显式指定模式(所见=选中模式,不依赖 PATCH 落库时序);None/非法 → 读持久化配置。
-    mode = mode if mode in FEED_MODES else get_feed_mode(db)  # mixed / real / seed
+    mode = get_feed_mode(db)  # mixed / real / seed(运营可在「首页轮播种子」页切换)
 
     items: list[dict] = []
     used_names: set[str] = set()
 
-    # 真实条(mixed / real):**不按 user 去重**——打乱 + 去连簇(相邻尽量不同用户、减少单人连刷)后取前
-    # limit;金额超上限的异常值已在查询剔除。同一用户可多条露出,但被打散、尽量不连续。
+    # 真实条(mixed / real):取较多近期记录(带 ~30s 缓存)后按 user 去重;金额超上限的异常值已在查询剔除。
     if mode != "seed":
-        for uid, sc, nick in _shuffle_declustered(_recent_real_rows(db))[:limit]:
+        rows = _recent_real_rows(db)
+        seen_users: set[int] = set()
+        for uid, sc, nick in rows:
+            if uid in seen_users:
+                continue
+            seen_users.add(uid)
             name = _mask_real(nick, uid)
-            used_names.add(name)  # 同一用户可多条,名字重复无害(used_names 仅供种子避重)
+            used_names.add(name)  # 真实名按昵称/id 稳定;偶发撞名可接受
             items.append({"masked_user": name, "saved_amount_cents": int(sc)})
+            if len(items) >= limit:
+                break
 
     # 种子补位 + 合成兜底(mixed / seed):mixed 下补真实不足的部分,seed 下全量用种子/合成。
     # real 模式**跳过**——只出真实,不掺任何假数据(真实不足则少于 limit,为 0 时返回空)。
@@ -310,78 +293,6 @@ def get_feed(db: Session, limit: int = 8, mode: str | None = None) -> list[dict]
     return items
 
 
-# ===== admin 侧:分页浏览「当前模式下可展示的记录」(审核用,不去重) =====
-_REAL_BROWSE_CAP = 1000   # 真实记录一次最多纳入这么多去洗牌+分页(足够审核;防超大库全量洗牌)
-_BROWSE_SEED = 20260707   # 固定洗牌种子:同一批数据下排列恒定 → 翻页稳定、能翻遍全部
-
-
-def _seed_browse_row(seed: OpsMarqueeSeed) -> dict:
-    """把一条种子按其「生成逻辑」**确定性**生成一行浏览项(名字/金额按 seed.id 派生固定种子 → 翻页稳定)。
-    名字:运营手填的非模板名原样用,否则本地合成;金额:区间内确定性长尾取值。user_id=0(种子无真实用户)。"""
-    r = random.Random(_BROWSE_SEED * 1_000_003 + int(seed.id))
-    fixed = (seed.masked_user or "").strip()
-    name = fixed if (fixed and not fixed.startswith("用户****")) else _synth_name(r)
-    lo = max(0, int(seed.min_cents))
-    hi = max(lo, int(seed.max_cents))
-    amt = lo if hi <= lo else lo + int(round((hi - lo) * (r.random() ** 2.2)))
-    return {"masked_user": name, "saved_amount_cents": amt, "created_at": "", "user_id": 0}
-
-
-def list_real_records(
-    db: Session, mode: str | None = None, offset: int = 0, limit: int = 8
-) -> tuple[list[dict], int]:
-    """分页浏览「**当前模式**下可在 app 轮播展示的全部记录」,供 admin 逐页审核(**不去重**):
-      - 只真实:全部 success+省>0 的真实记录;
-      - 只种子:每条启用种子按生成逻辑各出一行;
-      - 混播:真实 + 种子 全部合在一起。
-    与 app 轮播同口径:先洗牌 + 去连簇(相邻尽量不同 user;种子各自独立、不算同 user);**固定种子** →
-    同批数据下排列恒定,翻页不跳、能翻遍全部。返回 (items, total);item.user_id=0 表示种子。"""
-    mode = mode if mode in FEED_MODES else get_feed_mode(db)
-    # pool: [(cluster_key, item)];cluster_key 供去连簇——真实=user_id、种子=各自唯一负数(互不聚簇)
-    pool: list[tuple[int, dict]] = []
-    if mode != "seed":
-        rows = db.execute(
-            select(
-                ComparisonRecord.user_id,
-                ComparisonRecord.saved_amount_cents,
-                User.nickname,
-                ComparisonRecord.created_at,
-            )
-            .join(User, User.id == ComparisonRecord.user_id)
-            .where(
-                ComparisonRecord.status == "success",
-                ComparisonRecord.saved_amount_cents > 0,
-                ComparisonRecord.saved_amount_cents <= _REAL_MAX_CENTS,
-            )
-            .order_by(ComparisonRecord.created_at.desc())
-            .limit(_REAL_BROWSE_CAP)
-        ).all()
-        for uid, sc, nick, ca in rows:
-            pool.append((
-                int(uid),
-                {
-                    "masked_user": _mask_real(nick, int(uid)),
-                    "saved_amount_cents": int(sc),
-                    "created_at": str(ca)[:16] if ca is not None else "",
-                    "user_id": int(uid),
-                },
-            ))
-    if mode != "real":
-        seeds = (
-            db.execute(select(OpsMarqueeSeed).where(OpsMarqueeSeed.enabled.is_(True)))
-            .scalars()
-            .all()
-        )
-        for i, s in enumerate(seeds):
-            pool.append((-(i + 1), _seed_browse_row(s)))  # 每个种子唯一 key → 互不聚簇
-    total = len(pool)
-    # 每次用同一固定种子新建 Random → 同批数据排列恒定(翻页稳定);同时相邻尽量不同 user。
-    ordered = _shuffle_declustered(pool, random.Random(_BROWSE_SEED))
-    off = max(0, offset)
-    items = [item for _key, item in ordered[off : off + limit]]
-    return items, total
-
-
 # ===== 运营侧:种子 CRUD =====
 def list_seeds(db: Session) -> list[OpsMarqueeSeed]:
     return list(

app/repositories/user.py

@@ -42,22 +42,6 @@ def _gen_nickname() -> str:
     )
 
 
-def is_default_nickname(nickname: str | None) -> bool:
-    """是否为创建时自动分配的默认昵称(= "用户" + 9 位字母数字,见 [_gen_nickname])。
-
-    这类不是用户主动设置的昵称,展示脱敏时按「无昵称」处理(走 id 规则,见 ops_marquee._mask_real)。
-    精确匹配生成格式(前缀 + 定长字母数字集),不误伤真人以「用户」开头的昵称(如「用户体验师」含汉字、
-    长度也不符)。用户改过昵称即不再匹配。"""
-    if not nickname:
-        return False
-    s = nickname.strip()
-    return (
-        len(s) == len(_NICKNAME_PREFIX) + _NICKNAME_LEN
-        and s.startswith(_NICKNAME_PREFIX)
-        and all(c in _NICKNAME_ALPHABET for c in s[len(_NICKNAME_PREFIX):])
-    )
-
-
 def get_user_by_username(db: Session, username: str) -> User | None:
     return db.execute(
         select(User).where(User.username == username)

app/schemas/analytics_selfstat.py

@@ -1,34 +0,0 @@
-"""自报计数上报 schema。字段名对齐客户端 payload(snake_case),累计值语义。"""
-from __future__ import annotations
-
-from pydantic import BaseModel, Field
-
-
-class SelfStatEventIn(BaseModel):
-    event: str = Field(max_length=64)
-    attempted: int = 0
-    drop_capture: int = 0
-    delivered: int = 0
-    drop_undelivered: int = 0
-
-
-class SelfStatBatchIn(BaseModel):
-    device_id: str = Field(max_length=64)
-    epoch_id: str = Field(max_length=64)
-    sent_at: int | None = None
-    app_ver: str | None = Field(default=None, max_length=32)
-    oem: str | None = Field(default=None, max_length=32)
-    os: str | None = Field(default=None, max_length=32)
-    batches_attempted: int = 0
-    batches_ok: int = 0
-    batches_fail: int = 0
-    retries: int = 0
-    queue_depth: int = 0
-    # 允许空列表:某次快照只上报设备级计数(batches_*/retries/queue_depth)、无 event 细分时也合法
-    # (与 AnalyticsBatchIn 的 min_length=1 有意不同——那是行为事件、必须至少一条)。
-    events: list[SelfStatEventIn] = Field(default_factory=list, max_length=200)
-
-
-class SelfStatIngestOut(BaseModel):
-    ok: bool = True
-    snapshot_id: int

scripts/seed_coinhistory_labels_test.py

@@ -1,166 +0,0 @@
-"""收益明细「金币记录」文案验证脚手架(2026-07 文案改版验收用)。
-
-问题:客户端按 bizType 显示固定文案(路线B,强制覆盖后端 remark),但账号若没有对应
-bizType 的流水,收益明细页就空着,无从验证。本脚本往指定测试用户塞每种 bizType 各一条
-流水,让你在手机收益明细页一屏核对全部新文案;验收完 --clean 一键删除,不污染数据。
-
-⚠️ 仅 dev 库用(APP_ENV=dev 时才允许 --seed/--clean)。所有测试流水 ref_id 前缀 TESTDOC,
-按前缀精确清理,不会误删真实流水。
-
-用法(pricebot env 直调,见项目 CLAUDE.md):
-  D:/miniconda/envs/pricebot/python.exe scripts/seed_coinhistory_labels_test.py --show
-  D:/miniconda/envs/pricebot/python.exe scripts/seed_coinhistory_labels_test.py --seed
-  D:/miniconda/envs/pricebot/python.exe scripts/seed_coinhistory_labels_test.py --clean
-"""
-from __future__ import annotations
-
-import argparse
-import sys
-
-from app.core.config import settings
-from app.core.rewards import CN_TZ
-from datetime import datetime
-
-from app.db.session import SessionLocal
-from app.models.user import User
-from app.models.wallet import CoinAccount, CoinTransaction
-
-TEST_PHONE = "11111111111"
-REF_PREFIX = "TESTDOC"  # 所有本脚本造的流水都带这个 ref_id 前缀,便于精确清理
-
-# 每条 = (bizType, 故意写错/留空的 remark, 金币数)。
-# remark 故意填「错的」→ 若客户端仍显示新文案 = 证明路线B强制覆盖生效(无视后端 remark)。
-# task_ 一条留空 remark → 走客户端兜底映射。顺序即手机上从新到旧的展示顺序(后塞的在最上)。
-CASES: list[tuple[str, str, int]] = [
-    ("signin", "每日签到 第99天(旧文案,应被覆盖)", 220),
-    ("signin_boost", "签到膨胀 第99天", 3000),
-    ("reward_video", "看视频奖励金币(旧文案,应被覆盖)", 200),
-    ("feed_ad_reward_comparison", "", 50),   # 后端新拆:比价场景 → 比价奖励
-    ("feed_ad_reward_coupon", "", 50),       # 后端新拆:领券场景 → 领券奖励
-    ("feed_ad_reward", "信息流广告奖励(welfare/旧数据兜底)", 50),
-    ("price_report_reward", "上报更低价审核通过(旧文案,应被覆盖)", 1000),
-    ("feedback_reward", "意见反馈被采纳(旧文案,应被覆盖)", 10000),
-    ("task_enable_notification", "", 750),   # remark 留空 → 客户端「打开消息提醒奖励」
-    # 下两条现实中不会进金币记录(invite 发现金进邀请钱包 / compare_milestone 后端死代码不发钱),
-    # 仅用于验证「杀掉好友比价奖励 + 兜底改任务奖励」:两条都应显示「任务奖励」(remark 被强制无视)。
-    ("invite", "好友比价奖励(旧文案,应被杀→任务奖励)", 200),
-    ("compare_milestone", "", 120),
-]
-
-
-def _client_coin_title(biz_type: str, remark: str | None) -> str:
-    """复刻 CoinHistoryViewModel.coinTitle 的最新逻辑(路线B),用于 --show 预览。
-
-    ⚠️ 必须与客户端保持一致;客户端改了这里也要同步,否则预览会骗人。
-    """
-    fixed = {
-        "exchange_out": "金币兑换现金",
-        "signin": "每日签到奖励",
-        "signin_boost": "签到膨胀奖励",
-        "reward_video": "看视频赚金币",
-        "ad_reward": "看视频赚金币",
-        "feed_ad_reward_comparison": "比价奖励",
-        "feed_ad_reward_coupon": "领券奖励",
-        "feed_ad_reward": "信息流广告奖励",
-        "price_report_reward": "爆料奖励",
-        "feedback_reward": "反馈奖励",
-        "invite": "任务奖励",  # 杀掉"好友比价奖励",归兜底
-    }
-    if biz_type in fixed:
-        return fixed[biz_type]
-    if remark:
-        return remark
-    if biz_type == "task_enable_notification":
-        return "打开消息提醒奖励"
-    return "任务奖励"
-
-
-def _get_user(db) -> User:
-    u = db.query(User).filter(User.phone == TEST_PHONE).first()
-    if not u:
-        print(f"✗ 库里没有测试号 {TEST_PHONE} —— 先在手机上用这个号登录一次再跑本脚本。")
-        sys.exit(1)
-    return u
-
-
-def cmd_show() -> None:
-    """只打印:每种 bizType 经客户端映射后会显示成什么(不写库)。"""
-    print("bizType 造流水后,收益明细页预期显示的文案:\n")
-    print(f"  {'bizType':32} {'后端remark(故意填的)':32} → 手机显示")
-    print("  " + "-" * 90)
-    for biz, remark, _coin in CASES:
-        shown = _client_coin_title(biz, remark or None)
-        rk = (remark or "(空)")
-        print(f"  {biz:32} {rk:32} → {shown}")
-    print("\n注:remark 列是故意填的『旧/错』文案;'手机显示'若为新文案 = 强制覆盖生效。")
-    print("invite / compare_milestone 已从客户端映射删除:invite 有 remark 故显示原样,")
-    print("compare_milestone remark 空故落兜底『奖励』—— 两者都不再有专属新文案(符合『去掉』)。")
-
-
-def cmd_seed() -> None:
-    if settings.APP_ENV != "dev":
-        print(f"✗ 拒绝:APP_ENV={settings.APP_ENV},本脚本只在 dev 库造测试数据。")
-        sys.exit(1)
-    db = SessionLocal()
-    u = _get_user(db)
-    acc = db.query(CoinAccount).filter(CoinAccount.user_id == u.id).first()
-    if acc is None:
-        acc = CoinAccount(user_id=u.id, coin_balance=0, cash_balance_cents=0)
-        db.add(acc)
-        db.flush()
-    now = datetime.now(CN_TZ).replace(tzinfo=None)
-    made = 0
-    for i, (biz, remark, coin) in enumerate(CASES):
-        ref = f"{REF_PREFIX}:{biz}:{i}"
-        exists = db.query(CoinTransaction).filter(CoinTransaction.ref_id == ref).first()
-        if exists:
-            continue
-        acc.coin_balance += coin
-        db.add(CoinTransaction(
-            user_id=u.id, amount=coin, balance_after=acc.coin_balance,
-            biz_type=biz, ref_id=ref, remark=remark or None, created_at=now,
-        ))
-        made += 1
-    db.commit()
-    print(f"✓ 已给 user_id={u.id}({TEST_PHONE})造 {made} 条测试流水,当前金币余额 {acc.coin_balance}。")
-    print("  → 打开手机 App「收益明细 / 金币记录」下拉刷新,逐条核对文案。")
-    print("  → 验收完跑 --clean 删除这些测试流水。")
-
-
-def cmd_clean() -> None:
-    if settings.APP_ENV != "dev":
-        print(f"✗ 拒绝:APP_ENV={settings.APP_ENV}。")
-        sys.exit(1)
-    db = SessionLocal()
-    u = _get_user(db)
-    rows = db.query(CoinTransaction).filter(
-        CoinTransaction.user_id == u.id,
-        CoinTransaction.ref_id.like(f"{REF_PREFIX}:%"),
-    ).all()
-    total = sum(r.amount for r in rows)
-    for r in rows:
-        db.delete(r)
-    acc = db.query(CoinAccount).filter(CoinAccount.user_id == u.id).first()
-    if acc is not None:
-        acc.coin_balance -= total  # 把造流水时加的余额扣回,还原
-    db.commit()
-    print(f"✓ 已删除 {len(rows)} 条 TESTDOC 测试流水,余额回扣 {total},当前 {acc.coin_balance if acc else 0}。")
-
-
-def main() -> None:
-    ap = argparse.ArgumentParser(description="收益明细金币文案验证脚手架")
-    g = ap.add_mutually_exclusive_group(required=True)
-    g.add_argument("--show", action="store_true", help="只打印每种 bizType 的预期显示文案,不写库")
-    g.add_argument("--seed", action="store_true", help="往测试号造每种 bizType 各一条流水")
-    g.add_argument("--clean", action="store_true", help="删除本脚本造的所有测试流水")
-    args = ap.parse_args()
-    if args.show:
-        cmd_show()
-    elif args.seed:
-        cmd_seed()
-    elif args.clean:
-        cmd_clean()
-
-
-if __name__ == "__main__":
-    main()

tests/test_admin_read.py

@@ -320,92 +320,3 @@ def test_read_apis_require_auth(admin_client: TestClient) -> None:
         "/admin/api/feedbacks",
     ]:
         assert admin_client.get(path).status_code == 401, path
-
-
-def test_period_comparison_reward_coin_from_feed_scene(
-    admin_client: TestClient, admin_token: str
-) -> None:
-    """比价奖励金币口径:按 ad_feed_reward_record.feed_scene='comparison' 的实发金币
-    (status=granted)汇总,而非查从不写入的 biz_type 桶(修复大盘该卡恒 0)。
-    too_short(未发奖)不计。"""
-    from app.models.ad_feed_reward import AdFeedRewardRecord
-
-    d = "2021-06-15"  # 独立历史日,隔离其它用例数据
-    db = SessionLocal()
-    try:
-        uid = user_repo.upsert_user_for_login(db, phone="13800008801", register_channel="sms").id
-        db.add(AdFeedRewardRecord(
-            client_event_id="cmp-fs-granted", user_id=uid, reward_date=d,
-            ecpm_raw="0", coin=123, feed_scene="comparison", status="granted",
-        ))
-        db.add(AdFeedRewardRecord(
-            client_event_id="cmp-fs-tooshort", user_id=uid, reward_date=d,
-            ecpm_raw="0", coin=99, feed_scene="comparison", status="too_short",
-        ))
-        db.commit()
-    finally:
-        db.close()
-
-    r = admin_client.get(
-        "/admin/api/stats/overview", params={"date_from": d, "date_to": d},
-        headers=_auth(admin_token),
-    )
-    assert r.status_code == 200, r.text
-    assert r.json()["period"]["coins"]["comparison_reward_coin_total"] == 123
-
-
-def test_period_coupon_reward_coin_from_feed_scene(
-    admin_client: TestClient, admin_token: str
-) -> None:
-    """领券奖励金币口径:按 ad_feed_reward_record.feed_scene='coupon' 的实发金币汇总。"""
-    from app.models.ad_feed_reward import AdFeedRewardRecord
-
-    d = "2021-06-16"
-    db = SessionLocal()
-    try:
-        uid = user_repo.upsert_user_for_login(db, phone="13800008802", register_channel="sms").id
-        db.add(AdFeedRewardRecord(
-            client_event_id="cpn-fs-granted", user_id=uid, reward_date=d,
-            ecpm_raw="0", coin=456, feed_scene="coupon", status="granted",
-        ))
-        db.commit()
-    finally:
-        db.close()
-
-    r = admin_client.get(
-        "/admin/api/stats/overview", params={"date_from": d, "date_to": d},
-        headers=_auth(admin_token),
-    )
-    assert r.status_code == 200, r.text
-    assert r.json()["period"]["coins"]["coupon_reward_coin_total"] == 456
-
-
-def test_period_coupon_reward_excludes_reward_video(
-    admin_client: TestClient, admin_token: str
-) -> None:
-    """领券奖励金币不再把激励视频金币算进来(reward_video/ad_reward 从领券桶拆出);
-    激励视频仍单独计入 reward_video_coin_total。"""
-    from datetime import datetime
-
-    from app.models.wallet import CoinTransaction
-
-    d = "2021-06-17"
-    db = SessionLocal()
-    try:
-        uid = user_repo.upsert_user_for_login(db, phone="13800008803", register_channel="sms").id
-        db.add(CoinTransaction(
-            user_id=uid, amount=50, balance_after=50, biz_type="reward_video",
-            ref_id="rv-split-1", created_at=datetime(2021, 6, 17, 12, 0, 0),
-        ))
-        db.commit()
-    finally:
-        db.close()
-
-    r = admin_client.get(
-        "/admin/api/stats/overview", params={"date_from": d, "date_to": d},
-        headers=_auth(admin_token),
-    )
-    assert r.status_code == 200, r.text
-    coins = r.json()["period"]["coins"]
-    assert coins["coupon_reward_coin_total"] == 0        # 激励视频不计入领券奖励
-    assert coins["reward_video_coin_total"] == 50        # 仍计入激励视频卡

tests/test_admin_roles.py

@@ -150,80 +150,3 @@ def test_create_admin_rejects_unknown_role(admin_client, super_token) -> None:
         headers=_auth(super_token),
     )
     assert r.status_code == 400
-
-
-def _login_pages(username: str, password: str = "pass1234") -> list[str]:
-    """以某账号登录,取下发的有效可见页(左侧导航过滤依据)。"""
-    c = TestClient(admin_app)
-    return c.post(
-        "/admin/api/auth/login", json={"username": username, "password": password}
-    ).json()["admin"]["pages"]
-
-
-def test_custom_role_create_pages_take_effect(admin_client, super_token) -> None:
-    # 建「自定义」账号:role=custom + 勾选页(含一个非法 key,应被过滤)
-    r = admin_client.post(
-        "/admin/api/admins",
-        json={
-            "username": "cust_user", "password": "pass1234", "role": "custom",
-            "pages_override": ["dashboard", "withdraws", "xx-bad"],
-        },
-        headers=_auth(super_token),
-    )
-    assert r.status_code == 200, r.text
-    # 登录下发的 pages == 勾选集(非法 key 过滤),真正驱动左侧导航
-    assert set(_login_pages("cust_user")) == {"dashboard", "withdraws"}
-    # 账号列表回显原始勾选集(供编辑回填),同样已过滤
-    admins = {a["username"]: a for a in admin_client.get("/admin/api/admins", headers=_auth(super_token)).json()}
-    assert set(admins["cust_user"]["pages_override"]) == {"dashboard", "withdraws"}
-    assert admins["cust_user"]["role"] == "custom"
-
-
-def test_custom_role_update_and_switch_back_clears_override(admin_client, super_token) -> None:
-    admin_client.post(
-        "/admin/api/admins",
-        json={
-            "username": "cust_sw", "password": "pass1234", "role": "custom",
-            "pages_override": ["dashboard"],
-        },
-        headers=_auth(super_token),
-    )
-    aid = next(
-        a["id"] for a in admin_client.get("/admin/api/admins", headers=_auth(super_token)).json()
-        if a["username"] == "cust_sw"
-    )
-    # 只改勾选集(角色不变)→ 生效
-    u = admin_client.patch(
-        f"/admin/api/admins/{aid}", json={"pages_override": ["dashboard", "feedbacks"]},
-        headers=_auth(super_token),
-    )
-    assert u.status_code == 200, u.text
-    assert set(_login_pages("cust_sw")) == {"dashboard", "feedbacks"}
-    # 切回普通角色 operator → override 清空,pages 跟随角色(运营页集,且不含 admins)
-    u2 = admin_client.patch(
-        f"/admin/api/admins/{aid}", json={"role": "operator"}, headers=_auth(super_token)
-    )
-    assert u2.status_code == 200, u2.text
-    admins = {a["username"]: a for a in admin_client.get("/admin/api/admins", headers=_auth(super_token)).json()}
-    assert admins["cust_sw"]["pages_override"] is None
-    pages = set(_login_pages("cust_sw"))
-    assert "feedbacks" in pages and "admins" not in pages  # 运营口径,自定义页已不生效
-
-
-def test_switch_operator_to_custom_sets_override(admin_client, super_token) -> None:
-    admin_client.post(
-        "/admin/api/admins",
-        json={"username": "op2cust", "password": "pass1234", "role": "operator"},
-        headers=_auth(super_token),
-    )
-    aid = next(
-        a["id"] for a in admin_client.get("/admin/api/admins", headers=_auth(super_token)).json()
-        if a["username"] == "op2cust"
-    )
-    u = admin_client.patch(
-        f"/admin/api/admins/{aid}",
-        json={"role": "custom", "pages_override": ["config"]},
-        headers=_auth(super_token),
-    )
-    assert u.status_code == 200, u.text
-    assert set(_login_pages("op2cust")) == {"config"}

tests/test_analytics_health.py

@@ -1,160 +0,0 @@
-"""埋点健康度聚合:纯差分函数 + admin 端点。"""
-from __future__ import annotations
-
-from datetime import datetime, timezone
-
-from app.admin.repositories.analytics_health import _cn_day, _rates, diff_snapshots
-
-
-def _row(device, epoch, event, ts, **cum) -> dict:
-    base = {"attempted": 0, "drop_capture": 0, "delivered": 0, "drop_undelivered": 0}
-    base.update(cum)
-    return {
-        "device_id": device, "epoch_id": epoch, "event": event,
-        "created_at": datetime(2026, 7, 1, ts, 0, tzinfo=timezone.utc),
-        "app_ver": "0.2.12(62)", "oem": "ColorOS", "os": "Android 14",
-        **base,
-    }
-
-
-def test_diff_first_row_is_full_cumulative() -> None:
-    rows = [_row("d", "e", "video_play", 1, attempted=100, delivered=90)]
-    out = diff_snapshots(rows)
-    assert len(out) == 1
-    assert out[0]["d_attempted"] == 100
-    assert out[0]["d_delivered"] == 90
-
-
-def test_diff_consecutive_delta() -> None:
-    rows = [
-        _row("d", "e", "video_play", 1, attempted=100, delivered=90),
-        _row("d", "e", "video_play", 2, attempted=150, delivered=140),
-    ]
-    out = sorted(diff_snapshots(rows), key=lambda r: r["created_at"])
-    assert out[1]["d_attempted"] == 50
-    assert out[1]["d_delivered"] == 50
-
-
-def test_diff_epoch_reset_new_partition() -> None:
-    rows = [
-        _row("d", "e1", "video_play", 1, attempted=100),
-        _row("d", "e2", "video_play", 2, attempted=5),
-    ]
-    out = {(r["epoch_id"]): r["d_attempted"] for r in diff_snapshots(rows)}
-    assert out["e1"] == 100
-    assert out["e2"] == 5
-
-
-def test_diff_clamps_negative_on_reorder() -> None:
-    rows = [
-        _row("d", "e", "video_play", 1, attempted=100),
-        _row("d", "e", "video_play", 2, attempted=80),
-    ]
-    out = sorted(diff_snapshots(rows), key=lambda r: r["created_at"])
-    assert out[1]["d_attempted"] == 0
-
-
-def test_rates_computes_both_formulas() -> None:
-    out = _rates({"attempted": 150, "drop_capture": 0, "delivered": 140, "drop_undelivered": 10})
-    assert out["track_success_rate"] == 1.0
-    assert out["report_success_rate"] == 140 / 150
-
-
-def test_rates_zero_denominator_yields_none() -> None:
-    out = _rates({"attempted": 0, "drop_capture": 0, "delivered": 0, "drop_undelivered": 0})
-    assert out["track_success_rate"] is None
-    assert out["report_success_rate"] is None
-
-
-def test_cn_day_beijing_boundary() -> None:
-    # UTC 15:59 → 北京 23:59 同日;UTC 16:00 → 北京 次日 00:00
-    assert _cn_day(datetime(2026, 7, 1, 15, 59, tzinfo=timezone.utc)) == "2026-07-01"
-    assert _cn_day(datetime(2026, 7, 1, 16, 0, tzinfo=timezone.utc)) == "2026-07-02"
-
-
-def test_diff_same_timestamp_ordered_by_id() -> None:
-    ts = datetime(2026, 7, 1, 1, 0, tzinfo=timezone.utc)
-    # 相同 created_at,乱序传入(id=2 在前);应按 id 排序 → id=1(cum100) 在前
-    rows = [
-        {"id": 2, "device_id": "d", "epoch_id": "e", "event": "vp", "created_at": ts,
-         "app_ver": "v", "oem": "o", "os": "s",
-         "attempted": 150, "drop_capture": 0, "delivered": 0, "drop_undelivered": 0},
-        {"id": 1, "device_id": "d", "epoch_id": "e", "event": "vp", "created_at": ts,
-         "app_ver": "v", "oem": "o", "os": "s",
-         "attempted": 100, "drop_capture": 0, "delivered": 0, "drop_undelivered": 0},
-    ]
-    out = diff_snapshots(rows)
-    assert out[0]["d_attempted"] == 100  # id=1 sorts first → its cumulative
-    assert out[1]["d_attempted"] == 50   # id=2 second → 150-100
-
-
-import pytest
-from fastapi.testclient import TestClient
-
-from app.admin.main import admin_app
-from app.admin.repositories import admin_user as admin_repo
-from app.db.session import SessionLocal
-
-
-@pytest.fixture()
-def admin_client() -> TestClient:
-    return TestClient(admin_app)
-
-
-@pytest.fixture()
-def admin_token() -> str:
-    db = SessionLocal()
-    try:
-        if admin_repo.get_by_username(db, "health_admin") is None:
-            admin_repo.create_admin(db, username="health_admin", password="pw", role="super_admin")
-    finally:
-        db.close()
-    c = TestClient(admin_app)
-    r = c.post("/admin/api/auth/login", json={"username": "health_admin", "password": "pw"})
-    return r.json()["access_token"]
-
-
-def _auth(t: str) -> dict:
-    return {"Authorization": f"Bearer {t}"}
-
-
-def _seed_two_snapshots(client: TestClient) -> None:
-    for attempted, delivered in ((100, 90), (150, 140)):
-        client.post("/api/v1/analytics/selfstat", json={
-            "device_id": "d-health", "epoch_id": "e-health",
-            "app_ver": "0.2.12(62)", "oem": "ColorOS", "os": "Android 14",
-            "events": [{"event": "video_play", "attempted": attempted,
-                        "drop_capture": 0, "delivered": delivered, "drop_undelivered": 0}],
-        })
-
-
-def test_health_overview_requires_auth(admin_client: TestClient) -> None:
-    r = admin_client.get("/admin/api/analytics-health/overview",
-                         params={"date_from": "2026-07-01T00:00:00Z", "date_to": "2030-01-01T00:00:00Z"})
-    assert r.status_code == 401
-
-
-def test_health_overview(client: TestClient, admin_client: TestClient, admin_token: str) -> None:
-    _seed_two_snapshots(client)  # 播种走公开 app(ingest 端点在 app,不在 admin_app)
-    r = admin_client.get(
-        "/admin/api/analytics-health/overview",
-        params={"date_from": "2026-07-01T00:00:00Z", "date_to": "2030-01-01T00:00:00Z"},
-        headers=_auth(admin_token),
-    )
-    assert r.status_code == 200, r.text
-    data = r.json()
-    # 两快照累计 100→150,差分后 attempted 总 150(首快照 100 + 增量 50)
-    assert data["attempted"] == 150
-    assert data["track_success_rate"] == 1.0
-
-
-def test_health_breakdown(client: TestClient, admin_client: TestClient, admin_token: str) -> None:
-    _seed_two_snapshots(client)  # 播种走公开 app
-    r = admin_client.get(
-        "/admin/api/analytics-health/breakdown",
-        params={"date_from": "2026-07-01T00:00:00Z", "date_to": "2030-01-01T00:00:00Z", "dim": "event"},
-        headers=_auth(admin_token),
-    )
-    assert r.status_code == 200, r.text
-    rows = r.json()
-    assert any(row["key"] == "video_play" for row in rows)

tests/test_analytics_selfstat.py

@@ -1,43 +0,0 @@
-"""自报计数上报端点测试。"""
-from __future__ import annotations
-
-from fastapi.testclient import TestClient
-
-
-def _payload(**over) -> dict:
-    base = {
-        "device_id": "dev-1", "epoch_id": "ep-1", "sent_at": 1700000000000,
-        "app_ver": "0.2.12(62)", "oem": "ColorOS", "os": "Android 14",
-        "batches_attempted": 10, "batches_ok": 9, "batches_fail": 1,
-        "retries": 1, "queue_depth": 2,
-        "events": [
-            {"event": "video_play", "attempted": 100, "drop_capture": 1,
-             "delivered": 95, "drop_undelivered": 2},
-        ],
-    }
-    base.update(over)
-    return base
-
-
-def test_selfstat_ingest_ok(client: TestClient) -> None:
-    r = client.post("/api/v1/analytics/selfstat", json=_payload())
-    assert r.status_code == 200, r.text
-    body = r.json()
-    assert body["ok"] is True
-    assert isinstance(body["snapshot_id"], int)
-
-
-def test_selfstat_ingest_empty_events(client: TestClient) -> None:
-    r = client.post("/api/v1/analytics/selfstat", json=_payload(events=[]))
-    assert r.status_code == 200, r.text
-    assert r.json()["ok"] is True
-
-
-def test_selfstat_ingest_db_error_returns_503(client: TestClient, monkeypatch) -> None:
-    """落库异常时端点回 503(计数链路稳定性守卫),不裸奔 500。"""
-    def _boom(*_args, **_kwargs):
-        raise RuntimeError("db gone")
-
-    monkeypatch.setattr("app.repositories.analytics_selfstat.record_selfstat", _boom)
-    r = client.post("/api/v1/analytics/selfstat", json=_payload())
-    assert r.status_code == 503, r.text

tests/test_withdraw_ledger_check.py

@@ -1,150 +0,0 @@
-"""提现现金账本校验(admin ledger-check)测试:两本物理隔离的账各自对账。
-
-历史盲区:`withdraw_ledger_check` 曾拿全部提现单去和**普通现金流水**(cash_transaction)比对,
-而 source=invite_cash 的提现单流水其实在 invite_cash_transaction 表,导致每笔邀请提现单都被
-误报「缺扣款/缺退款流水」。这里用真实提现 API 造单 + before/after 差值断言锁定修复:
-  1) 邀请提现单不再污染普通现金账的缺流水计数;
-  2) 邀请账户已被纳入对账(能抓到它自己的缺流水);
-  3) 普通现金账的原有对账未被改坏。
-
-conftest 的库是 session 级共享、测试间不清,故一律用 before/after 差值,只反映本用例造的数据。
-"""
-from __future__ import annotations
-
-from sqlalchemy import delete, select
-
-from app.admin.repositories.queries import withdraw_ledger_check
-from app.db.session import SessionLocal
-from app.models.user import User
-from app.models.wallet import CoinAccount, InviteCashTransaction
-from app.repositories import wallet as crud_wallet
-
-
-def _login(client, phone: str) -> str:
-    client.post("/api/v1/auth/sms/send", json={"phone": phone})
-    r = client.post("/api/v1/auth/sms/login", json={"phone": phone, "code": "123456"})
-    assert r.status_code == 200, r.text
-    return r.json()["access_token"]
-
-
-def _auth(token: str) -> dict[str, str]:
-    return {"Authorization": f"Bearer {token}"}
-
-
-def _patch_userinfo(monkeypatch, openid: str) -> None:
-    monkeypatch.setattr(
-        "app.integrations.wxpay.code_to_userinfo",
-        lambda code: {"openid": openid, "nickname": None, "avatar_url": None, "raw": {}},
-    )
-
-
-def _seed_balances(client, token: str, phone: str, *, cash: int = 0, invite_cash: int = 0) -> None:
-    client.get("/api/v1/wallet/account", headers=_auth(token))
-    db = SessionLocal()
-    try:
-        user = db.execute(select(User).where(User.phone == phone)).scalar_one()
-        acc = db.get(CoinAccount, user.id)
-        acc.cash_balance_cents = cash
-        acc.invite_cash_balance_cents = invite_cash
-        db.commit()
-    finally:
-        db.close()
-
-
-def _reject(bill: str, reason: str = "测试拒绝") -> None:
-    db = SessionLocal()
-    try:
-        crud_wallet.reject_withdraw(db, bill, reason)
-    finally:
-        db.close()
-
-
-def _ledger() -> dict:
-    db = SessionLocal()
-    try:
-        return withdraw_ledger_check(db)
-    finally:
-        db.close()
-
-
-def test_rejected_invite_withdraw_not_flagged_missing(client, monkeypatch) -> None:
-    """核心回归:一笔被拒绝的 invite_cash 提现单,扣款/退款流水都在 invite_cash_transaction,
-    不应让普通现金账的缺扣款/缺退款计数增加(修复前每笔会各 +1)。"""
-    before = _ledger()
-
-    _patch_userinfo(monkeypatch, "openid_lc_1")
-    token = _login(client, "13800005001")
-    _seed_balances(client, token, "13800005001", cash=0, invite_cash=500)
-    client.post("/api/v1/wallet/bind-wechat", json={"code": "c"}, headers=_auth(token))
-    r = client.post(
-        "/api/v1/wallet/withdraw",
-        json={"amount_cents": 200, "source": "invite_cash"},
-        headers=_auth(token),
-    )
-    assert r.status_code == 200, r.text
-    _reject(r.json()["out_bill_no"])  # rejected + 退款流水落 invite_cash_transaction
-
-    after = _ledger()
-
-    # 普通现金账不该因这笔 invite 单产生缺流水(修复前会各 +1 → 就是页面上看到的误报)
-    assert after["missing_withdraw_txn_count"] == before["missing_withdraw_txn_count"]
-    assert after["missing_refund_txn_count"] == before["missing_refund_txn_count"]
-    # 邀请账扣款 + 退款流水齐全,邀请账自身也不该缺
-    assert after["invite_missing_withdraw_txn_count"] == before["invite_missing_withdraw_txn_count"]
-    assert after["invite_missing_refund_txn_count"] == before["invite_missing_refund_txn_count"]
-
-
-def test_invite_ledger_detects_missing_withdraw_txn(client, monkeypatch) -> None:
-    """删掉一笔 invite 提现单的扣款流水 → 邀请账缺扣款计数 +1、ok=False,
-    证明邀请账户已真正纳入对账(修复前邀请账完全不校验、永远报不出问题)。"""
-    _patch_userinfo(monkeypatch, "openid_lc_2")
-    token = _login(client, "13800005002")
-    _seed_balances(client, token, "13800005002", cash=0, invite_cash=500)
-    client.post("/api/v1/wallet/bind-wechat", json={"code": "c"}, headers=_auth(token))
-    r = client.post(
-        "/api/v1/wallet/withdraw",
-        json={"amount_cents": 200, "source": "invite_cash"},
-        headers=_auth(token),
-    )
-    bill = r.json()["out_bill_no"]
-
-    before = _ledger()
-    db = SessionLocal()
-    try:
-        db.execute(
-            delete(InviteCashTransaction).where(
-                InviteCashTransaction.ref_id == bill,
-                InviteCashTransaction.biz_type == "invite_withdraw",
-            )
-        )
-        db.commit()
-    finally:
-        db.close()
-    after = _ledger()
-
-    assert (
-        after["invite_missing_withdraw_txn_count"]
-        == before["invite_missing_withdraw_txn_count"] + 1
-    )
-    assert after["ok"] is False
-
-
-def test_coin_cash_withdraw_still_reconciled(client, monkeypatch) -> None:
-    """普通现金 coin_cash 提现单齐全时不新增缺流水(确保分账改造没弄坏原有普通现金对账)。"""
-    _patch_userinfo(monkeypatch, "openid_lc_3")
-    token = _login(client, "13800005003")
-    _seed_balances(client, token, "13800005003", cash=500, invite_cash=0)
-    client.post("/api/v1/wallet/bind-wechat", json={"code": "c"}, headers=_auth(token))
-
-    before = _ledger()
-    r = client.post(
-        "/api/v1/wallet/withdraw",
-        json={"amount_cents": 200, "source": "coin_cash"},
-        headers=_auth(token),
-    )
-    assert r.status_code == 200, r.text
-    after = _ledger()
-
-    # 普通现金提现扣款流水随单写入 cash_transaction,缺扣款计数不变;邀请账更不受影响
-    assert after["missing_withdraw_txn_count"] == before["missing_withdraw_txn_count"]
-    assert after["invite_missing_withdraw_txn_count"] == before["invite_missing_withdraw_txn_count"]