From 27f76918b2e1dfc3d937f9c73b54ea981acb167e Mon Sep 17 00:00:00 2001 From: ouzhou Date: Sun, 14 Jun 2026 22:54:07 +0800 Subject: [PATCH 1/3] =?UTF-8?q?feat(admin):=20=E8=BF=90=E8=90=A5=E5=90=8E?= =?UTF-8?q?=E5=8F=B0=E8=B7=9F=E8=BF=9B=E2=80=94=E2=80=94review=20=E5=8A=A0?= =?UTF-8?q?=E5=9B=BA=20+=20=E5=8F=8D=E9=A6=88=E6=94=B9=E7=89=88=20+=20?= =?UTF-8?q?=E5=88=97=E8=A1=A8=E9=A1=B5=E7=A0=81=E5=88=86=E9=A1=B5=20(#51)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 本 PR 汇合三块运营后台改动(原 #50 仅含其中「加固」一块,已并入本 PR 并关闭)。 ## 1. review 加固 (436b2a3) - 超管防自锁:降级/禁用最后一个 active super_admin 前校验,杜绝零超管死局 - 时间筛选统一 tz-aware(列为 timestamptz),比较绝对时刻、不依赖 DB 会话时区 - 上报审核 / 调余额(set·扣减)加行锁,防并发/连点重复发钱 - ad_audit 复算排序补 id 次级键;health-check 限 finance;调账/拒绝 reason 去空白校验 ## 2. 反馈改版 (5a18dbb) - contact 可选、截图≤6;admin 反馈列表筛选/排序;admin·wallet 接口调整 + docs ## 3. 列表页码分页 (1a7a624) - CursorPage 加 total;新增 offset_paginate(count 与分页同源) - 上报/审计日志从 id 游标改 offset 分页(支持跳页) - 用户 / 提现 / 上报 / 审计日志 四页接入页码分页 测试:admin 套件 47 passed。前端配套改动见 shaguabijia-admin-web。 --------- Co-authored-by: OuYingJun1024 <1034284404@qq.com> Reviewed-on: https://gitea.shaguabijia.com/WonderableAI/shaguabijia-app-server/pulls/51 Co-authored-by: ouzhou Co-committed-by: ouzhou --- app/admin/repositories/ad_audit.py | 6 +- app/admin/repositories/audit_log.py | 25 +++--- app/admin/repositories/queries.py | 129 ++++++++++++++++++---------- app/admin/routers/admins.py | 29 +++++-- app/admin/routers/audit.py | 6 +- app/admin/routers/feedback.py | 19 +++- app/admin/routers/price_report.py | 12 ++- app/admin/routers/users.py | 19 ++-- app/admin/routers/withdraw.py | 15 +++- app/admin/schemas/common.py | 7 +- app/admin/schemas/price_report.py | 10 ++- app/admin/schemas/user.py | 13 ++- app/api/v1/feedback.py | 9 +- app/core/config.py | 30 ++++++- app/core/rewards.py | 13 ++- app/models/feedback.py | 5 +- app/repositories/ad_feed_reward.py | 14 ++- app/repositories/wallet.py | 12 ++- docs/api/admin-feedbacks-list.md | 16 ++-- docs/api/feedback.md | 8 +- docs/database/feedback.md | 6 +- tests/test_admin.py | 7 +- 22 files changed, 295 insertions(+), 115 deletions(-) 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}" From 8fa55eec3e53e98a58eb62548b038be7ba408e1c Mon Sep 17 00:00:00 2001 From: liujiahui Date: Sun, 14 Jun 2026 23:41:54 +0800 Subject: [PATCH 2/3] =?UTF-8?q?fix(coupon):=20=E9=A2=86=E5=88=B8=E5=BC=B9?= =?UTF-8?q?=E7=AA=97=E9=A2=91=E6=8E=A7=E6=8C=89=20App=20=E7=8B=AC=E7=AB=8B?= =?UTF-8?q?=20+=20debug=20=E5=85=A8=E9=87=8D=E7=BD=AE=E8=A1=A5=E5=85=A8=20?= =?UTF-8?q?(#52)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit bug: 弹窗 engagement 表只按 (device, 日) 全局记一条,美团弹过/领过就把整台设备当天 标记 engage,淘宝/京东被压住不弹。需求是美团/淘宝/京东各自独立、每日各弹一次。 改动: - model: CouponPromptEngagement 加 package 列,唯一约束 (device,日) → (device,package,日) - alembic: 新增迁移 coupon_engage_per_package(加列 + 改唯一约束, batch_alter_table) - repository: has_engaged_today / mark_engagement 加 package 维度;新增 reset_today_completion - api: should-show / dismiss 接收 package;coupon_step(step=0) 按 App 记 engagement; 补 /prompt/shown 接口(客户端一直在调但后端缺失, 原 404); 补 /completed-today/reset(开发设置全重置用, 解首页卡置灰) 验证: curl 端到端 —— 美团弹过后 should-show 美团=false 淘宝/京东=true; shown/reset/completion-reset 端点全 ok。 Co-Authored-By: Claude Opus 4.8 --------- Co-authored-by: no_gen_mu Reviewed-on: https://gitea.shaguabijia.com/WonderableAI/shaguabijia-app-server/pulls/52 Co-authored-by: liujiahui Co-committed-by: liujiahui --- alembic/versions/coupon_engage_per_package.py | 53 ++++++++++++++++++ app/api/v1/coupon.py | 54 +++++++++++++++---- app/models/coupon_state.py | 22 +++++--- app/repositories/coupon_state.py | 37 ++++++++++--- app/schemas/coupon_state.py | 18 ++++++- 5 files changed, 157 insertions(+), 27 deletions(-) create mode 100644 alembic/versions/coupon_engage_per_package.py diff --git a/alembic/versions/coupon_engage_per_package.py b/alembic/versions/coupon_engage_per_package.py new file mode 100644 index 0000000..58d1e70 --- /dev/null +++ b/alembic/versions/coupon_engage_per_package.py @@ -0,0 +1,53 @@ +"""coupon_prompt_engagement 频控加 package 维度(美团/淘宝/京东各自独立弹) + +Revision ID: coupon_engage_per_package +Revises: store_mapping_jd_cols +Create Date: 2026-06-14 11:00:00.000000 + +需求:领券引导窗按 (device, App, 自然日) 频控——在美团弹过/领过,不影响淘宝、京东今天 +仍各弹一次。原表唯一键是 (device_id, engage_date),缺 package → 任一 App 弹过就把整台 +设备当天标记 engage,其余 App 被压住不弹(bug)。 + +本迁移: + 1. 加 package 列(NOT NULL,旧行用 server_default "" 填占位,不影响新逻辑判断)。 + 2. 旧唯一约束 (device_id, engage_date) → 新 (device_id, package, engage_date)。 + +SQLite 不支持直接 drop/add 约束,用 batch_alter_table(建临时表 + 拷数据 + 换名, +与 store_mapping_* 同款)。 +""" +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa + + +# revision identifiers, used by Alembic. +revision: str = 'coupon_engage_per_package' +down_revision: Union[str, Sequence[str], None] = 'store_mapping_jd_cols' +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + with op.batch_alter_table('coupon_prompt_engagement', schema=None) as batch_op: + # 加 package 列。旧行(改造前的全局记录)填 "" 占位:它们对应的是"老的全局态", + # 新逻辑按 (device, package, 日) 判,占位 "" 不会与真实包名(com.xxx)碰撞。 + batch_op.add_column( + sa.Column('package', sa.String(length=64), nullable=False, server_default='') + ) + # 旧唯一约束 (device_id, engage_date) → 新三元组 (device_id, package, engage_date)。 + batch_op.drop_constraint('uq_coupon_engage_device_date', type_='unique') + batch_op.create_unique_constraint( + 'uq_coupon_engage_device_pkg_date', + ['device_id', 'package', 'engage_date'], + ) + + +def downgrade() -> None: + with op.batch_alter_table('coupon_prompt_engagement', schema=None) as batch_op: + batch_op.drop_constraint('uq_coupon_engage_device_pkg_date', type_='unique') + batch_op.create_unique_constraint( + 'uq_coupon_engage_device_date', + ['device_id', 'engage_date'], + ) + batch_op.drop_column('package') diff --git a/app/api/v1/coupon.py b/app/api/v1/coupon.py index e44adb5..6e07120 100644 --- a/app/api/v1/coupon.py +++ b/app/api/v1/coupon.py @@ -27,6 +27,7 @@ from app.schemas.coupon_state import ( CouponCompletedTodayOut, CouponPromptDismissIn, CouponPromptShouldShowOut, + CouponPromptShownIn, ) logger = logging.getLogger("shagua.coupon") @@ -66,11 +67,11 @@ def _extract_coupon_results(resp_json: dict) -> list[dict]: def _mark_engagement_blocking( - device_id: str, user_id: int | None, engage_type: str + device_id: str, package: str, user_id: int | None, engage_type: str ) -> None: """独立 session 写 engagement(async 端点经 run_in_threadpool 调,不阻塞事件循环)。""" with SessionLocal() as db: - coupon_repo.mark_engagement(db, device_id, user_id, engage_type) + coupon_repo.mark_engagement(db, device_id, package, user_id, engage_type) def _record_claims_blocking( @@ -111,14 +112,17 @@ async def coupon_step( device_id = meta.get("device_id") user_id = _to_int(meta.get("user_id")) # 登录态才带;判断不靠它,资产留痕用 trace_id = meta.get("trace_id") + # 发起领券时前台 App 包名(step body 带 "package")。频控按 App,这条 engagement 要记到 + # 对应 App 上。App 内「去领取」发起时 package 可能缺/为空 → 退化为 "" 占位(全局态)。 + pkg = meta.get("package") or "" # 领券任务首帧(step=0)= 用户已发起领券 → 记一条今日 engagement(claim_started), - # 今天这台设备不再弹引导窗(对齐前台"点一键领取即 markEngaged")。写库失败绝不能 + # 今天**这个 App** 不再弹引导窗(对齐前台"点一键领取即 markEngaged")。写库失败绝不能 # 连累领券主流程,整段吞掉。 if device_id and meta.get("step") == 0: try: await run_in_threadpool( - _mark_engagement_blocking, device_id, user_id, "claim_started" + _mark_engagement_blocking, device_id, pkg, user_id, "claim_started" ) except Exception as e: # noqa: BLE001 logger.warning("coupon engagement write failed: %s", e) @@ -188,14 +192,29 @@ async def coupon_step( return resp_json +@router.post("/prompt/shown", summary="领券引导窗弹出即上报(按 App 记 shown)") +def coupon_prompt_shown(payload: CouponPromptShownIn, db: DbSession) -> dict[str, bool]: + """客户端弹出引导窗那刻调 → 记一条今日 engagement(shown),今天**这个 App** 不再自动弹。 + + 频控主判据(管跨重装):弹出即占用今天这个 App 的"一次"。用户领/拒/无视都算用掉。 + 后续点领取/拒绝再由 step/dismiss 把 type 升级。按 (device, package, 日) 记。 + """ + coupon_repo.mark_engagement( + db, payload.device_id, payload.package, payload.user_id, "shown" + ) + return {"ok": True} + + @router.post("/prompt/dismiss", summary="用户拒绝/关闭领券引导窗(记今日已 engage)") def coupon_prompt_dismiss(payload: CouponPromptDismissIn, db: DbSession) -> dict[str, bool]: - """客户端点关闭引导窗时调用 → 记一条今日 engagement(dismissed),今天不再弹。 + """客户端点关闭引导窗时调用 → 记一条今日 engagement(dismissed),今天**这个 App** 不再弹。 server 在透传链路里看不到"用户拒绝"(拒绝不发起领券),故必须客户端通知。 - MVP 不鉴权,按 device_id 记。 + 频控按 (device, package, 日),各 App 独立。MVP 不鉴权,按 device_id 记。 """ - coupon_repo.mark_engagement(db, payload.device_id, payload.user_id, "dismissed") + coupon_repo.mark_engagement( + db, payload.device_id, payload.package, payload.user_id, "dismissed" + ) return {"ok": True} @@ -205,12 +224,12 @@ def coupon_prompt_dismiss(payload: CouponPromptDismissIn, db: DbSession) -> dict summary="切到外卖 App 时是否还应弹领券引导窗", ) def coupon_prompt_should_show( - device_id: str, db: DbSession + device_id: str, db: DbSession, package: str = "" ) -> CouponPromptShouldShowOut: - """今天这台设备已 engage(领或拒)过 → should_show=false。客户端据此决定弹不弹 - (纯后台判据,客户端不再做前台 SP 缓存判断)。""" + """今天这台设备**这个 App** 已 engage(弹/领/拒)过 → should_show=false。各 App 独立: + 美团弹过不压淘宝/京东。客户端切到目标 App 时带 package 查(老客户端不带 → "" 全局态)。""" return CouponPromptShouldShowOut( - should_show=not coupon_repo.has_engaged_today(db, device_id) + should_show=not coupon_repo.has_engaged_today(db, device_id, package) ) @@ -236,3 +255,16 @@ def coupon_completed_today( return CouponCompletedTodayOut( completed=coupon_repo.has_completed_today(db, device_id) ) + + +@router.post( + "/completed-today/reset", + summary="重置今日已完成(开发设置全重置用,恢复首页「去领取」卡可点)", +) +def coupon_completed_today_reset( + payload: CouponPromptDismissIn, db: DbSession +) -> dict[str, bool]: + """删这台设备今天的 completion → has_completed_today 变 false,首页「去领取」卡恢复可点。 + 与 /prompt/reset 配套:开发设置「重置今日领券弹窗状态」一键把今日状态全清。MVP 不鉴权。""" + coupon_repo.reset_today_completion(db, payload.device_id) + return {"ok": True} diff --git a/app/models/coupon_state.py b/app/models/coupon_state.py index bd2aa2d..3f9364d 100644 --- a/app/models/coupon_state.py +++ b/app/models/coupon_state.py @@ -137,25 +137,35 @@ class CouponDailyCompletion(Base): class CouponPromptEngagement(Base): - """按 (device, 自然日) 记"今天是否对领券引导窗表达过意向"——弹窗频控源。""" + """按 (device, **App**, 自然日) 记"今天这个 App 是否对领券引导窗表达过意向"——弹窗频控源。 + + 2026-06-14:频控维度从 (device, 日) 改为 (device, package, 日)。需求是美团/淘宝/京东 + 各自独立——在美团弹过/领过,不影响淘宝、京东今天仍各弹一次。原来缺 package → 任一 App + 弹过就把整台设备当天标记 engage,其余 App 被压住不弹(bug)。 + """ __tablename__ = "coupon_prompt_engagement" __table_args__ = ( - # 一台设备一天一条:今天 engage 过(领或拒)就不再弹。 + # 一台设备、一个 App、一天一条:今天**这个 App** engage 过(领或拒)才不再弹该 App。 UniqueConstraint( - "device_id", "engage_date", - name="uq_coupon_engage_device_date", + "device_id", "package", "engage_date", + name="uq_coupon_engage_device_pkg_date", ), ) id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) device_id: Mapped[str] = mapped_column(String(64), nullable=False) + # 触发弹窗的目标 App 包名(com.sankuai.meituan / com.taobao.taobao / com.jingdong.app.mall)。 + # 频控维度,各 App 独立。旧行(改造前)无此值 → 迁移用占位 "" 填,不影响新逻辑判断。 + package: Mapped[str] = mapped_column( + String(64), nullable=False, server_default="" + ) user_id: Mapped[int | None] = mapped_column(Integer, index=True, nullable=True) # Asia/Shanghai 自然日。 engage_date: Mapped[date] = mapped_column(Date, nullable=False) - # claim_started(点了一键领取)/ dismissed(点了拒绝/关闭)。仅记录区分, - # 判断只看"今天有没有这条",type 不影响弹不弹。 + # claim_started(点了一键领取)/ dismissed(点了拒绝/关闭)/ shown(自动弹出即记)。 + # 仅记录区分,判断只看"今天这个 App 有没有这条",type 不影响弹不弹。 engage_type: Mapped[str] = mapped_column(String(16), nullable=False) created_at: Mapped[datetime] = mapped_column( diff --git a/app/repositories/coupon_state.py b/app/repositories/coupon_state.py index c5012b2..33f4781 100644 --- a/app/repositories/coupon_state.py +++ b/app/repositories/coupon_state.py @@ -31,11 +31,13 @@ def today_cn() -> date: # ===== 弹窗频控(coupon_prompt_engagement)===== -def has_engaged_today(db: Session, device_id: str) -> bool: - """这台设备今天是否已对领券引导窗表达过意向(领或拒)。有 = 不再弹。""" +def has_engaged_today(db: Session, device_id: str, package: str) -> bool: + """这台设备今天**这个 App** 是否已对领券引导窗表达过意向(领/拒/弹出)。有 = 该 App 不再弹。 + 频控按 (device, package, 日):美团弹过不影响淘宝/京东今天各自仍弹一次。""" row = db.execute( select(CouponPromptEngagement.id).where( CouponPromptEngagement.device_id == device_id, + CouponPromptEngagement.package == package, CouponPromptEngagement.engage_date == today_cn(), ) ).first() @@ -43,13 +45,18 @@ def has_engaged_today(db: Session, device_id: str) -> bool: def mark_engagement( - db: Session, device_id: str, user_id: int | None, engage_type: str + db: Session, device_id: str, package: str, user_id: int | None, engage_type: str ) -> None: - """记今日意向(claim_started / dismissed)。(device, 今天) 唯一,幂等 upsert。""" + """记今日意向(shown / claim_started / dismissed)。(device, package, 今天) 唯一,幂等 upsert。 + + engage_type 升级口径(同一 (device,package,日) 多次调,只覆盖 type,不新增行): + shown(自动弹出)→ claim_started(点一键领取)/ dismissed(点关闭)。判断只看"有没有这条"。 + """ today = today_cn() row = db.execute( select(CouponPromptEngagement).where( CouponPromptEngagement.device_id == device_id, + CouponPromptEngagement.package == package, CouponPromptEngagement.engage_date == today, ) ).scalar_one_or_none() @@ -59,19 +66,20 @@ def mark_engagement( row.user_id = user_id else: db.add(CouponPromptEngagement( - device_id=device_id, user_id=user_id, + device_id=device_id, package=package, user_id=user_id, engage_date=today, engage_type=engage_type, )) try: db.commit() except IntegrityError: - # 并发下另一请求刚插了同 (device, 日) → 唯一约束撞,回滚忽略(本就幂等)。 + # 并发下另一请求刚插了同 (device, package, 日) → 唯一约束撞,回滚忽略(本就幂等)。 db.rollback() def reset_today_engagement(db: Session, device_id: str) -> int: - """删这台设备今天的 engagement(开发设置「重置今日领券弹窗状态」调,测频控用)。 - 删后 has_engaged_today → false,今天又能弹。返回删除行数。""" + """删这台设备今天**所有 App** 的 engagement(开发设置「重置今日领券弹窗状态」调,测频控用)。 + 删后各 App has_engaged_today → false,今天又都能弹。返回删除行数。 + (不按 package 过滤:重置是"把今天清干净从头测",清全部 App 最符合预期。)""" result = db.execute( delete(CouponPromptEngagement).where( CouponPromptEngagement.device_id == device_id, @@ -123,6 +131,19 @@ def mark_completed_today( db.rollback() +def reset_today_completion(db: Session, device_id: str) -> int: + """删这台设备今天的"已完成"记录(开发设置「重置今日领券弹窗状态」全重置时调)。 + 删后 has_completed_today → false,首页「去领取」卡恢复可点。返回删除行数。""" + result = db.execute( + delete(CouponDailyCompletion).where( + CouponDailyCompletion.device_id == device_id, + CouponDailyCompletion.complete_date == today_cn(), + ) + ) + db.commit() + return result.rowcount or 0 + + # ===== 领券记录(coupon_claim_record)===== def record_claims( diff --git a/app/schemas/coupon_state.py b/app/schemas/coupon_state.py index fd66cb8..15e9c16 100644 --- a/app/schemas/coupon_state.py +++ b/app/schemas/coupon_state.py @@ -7,16 +7,30 @@ from pydantic import BaseModel class CouponPromptDismissIn(BaseModel): """客户端拒绝/关闭领券引导窗的通知体。 - server 据此记一条今日 engagement(dismissed)→ 今天这台设备不再弹引导窗。 + server 据此记一条今日 engagement(dismissed)→ 今天**这个 App** 不再弹引导窗。 + 频控按 (device, package, 日):各 App 独立。package 缺省 ""(老客户端兼容,退化为全局态)。 MVP 不鉴权,按 device_id 判断;user_id 登录态带上就一并记(资产),可空。 """ device_id: str + package: str = "" + user_id: int | None = None + + +class CouponPromptShownIn(BaseModel): + """客户端弹出领券引导窗即上报(记 shown)。 + + 弹出那刻就记一条今日 engagement(shown)→ 今天**这个 App** 不再自动弹(频控主判据, + 管跨重装;本地 SP 兜后台抖动)。后续用户点领取/拒绝再把 type 升级成 claim_started/dismissed。 + """ + + device_id: str + package: str user_id: int | None = None class CouponPromptShouldShowOut(BaseModel): - """切到外卖 App 时是否还应弹领券引导窗。今天已 engage(领或拒)过 → false。""" + """切到外卖 App 时是否还应弹领券引导窗。今天**这个 App** 已 engage(弹/领/拒)过 → false。""" should_show: bool From 9ec9d2389dac1d6f3024e6bd98723a64985eaebc Mon Sep 17 00:00:00 2001 From: marco Date: Mon, 15 Jun 2026 18:07:32 +0800 Subject: [PATCH 3/3] =?UTF-8?q?feat(store=5Fmapping):=20=E6=B7=98=E5=AE=9D?= =?UTF-8?q?=E7=BC=93=E5=AD=98=20deeplink=20=E5=A4=B1=E6=95=88=E6=A3=80?= =?UTF-8?q?=E6=B5=8B=E2=86=92=E6=A0=87=E8=AE=B0=E2=86=92=E5=9B=9E=E9=80=80?= =?UTF-8?q?(app-server=20=E4=BE=A7)=20(#53)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 加 taobao_deeplink_invalid_at 列(迁移+model);mark_taobao_deeplink_invalid 按 shopId 标记所有行;lookup_nearest 过滤失效淘宝候选;新增 POST /internal/store-mapping/invalidate。+ 7 单测。 Co-Authored-By: Claude Opus 4.8 Reviewed-on: https://gitea.shaguabijia.com/WonderableAI/shaguabijia-app-server/pulls/53 --- .../versions/store_mapping_tb_dl_invalid.py | 33 +++++ app/api/internal/store.py | 30 ++++- app/models/store_mapping.py | 3 + app/repositories/store_mapping.py | 22 +++- app/schemas/store_mapping.py | 14 +++ tests/test_store_mapping_invalidate.py | 118 ++++++++++++++++++ 6 files changed, 218 insertions(+), 2 deletions(-) create mode 100644 alembic/versions/store_mapping_tb_dl_invalid.py create mode 100644 tests/test_store_mapping_invalidate.py diff --git a/alembic/versions/store_mapping_tb_dl_invalid.py b/alembic/versions/store_mapping_tb_dl_invalid.py new file mode 100644 index 0000000..cea52a4 --- /dev/null +++ b/alembic/versions/store_mapping_tb_dl_invalid.py @@ -0,0 +1,33 @@ +"""store_mapping 加淘宝 deeplink 失效标记列 taobao_deeplink_invalid_at + +Revision ID: store_mapping_tb_dl_invalid +Revises: coupon_engage_per_package +Create Date: 2026-06-15 00:00:00.000000 + +缓存的淘宝店内搜索 deeplink 会失效(打开是"页面出错了"降级页)。pricebot 比价撞到错误页时 +回退正常搜店, 并 server→server 通知把该 shopId 的 deeplink 标记失效。本列记失效时刻 +(NULL=有效); lookup 反查时过滤掉已失效的淘宝候选, 不再返回坏 deeplink。 +只加淘宝一列(当前只接淘宝); 用时间戳而非布尔: 留痕可审计、可统计失效率, 且不销毁原 deeplink。 +无需索引: 过滤总叠在 name_taobao== 之后, 候选集已小。 +""" +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa + + +# revision identifiers, used by Alembic. +revision: str = 'store_mapping_tb_dl_invalid' +down_revision: Union[str, Sequence[str], None] = 'coupon_engage_per_package' +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + with op.batch_alter_table('store_mapping', schema=None) as batch_op: + batch_op.add_column(sa.Column('taobao_deeplink_invalid_at', sa.DateTime(timezone=True), nullable=True)) + + +def downgrade() -> None: + with op.batch_alter_table('store_mapping', schema=None) as batch_op: + batch_op.drop_column('taobao_deeplink_invalid_at') diff --git a/app/api/internal/store.py b/app/api/internal/store.py index 916398a..b7959e7 100644 --- a/app/api/internal/store.py +++ b/app/api/internal/store.py @@ -17,7 +17,12 @@ from fastapi import APIRouter, Header from app.api.deps import DbSession from app.api.internal.price import _check_secret from app.repositories import store_mapping as repo -from app.schemas.store_mapping import StoreMappingIn, StoreMappingOut +from app.schemas.store_mapping import ( + StoreMappingIn, + StoreMappingInvalidateIn, + StoreMappingInvalidateOut, + StoreMappingOut, +) logger = logging.getLogger("shagua.internal.store") @@ -77,3 +82,26 @@ def report_store_mapping( payload.source_device_id, payload.source_user_id, ) return StoreMappingOut(inserted=created, row_id=row_id) + + +@router.post( + "/store-mapping/invalidate", + response_model=StoreMappingInvalidateOut, + summary="标记某平台 shopId 的缓存 deeplink 失效(pricebot 撞错误页回退时上报,lookup 不再返回)", +) +def invalidate_store_mapping( + payload: StoreMappingInvalidateIn, + db: DbSession, + x_internal_secret: Annotated[str | None, Header()] = None, +) -> StoreMappingInvalidateOut: + _check_secret(x_internal_secret) + if payload.platform != "taobao": + # 当前只接淘宝; 其它平台先 no-op(affected=0), 不报错 — 向后兼容 pricebot 将来扩展。 + logger.info("store_mapping invalidate 跳过: platform=%s 暂不支持", payload.platform) + return StoreMappingInvalidateOut(ok=True, affected=0) + affected = repo.mark_taobao_deeplink_invalid(db, payload.shop_id) + logger.info( + "store_mapping invalidate platform=%s shop_id=%s → 标记失效 %d 行", + payload.platform, payload.shop_id, affected, + ) + return StoreMappingInvalidateOut(ok=True, affected=affected) diff --git a/app/models/store_mapping.py b/app/models/store_mapping.py index a76f8aa..35272d0 100644 --- a/app/models/store_mapping.py +++ b/app/models/store_mapping.py @@ -86,6 +86,9 @@ class StoreMapping(Base): taobao_share_url: Mapped[str | None] = mapped_column(String(256), nullable=True) # m.tb.cn 短链 taobao_resolved_url: Mapped[str | None] = mapped_column(Text, nullable=True) # 解析出的目标 URL(含 shopId) taobao_deeplink: Mapped[str | None] = mapped_column(Text, nullable=True) # 拼好的 et-store/search deeplink + # 淘宝 deeplink 失效标记:比价撞"页面出错了"降级页时被置(pricebot server→server invalidate), + # NULL=有效。lookup 反查过滤掉非 NULL 的淘宝候选,不再返回坏 deeplink(重搜会写新行覆盖)。 + taobao_deeplink_invalid_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) # ===== 美团原料(同淘宝;dpurl.cn 短链 → 302 反查 poi_id_str → imeituan:// deeplink)===== # ⚠️ poi_id_str 每次分享重新加密、非稳定主键(调研文档 §八), 故单列存"可复跳的一次性票据", diff --git a/app/repositories/store_mapping.py b/app/repositories/store_mapping.py index 8ebc9fc..358aa7d 100644 --- a/app/repositories/store_mapping.py +++ b/app/repositories/store_mapping.py @@ -15,7 +15,7 @@ from __future__ import annotations import logging import math -from sqlalchemy import select +from sqlalchemy import func, select, update from sqlalchemy.exc import IntegrityError from sqlalchemy.orm import Session @@ -107,6 +107,23 @@ def upsert(db: Session, payload: StoreMappingIn) -> tuple[int, int | None]: return 0, existing.id +def mark_taobao_deeplink_invalid(db: Session, shop_id: str) -> int: + """把所有 id_taobao=shop_id 的行标记淘宝 deeplink 失效(置 invalid_at=now)。返回本次新标记的行数。 + + 按 shopId 标记**所有**行 —— 同一个坏 shopId(撞淘宝"页面出错了"降级页)可能散在多次比价的 + 多行里, 全标掉才能让后续 lookup 不再返回它。幂等: 已标记的行(invalid_at 非 NULL)跳过。""" + result = db.execute( + update(StoreMapping) + .where( + StoreMapping.id_taobao == shop_id, + StoreMapping.taobao_deeplink_invalid_at.is_(None), + ) + .values(taobao_deeplink_invalid_at=func.now()) + ) + db.commit() + return result.rowcount or 0 + + # ============================================================ # 缓存查询: 比价前按"源平台店名"反查已沉淀的各目标平台店铺 id, 命中就让 pricebot 直接 # deeplink 跳店内搜索, 省掉"开平台→进店→分享反查"整段。 @@ -161,6 +178,9 @@ def lookup_nearest( if tgt == src_key: continue # 不返回源平台自己 cands = [r for r in rows if getattr(r, id_attr)] + if tgt == "taobao": + # 失效的淘宝 deeplink(撞过错误页被 invalidate)整条排除 = 当没缓存, pricebot 走正常搜店。 + cands = [r for r in cands if r.taobao_deeplink_invalid_at is None] if not cands: continue best = _pick_best(cands, lat, lng) diff --git a/app/schemas/store_mapping.py b/app/schemas/store_mapping.py index 2d84c75..c9a20d9 100644 --- a/app/schemas/store_mapping.py +++ b/app/schemas/store_mapping.py @@ -60,3 +60,17 @@ class StoreMappingOut(BaseModel): inserted: int row_id: int | None = None + + +class StoreMappingInvalidateIn(BaseModel): + """标记某平台某 shopId 的缓存 deeplink 失效(pricebot 撞错误页回退时上报)。""" + + platform: str # 目前只支持 "taobao" + shop_id: str # 失效的店铺 id(淘宝 = shopId = id_taobao) + + +class StoreMappingInvalidateOut(BaseModel): + """标记结果。affected = 本次新标记失效的行数(按 shopId 标记所有匹配行)。""" + + ok: bool + affected: int diff --git a/tests/test_store_mapping_invalidate.py b/tests/test_store_mapping_invalidate.py new file mode 100644 index 0000000..e1930f9 --- /dev/null +++ b/tests/test_store_mapping_invalidate.py @@ -0,0 +1,118 @@ +"""淘宝 deeplink 失效标记 + lookup 过滤 + invalidate 端点测试。 + +覆盖:按 shopId 标记所有行(幂等)、lookup 过滤失效淘宝候选、失效后新行仍可命中、 +invalidate 端点鉴权(503 未配 / 401 头错 / 200)、端到端标记后 lookup MISS、非淘宝 no-op。 +""" +from __future__ import annotations + +import pytest + +from app.core.config import settings +from app.db.session import SessionLocal +from app.models.store_mapping import StoreMapping +from app.repositories import store_mapping as repo + +_SECRET = "test-internal-secret-only-for-pytest" + + +def _mk_row(db, *, trace_id, name_meituan=None, name_taobao=None, + id_taobao=None, deeplink=None, lat=None, lng=None): + row = StoreMapping( + trace_id=trace_id, business_type="food", + name_meituan=name_meituan, name_taobao=name_taobao, + id_taobao=id_taobao, taobao_deeplink=deeplink, lat=lat, lng=lng, + ) + db.add(row) + db.commit() + db.refresh(row) + return row + + +@pytest.fixture() +def db(): + """每个用例前后清空 store_mapping(表由 conftest 的 create_all 建好,session 级共享)。""" + s = SessionLocal() + s.query(StoreMapping).delete() + s.commit() + yield s + s.query(StoreMapping).delete() + s.commit() + s.close() + + +# ---------- repo 层 ---------- + +def test_mark_invalid_marks_all_rows_with_shop_id(db): + # 同一个坏 shopId 散在两行(两次比价),另一行不同 shopId + _mk_row(db, trace_id="t1", name_meituan="绝味鸭脖", name_taobao="绝味鸭脖", id_taobao="BAD1", deeplink="dl_a") + _mk_row(db, trace_id="t2", name_meituan="绝味鸭脖", name_taobao="绝味鸭脖", id_taobao="BAD1", deeplink="dl_b") + _mk_row(db, trace_id="t3", name_meituan="绝味鸭脖", name_taobao="绝味鸭脖", id_taobao="GOOD9", deeplink="dl_g") + + # 按 shopId 标记所有行 + assert repo.mark_taobao_deeplink_invalid(db, "BAD1") == 2 + # 幂等:已标记的不重复 + assert repo.mark_taobao_deeplink_invalid(db, "BAD1") == 0 + + rows = {r.trace_id: r for r in db.query(StoreMapping).all()} + assert rows["t1"].taobao_deeplink_invalid_at is not None + assert rows["t2"].taobao_deeplink_invalid_at is not None + assert rows["t3"].taobao_deeplink_invalid_at is None # 不同 shopId 不动 + + +def test_lookup_filters_invalid_taobao(db): + _mk_row(db, trace_id="t1", name_meituan="绝味鸭脖", name_taobao="绝味鸭脖", id_taobao="BAD1", deeplink="dl") + # 标记前命中 + assert repo.lookup_nearest(db, "meituan", "绝味鸭脖")["taobao"]["shop_id"] == "BAD1" + # 标记失效后淘宝候选被过滤 → MISS(仅此一条淘宝) + repo.mark_taobao_deeplink_invalid(db, "BAD1") + assert "taobao" not in repo.lookup_nearest(db, "meituan", "绝味鸭脖") + + +def test_lookup_picks_new_valid_row_after_invalidate(db): + # 失效旧行 + 重搜写的新行(新 shopId,invalid_at=NULL)共存 → lookup 选到新行 + _mk_row(db, trace_id="t1", name_meituan="店A", name_taobao="店A", id_taobao="BAD1", deeplink="dl_bad") + repo.mark_taobao_deeplink_invalid(db, "BAD1") + _mk_row(db, trace_id="t2", name_meituan="店A", name_taobao="店A", id_taobao="NEW2", deeplink="dl_new") + assert repo.lookup_nearest(db, "meituan", "店A")["taobao"]["shop_id"] == "NEW2" + + +# ---------- invalidate 端点 ---------- + +def test_invalidate_endpoint_secret_unset_503(client, monkeypatch): + monkeypatch.setattr(settings, "INTERNAL_API_SECRET", "") # 未配 = 端点关闭 + r = client.post("/internal/store-mapping/invalidate", + json={"platform": "taobao", "shop_id": "X"}) + assert r.status_code == 503 + + +def test_invalidate_endpoint_auth(client, monkeypatch): + monkeypatch.setattr(settings, "INTERNAL_API_SECRET", _SECRET) + body = {"platform": "taobao", "shop_id": "X"} + assert client.post("/internal/store-mapping/invalidate", json=body).status_code == 401 + assert client.post("/internal/store-mapping/invalidate", json=body, + headers={"X-Internal-Secret": "wrong"}).status_code == 401 + r = client.post("/internal/store-mapping/invalidate", json=body, + headers={"X-Internal-Secret": _SECRET}) + assert r.status_code == 200 and r.json()["ok"] is True + + +def test_invalidate_endpoint_marks_and_lookup_miss(client, db, monkeypatch): + monkeypatch.setattr(settings, "INTERNAL_API_SECRET", _SECRET) + _mk_row(db, trace_id="t1", name_meituan="店B", name_taobao="店B", id_taobao="BADX", deeplink="dl") + _mk_row(db, trace_id="t2", name_meituan="店B", name_taobao="店B", id_taobao="BADX", deeplink="dl2") + r = client.post("/internal/store-mapping/invalidate", + json={"platform": "taobao", "shop_id": "BADX"}, + headers={"X-Internal-Secret": _SECRET}) + assert r.status_code == 200 and r.json()["affected"] == 2 + lk = client.get("/internal/store-mapping/lookup", + params={"source_platform": "meituan", "name": "店B"}, + headers={"X-Internal-Secret": _SECRET}) + assert "taobao" not in lk.json() + + +def test_invalidate_endpoint_non_taobao_noop(client, monkeypatch): + monkeypatch.setattr(settings, "INTERNAL_API_SECRET", _SECRET) + r = client.post("/internal/store-mapping/invalidate", + json={"platform": "jd", "shop_id": "X"}, + headers={"X-Internal-Secret": _SECRET}) + assert r.status_code == 200 and r.json()["affected"] == 0