Compare commits

...

4 Commits

Author SHA1 Message Date
marco 43db832b2c feat(store-mapping): JD 缓存 deeplink 失效标记(对标淘宝)
store_mapping 加 jd_deeplink_invalid_at 列 + 迁移(down_revision=store_mapping_tb_dl_invalid);
mark_jd_deeplink_invalid 按 id_jd 标记 + lookup_nearest 过滤已失效 JD 候选;
/internal/store-mapping/invalidate 接 jd 分支。配 pricebot 京东缓存 deeplink 失效回退。

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-15 20:41:39 +08:00
marco 9ec9d2389d feat(store_mapping): 淘宝缓存 deeplink 失效检测→标记→回退(app-server 侧) (#53)
加 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 <noreply@anthropic.com>

Reviewed-on: #53
2026-06-15 18:07:32 +08:00
liujiahui 8fa55eec3e fix(coupon): 领券弹窗频控按 App 独立 + debug 全重置补全 (#52)
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 <noreply@anthropic.com>

---------

Co-authored-by: no_gen_mu <liujianhishen@gmail.com>
Reviewed-on: #52
Co-authored-by: liujiahui <liujiahui@wonderable.ai>
Co-committed-by: liujiahui <liujiahui@wonderable.ai>
2026-06-14 23:41:54 +08:00
ouzhou 27f76918b2 feat(admin): 运营后台跟进——review 加固 + 反馈改版 + 列表页码分页 (#51)
本 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: #51
Co-authored-by: ouzhou <ouzhou@wonderable.ai>
Co-committed-by: ouzhou <ouzhou@wonderable.ai>
2026-06-14 22:54:07 +08:00
34 changed files with 765 additions and 144 deletions
@@ -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')
@@ -0,0 +1,32 @@
"""store_mapping 加京东 deeplink 失效标记列 jd_deeplink_invalid_at
Revision ID: store_mapping_jd_dl_invalid
Revises: store_mapping_tb_dl_invalid
Create Date: 2026-06-15 00:00:00.000000
京东同淘宝: 缓存的店内搜索 deeplink 会失效(打开是"当前门店超出配送范围"页)。pricebot 比价撞到
失效页时回退正常搜店, 并 server→server 通知把该 storeId 的 deeplink 标记失效。本列记失效时刻
(NULL=有效); lookup 反查时过滤掉已失效的京东候选, 不再返回坏 deeplink。
与淘宝 taobao_deeplink_invalid_at 对称; 用时间戳而非布尔: 留痕可审计、可统计失效率, 不销毁原 deeplink。
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision: str = 'store_mapping_jd_dl_invalid'
down_revision: Union[str, Sequence[str], None] = 'store_mapping_tb_dl_invalid'
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('jd_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('jd_deeplink_invalid_at')
@@ -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')
+3 -3
View File
@@ -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"])
+14 -11
View File
@@ -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
+84 -45
View File
@@ -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:
+24 -5
View File
@@ -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()
+4 -2
View File
@@ -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,
)
+17 -2
View File
@@ -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,
+8 -4
View File
@@ -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":
+12 -7
View File
@@ -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} 分)"
+11 -4
View File
@@ -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(
+6 -1
View File
@@ -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):
+9 -1
View File
@@ -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):
"""审核台顶部各状态计数。"""
+12 -1
View File
@@ -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(
+32 -1
View File
@@ -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,29 @@ 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":
affected = repo.mark_taobao_deeplink_invalid(db, payload.shop_id)
elif payload.platform == "jd":
affected = repo.mark_jd_deeplink_invalid(db, payload.shop_id)
else:
# 当前只接淘宝/京东; 其它平台先 no-op(affected=0), 不报错 — 向后兼容 pricebot 将来扩展。
logger.info("store_mapping invalidate 跳过: platform=%s 暂不支持", payload.platform)
return StoreMappingInvalidateOut(ok=True, affected=0)
logger.info(
"store_mapping invalidate platform=%s shop_id=%s → 标记失效 %d",
payload.platform, payload.shop_id, affected,
)
return StoreMappingInvalidateOut(ok=True, affected=affected)
+43 -11
View File
@@ -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}
+4 -5
View File
@@ -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="联系方式过长")
+29 -1
View File
@@ -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:
+12 -1
View File
@@ -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))
+16 -6
View File
@@ -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(
+3 -2
View File
@@ -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
+5
View File
@@ -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 每次分享重新加密、非稳定主键(调研文档 §八), 故单列存"可复跳的一次性票据",
@@ -102,6 +105,8 @@ class StoreMapping(Base):
jd_share_url: Mapped[str | None] = mapped_column(String(256), nullable=True) # 3.cn 短链
jd_resolved_url: Mapped[str | None] = mapped_column(Text, nullable=True) # 反查出的目标 openapp.jdmobile:// deeplink
jd_deeplink: Mapped[str | None] = mapped_column(Text, nullable=True) # 拼好的 pages/search 店内搜索 deeplink
# 京东 deeplink 失效标记(同 taobao_deeplink_invalid_at):撞"当前门店超出配送范围"页时被置,NULL=有效。
jd_deeplink_invalid_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
# 灵活字段兜底(免得加字段就迁移)
attrs: Mapped[dict | None] = mapped_column(_JSON, nullable=True)
+12 -2
View File
@@ -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):
+29 -8
View File
@@ -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(
+41 -1
View File
@@ -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,40 @@ 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
def mark_jd_deeplink_invalid(db: Session, store_id: str) -> int:
"""把所有 id_jd=store_id 的行标记京东 deeplink 失效(置 invalid_at=now)。返回本次新标记的行数。
同 mark_taobao_deeplink_invalid:按 storeId 标记**所有**行(撞京东"当前门店超出配送范围"页),
全标掉才能让后续 lookup 不再返回它。幂等:已标记的行(invalid_at 非 NULL)跳过。"""
result = db.execute(
update(StoreMapping)
.where(
StoreMapping.id_jd == store_id,
StoreMapping.jd_deeplink_invalid_at.is_(None),
)
.values(jd_deeplink_invalid_at=func.now())
)
db.commit()
return result.rowcount or 0
# ============================================================
# 缓存查询: 比价前按"源平台店名"反查已沉淀的各目标平台店铺 id, 命中就让 pricebot 直接
# deeplink 跳店内搜索, 省掉"开平台→进店→分享反查"整段。
@@ -161,6 +195,12 @@ 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]
elif tgt == "jd":
# 同上:失效的京东 deeplink(撞"当前门店超出配送范围"被 invalidate)整条排除。
cands = [r for r in cands if r.jd_deeplink_invalid_at is None]
if not cands:
continue
best = _pick_best(cands, lat, lng)
+9 -3
View File
@@ -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,
+16 -2
View File
@@ -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
+14
View File
@@ -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
+11 -5
View File
@@ -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 | 1100 |
| `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` 超出 1100 范围 / `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/<file>` 静态读。
+4 -4
View File
@@ -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`,便于回访
+3 -3
View File
@@ -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 | 提交时间 |
+4 -3
View File
@@ -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}"
+156
View File
@@ -0,0 +1,156 @@
"""淘宝 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,
name_jd=None, id_jd=None, jd_deeplink=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,
name_jd=name_jd, id_jd=id_jd, jd_deeplink=jd_deeplink,
)
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_unsupported_platform_noop(client, monkeypatch):
# 当前支持 taobao/jd;其它平台(如 meituan)no-op 返 affected=0
monkeypatch.setattr(settings, "INTERNAL_API_SECRET", _SECRET)
r = client.post("/internal/store-mapping/invalidate",
json={"platform": "meituan", "shop_id": "X"},
headers={"X-Internal-Secret": _SECRET})
assert r.status_code == 200 and r.json()["affected"] == 0
# ---------- 京东(对称淘宝)----------
def test_mark_jd_invalid_marks_all_rows_with_store_id(db):
_mk_row(db, trace_id="j1", name_meituan="兰州拉面", name_jd="兰州拉面", id_jd="JBAD", jd_deeplink="dl_a")
_mk_row(db, trace_id="j2", name_meituan="兰州拉面", name_jd="兰州拉面", id_jd="JBAD", jd_deeplink="dl_b")
_mk_row(db, trace_id="j3", name_meituan="兰州拉面", name_jd="兰州拉面", id_jd="JGOOD", jd_deeplink="dl_g")
assert repo.mark_jd_deeplink_invalid(db, "JBAD") == 2
assert repo.mark_jd_deeplink_invalid(db, "JBAD") == 0 # 幂等
rows = {r.trace_id: r for r in db.query(StoreMapping).all()}
assert rows["j1"].jd_deeplink_invalid_at is not None
assert rows["j2"].jd_deeplink_invalid_at is not None
assert rows["j3"].jd_deeplink_invalid_at is None # 不同 storeId 不动
def test_lookup_filters_invalid_jd(db):
_mk_row(db, trace_id="j1", name_meituan="兰州拉面", name_jd="兰州拉面", id_jd="JBAD", jd_deeplink="dl")
assert repo.lookup_nearest(db, "meituan", "兰州拉面")["jd"]["store_id"] == "JBAD"
repo.mark_jd_deeplink_invalid(db, "JBAD")
assert "jd" not in repo.lookup_nearest(db, "meituan", "兰州拉面")
def test_invalidate_endpoint_jd_marks_and_lookup_miss(client, db, monkeypatch):
monkeypatch.setattr(settings, "INTERNAL_API_SECRET", _SECRET)
_mk_row(db, trace_id="j1", name_meituan="店J", name_jd="店J", id_jd="JBADX", jd_deeplink="dl")
_mk_row(db, trace_id="j2", name_meituan="店J", name_jd="店J", id_jd="JBADX", jd_deeplink="dl2")
r = client.post("/internal/store-mapping/invalidate",
json={"platform": "jd", "shop_id": "JBADX"},
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": "店J"},
headers={"X-Internal-Secret": _SECRET})
assert "jd" not in lk.json()