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/audit_log.py b/app/admin/repositories/audit_log.py index 8b41bf4..7eec119 100644 --- a/app/admin/repositories/audit_log.py +++ b/app/admin/repositories/audit_log.py @@ -4,7 +4,7 @@ """ from __future__ import annotations -from sqlalchemy import select +from sqlalchemy import func, select from sqlalchemy.orm import Session from app.models.admin import AdminAuditLog @@ -49,8 +49,9 @@ def list_audit_logs( admin_id: int | None = None, limit: int = 50, cursor: int | None = None, -) -> tuple[list[AdminAuditLog], int | None]: - """游标分页(id 倒序),与现有 list_* 约定一致。返回 (rows, next_cursor)。""" +) -> tuple[list[AdminAuditLog], int | None, int]: + """offset 分页(id 倒序)+ total。cursor 即 offset((page-1)*pageSize),支持页码跳页。 + 返回 (rows, next_cursor, total)。""" stmt = select(AdminAuditLog) if action: stmt = stmt.where(AdminAuditLog.action == action) @@ -58,13 +59,15 @@ def list_audit_logs( stmt = stmt.where(AdminAuditLog.target_type == target_type) if admin_id is not None: stmt = stmt.where(AdminAuditLog.admin_id == admin_id) - if cursor is not None: - stmt = stmt.where(AdminAuditLog.id < cursor) - stmt = stmt.order_by(AdminAuditLog.id.desc()) - rows = list(db.execute(stmt.limit(limit + 1)).scalars().all()) + + total = int(db.execute(select(func.count()).select_from(stmt.subquery())).scalar_one()) + offset = max(cursor or 0, 0) + rows = list( + db.execute( + stmt.order_by(AdminAuditLog.id.desc()).offset(offset).limit(limit + 1) + ).scalars().all() + ) has_more = len(rows) > limit items = rows[:limit] - # next_cursor 必须是"本页返回的最后一条"的 id(下一页查 id < 它),不能用 rows[limit]—— - # rows[limit] 是探测下一页用的第 limit+1 条,它既不在本页也不在下页 → 每页边界丢一条。 - next_cursor = items[-1].id if has_more else None - return items, next_cursor + next_cursor = offset + limit if has_more else None + return items, next_cursor, total diff --git a/app/admin/repositories/queries.py b/app/admin/repositories/queries.py index d7293f0..a4721e0 100644 --- a/app/admin/repositories/queries.py +++ b/app/admin/repositories/queries.py @@ -38,6 +38,28 @@ def cursor_paginate( return items, next_cursor +def offset_paginate( + db: Session, stmt: Select, sort_clause: tuple, *, limit: int, cursor: int | None +) -> tuple[list, int | None, int]: + """offset 分页 + 总数。stmt 只含 where/join,不要预先带 order_by/offset/limit。 + + cursor 即 offset(页码分页:offset=(page-1)*pageSize)。返回 (items, next_cursor, total): + - total:符合筛选条件的总条数(供 antd pagination 渲染页码/共 N 条),count 在 P0 量级开销可忽略; + - next_cursor:下一页 offset(兼容「加载更多」),末页为 None。 + 多取 1 条探测下一页。sort_clause 为 order_by 表达式元组(末位应含 id 保证稳定排序)。""" + total = int( + db.execute(select(func.count()).select_from(stmt.subquery())).scalar_one() + ) + offset = max(cursor or 0, 0) + rows = list( + db.execute(stmt.order_by(*sort_clause).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, total + + def list_users( db: Session, *, @@ -53,11 +75,11 @@ def list_users( sort_order: str = "desc", limit: int = 20, cursor: int | None = None, -) -> tuple[list[User], int | None]: +) -> tuple[list[User], int | None, int]: """用户列表(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 +90,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, @@ -84,14 +106,7 @@ def list_users( sort_col = sort_cols.get(sort_by, User.id) order_fn = asc if sort_order == "asc" else desc id_order = asc(User.id) if sort_order == "asc" else desc(User.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 + return offset_paginate(db, stmt, (order_fn(sort_col), id_order), limit=limit, cursor=cursor) def list_onboarding_devices(db: Session, *, limit: int = 500) -> list[dict]: @@ -160,7 +175,7 @@ def list_all_withdraw_orders( quick_filter: str | None = None, limit: int = 20, cursor: int | None = None, -) -> tuple[list[WithdrawOrder], int | None]: +) -> tuple[list[WithdrawOrder], int | None, int]: stmt = select(WithdrawOrder) needs_user_join = bool(keyword and keyword.strip()) or quick_filter == "high_risk" if needs_user_join: @@ -190,16 +205,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( @@ -244,6 +259,53 @@ def list_all_withdraw_orders( sort_col = sort_cols.get(sort_by, WithdrawOrder.created_at) order_fn = asc if sort_order == "asc" else desc id_order = asc(WithdrawOrder.id) if sort_order == "asc" else desc(WithdrawOrder.id) + return offset_paginate(db, stmt, (order_fn(sort_col), id_order), limit=limit, cursor=cursor) + + +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.replace(tzinfo=timezone.utc) + return value.astimezone(timezone.utc) + + +def list_feedbacks( + db: Session, + *, + 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) + 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) @@ -254,29 +316,6 @@ def list_all_withdraw_orders( return items, next_cursor -def _as_utc_naive(value: datetime) -> datetime: - """前端传 ISO 时间;DB 当前按 UTC naive 比较最稳(SQLite/本地开发一致)。""" - if value.tzinfo is None: - return value - return value.astimezone(timezone.utc).replace(tzinfo=None) - - -def list_feedbacks( - db: Session, - *, - status: str | None = None, - user_id: int | None = None, - limit: int = 20, - cursor: int | None = None, -) -> tuple[list[Feedback], int | None]: - 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) - - def get_withdraw_by_out_bill_no(db: Session, out_bill_no: str) -> WithdrawOrder | None: """按商户单号查提现单(admin 重试打款先拿 user_id 用,M3)。""" return db.execute( @@ -463,14 +502,14 @@ def list_price_reports( user_id: int | None = None, limit: int = 20, cursor: int | None = None, -) -> tuple[list[PriceReport], int | None]: - """上报更低价列表(admin 全量,可按状态/用户筛)。游标同 feedback:id 倒序。""" +) -> tuple[list[PriceReport], int | None, int]: + """上报更低价列表(admin 全量,可按状态/用户筛)。offset 分页 + total,id 倒序。""" stmt = select(PriceReport) if status: stmt = stmt.where(PriceReport.status == status) if user_id is not None: stmt = stmt.where(PriceReport.user_id == user_id) - return cursor_paginate(db, stmt, PriceReport.id, limit=limit, cursor=cursor) + return offset_paginate(db, stmt, (PriceReport.id.desc(),), limit=limit, cursor=cursor) def price_report_summary(db: Session) -> dict: 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/audit.py b/app/admin/routers/audit.py index 580e03c..dd7cace 100644 --- a/app/admin/routers/audit.py +++ b/app/admin/routers/audit.py @@ -26,9 +26,11 @@ def list_audit_logs( limit: Annotated[int, Query(ge=1, le=100)] = 50, cursor: Annotated[int | None, Query()] = None, ) -> CursorPage[AdminAuditLogOut]: - items, next_cursor = audit_repo.list_audit_logs( + items, next_cursor, total = audit_repo.list_audit_logs( db, action=action, target_type=target_type, admin_id=admin_id, limit=limit, cursor=cursor, ) return CursorPage( - items=[AdminAuditLogOut.model_validate(x) for x in items], next_cursor=next_cursor, + items=[AdminAuditLogOut.model_validate(x) for x in items], + next_cursor=next_cursor, + total=total, ) diff --git a/app/admin/routers/feedback.py b/app/admin/routers/feedback.py index 23f21de..51a2a29 100644 --- a/app/admin/routers/feedback.py +++ b/app/admin/routers/feedback.py @@ -1,6 +1,7 @@ -"""admin 反馈工单:列表(读)+ 标记已处理(写,带审计)。""" +"""admin 反馈工单:列表(读,支持 状态/用户ID/内容/时间 筛选 + 排序)+ 标记已处理(写,带审计)。""" from __future__ import annotations +from datetime import datetime from typing import Annotated from fastapi import APIRouter, Depends, HTTPException, Query, Request @@ -25,11 +26,25 @@ def list_feedbacks( db: AdminDb, status: Annotated[str | None, Query()] = None, user_id: Annotated[int | None, Query()] = None, + content: Annotated[str | None, Query(max_length=100)] = None, + created_from: Annotated[datetime | None, Query()] = None, + created_to: Annotated[datetime | None, Query()] = None, + sort_by: Annotated[str, Query(pattern="^(id|created_at)$")] = "id", + sort_order: Annotated[str, Query(pattern="^(asc|desc)$")] = "desc", limit: Annotated[int, Query(ge=1, le=100)] = 20, cursor: Annotated[int | None, Query()] = None, ) -> CursorPage[FeedbackOut]: items, next_cursor = queries.list_feedbacks( - db, status=status, user_id=user_id, limit=limit, cursor=cursor, + db, + status=status, + user_id=user_id, + content=content, + created_from=created_from, + created_to=created_to, + sort_by=sort_by, + sort_order=sort_order, + limit=limit, + cursor=cursor, ) return CursorPage( items=[FeedbackOut.model_validate(f) for f in items], next_cursor=next_cursor, diff --git a/app/admin/routers/price_report.py b/app/admin/routers/price_report.py index b29f271..8e04b6d 100644 --- a/app/admin/routers/price_report.py +++ b/app/admin/routers/price_report.py @@ -40,11 +40,13 @@ def list_price_reports( limit: Annotated[int, Query(ge=1, le=100)] = 20, cursor: Annotated[int | None, Query()] = None, ) -> CursorPage[PriceReportOut]: - items, next_cursor = queries.list_price_reports( + items, next_cursor, total = queries.list_price_reports( db, status=status, user_id=user_id, limit=limit, cursor=cursor, ) return CursorPage( - items=[PriceReportOut.model_validate(r) for r in items], next_cursor=next_cursor, + items=[PriceReportOut.model_validate(r) for r in items], + next_cursor=next_cursor, + total=total, ) @@ -60,7 +62,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 +92,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..c631b5d 100644 --- a/app/admin/routers/users.py +++ b/app/admin/routers/users.py @@ -45,7 +45,7 @@ def list_users( limit: Annotated[int, Query(ge=1, le=100)] = 20, cursor: Annotated[int | None, Query()] = None, ) -> CursorPage[AdminUserListItem]: - items, next_cursor = queries.list_users( + items, next_cursor, total = queries.list_users( db, phone=phone, register_channel=register_channel, status=status, nickname=nickname, created_from=created_from, created_to=created_to, last_login_from=last_login_from, last_login_to=last_login_to, @@ -54,6 +54,7 @@ def list_users( return CursorPage( items=[AdminUserListItem.model_validate(u) for u in items], next_cursor=next_cursor, + total=total, ) @@ -126,7 +127,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 +136,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 +176,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 +189,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..732159f 100644 --- a/app/admin/routers/withdraw.py +++ b/app/admin/routers/withdraw.py @@ -63,7 +63,7 @@ def list_withdraws( limit: Annotated[int, Query(ge=1, le=100)] = 20, cursor: Annotated[int | None, Query()] = None, ) -> CursorPage[WithdrawOrderOut]: - items, next_cursor = queries.list_all_withdraw_orders( + items, next_cursor, total = queries.list_all_withdraw_orders( db, user_id=user_id, status=status, @@ -78,7 +78,9 @@ def list_withdraws( cursor=cursor, ) return CursorPage( - items=[WithdrawOrderOut.model_validate(o) for o in items], next_cursor=next_cursor, + items=[WithdrawOrderOut.model_validate(o) for o in items], + next_cursor=next_cursor, + total=total, ) @@ -87,7 +89,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 @@ -160,7 +167,7 @@ def withdraw_detail(out_bill_no: str, db: AdminDb) -> WithdrawDetailOut: withdraw_success_cents=overview["withdraw_success_cents"], ) - recent_withdraws, _ = queries.list_all_withdraw_orders( + recent_withdraws, _, _ = queries.list_all_withdraw_orders( db, user_id=order.user_id, limit=5, cursor=None, ) recent_cash_transactions, _ = queries.list_all_cash_transactions( diff --git a/app/admin/schemas/common.py b/app/admin/schemas/common.py index 6987eeb..9f283c6 100644 --- a/app/admin/schemas/common.py +++ b/app/admin/schemas/common.py @@ -9,10 +9,15 @@ T = TypeVar("T") class CursorPage(BaseModel, Generic[T]): - """游标分页响应:items + 下一页游标(next_cursor=None 表示末页)。""" + """分页响应:items + 下一页游标(next_cursor=None 表示末页)+ 可选 total。 + + next_cursor:offset 分页时即下一页 offset,「加载更多」用;末页为 None。 + total:符合筛选条件的总条数,页码分页(antd pagination)用;不需要总数的接口可不传(None)。 + """ items: list[T] next_cursor: int | None = None + total: int | None = None class OkResponse(BaseModel): 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/api/v1/feedback.py b/app/api/v1/feedback.py index cc30d86..c085780 100644 --- a/app/api/v1/feedback.py +++ b/app/api/v1/feedback.py @@ -1,7 +1,7 @@ """帮助与反馈 endpoint。 路由前缀 `/api/v1/feedback`,需 Bearer 鉴权(反馈绑到登录用户,便于回访)。 - POST / 提交反馈(multipart:content / contact 必填,images 可选 ≤4 张) + POST / 提交反馈(multipart:content 必填;contact 可选(原型改版后客户端已不再采集);images 可选 ≤6 张) 截图复用 [app.core.media] 落盘到 /media/feedback/。 """ @@ -20,7 +20,7 @@ logger = logging.getLogger("shagua.feedback") router = APIRouter(prefix="/api/v1/feedback", tags=["feedback"]) -_MAX_IMAGES = 4 +_MAX_IMAGES = 6 _CONTENT_MAX = 2000 _CONTACT_MAX = 128 @@ -30,7 +30,8 @@ async def submit_feedback( user: CurrentUser, db: DbSession, content: str = Form(...), - contact: str = Form(...), + # 原型改版后客户端不再采集联系方式;保留字段以兼容旧端 + 后续可能复用,默认空串。 + contact: str = Form(default=""), images: list[UploadFile] = File(default=[]), ) -> FeedbackOut: content = content.strip() @@ -39,8 +40,6 @@ async def submit_feedback( raise HTTPException(status_code=400, detail="反馈内容不能为空") if len(content) > _CONTENT_MAX: raise HTTPException(status_code=400, detail="反馈内容过长") - if not contact: - raise HTTPException(status_code=400, detail="联系方式不能为空") if len(contact) > _CONTACT_MAX: raise HTTPException(status_code=400, detail="联系方式过长") diff --git a/app/core/config.py b/app/core/config.py index a64c3a8..a06ebd0 100644 --- a/app/core/config.py +++ b/app/core/config.py @@ -9,11 +9,16 @@ from functools import lru_cache from pathlib import Path from typing import Literal -from pydantic import Field +from pydantic import Field, model_validator from pydantic_settings import BaseSettings, SettingsConfigDict _PROJECT_ROOT = Path(__file__).resolve().parent.parent.parent +# 生产环境 JWT secret 的最小可接受长度(字节)。HS256 推荐高熵随机串;<16 视为弱密钥。 +_MIN_PROD_SECRET_LEN = 16 +# 已知的占位默认值(代码里写死的 default),prod 下绝不能沿用。 +_INSECURE_SECRET_DEFAULTS = frozenset({"change-me", "change-me-admin", ""}) + class Settings(BaseSettings): model_config = SettingsConfigDict( @@ -198,6 +203,29 @@ class Settings(BaseSettings): def is_prod(self) -> bool: return self.APP_ENV == "prod" + @model_validator(mode="after") + def _enforce_prod_secrets(self) -> "Settings": + """prod 下强校验 JWT secret,弱/默认/空即启动报错(fail-fast,挡住 token 被伪造)。 + + 只校验两个签发凭证:App 用户的 JWT_SECRET_KEY、后台的 ADMIN_JWT_SECRET——它们沿用默认值 + 时任何人都能伪造 access/admin token → 账号与后台失陷。INTERNAL_API_SECRET 默认空 = 内部端点 + 关闭(返 503),是安全的默认态,故不在此强制。dev 不触发,便于本地直接起。 + """ + if not self.is_prod: + return self + weak: list[str] = [] + for name in ("JWT_SECRET_KEY", "ADMIN_JWT_SECRET"): + value = getattr(self, name) + if value in _INSECURE_SECRET_DEFAULTS or len(value) < _MIN_PROD_SECRET_LEN: + weak.append(name) + if weak: + raise ValueError( + f"APP_ENV=prod 但检测到弱/默认密钥: {', '.join(weak)} —— 必须改成 " + f"≥{_MIN_PROD_SECRET_LEN} 位高熵随机串(否则 JWT 可被伪造 → 用户/后台账号失陷)。" + f"生成示例: python -c \"import secrets; print(secrets.token_urlsafe(48))\"" + ) + return self + @lru_cache(maxsize=1) def get_settings() -> Settings: diff --git a/app/core/rewards.py b/app/core/rewards.py index 3870af0..e9eac9a 100644 --- a/app/core/rewards.py +++ b/app/core/rewards.py @@ -143,6 +143,13 @@ AD_LT_FACTOR_TABLE: tuple[tuple[float, int, int | None], ...] = ( (1.0, 11, None), ) +# 客户端可影响的 eCPM 可信上限(分/千次展示):信息流广告一期由客户端上报 eCPM,伪造天价 eCPM +# 可铸出天量金币(见 calculate_ad_reward_coin)。真实 eCPM 一般 <¥100 CPM(=10000 分),档位表顶档 +# 为 >¥400(=40000 分);取 ¥500 CPM=50000 分,留足真实头部余量又封死伪造值。钳在唯一计算口 +# calculate_ad_reward_coin,故 feed 与 reward_video(回退客户端上报 eCPM 时)一并护住;阈值设在所有 +# 真实值之上,不会少发正规奖励。 +AD_ECPM_MAX_FEN: int = 50_000 + def parse_ecpm_fen(ecpm: str | int | float | None) -> float: """解析 eCPM 原始值(穿山甲 getEcpm 原值,单位=分/千次展示)。非法/缺失→0。""" @@ -187,8 +194,12 @@ def calculate_ad_reward_coin(ecpm: str | int | float | None, count_after_this: i eCPM 是穿山甲 getEcpm 原值,单位【分/千次展示】;先 ÷100 转成元(因子判档 + 收益换算都用元)。 单次收益(元)= eCPM元 ÷ 1000(每千次→单次) × 因子1(eCPM 元档) × 因子2(LT); 再按 1 元=10000 金币取整。count_after_this 为账号累计第 N 次看视频(LT 因子用,不按天重置)。 + + eCPM 在此先钳到 AD_ECPM_MAX_FEN(¥500 CPM):信息流广告一期 eCPM 由客户端上报,伪造天价值 + 会铸天量金币;钳在这唯一入口,feed 与 reward_video 回退客户端 eCPM 的路径都护住,且阈值高于 + 所有真实值,不影响正规发奖。 """ - ecpm_yuan = parse_ecpm_yuan(ecpm) + ecpm_yuan = min(parse_ecpm_yuan(ecpm), AD_ECPM_MAX_FEN / 100.0) yuan = (ecpm_yuan / 1000.0) * ad_ecpm_factor(ecpm_yuan) * ad_lt_factor(count_after_this) return max(0, round(yuan * COIN_PER_YUAN)) diff --git a/app/models/feedback.py b/app/models/feedback.py index 004d9e2..caef033 100644 --- a/app/models/feedback.py +++ b/app/models/feedback.py @@ -1,7 +1,8 @@ """用户反馈表(帮助与反馈)。 -每条 = 用户一次提交。content 必填,contact 必填(微信/QQ/手机,便于回访),images 为可选的 -截图 URL 列表(/media/feedback/...,JSON 存)。status: new(待处理)/ handled(已处理)。 +每条 = 用户一次提交。content 必填;contact 原为必填(微信/QQ/手机),原型改版后客户端不再采集, +新数据存空串(列保持 NOT NULL,免迁移;历史数据仍有值);images 为可选的截图 URL 列表 +(/media/feedback/...,JSON 存)。status: new(待处理)/ handled(已处理)。 """ from __future__ import annotations diff --git a/app/repositories/ad_feed_reward.py b/app/repositories/ad_feed_reward.py index 894c928..c2e367c 100644 --- a/app/repositories/ad_feed_reward.py +++ b/app/repositories/ad_feed_reward.py @@ -16,6 +16,10 @@ from app.repositories import wallet as crud_wallet FEED_REWARD_UNIT_SECONDS = 10 +# 单个 feed 事件的时长上限(秒):一期 duration_seconds 由客户端上报,伪造超长时长会刷份数 +# (每 10 秒 1 份)。真实单条信息流视频远小于此;取 120s=12 份封顶,挡刷量、不影响正规单。 +# 与 rewards.AD_ECPM_MAX_FEN(eCPM 钳顶)合起来,把单事件可铸金币锁进有限区间。 +FEED_MAX_DURATION_SECONDS = 120 def _find_by_event(db: Session, client_event_id: str) -> AdFeedRewardRecord | None: @@ -66,13 +70,19 @@ def grant_feed_reward( adn: str | None = None, slot_id: str | None = None, ) -> AdFeedRewardRecord: - """完成一条信息流广告后结算奖励。client_event_id 幂等,同号重试不重复发。""" + """完成一条信息流广告后结算奖励。client_event_id 幂等,同号重试不重复发。 + + 一期 eCPM/时长均由客户端上报,故服务端两道硬闸防刷:时长钳到 FEED_MAX_DURATION_SECONDS + 限单事件份数,eCPM 在 rewards.calculate_ad_reward_coin 内钳到 AD_ECPM_MAX_FEN 限单份金额; + 叠加每日 get_ad_daily_limit 条数上限,把单用户日产出锁进有限区间。 + """ existing = _find_by_event(db, client_event_id) if existing is not None: return existing today = cn_today().isoformat() - safe_duration = max(0, min(duration_seconds, 24 * 60 * 60)) + # 客户端上报时长先钳到 FEED_MAX_DURATION_SECONDS,防伪造超长时长刷份数(见常量注释)。 + safe_duration = max(0, min(duration_seconds, FEED_MAX_DURATION_SECONDS)) unit_count = safe_duration // FEED_REWARD_UNIT_SECONDS if _granted_today(db, user_id, today) >= rewards.get_ad_daily_limit(db): 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, diff --git a/docs/api/admin-feedbacks-list.md b/docs/api/admin-feedbacks-list.md index 2a3d002..af0460f 100644 --- a/docs/api/admin-feedbacks-list.md +++ b/docs/api/admin-feedbacks-list.md @@ -1,4 +1,4 @@ -# GET /admin/api/feedbacks — 反馈工单列表(游标分页) +# GET /admin/api/feedbacks — 反馈工单列表(offset 分页 + 筛选/排序) > 所属:Admin·反馈 组(前缀 `/admin/api/feedbacks`) | 鉴权:Bearer admin_token(角色:任意已登录管理员,无 `require_role`,仅 `get_current_admin`) | [← 返回 API 索引](./README.md) @@ -7,8 +7,13 @@ |---|---|---|---|---| | `status` | string | ❌ | null | 反馈状态,精确匹配:`new`(待处理) / `handled`(已处理);传空/不传则不筛 | | `user_id` | int | ❌ | null | 按提交用户 id 精确筛 | +| `content` | string | ❌ | null | 反馈内容模糊匹配(ilike,≤100 字) | +| `created_from` | datetime | ❌ | null | 提交时间 ≥(ISO,统一按 UTC 比较) | +| `created_to` | datetime | ❌ | null | 提交时间 ≤(ISO,统一按 UTC 比较) | +| `sort_by` | string | ❌ | `id` | 排序列:`id` / `created_at` | +| `sort_order` | string | ❌ | `desc` | `asc` / `desc` | | `limit` | int | ❌ | 20 | 1–100 | -| `cursor` | int | ❌ | null | 上一页 next_cursor(按 feedback id 倒序,查 `id < cursor`) | +| `cursor` | int | ❌ | null | 上一页 next_cursor(**offset 分页**,cursor=offset) | ## 出参 响应 `200`:`{ items: FeedbackOut[], next_cursor: int|null }`(`next_cursor=null` 表示末页) @@ -26,9 +31,10 @@ ## 错误码 - `401` 未带/无效/过期 admin token、管理员被禁用(头带 `WWW-Authenticate: Bearer`) -- `422` `limit` 超出 1–100 范围 / 字段类型不合法 +- `422` `limit` 超出 1–100 范围 / `sort_by`·`sort_order` 不在允许集 / 字段类型不合法 ## 说明 -- 游标分页约定:结果按 feedback `id` 倒序;`cursor` 传上一页返回的 `next_cursor`;`next_cursor=null` 即末页。 -- `status` / `user_id` 均为精确匹配,可叠加。 +- **offset 分页**(同 [admin-users-list](./admin-users-list.md)):`cursor` 即 offset,传上一页返回的 `next_cursor`;`next_cursor=null` 即末页。改用 offset 是为了在任意列排序下游标语义统一,代价是翻页期间数据变动可能错位一条(admin 低频可接受)。 +- 排序:`sort_by`(id/created_at)× `sort_order`(asc/desc),恒以 `id` 同向兜底次序。 +- `status` / `user_id` 精确匹配、`content` 模糊、`created_from`/`created_to` 时间范围,均可叠加。 - 关联表 [feedback](../database/feedback.md);截图为相对路径,经 `GET /media/feedback/` 静态读。 diff --git a/docs/api/feedback.md b/docs/api/feedback.md index 8b65588..259df11 100644 --- a/docs/api/feedback.md +++ b/docs/api/feedback.md @@ -7,8 +7,8 @@ | 字段 | 类型 | 必填 | 说明 | |---|---|---|---| | `content` | string | ✓ | 反馈正文,**1-2000 字**(strip 后) | -| `contact` | string | ✓ | 联系方式(微信/QQ/手机号),**1-128 字**,便于回访 | -| `images` | file[] | ✗ | 截图,**最多 4 张**,每张走头像同款校验(JPEG/PNG/WebP,≤ 5 MB,魔数嗅探) | +| `contact` | string | ✗ | 联系方式(微信/QQ/手机号),**≤128 字**。原型改版后客户端已不再采集、不传该字段(后端默认空串);保留字段兼容旧端 | +| `images` | file[] | ✗ | 截图,**最多 6 张**,每张走头像同款校验(JPEG/PNG/WebP,≤ 5 MB,魔数嗅探) | ## 出参 响应 `200`: @@ -29,9 +29,9 @@ > 不返回上传的 image URL——这是给运营后台看的,客户端通常不需要。 ## 错误码 -- `400` 内容为空 / 内容超 2000 字 / 联系方式为空 / 联系方式超 128 字 / 图片超 4 张 / 单图非法(空/过大/格式不对) +- `400` 内容为空 / 内容超 2000 字 / 联系方式超 128 字 / 图片超 6 张 / 单图非法(空/过大/格式不对) - `401` 未带 token / token 无效或过期 / 用户被禁用 -- `422` 缺 `content` 或 `contact` 字段 +- `422` 缺 `content` 字段 ## 说明 - **反馈绑用户**:`feedback.user_id = current_user.id`,便于回访 diff --git a/docs/database/feedback.md b/docs/database/feedback.md index 2261af2..d916b30 100644 --- a/docs/database/feedback.md +++ b/docs/database/feedback.md @@ -2,10 +2,10 @@ > 模型 `app/models/feedback.py` · 仓库 `app/repositories/feedback.py` · 接口 [feedback](../api/feedback.md) · admin [admin-feedbacks-list](../api/admin-feedbacks-list.md) / [admin-feedback-handle](../api/admin-feedback-handle.md) · [← 索引](./README.md) · [总览](./OVERVIEW.md) -App「帮助与反馈」每次提交写一行。`content` 与 `contact` 必填,`images` 为可选截图。后台人工处理后置 `handled`。与 `price_report`(结构化上报更低价)不同,本表是**自由文本**反馈。 +App「帮助与反馈」每次提交写一行。`content` 必填;`contact` 原必填,**原型改版后客户端不再采集,新数据存空串**(列保持 NOT NULL、免迁移,历史数据仍有值);`images` 为可选截图(≤6 张)。后台人工处理后置 `handled`。与 `price_report`(结构化上报更低价)不同,本表是**自由文本**反馈。 ## 用在哪 / 增删改查 -- **C(插入)**:`POST /api/v1/feedback`(multipart:`content` + `contact` + 可选 `images`;`create_feedback`)。截图先经 `core.media` 落 `/media/feedback/` 拿相对路径,再随反馈写入,`status='new'`。 +- **C(插入)**:`POST /api/v1/feedback`(multipart:`content` + 可选 `contact` + 可选 `images`;`create_feedback`)。截图先经 `core.media` 落 `/media/feedback/` 拿相对路径,再随反馈写入,`status='new'`。 - **U(更新)**:admin 处理反馈 `update_feedback_status` → `status='handled'`(同事务写 `admin_audit_log`)。 - **D**:无。 - **R**:admin 反馈列表(可按 `status` 筛)。C 端当前无"我的反馈列表"读接口。 @@ -16,7 +16,7 @@ App「帮助与反馈」每次提交写一行。`content` 与 `contact` 必填,` | `id` | Integer | PK, autoincrement | | | `user_id` | Integer | FK→user.id, index, NOT NULL | 提交用户 | | `content` | Text | NOT NULL | 反馈正文 | -| `contact` | String(128) | NOT NULL | 联系方式(微信/QQ/手机,便于回访) | +| `contact` | String(128) | NOT NULL | 联系方式(微信/QQ/手机)。客户端改版后不再采集,新数据为空串;列仍 NOT NULL | | `images` | JSON | nullable | 截图相对 URL 列表 `/media/feedback/...`;无图为 NULL | | `status` | String(16) | NOT NULL, default `new` | 取值:`new`(待处理)/ `handled`(已处理) | | `created_at` | DateTime(tz) | server_default now(), index | 提交时间 | diff --git a/tests/test_admin.py b/tests/test_admin.py index dea292f..7932a7f 100644 --- a/tests/test_admin.py +++ b/tests/test_admin.py @@ -125,7 +125,7 @@ def test_long_password_does_not_crash(admin_client: TestClient) -> None: def test_audit_log_pagination_no_gap() -> None: - """审计游标分页跨页不丢/不重(回归 next_cursor off-by-one)。""" + """审计分页跨页不丢/不重(offset 分页:cursor 即 offset,翻完覆盖全部)。""" from app.admin.repositories import admin_user as admin_repo from app.admin.repositories import audit_log as audit_repo @@ -142,12 +142,13 @@ def test_audit_log_pagination_no_gap() -> None: ) created_ids.append(log.id) - # limit=2 翻 5 条,收集所有 id,应正好覆盖创建的 5 条(无丢无重) + # limit=2 翻 5 条,收集所有 id,应正好覆盖创建的 5 条(无丢无重);total 恒为符合条件总数 seen: list[int] = [] cursor = None for _ in range(10): # 上限防死循环 - rows, cursor = audit_repo.list_audit_logs(db, action=action, limit=2, cursor=cursor) + rows, cursor, total = audit_repo.list_audit_logs(db, action=action, limit=2, cursor=cursor) seen.extend(r.id for r in rows) + assert total == len(created_ids), f"total 应为 {len(created_ids)},得 {total}" if cursor is None: break assert sorted(seen) == sorted(created_ids), f"分页丢/重: want={created_ids} got={seen}"