diff --git a/app/admin/repositories/ad_audit.py b/app/admin/repositories/ad_audit.py index 064b036..e08ae68 100644 --- a/app/admin/repositories/ad_audit.py +++ b/app/admin/repositories/ad_audit.py @@ -50,7 +50,7 @@ def _reward_video_rows( AdRewardRecord.reward_date == date, AdRewardRecord.reward_scene == "reward_video", ) - .order_by(AdRewardRecord.user_id, AdRewardRecord.created_at) + .order_by(AdRewardRecord.user_id, AdRewardRecord.created_at, AdRewardRecord.id) ) if user_id is not None: stmt = stmt.where(AdRewardRecord.user_id == user_id) @@ -127,7 +127,7 @@ def _feed_rows(db: Session, *, date: str, user_id: int | None) -> list[dict]: stmt = ( select(AdFeedRewardRecord) .where(AdFeedRewardRecord.reward_date == date) - .order_by(AdFeedRewardRecord.user_id, AdFeedRewardRecord.created_at) + .order_by(AdFeedRewardRecord.user_id, AdFeedRewardRecord.created_at, AdFeedRewardRecord.id) ) if user_id is not None: stmt = stmt.where(AdFeedRewardRecord.user_id == user_id) @@ -205,7 +205,7 @@ def ad_coin_audit( rows.extend(_reward_video_rows(db, date=date, user_id=user_id)) if scene in (None, "feed"): rows.extend(_feed_rows(db, date=date, user_id=user_id)) - rows.sort(key=lambda r: r["created_at"], reverse=True) + rows.sort(key=lambda r: (r["created_at"], r["record_id"]), reverse=True) total = len(rows) mismatch_count = sum(1 for r in rows if not r["matched"]) diff --git a/app/admin/repositories/queries.py b/app/admin/repositories/queries.py index d7293f0..02b23f4 100644 --- a/app/admin/repositories/queries.py +++ b/app/admin/repositories/queries.py @@ -57,7 +57,7 @@ def list_users( """用户列表(admin 全量)。支持手机号前缀 / 渠道 / 状态 / 昵称模糊 / 注册·最近登录时间范围筛选, 按 id·注册时间·最近登录排序。**offset 分页**(cursor=offset):任意列排序下游标语义统一, 代价是翻页期间数据变动可能错位一条——admin 低频场景可接受(同 [list_all_withdraw_orders])。 - 日期入参统一转 UTC naive 比较(User 时间均为 UTC naive,见 _as_utc_naive)。""" + 日期入参统一转 tz-aware UTC 比较(列为 timestamptz,见 _as_utc)。""" stmt = select(User) if phone: stmt = stmt.where(User.phone.like(f"{phone}%")) # 前缀匹配 @@ -68,13 +68,13 @@ def list_users( if nickname and nickname.strip(): stmt = stmt.where(User.nickname.ilike(f"%{nickname.strip()}%")) if created_from is not None: - stmt = stmt.where(User.created_at >= _as_utc_naive(created_from)) + stmt = stmt.where(User.created_at >= _as_utc(created_from)) if created_to is not None: - stmt = stmt.where(User.created_at <= _as_utc_naive(created_to)) + stmt = stmt.where(User.created_at <= _as_utc(created_to)) if last_login_from is not None: - stmt = stmt.where(User.last_login_at >= _as_utc_naive(last_login_from)) + stmt = stmt.where(User.last_login_at >= _as_utc(last_login_from)) if last_login_to is not None: - stmt = stmt.where(User.last_login_at <= _as_utc_naive(last_login_to)) + stmt = stmt.where(User.last_login_at <= _as_utc(last_login_to)) sort_cols = { "id": User.id, @@ -190,16 +190,16 @@ def list_all_withdraw_orders( date_col = WithdrawOrder.updated_at if date_field == "updated_at" else WithdrawOrder.created_at if date_from is not None: - stmt = stmt.where(date_col >= _as_utc_naive(date_from)) + stmt = stmt.where(date_col >= _as_utc(date_from)) if date_to is not None: - stmt = stmt.where(date_col <= _as_utc_naive(date_to)) + stmt = stmt.where(date_col <= _as_utc(date_to)) - now = datetime.now(timezone.utc).replace(tzinfo=None) + # tz-aware:列为 timestamptz,比较绝对时刻、与 DB 会话时区无关(同 _as_utc / stats.py) + now = datetime.now(timezone.utc) today_start = ( datetime.now(ZoneInfo("Asia/Shanghai")) .replace(hour=0, minute=0, second=0, microsecond=0) .astimezone(timezone.utc) - .replace(tzinfo=None) ) if quick_filter == "abnormal": stmt = stmt.where( @@ -254,11 +254,16 @@ def list_all_withdraw_orders( return items, next_cursor -def _as_utc_naive(value: datetime) -> datetime: - """前端传 ISO 时间;DB 当前按 UTC naive 比较最稳(SQLite/本地开发一致)。""" +def _as_utc(value: datetime) -> datetime: + """前端传 ISO 时间 → 统一成 tz-aware UTC 再比较。 + + 所有时间列均为 `DateTime(timezone=True)`(Postgres timestamptz);用 tz-aware 绑定参数 + 比较的是绝对时刻,与 DB 会话时区无关、恒正确。曾用 naive UTC,正确性依赖会话 TimeZone=UTC, + 生产会话非 UTC 时筛选边界会整体偏移——故统一 tz-aware(与 stats.py / withdraw_summary 一致)。 + 无时区入参按 UTC 解释。""" if value.tzinfo is None: - return value - return value.astimezone(timezone.utc).replace(tzinfo=None) + return value.replace(tzinfo=timezone.utc) + return value.astimezone(timezone.utc) def list_feedbacks( @@ -266,15 +271,41 @@ def list_feedbacks( *, status: str | None = None, user_id: int | None = None, + content: str | None = None, + created_from: datetime | None = None, + created_to: datetime | None = None, + sort_by: str = "id", + sort_order: str = "desc", limit: int = 20, cursor: int | None = None, ) -> tuple[list[Feedback], int | None]: + """反馈工单列表。支持 状态 / 用户ID / 内容模糊 / 提交时间范围 筛选,按 id·提交时间排序。 + **offset 分页**(cursor=offset):任意列排序下游标语义统一(同 [list_users]),代价是翻页期间 + 数据变动可能错位一条——admin 低频场景可接受。created_at 为 timestamptz,日期入参统一转 tz-aware UTC 比较。""" stmt = select(Feedback) if status: stmt = stmt.where(Feedback.status == status) if user_id is not None: stmt = stmt.where(Feedback.user_id == user_id) - return cursor_paginate(db, stmt, Feedback.id, limit=limit, cursor=cursor) + if content and content.strip(): + stmt = stmt.where(Feedback.content.ilike(f"%{content.strip()}%")) + if created_from is not None: + stmt = stmt.where(Feedback.created_at >= _as_utc(created_from)) + if created_to is not None: + stmt = stmt.where(Feedback.created_at <= _as_utc(created_to)) + + sort_cols = {"id": Feedback.id, "created_at": Feedback.created_at} + sort_col = sort_cols.get(sort_by, Feedback.id) + order_fn = asc if sort_order == "asc" else desc + id_order = asc(Feedback.id) if sort_order == "asc" else desc(Feedback.id) + stmt = stmt.order_by(order_fn(sort_col), id_order) + + offset = max(cursor or 0, 0) + rows = list(db.execute(stmt.offset(offset).limit(limit + 1)).scalars().all()) + has_more = len(rows) > limit + items = rows[:limit] + next_cursor = offset + limit if has_more else None + return items, next_cursor def get_withdraw_by_out_bill_no(db: Session, out_bill_no: str) -> WithdrawOrder | None: diff --git a/app/admin/routers/admins.py b/app/admin/routers/admins.py index c3b8483..9f7ac15 100644 --- a/app/admin/routers/admins.py +++ b/app/admin/routers/admins.py @@ -17,6 +17,12 @@ router = APIRouter( ) +def _active_super_count(db: AdminDb) -> int: + return sum( + 1 for a in admin_repo.list_admins(db) if a.role == "super_admin" and a.status == "active" + ) + + @router.get("", response_model=list[AdminOut], summary="管理员列表") def list_admins(db: AdminDb) -> list[AdminOut]: return [AdminOut.model_validate(a) for a in admin_repo.list_admins(db)] @@ -48,16 +54,29 @@ def update_admin( if admin_id == admin.id and body.status == "disabled": raise HTTPException(status_code=400, detail="不能禁用自己") + # 防自锁:降级 / 禁用某个 super_admin 前,确认操作后仍至少剩 1 个 active super_admin, + # 否则会进入「零可用超管」死局——本路由仅 super 可进,只能改库恢复。 + demotes_super = ( + target.role == "super_admin" + and target.status == "active" + and ( + (body.role is not None and body.role != "super_admin") + or body.status == "disabled" + ) + ) + if demotes_super and _active_super_count(db) <= 1: + raise HTTPException(status_code=400, detail="不能降级/禁用最后一个超级管理员") + changes: dict = {} - if body.role is not None: + if body.role is not None and body.role != target.role: + changes["role"] = {"before": target.role, "after": body.role} target.role = body.role - changes["role"] = body.role - if body.status is not None: + if body.status is not None and body.status != target.status: + changes["status"] = {"before": target.status, "after": body.status} target.status = body.status - changes["status"] = body.status if body.password is not None: - target.password_hash = hash_password(body.password) changes["password"] = "reset" + target.password_hash = hash_password(body.password) if not changes: raise HTTPException(status_code=400, detail="无任何变更字段") db.commit() diff --git a/app/admin/routers/price_report.py b/app/admin/routers/price_report.py index b29f271..f036605 100644 --- a/app/admin/routers/price_report.py +++ b/app/admin/routers/price_report.py @@ -60,7 +60,9 @@ def approve_price_report( admin: Annotated[AdminUser, Depends(require_role("operator"))], db: AdminDb, ) -> OkResponse: - rep = db.get(PriceReport, report_id) + # 行锁(SELECT FOR UPDATE):并发/连点双请求会都读到 pending → 各发一次金币双倍发奖, + # 锁住该行串行化,第二个请求拿锁后看到 approved → 走 400。SQLite 下 FOR UPDATE 为 no-op。 + rep = db.get(PriceReport, report_id, with_for_update=True) if rep is None: raise HTTPException(status_code=404, detail="上报记录不存在") if rep.status != "pending": @@ -88,7 +90,7 @@ def reject_price_report( admin: Annotated[AdminUser, Depends(require_role("operator"))], db: AdminDb, ) -> OkResponse: - rep = db.get(PriceReport, report_id) + rep = db.get(PriceReport, report_id, with_for_update=True) # 行锁,同 approve(防并发重复审核) if rep is None: raise HTTPException(status_code=404, detail="上报记录不存在") if rep.status != "pending": diff --git a/app/admin/routers/users.py b/app/admin/routers/users.py index e992e20..9a3ed01 100644 --- a/app/admin/routers/users.py +++ b/app/admin/routers/users.py @@ -126,7 +126,8 @@ def grant_user_coins( if body.mode == "set": if body.amount < 0: raise HTTPException(status_code=400, detail="目标金币值不能为负") - before = wallet_repo.get_or_create_account(db, user_id, commit=False).coin_balance + # lock=True:锁账户行,防连点/并发各读同一 before 算同一 delta 双写,余额错位 + before = wallet_repo.get_or_create_account(db, user_id, commit=False, lock=True).coin_balance delta = body.amount - before if delta == 0: raise HTTPException(status_code=400, detail=f"当前金币已为 {body.amount},无需调整") @@ -134,9 +135,9 @@ def grant_user_coins( if body.amount == 0: raise HTTPException(status_code=400, detail="amount 不能为 0") delta = body.amount - # 负数扣减时不允许扣成负余额(运营误操作保护) + # 负数扣减时不允许扣成负余额(运营误操作保护);lock=True 防并发扣穿 if delta < 0: - acc_now = wallet_repo.get_or_create_account(db, user_id, commit=False) + acc_now = wallet_repo.get_or_create_account(db, user_id, commit=False, lock=True) if acc_now.coin_balance + delta < 0: raise HTTPException( status_code=400, detail=f"扣减后金币为负(当前余额 {acc_now.coin_balance})" @@ -174,7 +175,10 @@ def grant_user_cash( if body.mode == "set": if body.amount_cents < 0: raise HTTPException(status_code=400, detail="目标现金值不能为负") - before = wallet_repo.get_or_create_account(db, user_id, commit=False).cash_balance_cents + # lock=True:锁账户行,防连点/并发各读同一 before 算同一 delta 双写,余额错位 + before = wallet_repo.get_or_create_account( + db, user_id, commit=False, lock=True + ).cash_balance_cents delta = body.amount_cents - before if delta == 0: raise HTTPException( @@ -184,9 +188,9 @@ def grant_user_cash( if body.amount_cents == 0: raise HTTPException(status_code=400, detail="amount_cents 不能为 0") delta = body.amount_cents - # 负数扣减时不允许扣成负余额(运营误操作保护) + # 负数扣减时不允许扣成负余额(运营误操作保护);lock=True 防并发扣穿 if delta < 0: - acc_now = wallet_repo.get_or_create_account(db, user_id, commit=False) + acc_now = wallet_repo.get_or_create_account(db, user_id, commit=False, lock=True) if acc_now.cash_balance_cents + delta < 0: raise HTTPException( status_code=400, detail=f"扣减后现金为负(当前余额 {acc_now.cash_balance_cents} 分)" diff --git a/app/admin/routers/withdraw.py b/app/admin/routers/withdraw.py index 7fcfff6..70be4b4 100644 --- a/app/admin/routers/withdraw.py +++ b/app/admin/routers/withdraw.py @@ -87,7 +87,12 @@ def withdraws_summary(db: AdminDb) -> WithdrawSummaryOut: return WithdrawSummaryOut(**queries.withdraw_summary(db)) -@router.get("/health-check", response_model=WxpayHealthCheckOut, summary="提现配置健康检查") +@router.get( + "/health-check", + response_model=WxpayHealthCheckOut, + summary="提现配置健康检查", + dependencies=[Depends(require_role("finance"))], # 暴露密钥路径/配置,限财务+super +) def withdraw_health_check() -> WxpayHealthCheckOut: private_path = wxpay._resolve_config_path(settings.WXPAY_MCH_PRIVATE_KEY_PATH) # noqa: SLF001 public_path = wxpay._resolve_config_path(settings.WXPAY_PUBLIC_KEY_PATH) # noqa: SLF001 diff --git a/app/admin/schemas/price_report.py b/app/admin/schemas/price_report.py index 30ea8ad..ddf3278 100644 --- a/app/admin/schemas/price_report.py +++ b/app/admin/schemas/price_report.py @@ -3,7 +3,7 @@ from __future__ import annotations from datetime import datetime -from pydantic import BaseModel, ConfigDict, Field +from pydantic import BaseModel, ConfigDict, Field, field_validator class PriceReportOut(BaseModel): @@ -36,6 +36,14 @@ class PriceReportOut(BaseModel): class PriceReportRejectRequest(BaseModel): reason: str = Field(min_length=1, max_length=256, description="拒绝理由,用户端记录页会看到") + @field_validator("reason") + @classmethod + def _reason_not_blank(cls, v: str) -> str: + # min_length=1 放过纯空白(" "),trim 后再校验非空,避免审计/用户端记录到空理由 + if not v.strip(): + raise ValueError("拒绝理由不能为空") + return v.strip() + class PriceReportSummary(BaseModel): """审核台顶部各状态计数。""" diff --git a/app/admin/schemas/user.py b/app/admin/schemas/user.py index f781fe5..6f1c352 100644 --- a/app/admin/schemas/user.py +++ b/app/admin/schemas/user.py @@ -4,7 +4,7 @@ from __future__ import annotations from datetime import datetime from typing import Literal -from pydantic import BaseModel, ConfigDict, Field +from pydantic import BaseModel, ConfigDict, Field, field_validator class AdminUserListItem(BaseModel): @@ -37,6 +37,13 @@ class AdminUserOverview(BaseModel): feedback_total: int +def _strip_reason(v: str) -> str: + # min_length=1 放过纯空白(" "),trim 后再校验非空,避免审计记到空原因 + if not v.strip(): + raise ValueError("操作原因不能为空") + return v.strip() + + class GrantCoinsRequest(BaseModel): mode: Literal["delta", "set"] = Field( "delta", description="delta=增减(amount 为变动量) / set=设为(amount 为目标值,须≥0)" @@ -47,6 +54,8 @@ class GrantCoinsRequest(BaseModel): ) reason: str = Field(..., min_length=1, max_length=128, description="操作原因(必填,入审计)") + _v_reason = field_validator("reason")(_strip_reason) + class GrantCashRequest(BaseModel): mode: Literal["delta", "set"] = Field( @@ -58,6 +67,8 @@ class GrantCashRequest(BaseModel): ) reason: str = Field(..., min_length=1, max_length=128, description="操作原因(必填,入审计)") + _v_reason = field_validator("reason")(_strip_reason) + class SetUserStatusRequest(BaseModel): status: Literal["active", "disabled"] = Field( diff --git a/app/repositories/wallet.py b/app/repositories/wallet.py index 3438928..615f0eb 100644 --- a/app/repositories/wallet.py +++ b/app/repositories/wallet.py @@ -78,9 +78,15 @@ class WithdrawNotReviewable(Exception): """提现单当前状态不可审核(非 reviewing,可能已被处理过)。""" -def get_or_create_account(db: Session, user_id: int, *, commit: bool = True) -> CoinAccount: - """取用户金币账户,不存在则建一个空账户。""" - acc = db.get(CoinAccount, user_id) +def get_or_create_account( + db: Session, user_id: int, *, commit: bool = True, lock: bool = False +) -> CoinAccount: + """取用户金币账户,不存在则建一个空账户。 + + lock=True 时对已存在的账户行加 SELECT FOR UPDATE(读-算-写余额的调用方串行化,防并发 + 双写余额错位,如 admin set 模式连点);默认 False 不改 C 端发奖行为。SQLite 下为 no-op。 + """ + acc = db.get(CoinAccount, user_id, with_for_update=True) if lock else db.get(CoinAccount, user_id) if acc is None: acc = CoinAccount( user_id=user_id,