提现档位后端权威化:withdraw-info按source下发tiers,新人档历史一次性+常规档每日限次/选一额度,下单加档位闸防绕过(7-9)

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

alembic/versions/admin_user_pages_override.py

@@ -0,0 +1,35 @@
+"""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/permissions.py

@@ -9,6 +9,9 @@ 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,14 +26,17 @@ def create_admin(
     password: str,
     role: str = "operator",
     plain_password: str | None = None,
+    pages_override: list[str] | None = None,
 ) -> AdminUser:
     """建管理员。plain_password 非空则额外留存明文(后台 UI 建的账号传,供权限管理页复看);
-    脚本/起后台建账号不传(留 None → 前端不显示密码)。"""
+    脚本/起后台建账号不传(留 None → 前端不显示密码)。
+    pages_override:role=="custom" 时传专属可见页 key 列表;普通角色为 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/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 SUPER_ADMIN_ROLE
+from app.admin.permissions import 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.schemas.admin import AdminCreateRequest, AdminUpdateRequest
@@ -26,14 +26,22 @@ def _active_super_count(db: AdminDb) -> int:
 
 
 def _validate_role(db: AdminDb, role: str) -> None:
-    """角色必须是 super_admin 或 admin_role 表里已存在的角色,否则 400。"""
+    """角色必须是 super_admin / custom(自定义) / admin_role 表里已存在的角色,否则 400。"""
-    if role == SUPER_ADMIN_ROLE:
+    if role in (SUPER_ADMIN_ROLE, CUSTOM_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] = []
@@ -51,13 +59,16 @@ 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}, ip=get_client_ip(request), commit=True,
+        detail={"username": new.username, "role": new.role, "pages_override": override},
+        ip=get_client_ip(request), commit=True,
     )
     return AdminOut.model_validate(new)
 
@@ -89,7 +100,8 @@ def update_admin(
         _validate_role(db, body.role)
 
     changes: dict = {}
-    if body.role is not None and body.role != target.role:
+    role_changed = body.role is not None and body.role != target.role
+    if role_changed:
         changes["role"] = {"before": target.role, "after": body.role}
         target.role = body.role
     if body.status is not None and body.status != target.status:
@@ -99,6 +111,17 @@ 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/auth.py

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

app/admin/schemas/admin.py

@@ -13,14 +13,18 @@ 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/auth.py

@@ -21,8 +21,11 @@ class AdminOut(BaseModel):
     created_at: datetime
     last_login_at: datetime | None = None
     # 该管理员当前角色的有效可见页(= 左侧导航项 key);仅登录 / /me 填充,列表接口默认空。
-    # 前端据此过滤左侧导航(super_admin = 全部页)。见 app/admin/permissions.py。
+    # 前端据此过滤左侧导航(super_admin = 全部页;role=="custom" = pages_override)。见 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/api/v1/wallet.py

@@ -43,6 +43,7 @@ from app.schemas.welfare import (
     WithdrawRequest,
     WithdrawResultOut,
     WithdrawStatusOut,
+    WithdrawTierOut,
 )
 
 logger = logging.getLogger("shagua.wallet")
@@ -173,8 +174,15 @@ def unbind_wechat(
     return UnbindWechatResultOut(bound=False)
 
 
-@router.get("/withdraw-info", response_model=WithdrawInfoOut, summary="提现额度/绑定状态/免确认开关")
+@router.get("/withdraw-info", response_model=WithdrawInfoOut, summary="提现额度/绑定状态/免确认开关/档位")
-def withdraw_info(user: CurrentUser, db: DbSession) -> WithdrawInfoOut:
+def withdraw_info(
+    user: CurrentUser,
+    db: DbSession,
+    source: str = Query(
+        "coin_cash",
+        description="提现账户:coin_cash(福利页,下发 tiers 档位) / invite_cash(邀请页,tiers 为空走旧逻辑)",
+    ),
+) -> WithdrawInfoOut:
     u = db.get(User, user.id)
     # 顺带同步免确认授权状态(捕获首单确认后已生效的授权 pending→active),让开关展示实时
     auth = crud_wallet.sync_transfer_auth(db, user.id)
@@ -185,6 +193,7 @@ def withdraw_info(user: CurrentUser, db: DbSession) -> WithdrawInfoOut:
         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"),
+        tiers=[WithdrawTierOut(**t) for t in crud_wallet.withdraw_tier_states(db, user.id, source)],
     )
 
 
@@ -218,6 +227,9 @@ def withdraw(req: WithdrawRequest, user: CurrentUser, db: DbSession) -> Withdraw
             status_code=status.HTTP_409_CONFLICT,
             detail="已有提现申请正在审核或打款中,请处理完成后再申请",
         ) from e
+    except crud_wallet.WithdrawTierUnavailableError as e:
+        # 福利页档位闸(7-9):次数满/已选其他额度。正常客户端已按 tiers 预拦,此处兜底防绕过。
+        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="今日额度已达上限") from e
     except crud_wallet.InsufficientCashError as e:
         raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="现金余额不足") from e
 

app/core/rewards.py

@@ -6,6 +6,7 @@
 from __future__ import annotations
 
 from datetime import date, datetime, timedelta, timezone
+from typing import NamedTuple
 
 # 业务时区:签到的"今天"按北京时间算,不能用 UTC。
 # 否则 UTC+8 的凌晨 0~8 点会被算成 UTC 的前一天,导致签到日期错位。
@@ -52,6 +53,30 @@ WITHDRAW_MIN_CENTS: int = 10
 WITHDRAW_MAX_CENTS: int = 5_000_000  # 5 万元
 
 
+# ===== 提现档位(福利页 coin_cash;7-9 对齐原型 withdrawal.html)=====
+# 后端是档位唯一真相源:withdraw-info 按此下发,create_withdraw 按此校验(防绕过客户端刷)。
+# 规则(2026-07-09 拍板):
+#   - 新人档(is_newbie):账号历史一次性,"发起就算用过"(任意状态含被拒),用过即不再下发;
+#     0.1 与 0.3 各自独立同天可各提一次,且不参与常规档"每日选一个额度"互斥。
+#   - 常规档:按北京日计次(0.5×3 / 10×1 / 20×1),三档每天只能选一个。
+# invite_cash(邀请页)本轮无档位概念,不在此表。改档位=改这里发版。
+class WithdrawTier(NamedTuple):
+    amount_cents: int
+    label: str          # 客户端档位方块展示文案
+    badge: str | None   # 角标文案;None=无角标
+    daily_limit: int    # 每日次数上限(新人档的"历史一次性"另由 is_newbie 判定)
+    is_newbie: bool
+
+
+WITHDRAW_TIERS_COIN_CASH: tuple[WithdrawTier, ...] = (
+    WithdrawTier(10, "0.1", "新人福利", 1, True),
+    WithdrawTier(30, "0.3", "新人福利", 1, True),
+    WithdrawTier(50, "0.5", None, 3, False),
+    WithdrawTier(1000, "10", None, 1, False),
+    WithdrawTier(2000, "20", None, 1, False),
+)
+
+
 # ===== 一次性任务(领一次,user_task 去重)=====
 TASK_ENABLE_NOTIFICATION = "enable_notification"
 

app/models/admin.py

@@ -28,8 +28,11 @@ class AdminUser(Base):
     # 明文登录密码:仅「后台 UI 创建/重置」的管理员留存,供超管在权限管理页复看转交。
     # 脚本/起后台时建的超管账号不写(为 None → 前端「不显示密码」)。⚠️ 内部工具便利取舍,见 create/list。
     plain_password: Mapped[str | None] = mapped_column(String(128), nullable=True)
-    # super_admin(全权+管账号)/ finance(钱:提现+金币)/ operator(用户+反馈+大盘)
+    # super_admin(全权+管账号)/ finance(钱:提现+金币)/ operator(用户+反馈+大盘)/ custom(按人自定义)
     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/repositories/ad_feed_reward.py

@@ -176,10 +176,19 @@ 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="feed_ad_reward", ref_id=client_event_id,
+        biz_type=reward_biz, ref_id=client_event_id,
-        remark="信息流广告奖励",
+        remark=reward_remark,
     )
     rec = AdFeedRewardRecord(
         client_event_id=client_event_id,

app/repositories/wallet.py

@@ -11,7 +11,7 @@ import unicodedata
 import uuid
 from datetime import datetime, timedelta, timezone
 
-from sqlalchemy import select, update
+from sqlalchemy import func, select, update
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm import Session
 
@@ -67,6 +67,10 @@ class WithdrawTooFrequentError(Exception):
     """提现申请过于频繁,或已有未完成提现单。"""
 
 
+class WithdrawTierUnavailableError(Exception):
+    """该档位今日不可提:次数已满,或今天已选了其他额度(7-9 福利页档位规则)。"""
+
+
 class WithdrawTransferError(Exception):
     """调用微信转账失败(已退回余额)。"""
 
@@ -605,6 +609,89 @@ def _settle_after_ambiguous(db: Session, order: WithdrawOrder, reason: str) -> N
         db.commit()
 
 
+def _beijing_today_start_utc() -> datetime:
+    """北京时今日 0 点(转 UTC)。WithdrawOrder.created_at 是 func.now()(UTC)存储,
+    比较时统一转 UTC,与 admin 看板 today_start 同口径(admin/repositories/queries.py)。"""
+    return (
+        datetime.now(rewards.CN_TZ)
+        .replace(hour=0, minute=0, second=0, microsecond=0)
+        .astimezone(timezone.utc)
+    )
+
+
+def withdraw_tier_states(db: Session, user_id: int, source: str = "coin_cash") -> list[dict]:
+    """福利页(coin_cash)提现档位的可提现状态。withdraw-info 下发与 create_withdraw 校验共用此口径。
+
+    规则(2026-07-09 拍板,7-9提现ui对齐):
+      - 新人档(0.1/0.3):账号历史一次性——只要发起过(**任意状态**,含被拒/失败,"发起就算")
+        即视为已用,直接**从返回列表消失**;两档各自独立互不影响,不参与"每日选一个额度"互斥。
+      - 常规档(0.5×3 / 10×1 / 20×1):按北京日计次,"发起就算占用"(当天创建的单不论最终状态
+        都计入,被拒/失败不退当天名额);三档每天只能选一个,选定后其余两档当天 other_tier_selected。
+      - invite_cash 本轮无档位概念 → 返回空列表(邀请页客户端仍用本地写死档位,行为不变)。
+    余额是否足够由客户端本地判断(余额随兑换实时变化,不在此快照)。
+    """
+    if source != "coin_cash":
+        return []
+    tiers = rewards.WITHDRAW_TIERS_COIN_CASH
+    amounts = [t.amount_cents for t in tiers]
+    newbie_amounts = [t.amount_cents for t in tiers if t.is_newbie]
+    # 新人档历史是否用过:任意时间、任意状态("发起就算")
+    used_newbie: set[int] = set(
+        db.execute(
+            select(WithdrawOrder.amount_cents)
+            .distinct()
+            .where(
+                WithdrawOrder.user_id == user_id,
+                WithdrawOrder.source == "coin_cash",
+                WithdrawOrder.amount_cents.in_(newbie_amounts),
+            )
+        ).scalars()
+    ) if newbie_amounts else set()
+    # 今日(北京日)每档已发起次数(任意状态)
+    today_counts: dict[int, int] = {
+        int(amount): int(cnt)
+        for amount, cnt in db.execute(
+            select(WithdrawOrder.amount_cents, func.count(WithdrawOrder.id))
+            .where(
+                WithdrawOrder.user_id == user_id,
+                WithdrawOrder.source == "coin_cash",
+                WithdrawOrder.amount_cents.in_(amounts),
+                WithdrawOrder.created_at >= _beijing_today_start_utc(),
+            )
+            .group_by(WithdrawOrder.amount_cents)
+        )
+    }
+    # "每日选一个额度":今天发起过的常规档(新人档不算)
+    selected_regular = next(
+        (t.amount_cents for t in tiers if not t.is_newbie and today_counts.get(t.amount_cents, 0) > 0),
+        None,
+    )
+    out: list[dict] = []
+    for t in tiers:
+        if t.is_newbie:
+            if t.amount_cents in used_newbie:
+                continue  # 用过即消失,不再下发
+            out.append({
+                "amount_cents": t.amount_cents, "label": t.label, "badge": t.badge,
+                "is_newbie": True, "available": True, "disabled_reason": None,
+                "remaining_today": 1,
+            })
+            continue
+        used = today_counts.get(t.amount_cents, 0)
+        if selected_regular is not None and selected_regular != t.amount_cents:
+            available, reason, remaining = False, "other_tier_selected", 0
+        elif used >= t.daily_limit:
+            available, reason, remaining = False, "quota_exhausted", 0
+        else:
+            available, reason, remaining = True, None, t.daily_limit - used
+        out.append({
+            "amount_cents": t.amount_cents, "label": t.label, "badge": t.badge,
+            "is_newbie": False, "available": available, "disabled_reason": reason,
+            "remaining_today": remaining,
+        })
+    return out
+
+
 def create_withdraw(
     db: Session,
     user_id: int,
@@ -660,6 +747,19 @@ def create_withdraw(
     if active_order_id is not None:
         raise WithdrawTooFrequentError
 
+    # 福利页档位闸(7-9):coin_cash 只能提预设档位,且该档今日可提(服务端权威口径,防绕过
+    # 客户端刷)。放在幂等返回/在途互斥之后:同号重试仍原样返回旧单,不被档位闸误杀。
+    # allow_sub_min(0.01 调试直发)保持原样放行,不受档位约束;invite_cash 本轮无档位概念不校验。
+    if source == "coin_cash" and not allow_sub_min:
+        tier_state = next(
+            (t for t in withdraw_tier_states(db, user_id, source) if t["amount_cents"] == amount_cents),
+            None,
+        )
+        if tier_state is None:  # 非预设档位金额,或新人档已用过(已从列表消失)
+            raise InvalidWithdrawAmountError
+        if not tier_state["available"]:
+            raise WithdrawTierUnavailableError
+
     # 账户须存在(原子扣款的 UPDATE 不会建账户)
     get_or_create_account(db, user_id, commit=True)
 

app/schemas/welfare.py

@@ -75,6 +75,21 @@ class ExchangeResultOut(BaseModel):
 
 # ===== 提现(现金 → 微信零钱) =====
 
+class WithdrawTierOut(BaseModel):
+    """提现档位(福利页 coin_cash;7-9 对齐原型)。served by rewards.WITHDRAW_TIERS_COIN_CASH。"""
+
+    amount_cents: int = Field(..., description="档位金额(分)")
+    label: str = Field(..., description="档位方块展示文案,如 0.1 / 10")
+    badge: str | None = Field(None, description="角标文案(如 新人福利);无则空")
+    is_newbie: bool = Field(False, description="新人档:历史一次性,用过后不再下发;免广告直提")
+    available: bool = Field(True, description="当前是否可提(次数/选一额度口径;余额由客户端自判)")
+    disabled_reason: str | None = Field(
+        None,
+        description="不可提原因:quota_exhausted(今日次数满) / other_tier_selected(今日已选其他额度)",
+    )
+    remaining_today: int = Field(0, description="今日剩余可提次数")
+
+
 class WithdrawInfoOut(BaseModel):
     min_cents: int = Field(..., description="单次最低提现(分)")
     max_cents: int = Field(..., description="单次最高提现(分)")
@@ -84,6 +99,10 @@ class WithdrawInfoOut(BaseModel):
     transfer_auth_enabled: bool = Field(
         False, description="是否已开启免确认到账(开启后提现免跳微信确认,直接到账)"
     )
+    tiers: list[WithdrawTierOut] = Field(
+        default_factory=list,
+        description="提现档位(source=coin_cash 下发;invite_cash 为空,客户端走旧逻辑)",
+    )
 
 
 # ===== 免确认收款授权(用户授权免确认模式)=====

scripts/seed_coinhistory_labels_test.py

@@ -0,0 +1,166 @@
+"""收益明细「金币记录」文案验证脚手架(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_roles.py

@@ -150,3 +150,80 @@ 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_invite_cash_withdraw.py

@@ -143,14 +143,15 @@ def test_two_accounts_withdraw_independent(client, monkeypatch) -> None:
     _reject(r1.json()["out_bill_no"])    # 退回 invite_cash + 结清活跃单
     r2 = client.post(
         "/api/v1/wallet/withdraw",
-        json={"amount_cents": 100, "source": "coin_cash"},
+        # 50 分 = 0.5 元档(7-9 起 coin_cash 只能提预设档位)
+        json={"amount_cents": 50, "source": "coin_cash"},
         headers=_auth(token),
     )
     assert r2.json()["status"] == "reviewing"
 
     cash, invite_cash = _balances(client, token)
     assert invite_cash == 500    # 已退回
-    assert cash == 300           # 扣了 cash 100
+    assert cash == 350           # 扣了 cash 50
 
 
 def test_invite_me_returns_reward_stats(client) -> None:
@@ -177,7 +178,8 @@ def test_withdraw_orders_source_filter(client, monkeypatch) -> None:
     _reject(r1.json()["out_bill_no"])    # 结清,才能提第二笔
     client.post(
         "/api/v1/wallet/withdraw",
-        json={"amount_cents": 100, "source": "coin_cash"},
+        # 50 分 = 0.5 元档(7-9 起 coin_cash 只能提预设档位)
+        json={"amount_cents": 50, "source": "coin_cash"},
         headers=_auth(token),
     )
 

tests/test_withdraw_ledger_check.py

@@ -139,7 +139,8 @@ def test_coin_cash_withdraw_still_reconciled(client, monkeypatch) -> None:
     before = _ledger()
     r = client.post(
         "/api/v1/wallet/withdraw",
-        json={"amount_cents": 200, "source": "coin_cash"},
+        # 50 分 = 0.5 元档(7-9 起 coin_cash 只能提预设档位)
+        json={"amount_cents": 50, "source": "coin_cash"},
         headers=_auth(token),
     )
     assert r.status_code == 200, r.text

tests/test_withdraw_tiers.py

@@ -0,0 +1,186 @@
+"""福利页(coin_cash)提现档位规则测试(7-9提现ui对齐)。
+
+规则(2026-07-09 拍板):
+  - 档位 0.1/0.3(新人,历史一次性,免广告)+ 0.5(日3次)/10/20(日1次)。
+  - 计次口径"发起就算":当天创建的单不论最终状态(含被拒)都占名额;新人档任何状态都算用过。
+  - 常规三档每天只能选一个;新人档不参与该互斥,两个新人档同天可各提一次。
+  - invite_cash 无档位概念:tiers 为空、下单不走档位闸(邀请页行为不变)。
+wxpay 调用全部 monkeypatch;现金余额 DB 直灌(同 test_withdraw.py 套路)。
+"""
+from __future__ import annotations
+
+from sqlalchemy import select
+
+from app.db.session import SessionLocal
+from app.models.user import User
+from app.models.wallet import CoinAccount
+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 _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 _patch_userinfo(monkeypatch, openid: str) -> None:
+    monkeypatch.setattr(
+        "app.integrations.wxpay.code_to_userinfo",
+        lambda code: {"openid": openid, "nickname": "昵称", "avatar_url": None, "raw": {}},
+    )
+
+
+def _reject(bill: str) -> None:
+    """管理员拒绝:退款结清活跃单(便于同用户继续发起下一笔;名额按'发起就算'仍占)。"""
+    db = SessionLocal()
+    try:
+        crud_wallet.reject_withdraw(db, bill, "test")
+    finally:
+        db.close()
+
+
+def _withdraw(client, token: str, cents: int):
+    return client.post(
+        "/api/v1/wallet/withdraw",
+        json={"amount_cents": cents, "source": "coin_cash"},
+        headers=_auth(token),
+    )
+
+
+def _tiers(client, token: str, source: str = "coin_cash") -> list[dict]:
+    r = client.get(
+        "/api/v1/wallet/withdraw-info", params={"source": source}, headers=_auth(token)
+    )
+    assert r.status_code == 200, r.text
+    return r.json()["tiers"]
+
+
+def test_withdraw_info_tiers_full_and_invite_empty(client, monkeypatch) -> None:
+    """新用户 coin_cash 下发 5 档(新人角标齐);invite_cash tiers 为空。"""
+    token = _login(client, "13800006001")
+    tiers = _tiers(client, token)
+    assert [t["amount_cents"] for t in tiers] == [10, 30, 50, 1000, 2000]
+    assert [t["label"] for t in tiers] == ["0.1", "0.3", "0.5", "10", "20"]
+    assert tiers[0]["badge"] == "新人福利" and tiers[0]["is_newbie"] is True
+    assert tiers[1]["badge"] == "新人福利" and tiers[1]["is_newbie"] is True
+    assert all(t["available"] for t in tiers)
+    assert tiers[2]["remaining_today"] == 3  # 0.5 日 3 次
+    assert _tiers(client, token, source="invite_cash") == []
+
+
+def test_newbie_tiers_independent_and_once_forever(client, monkeypatch) -> None:
+    """0.1 提过(即使被拒)→ 永久消失;同天 0.3 仍可提;新人档不锁常规档。"""
+    _patch_userinfo(monkeypatch, "openid_tier_2")
+    token = _login(client, "13800006002")
+    _seed_balances(client, token, "13800006002", cash=5000)
+    client.post("/api/v1/wallet/bind-wechat", json={"code": "c"}, headers=_auth(token))
+
+    r = _withdraw(client, token, 10)  # 0.1 新人档
+    assert r.status_code == 200, r.text
+    _reject(r.json()["out_bill_no"])  # 被拒也算用过("发起就算")
+
+    tiers = _tiers(client, token)
+    amounts = [t["amount_cents"] for t in tiers]
+    assert 10 not in amounts          # 0.1 消失
+    assert 30 in amounts              # 0.3 还在,同天仍可提
+    # 新人档不参与"选一个额度":常规三档全部仍可提
+    regular = {t["amount_cents"]: t for t in tiers if not t["is_newbie"]}
+    assert all(regular[a]["available"] for a in (50, 1000, 2000))
+
+    r = _withdraw(client, token, 30)  # 同天 0.3 照提
+    assert r.status_code == 200, r.text
+    _reject(r.json()["out_bill_no"])
+
+    # 0.1 已用过,重提 → 非法金额(档位已消失)
+    r = _withdraw(client, token, 10)
+    assert r.status_code == 400, r.text
+
+
+def test_regular_daily_select_one_tier(client, monkeypatch) -> None:
+    """当天提过 0.5 → 10/20 置灰 other_tier_selected,下单 409;0.5 还能继续提(3 次内)。"""
+    _patch_userinfo(monkeypatch, "openid_tier_3")
+    token = _login(client, "13800006003")
+    _seed_balances(client, token, "13800006003", cash=10_000)
+    client.post("/api/v1/wallet/bind-wechat", json={"code": "c"}, headers=_auth(token))
+
+    r = _withdraw(client, token, 50)
+    assert r.status_code == 200, r.text
+    _reject(r.json()["out_bill_no"])  # 结清活跃单;当天名额仍占("发起就算")
+
+    tiers = {t["amount_cents"]: t for t in _tiers(client, token)}
+    assert tiers[50]["available"] and tiers[50]["remaining_today"] == 2
+    assert not tiers[1000]["available"] and tiers[1000]["disabled_reason"] == "other_tier_selected"
+    assert not tiers[2000]["available"] and tiers[2000]["disabled_reason"] == "other_tier_selected"
+
+    r = _withdraw(client, token, 1000)  # 选一额度互斥 → 409
+    assert r.status_code == 409, r.text
+    assert "今日额度已达上限" in r.json()["detail"]
+
+    r = _withdraw(client, token, 50)  # 0.5 第 2 次照常
+    assert r.status_code == 200, r.text
+
+
+def test_regular_daily_quota_exhausted(client, monkeypatch) -> None:
+    """0.5 日 3 次:第 4 次 409;tiers 显示 quota_exhausted。"""
+    _patch_userinfo(monkeypatch, "openid_tier_4")
+    token = _login(client, "13800006004")
+    _seed_balances(client, token, "13800006004", cash=10_000)
+    client.post("/api/v1/wallet/bind-wechat", json={"code": "c"}, headers=_auth(token))
+
+    for _ in range(3):
+        r = _withdraw(client, token, 50)
+        assert r.status_code == 200, r.text
+        _reject(r.json()["out_bill_no"])
+
+    tiers = {t["amount_cents"]: t for t in _tiers(client, token)}
+    assert not tiers[50]["available"]
+    assert tiers[50]["disabled_reason"] == "quota_exhausted"
+    assert tiers[50]["remaining_today"] == 0
+
+    r = _withdraw(client, token, 50)
+    assert r.status_code == 409, r.text
+
+
+def test_non_tier_amount_rejected_for_coin_cash(client, monkeypatch) -> None:
+    """coin_cash 只能提预设档位:任意其他金额 400(防绕过客户端刷)。"""
+    _patch_userinfo(monkeypatch, "openid_tier_5")
+    token = _login(client, "13800006005")
+    _seed_balances(client, token, "13800006005", cash=10_000)
+    client.post("/api/v1/wallet/bind-wechat", json={"code": "c"}, headers=_auth(token))
+
+    r = _withdraw(client, token, 123)
+    assert r.status_code == 400, r.text
+
+
+def test_invite_cash_not_gated_by_tiers(client, monkeypatch) -> None:
+    """invite_cash 不走档位闸:非档位金额(200=2元)照常下单——邀请页行为不变。"""
+    _patch_userinfo(monkeypatch, "openid_tier_6")
+    token = _login(client, "13800006006")
+    _seed_balances(client, token, "13800006006", 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
+    assert r.json()["status"] == "reviewing"