Compare commits

...

19 Commits

Author SHA1 Message Date
marco 29225e9b65 feat(wxpay): 提现 debug 免人工审核直发 + 修复 user_display_name 被微信拒收
- WithdrawRequest 加 skip_review 字段;/withdraw 在 skip_review 且非生产(APP_ENV!=prod)
  时,建单后立即 execute_withdraw_transfer 直发、跳过人工审核。双闸:仅 debug 客户端下发
  + 服务端非 prod 才认,生产恒走人工审核。便于本地/联调直接测提现到账链路。
- 修 _auth_display_name:微信 user_display_name 校验极严,空格和标点也按"控制字符"拒收
  (实测 'wonderable ai' 因空格被 400,'wonderableai'/'周周'/'傻瓜比价用户8888' 才过)。
  原清洗只去星形面字符(ord>0xFFFF),漏掉 BMP emoji/空格/标点/控制符/零宽符。改为白名单:
  只保留 Unicode 文字(L*)+ 数字(N*),其余全剔,清空兜底手机号尾号。修复昵称含空格/emoji
  的用户发起免确认收款授权直接 400、提现打款失败的问题。

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-06 18:48:31 +08:00
marco 50ed726816 feat(wallet): 微信商家转账「用户授权免确认收款」(免确认到账)
提现从「每次跳微信确认」升级为免确认模式:首次授权一次,之后审核通过直接到账,用户零跳转。
方式一(首单转账顺带授权 pre-transfer-with-authorization)+ 方式二(显式开启 user-confirm-authorization)。

- 新增 wechat_transfer_authorization 表(user 1:1,out_authorization_no/authorization_id/state) + 迁移
- wxpay.py 新增 apply/query/close_transfer_authorization + pre_transfer_with_authorization + transfer_with_authorization
- repositories/wallet.py 授权 CRUD + sync/_refresh_active_auth + execute_withdraw_transfer 三分叉
  (active->免确认转账 / 无授权且配了回调->方式一 / 未配->退化原确认模式);失效回查授权权威判定,绝不盲退
- /wallet/transfer-auth(开启)/status/close 路由 + withdraw-info 增 transfer_auth_enabled
- 授权结果回调 stub /api/v1/wxpay/transfer-auth-notify(一期仅应答不验签,真值靠 query 轮询)
- config 增 WXPAY_AUTH_NOTIFY_URL

一期无回调验签;二期补 V3 平台证书验签 + AEAD 解密。

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-06 14:40:02 +08:00
marco 07c0b7502c feat(welfare): 提现人工审核流程 + 看广告每日时长限流(防刷主闸) (#15)
提现人工审核(提现不再即时打款):
- models/wallet.py: WithdrawOrder 状态机改为
  reviewing →(审核通过)→ pending → success/failed;reviewing →(驳回)→ rejected(已退款),
  默认态 pending → reviewing(扣现金建单但不打款);新增 user_name 列(实名, 微信达额转账要求)
- admin/routers/withdraw.py + admin/schemas/wallet.py: 运营后台审核接口(通过触发微信转账 / 驳回退款)
- api/v1/wallet.py + repositories/wallet.py: 发起提现进 reviewing 态, 驳回退回现金并写 withdraw_refund
- schemas/welfare.py: 提现出参增 reviewing/rejected 状态 + fail_reason

看广告每日时长限流(防刷主闸 + 次数兜底):
- 新增 models/ad_watch_log.py: 按 (user_id, watch_date) 聚合当日观看秒数
- 新增 repositories/ad_watch.py: watched_seconds_today / add_watch_seconds(夹 [0, MAX_SINGLE_WATCH_SECONDS])
- api/v1/ad.py: 新增 POST /ad/watch-report(JWT 取 user, 落时长返当日累计);
  /ad/reward-status 增 watched_seconds_today / limit / remaining
- schemas/ad.py: WatchReportIn/Out
- core/rewards.py: 新增 DAILY_AD_WATCH_SECONDS_LIMIT=50 分钟(主闸)+ MAX_SINGLE_WATCH_SECONDS=120;
  DAILY_AD_REWARD_LIMIT 20→200 改作次数兜底(防前端少报时长绕过时长闸, 又不误伤看短广告的正常用户)
- repositories/ad_reward.py + models/__init__.py: 接入新表与时长闸

alembic: withdraw_review_and_ad_watch 迁移(withdraw_order.user_name 列 + ad_watch_log 表)
tests: 补 test_ad_reward / test_withdraw

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: pure <pure@192.168.0.104>
Reviewed-on: #15
2026-06-06 04:51:50 +08:00
marco 67a9a15775 feat(admin): app_config 运营配置后台(rewards 等常量可后台配置) (#14)
- 新增 app_config 表 + model/repository/config_schema + admin config 路由与 schema + 测试
- rewards.py 及 ad_reward/signin/task/wallet/comparison_milestone repositories 接入可配置项;ad.py / admin/main.py 配套
- alembic merge(ebb6af5c0b56)合并本地 app_config(cfg1a2b3c4d5)与 incoming price_report/best_deeplink(b7e2c1a9f4d3)两 head

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

Reviewed-on: #14
2026-06-06 04:13:08 +08:00
zhangxianze 4506677483 feat(report): 上报更低价后端——price_report 表 + 提交/列表接口 (#13)
Co-authored-by: xianze <ze@192.168.0.138>
Reviewed-on: #13
Co-authored-by: zhangxianze <zhangxianze@wonderable.ai>
Co-committed-by: zhangxianze <zhangxianze@wonderable.ai>
2026-06-06 00:42:39 +08:00
marco b50495bebe feat: 短信接入极光真实发送 + 新增运营 admin 后台子应用
短信(SMS_MOCK 切 mock/real):
- integrations/sms.py 重写: real 模式走极光短信 REST /v1/messages 自定义验证码(本服务 secrets
  生成 6 位码 + 进程内存 + 本地校验一次性/防爆破), 鉴权复用极光一键登录 JG_APP_KEY/MASTER_SECRET
  (同一极光应用, 上线只需 SMS_MOCK=false); mock 仍"任意6位通过"不动其余测试
- 防刷四层: 单号冷却 + 单号每日上限 + 单IP rate_limit(/sms/send 10/min、/sms/login 20/min)
  + 单码失败次数作废; SmsError 带 status_code 映射 429/503/400
- config 增 SMS_SEND_ENDPOINT/SIGN_ID/TEMPLATE_ID/CODE_LENGTH/DAILY_LIMIT/MAX_VERIFY_ATTEMPTS;
  test_auth 加 real 模式单测; sms.md/后端技术实现/待办账本同步

admin 后台(app/admin/ 独立子应用, uvicorn app.admin.main:admin_app :8771):
- 复用主仓 models/repositories/integrations + 同库, 鉴权完全隔离(ADMIN_JWT_SECRET≠JWT_SECRET_KEY
  + payload typ=admin + bcrypt 密码 + 可选 IP 白名单); 主 app 不 import 本包, admin 崩不影响主进程
- 路由: 登录 / 账号管理(RBAC: super_admin·finance·operator) / 用户列表+360详情+封禁+手动调币 /
  钱包流水 / 提现重试对账 / 反馈工单 / 数据大盘; 全写操作落 admin_audit_log(涉钱与业务写同事务)
- 涉钱逻辑(调微信/退款/对账)复用 app.repositories.wallet 不重写
- 新增 models/admin.py(AdminUser/AdminAuditLog) + admin_tables 迁移 + create_admin.py +
  deploy/shaguabijia-admin.service; 依赖加 bcrypt

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-04 03:02:41 +08:00
ouzhou 90c0d65a16 chore(alembic): merge comparison_milestone 和 savings_report_fields 两个 head (#11)
Co-authored-by: OuYingJun1024 <1034284404@qq.com>
Reviewed-on: #11
Co-authored-by: ouzhou <ouzhou@wonderable.ai>
Co-committed-by: ouzhou <ouzhou@wonderable.ai>
2026-06-03 09:48:28 +08:00
OuYingJun1024 518f8c8b92 Merge branch 'main' of https://gitea.shaguabijia.com/WonderableAI/shaguabijia-app-server 2026-06-02 15:07:23 +08:00
marco e4588303fb feat(compare): 加 /api/v1/intent/step 透传到 pricebot(淘宝多帧意图识别)
配套 pricebot-backend 的 /api/intent/step(淘宝源 Phase 1 多帧意图识别: 展开 + 滚动
采集 → LLM 提取)。沿用 _passthrough 壳: 不鉴权、原样转发, 客户端循环调用直到 done
(done 帧顶层带 result + calibration)。目前仅淘宝源走这条, 其它源走单次 /intent/recognize。

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-02 14:40:19 +08:00
OuYingJun1024 9521cd96ce chore(alembic): merge comparison_milestone 和 savings_report_fields 两个 head 2026-06-02 09:53:25 +08:00
ouzhou da7ce69494 feat(ad): eCPM 上报接口 + 冷却策略解耦纯函数 + savings JSONB 跨库修复 (#8)
- 新增 POST /api/v1/ad/ecpm-report:激励视频展示后客户端上报本次 eCPM,落 ad_ecpm_record 做内部收益统计(model/repository/schema/alembic 迁移 + 接口文档)
- 看广告冷却策略抽到 app/core/ad_cooldown.py 纯函数;ad_reward.today_status 只取数据,换策略只改这一处
- savings.dishes 改 JSON().with_variant(JSONB,"postgresql"):SQLite 无 visit_JSONB 会让 create_all 编译崩,修复后测试套件恢复

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

---------

Co-authored-by: OuYingJun1024 <1034284404@qq.com>
Reviewed-on: #8
Co-authored-by: ouzhou <ouzhou@wonderable.ai>
Co-committed-by: ouzhou <ouzhou@wonderable.ai>
2026-06-01 22:23:15 +08:00
zhangxianze 288766443a feat(记账): 比价省钱记账后端 M1(savings_record 唯一真相表 + /order/report + demo 兜底) (#10)
从最新 origin/main 重拉的干净分支,只含本人记账改动,不含 marco 的归因订单(order_record)——后者是另一条未合并分支 feat/user-order-records 的功能,此前被误用 git add -A 和记账混进了 feat/savings-accounting-m1。本分支不带它:savings_record 升级为唯一真相表(新增原价/比价价/支付渠道/源平台/幂等键等列,迁移 savings_report_fields 直接挂 f01db5d77dac、不经 order_record_table);/order/report 改写 savings_record(source='compare'),省额=源平台原价−实付;展示 /api/v1/savings/* 真实优先 demo 兜底。无 order_record 的 model/repo/迁移/路由。alembic 单一 head + pytest 54 passed。

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: xianze <ze@192.168.0.138>
Reviewed-on: #10
Co-authored-by: zhangxianze <zhangxianze@wonderable.ai>
Co-committed-by: zhangxianze <zhangxianze@wonderable.ai>
2026-06-01 21:29:52 +08:00
marco a114a0f3f2 docs(integrations/wxpay): 删 code_to_openid 行,跟齐 9d1278c 死代码清理
9d1278c 已删 wxpay.py 的 code_to_openid 函数(能力被 code_to_userinfo 覆盖),
docs 函数表里这一行没跟着删,补齐。

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
2026-06-01 20:44:46 +08:00
marco 9d1278cb33 chore: 清理两处未使用代码
- app/db/base.py: 删 MappedAsDataclass 的 import (没人继承用)
- app/integrations/wxpay.py: 删 code_to_openid 函数 (code_to_userinfo 已覆盖能力, 18 行死代码)
2026-05-31 17:10:16 +08:00
marco 6c7eaa9734 docs: 跟齐 5 月 PR (#4 user/feedback / #5 ad-cooldown / #6 PG / abd8eab meituan 降级)
整轮 doc 跟齐 2026-05 落地但 doc 没补全/没更新的内容. 按 P0~P3 优先级处理.

P0 - 新 API doc + index 更新:
* docs/api/user-profile.md  (新) PATCH /api/v1/user/profile 改昵称
* docs/api/user-avatar.md   (新) POST /api/v1/user/avatar 上传头像(魔数嗅探/旧文件清理)
* docs/api/user-delete.md   (新) DELETE /api/v1/user 注销账号(软删+匿名化)
* docs/api/feedback.md      (新) POST /api/v1/feedback 提交反馈(含截图)
* docs/api/README.md           加 User/Feedback 组 + /media 静态服务段 +
                                修 UserOut "无接口可改恒为 null" 过时描述

P0 - 美团降级 / 现有 doc 校准:
* docs/api/meituan-coupons.md      加 "未配 MT_CPS 凭证降级返空" 说明
* docs/api/meituan-referral-link.md 同上 + 客户端兜底建议
* docs/api/meituan-feed.md         区分两种空结果路径(未配 vs 调用失败) +
                                    /health.mt_cps_configured 排障入口
* docs/integrations/meituan.md     "未配凭证降级" + "feed 静默吞错" 写进集成层
* ad-reward-status.md              PR #5 commit 已自带 doc 更新, 复核无需再改

P1 - 主文档大改:
* docs/后端技术实现.md
  - §1 能力表加 用户资料/外卖比价/钱包/福利/广告/反馈 (从 3 条扩到 9 条)
  - §1 数据模型纠正 "仅 1 张 user 表" -> 现有 10 张 + alembic_version
  - §2 DB 行 PG 主 / SQLite 兜底 + 加 psycopg3 行
  - §3 目录树补 user.py / feedback.py / media.py / repositories/feedback.py /
        scripts 子项 / docs/postgres-migration.md
  - §7 数据模型重写: 10 张业务表速查表 + alembic 合并迁移 f01db5d77dac 始末
  - §8 部署: 加 init_postgres.py 一句话流程 + checklist 加 "DATABASE_URL 切 PG"
        + "psycopg3 已装" + "MEDIA_ROOT 可写"
  - §10 已知问题: 加 美团降级 / agent 不鉴权 / MEDIA_ROOT serve / init_postgres bug /
        alembic 多 head 风险; 删过时 SQLite 行
  - 接口计数 11 -> 38

* docs/待办与技术债.md
  - 已解决追加: PG 切换 / 外卖比价透传 / 用户资料反馈接口 / 美团降级 /
                看广告 4 态冷却
  - 新增 "P2 DB 与 Alembic 治理" 段: alembic 多 head 团队规约 (含本次撞坑实录) +
    init_postgres.py 已知 5 处小 bug + MEDIA_ROOT serve

P2 - PG 迁移文档互指 + 迁移链补完:
* docs/数据库迁移.md
  - 顶部加 postgres-migration.md / 后端技术实现.md 互指
  - §6 迁移链 9 条 -> 12 条 (含 feedback_table / convert_dishes_jsonb /
        merge_pg_jsonb_and_feedback_heads)
  - §7 踩坑里 "多 head" 一行扩成完整故事 + 修复命令 + 团队预防规约
* docs/postgres-migration.md
  - 顶部加 数据库迁移.md / 后端技术实现.md §10 互指

P3 - 上线清单:
* docs/看广告赚金币上线清单.md
  - 设计要点加 PR #5 4 态 CTA + 后端权威派生说明 +
    VIDEO_ROUND_REQUIRED_COUNT/VIDEO_ROUND_COOLDOWN_SECONDS 常量
  - §E 端到端验收加 本轮冷却 / 弹窗 limit note / 跨设备一致 三项

所有最后更新日期同步到 2026-05-31.
2026-05-31 11:05:10 +08:00
marco c3c64fa06d chore: alembic merge revision + PG 迁移指南文档 (#7)
PostgreSQL 迁移 (#6) 之后两个分支引入了独立 heads (ef96beb47b1e +
d1e2f3a4b5c6), 必须合并成单一 head 才能 alembic upgrade head 顺利跑通。
- alembic/versions/f01db5d77dac_merge_pg_jsonb_and_feedback_heads.py:
  空 op 的 merge revision, 把两个 heads 收敛为 f01db5d77dac
- docs/postgres-migration.md: SQLite → PostgreSQL 迁移指南
  (背景/动机/操作步骤/无真实用户数据假设, 上线前补 pgloader 演练等)

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Reviewed-on: #7
2026-05-30 18:13:11 +08:00
marco e69e244de7 Merge pull request 'feat(ad-reward): reward-status 加本轮 3 次冷却字段 (round_count + cooldown_until)' (#5) from feat/welfare-video-round-cooldown into main
Reviewed-on: #5
2026-05-29 15:45:51 +08:00
marco 2ebde935f9 Merge pull request 'feat: SQLite → PostgreSQL 迁移代码改动' (#6) from feat/postgres-migration-and-bootstrap-script into main
Reviewed-on: #6
2026-05-29 15:44:40 +08:00
OuYingJun1024 56f4548654 feat(ad-reward): reward-status 加本轮 3 次冷却字段 (round_count + cooldown_until)
让客户端 4 态任务行 CTA (Normal/Loading/Capped/CoolingDown) 与 ExitReward
弹窗 limit note (本轮看完/今日看完 二选一) 都由后端权威派生, 跨设备 / 重装 /
杀进程都一致, 客户端不再依赖 in-memory 本地存储。

- core/rewards.py: 加 VIDEO_ROUND_REQUIRED_COUNT=3 / VIDEO_ROUND_COOLDOWN_SECONDS=600
- repositories/ad_reward.py: today_status 从 3-tuple 扩成 5-tuple,
  从已有 created_at 字段按 ORDER BY DESC + OFFSET round_count 派生最近一个
  已完成轮末尾时间; 零新字段零新表
- schemas/ad.py: AdRewardStatusOut + TestGrantOut 加 round_count + cooldown_until
- api/v1/ad.py: reward_status / test_grant 接 5-tuple
- tests: 加 4 个覆盖 (初始 / 未达一轮 / 刚达一轮进冷却 / 冷却已过)
- docs/api/ad-reward-status.md: 补字段说明 + 客户端 4 态 CTA 推导规则 +
  "冷却仅 UX, 发奖不受影响" 注解

冷却是 UX 约束: pangle-callback 发奖只看 daily_limit, 冷却期间穿山甲
回调照常发奖, 不另设拒发分支。

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
2026-05-29 13:27:56 +08:00
141 changed files with 8056 additions and 281 deletions
@@ -0,0 +1,26 @@
"""merge comparison and savings heads
Revision ID: 8ac524a8ea02
Revises: d4e5f6a7b8c9, savings_report_fields
Create Date: 2026-06-02 09:53:06.924912
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision: str = '8ac524a8ea02'
down_revision: Union[str, Sequence[str], None] = ('d4e5f6a7b8c9', 'savings_report_fields')
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
pass
def downgrade() -> None:
pass
@@ -0,0 +1,65 @@
"""新增 price_report 上报更低价表
Revision ID: 9258bddde4ea
Revises: ad60a1b2c3d4
Create Date: 2026-06-05 10:15:51.508598
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision: str = '9258bddde4ea'
down_revision: Union[str, Sequence[str], None] = 'ad60a1b2c3d4'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
# ### commands auto generated by Alembic - please adjust! ###
op.create_table('price_report',
sa.Column('id', sa.Integer(), autoincrement=True, nullable=False),
sa.Column('user_id', sa.Integer(), nullable=False),
sa.Column('comparison_record_id', sa.Integer(), nullable=True),
sa.Column('store_name', sa.String(length=128), nullable=True),
sa.Column('dish_summary', sa.String(length=256), nullable=True),
sa.Column('original_platform_id', sa.String(length=32), nullable=True),
sa.Column('original_platform_name', sa.String(length=32), nullable=True),
sa.Column('original_price_cents', sa.Integer(), nullable=True),
sa.Column('reported_platform_id', sa.String(length=32), nullable=False),
sa.Column('reported_platform_name', sa.String(length=32), nullable=False),
sa.Column('reported_price_cents', sa.Integer(), nullable=False),
sa.Column('images', sa.JSON(), nullable=False),
sa.Column('status', sa.String(length=16), nullable=False),
sa.Column('reject_reason', sa.String(length=256), nullable=True),
sa.Column('reward_coins', sa.Integer(), nullable=True),
sa.Column('reviewed_at', sa.DateTime(timezone=True), nullable=True),
sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('(CURRENT_TIMESTAMP)'), nullable=False),
sa.ForeignKeyConstraint(['comparison_record_id'], ['comparison_record.id'], ),
sa.ForeignKeyConstraint(['user_id'], ['user.id'], ),
sa.PrimaryKeyConstraint('id')
)
with op.batch_alter_table('price_report', schema=None) as batch_op:
batch_op.create_index(batch_op.f('ix_price_report_comparison_record_id'), ['comparison_record_id'], unique=False)
batch_op.create_index(batch_op.f('ix_price_report_created_at'), ['created_at'], unique=False)
batch_op.create_index(batch_op.f('ix_price_report_status'), ['status'], unique=False)
batch_op.create_index(batch_op.f('ix_price_report_user_id'), ['user_id'], unique=False)
# 注:autogenerate 另检测到「删除 order_record 表」——那是已降级为审计的历史表
# (model 已移除、库表保留),与本次无关,手动剔除,避免误删他人审计数据。
# ### end Alembic commands ###
def downgrade() -> None:
# ### commands auto generated by Alembic - please adjust! ###
# (对称:upgrade 未删 order_record,downgrade 也不重建它)
with op.batch_alter_table('price_report', schema=None) as batch_op:
batch_op.drop_index(batch_op.f('ix_price_report_user_id'))
batch_op.drop_index(batch_op.f('ix_price_report_status'))
batch_op.drop_index(batch_op.f('ix_price_report_created_at'))
batch_op.drop_index(batch_op.f('ix_price_report_comparison_record_id'))
op.drop_table('price_report')
# ### end Alembic commands ###
+47
View File
@@ -0,0 +1,47 @@
"""ad_ecpm_record table (广告展示 eCPM 上报记录,内部收益统计/对账)
Revision ID: a1b2c3d4e5f6
Revises: f01db5d77dac
Create Date: 2026-05-31 11:30:00.000000
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision: str = 'a1b2c3d4e5f6'
down_revision: Union[str, Sequence[str], None] = 'f01db5d77dac'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
op.create_table(
'ad_ecpm_record',
sa.Column('id', sa.Integer(), autoincrement=True, nullable=False),
sa.Column('user_id', sa.Integer(), nullable=False),
sa.Column('ad_type', sa.String(length=32), nullable=False),
sa.Column('adn', sa.String(length=32), nullable=True),
sa.Column('slot_id', sa.String(length=64), nullable=True),
sa.Column('ecpm_raw', sa.String(length=32), nullable=False),
sa.Column('report_date', sa.String(length=10), nullable=False),
sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('(CURRENT_TIMESTAMP)'), nullable=False),
sa.ForeignKeyConstraint(['user_id'], ['user.id'], ),
sa.PrimaryKeyConstraint('id'),
)
with op.batch_alter_table('ad_ecpm_record', schema=None) as batch_op:
batch_op.create_index(batch_op.f('ix_ad_ecpm_record_user_id'), ['user_id'], unique=False)
batch_op.create_index(batch_op.f('ix_ad_ecpm_record_report_date'), ['report_date'], unique=False)
batch_op.create_index(batch_op.f('ix_ad_ecpm_record_created_at'), ['created_at'], unique=False)
def downgrade() -> None:
with op.batch_alter_table('ad_ecpm_record', schema=None) as batch_op:
batch_op.drop_index(batch_op.f('ix_ad_ecpm_record_created_at'))
batch_op.drop_index(batch_op.f('ix_ad_ecpm_record_report_date'))
batch_op.drop_index(batch_op.f('ix_ad_ecpm_record_user_id'))
op.drop_table('ad_ecpm_record')
+65
View File
@@ -0,0 +1,65 @@
"""admin_user + admin_audit_log tables (运营 admin 后台:账号 + 操作审计)
Revision ID: ad60a1b2c3d4
Revises: 8ac524a8ea02
Create Date: 2026-06-03 10:00:00.000000
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision: str = 'ad60a1b2c3d4'
down_revision: Union[str, Sequence[str], None] = '8ac524a8ea02'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
op.create_table(
'admin_user',
sa.Column('id', sa.Integer(), autoincrement=True, nullable=False),
sa.Column('username', sa.String(length=64), nullable=False),
sa.Column('password_hash', sa.String(length=255), nullable=False),
sa.Column('role', sa.String(length=20), nullable=False, server_default='operator'),
sa.Column('status', sa.String(length=20), nullable=False, server_default='active'),
sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('(CURRENT_TIMESTAMP)'), nullable=False),
sa.Column('last_login_at', sa.DateTime(timezone=True), nullable=True),
sa.PrimaryKeyConstraint('id'),
)
with op.batch_alter_table('admin_user', schema=None) as batch_op:
batch_op.create_index(batch_op.f('ix_admin_user_username'), ['username'], unique=True)
op.create_table(
'admin_audit_log',
sa.Column('id', sa.Integer(), autoincrement=True, nullable=False),
sa.Column('admin_id', sa.Integer(), nullable=False),
sa.Column('admin_username', sa.String(length=64), nullable=False),
sa.Column('action', sa.String(length=64), nullable=False),
sa.Column('target_type', sa.String(length=32), nullable=False),
sa.Column('target_id', sa.String(length=64), nullable=True),
sa.Column('detail', sa.JSON().with_variant(sa.dialects.postgresql.JSONB(), 'postgresql'), nullable=True),
sa.Column('ip', sa.String(length=64), nullable=True),
sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('(CURRENT_TIMESTAMP)'), nullable=False),
sa.ForeignKeyConstraint(['admin_id'], ['admin_user.id'], ),
sa.PrimaryKeyConstraint('id'),
)
with op.batch_alter_table('admin_audit_log', schema=None) as batch_op:
batch_op.create_index(batch_op.f('ix_admin_audit_log_admin_id'), ['admin_id'], unique=False)
batch_op.create_index(batch_op.f('ix_admin_audit_log_action'), ['action'], unique=False)
batch_op.create_index(batch_op.f('ix_admin_audit_log_created_at'), ['created_at'], unique=False)
def downgrade() -> None:
with op.batch_alter_table('admin_audit_log', schema=None) as batch_op:
batch_op.drop_index(batch_op.f('ix_admin_audit_log_created_at'))
batch_op.drop_index(batch_op.f('ix_admin_audit_log_action'))
batch_op.drop_index(batch_op.f('ix_admin_audit_log_admin_id'))
op.drop_table('admin_audit_log')
with op.batch_alter_table('admin_user', schema=None) as batch_op:
batch_op.drop_index(batch_op.f('ix_admin_user_username'))
op.drop_table('admin_user')
+33
View File
@@ -0,0 +1,33 @@
"""app_config table (运营可配置项:rewards 常量等后台化)
Revision ID: cfg1a2b3c4d5
Revises: ad60a1b2c3d4
Create Date: 2026-06-04 12:00:00.000000
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision: str = 'cfg1a2b3c4d5'
down_revision: Union[str, Sequence[str], None] = 'ad60a1b2c3d4'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
op.create_table(
'app_config',
sa.Column('key', sa.String(length=64), nullable=False),
sa.Column('value', sa.JSON().with_variant(sa.dialects.postgresql.JSONB(), 'postgresql'), nullable=False),
sa.Column('updated_by_admin_id', sa.Integer(), nullable=True),
sa.Column('updated_at', sa.DateTime(timezone=True), server_default=sa.text('(CURRENT_TIMESTAMP)'), nullable=False),
sa.PrimaryKeyConstraint('key'),
)
def downgrade() -> None:
op.drop_table('app_config')
@@ -0,0 +1,32 @@
"""comparison_record 加 best_deeplink(再次比价直达商家深链)
Revision ID: b7e2c1a9f4d3
Revises: 9258bddde4ea
Create Date: 2026-06-05
「再次比价」要直达上次最低价平台的商家/商品页。深链(link)在比价时客户端已从剪贴板采到
(collectedLinks[best_index]),本列把它落库;再次比价时写剪贴板 + launch 该平台 App 直达。
旧记录无此列值(None) → 前端降级为打开对应 App 首页。
手写迁移(只加一列):autogenerate 会误报"删 order_record"(降级为审计的历史表,model 已移除、
库表保留),手写规避该噪音。
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
revision: str = "b7e2c1a9f4d3"
down_revision: Union[str, Sequence[str], None] = "9258bddde4ea"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
with op.batch_alter_table("comparison_record", schema=None) as batch_op:
batch_op.add_column(sa.Column("best_deeplink", sa.String(length=1024), nullable=True))
def downgrade() -> None:
with op.batch_alter_table("comparison_record", schema=None) as batch_op:
batch_op.drop_column("best_deeplink")
@@ -0,0 +1,41 @@
"""comparison_milestone_claim table (比价战绩里程碑领取记录)
Revision ID: d4e5f6a7b8c9
Revises: c3d4e5f6a7b8
Create Date: 2026-05-31 18:00:00.000000
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision: str = 'd4e5f6a7b8c9'
down_revision: Union[str, Sequence[str], None] = 'c3d4e5f6a7b8'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
op.create_table(
'comparison_milestone_claim',
sa.Column('id', sa.Integer(), autoincrement=True, nullable=False),
sa.Column('user_id', sa.Integer(), nullable=False),
sa.Column('milestone', sa.Integer(), nullable=False),
sa.Column('coin_awarded', sa.Integer(), nullable=False),
sa.Column('claimed_at', sa.DateTime(timezone=True), server_default=sa.text('(CURRENT_TIMESTAMP)'), nullable=False),
sa.ForeignKeyConstraint(['user_id'], ['user.id'], ),
sa.PrimaryKeyConstraint('id'),
sa.UniqueConstraint('user_id', 'milestone', name='uq_compare_milestone_user'),
)
with op.batch_alter_table('comparison_milestone_claim', schema=None) as batch_op:
batch_op.create_index(batch_op.f('ix_comparison_milestone_claim_user_id'), ['user_id'], unique=False)
def downgrade() -> None:
with op.batch_alter_table('comparison_milestone_claim', schema=None) as batch_op:
batch_op.drop_index(batch_op.f('ix_comparison_milestone_claim_user_id'))
op.drop_table('comparison_milestone_claim')
@@ -0,0 +1,28 @@
"""comparison_record.information (done 帧文案/失败原因)
Revision ID: c3d4e5f6a7b8
Revises: b2c3d4e5f6a7
Create Date: 2026-05-31 18:00:00.000000
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision: str = 'c3d4e5f6a7b8'
down_revision: Union[str, Sequence[str], None] = 'b2c3d4e5f6a7'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
with op.batch_alter_table('comparison_record', schema=None) as batch_op:
batch_op.add_column(sa.Column('information', sa.String(length=256), nullable=True))
def downgrade() -> None:
with op.batch_alter_table('comparison_record', schema=None) as batch_op:
batch_op.drop_column('information')
@@ -0,0 +1,64 @@
"""comparison_record table (比价记录:每次比价完整明细,「我的比价记录」数据源)
Revision ID: b2c3d4e5f6a7
Revises: a1b2c3d4e5f6
Create Date: 2026-05-31 15:00:00.000000
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision: str = 'b2c3d4e5f6a7'
down_revision: Union[str, Sequence[str], None] = 'a1b2c3d4e5f6'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
op.create_table(
'comparison_record',
sa.Column('id', sa.Integer(), autoincrement=True, nullable=False),
sa.Column('user_id', sa.Integer(), nullable=False),
sa.Column('device_id', sa.String(length=64), nullable=True),
sa.Column('business_type', sa.String(length=16), nullable=False),
sa.Column('trace_id', sa.String(length=64), nullable=False),
sa.Column('source_platform_id', sa.String(length=32), nullable=True),
sa.Column('source_platform_name', sa.String(length=32), nullable=True),
sa.Column('source_package', sa.String(length=128), nullable=True),
sa.Column('source_price_cents', sa.Integer(), nullable=True),
sa.Column('best_platform_id', sa.String(length=32), nullable=True),
sa.Column('best_platform_name', sa.String(length=32), nullable=True),
sa.Column('best_price_cents', sa.Integer(), nullable=True),
sa.Column('saved_amount_cents', sa.Integer(), nullable=True),
sa.Column('is_source_best', sa.Boolean(), nullable=True),
sa.Column('store_name', sa.String(length=128), nullable=True),
sa.Column('total_dish_count', sa.Integer(), nullable=True),
sa.Column('skipped_dish_count', sa.Integer(), nullable=True),
sa.Column('status', sa.String(length=16), nullable=False),
# PG 上为 JSONB,其它(SQLite)为 JSON——与模型层 with_variant 对齐
sa.Column('items', sa.JSON().with_variant(sa.dialects.postgresql.JSONB(), 'postgresql'), nullable=False),
sa.Column('comparison_results', sa.JSON().with_variant(sa.dialects.postgresql.JSONB(), 'postgresql'), nullable=False),
sa.Column('skipped_dish_names', sa.JSON().with_variant(sa.dialects.postgresql.JSONB(), 'postgresql'), nullable=False),
sa.Column('raw_payload', sa.JSON().with_variant(sa.dialects.postgresql.JSONB(), 'postgresql'), nullable=True),
sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('(CURRENT_TIMESTAMP)'), nullable=False),
sa.ForeignKeyConstraint(['user_id'], ['user.id'], ),
sa.PrimaryKeyConstraint('id'),
sa.UniqueConstraint('user_id', 'trace_id', name='uq_comparison_user_trace'),
)
with op.batch_alter_table('comparison_record', schema=None) as batch_op:
batch_op.create_index(batch_op.f('ix_comparison_record_user_id'), ['user_id'], unique=False)
batch_op.create_index(batch_op.f('ix_comparison_record_business_type'), ['business_type'], unique=False)
batch_op.create_index(batch_op.f('ix_comparison_record_created_at'), ['created_at'], unique=False)
def downgrade() -> None:
with op.batch_alter_table('comparison_record', schema=None) as batch_op:
batch_op.drop_index(batch_op.f('ix_comparison_record_created_at'))
batch_op.drop_index(batch_op.f('ix_comparison_record_business_type'))
batch_op.drop_index(batch_op.f('ix_comparison_record_user_id'))
op.drop_table('comparison_record')
@@ -0,0 +1,26 @@
"""merge app_config and price_report heads
Revision ID: ebb6af5c0b56
Revises: cfg1a2b3c4d5, b7e2c1a9f4d3
Create Date: 2026-06-06 02:29:13.475265
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision: str = 'ebb6af5c0b56'
down_revision: Union[str, Sequence[str], None] = ('cfg1a2b3c4d5', 'b7e2c1a9f4d3')
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
pass
def downgrade() -> None:
pass
@@ -0,0 +1,26 @@
"""merge pg_jsonb and feedback heads
Revision ID: f01db5d77dac
Revises: ef96beb47b1e, d1e2f3a4b5c6
Create Date: 2026-05-29 16:09:58.498935
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision: str = 'f01db5d77dac'
down_revision: Union[str, Sequence[str], None] = ('ef96beb47b1e', 'd1e2f3a4b5c6')
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
pass
def downgrade() -> None:
pass
+49
View File
@@ -0,0 +1,49 @@
"""savings_record:真实比价上报字段 + (user_id, client_event_id) 幂等唯一约束
把 savings_record 升级为「记账唯一真相表」:真实上报(source='compare')写入
原价/比价价/支付渠道/平台包名/源平台名/源链接/幂等键/device_id;
demo seeder 行(source='demo')这些列均为 NULL。
Revision ID: savings_report_fields
Revises: f01db5d77dac
Create Date: 2026-05-31 19:00:00.000000
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision: str = 'savings_report_fields'
down_revision: Union[str, Sequence[str], None] = 'f01db5d77dac'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
with op.batch_alter_table('savings_record', schema=None) as batch_op:
batch_op.add_column(sa.Column('original_price_cents', sa.Integer(), nullable=True))
batch_op.add_column(sa.Column('compared_price_cents', sa.Integer(), nullable=True))
batch_op.add_column(sa.Column('pay_channel', sa.String(length=16), nullable=True))
batch_op.add_column(sa.Column('platform_package', sa.String(length=128), nullable=True))
batch_op.add_column(sa.Column('source_platform_name', sa.String(length=32), nullable=True))
batch_op.add_column(sa.Column('source_deeplink', sa.String(length=512), nullable=True))
batch_op.add_column(sa.Column('client_event_id', sa.String(length=64), nullable=True))
batch_op.add_column(sa.Column('device_id', sa.String(length=128), nullable=True))
# (user_id, client_event_id) 幂等;demo 行 client_event_id 为 NULL 不冲突
batch_op.create_unique_constraint('uq_savings_user_event', ['user_id', 'client_event_id'])
def downgrade() -> None:
with op.batch_alter_table('savings_record', schema=None) as batch_op:
batch_op.drop_constraint('uq_savings_user_event', type_='unique')
batch_op.drop_column('device_id')
batch_op.drop_column('client_event_id')
batch_op.drop_column('source_deeplink')
batch_op.drop_column('source_platform_name')
batch_op.drop_column('platform_package')
batch_op.drop_column('pay_channel')
batch_op.drop_column('compared_price_cents')
batch_op.drop_column('original_price_cents')
@@ -0,0 +1,48 @@
"""wechat_transfer_authorization 表(免确认收款授权)
商家转账「用户授权免确认收款模式」:用户授权一次后,后续提现免逐笔确认直接到账。
一个用户一条(user_id 主键)。out_authorization_no 我方生成,authorization_id 微信 active 后返回。
Revision ID: wx_transfer_auth
Revises: withdraw_review_ad_watch
Create Date: 2026-06-06 12:00:00.000000
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision: str = 'wx_transfer_auth'
down_revision: Union[str, Sequence[str], None] = 'withdraw_review_ad_watch'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
op.create_table(
'wechat_transfer_authorization',
sa.Column('user_id', sa.Integer(), nullable=False),
sa.Column('openid', sa.String(length=64), nullable=False),
sa.Column('out_authorization_no', sa.String(length=64), nullable=False),
sa.Column('authorization_id', sa.String(length=64), nullable=True),
sa.Column('state', sa.String(length=16), server_default='pending', nullable=False),
sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('(CURRENT_TIMESTAMP)'), nullable=False),
sa.Column('updated_at', sa.DateTime(timezone=True), server_default=sa.text('(CURRENT_TIMESTAMP)'), nullable=False),
sa.ForeignKeyConstraint(['user_id'], ['user.id'], ),
sa.PrimaryKeyConstraint('user_id'),
)
with op.batch_alter_table('wechat_transfer_authorization', schema=None) as batch_op:
batch_op.create_index(
batch_op.f('ix_wechat_transfer_authorization_out_authorization_no'),
['out_authorization_no'],
unique=True,
)
def downgrade() -> None:
with op.batch_alter_table('wechat_transfer_authorization', schema=None) as batch_op:
batch_op.drop_index(batch_op.f('ix_wechat_transfer_authorization_out_authorization_no'))
op.drop_table('wechat_transfer_authorization')
@@ -0,0 +1,62 @@
"""withdraw_order.user_name 列 + ad_watch_log 表
提现审核功能:WithdrawOrder 加 user_name(审核异步打款时传给微信的实名,达额转账要求)。
看广告时长防刷:新增 ad_watch_log 表(前端 onAdClose 上报观看秒数,按 (user_id, watch_date)
聚合做"每天 50 分钟"硬上限)。
Revision ID: withdraw_review_ad_watch
Revises: ebb6af5c0b56
Create Date: 2026-06-05 12:00:00.000000
注:本迁移原 down_revision=ad60a1b2c3d4(分支创建时的 head);rebase 合 main 后 main 侧已新增
app_config / price_report 等迁移并 merge 出 head=ebb6af5c0b56,故改挂到其上,保持单 head(本迁移
与那些表互不相干,叠加顺序无影响)。
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision: str = 'withdraw_review_ad_watch'
down_revision: Union[str, Sequence[str], None] = 'ebb6af5c0b56'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
# ① 提现单加实名列:审核通过后异步打款时传给微信(达额转账要求),发起提现时存下
with op.batch_alter_table('withdraw_order', schema=None) as batch_op:
batch_op.add_column(sa.Column('user_name', sa.String(length=64), nullable=True))
# ② 看广告观看时长表:前端上报观看秒数,(user_id, watch_date) 聚合做每日 50 分钟防刷主闸
op.create_table(
'ad_watch_log',
sa.Column('id', sa.Integer(), autoincrement=True, nullable=False),
sa.Column('user_id', sa.Integer(), nullable=False),
sa.Column('watch_seconds', sa.Integer(), server_default='0', nullable=False),
sa.Column('watch_date', sa.String(length=10), nullable=False),
sa.Column(
'created_at', sa.DateTime(timezone=True),
server_default=sa.text('(CURRENT_TIMESTAMP)'), nullable=False,
),
sa.ForeignKeyConstraint(['user_id'], ['user.id'], ),
sa.PrimaryKeyConstraint('id'),
)
with op.batch_alter_table('ad_watch_log', schema=None) as batch_op:
batch_op.create_index(batch_op.f('ix_ad_watch_log_user_id'), ['user_id'], unique=False)
batch_op.create_index(batch_op.f('ix_ad_watch_log_watch_date'), ['watch_date'], unique=False)
batch_op.create_index(batch_op.f('ix_ad_watch_log_created_at'), ['created_at'], unique=False)
def downgrade() -> None:
with op.batch_alter_table('ad_watch_log', schema=None) as batch_op:
batch_op.drop_index(batch_op.f('ix_ad_watch_log_created_at'))
batch_op.drop_index(batch_op.f('ix_ad_watch_log_watch_date'))
batch_op.drop_index(batch_op.f('ix_ad_watch_log_user_id'))
op.drop_table('ad_watch_log')
with op.batch_alter_table('withdraw_order', schema=None) as batch_op:
batch_op.drop_column('user_name')
+6
View File
@@ -0,0 +1,6 @@
"""运营 Admin 后台子应用。
独立 FastAPI app(app.admin.main:admin_app),独立进程/端口运行,复用 App 的
models/repositories/integrations + 同一个 DB,但鉴权完全隔离(独立 JWT secret)。
现有 app.main:app 不 import 本包,admin 崩溃不影响 App 主进程。
"""
+38
View File
@@ -0,0 +1,38 @@
"""审计写入门面。
每个 admin 写操作调一次 write_audit,把"谁(admin)在哪个 IP 对什么(target)做了什么
(action)+ 前后值(detail)"落进 admin_audit_log。
⚠️ 涉钱/涉状态的写操作:传 commit=False,和业务写操作放同一事务一起 commit,
保证"改了就有痕、有痕就改了"原子(见 plan 风险点 1)。轻量操作可 commit=True。
"""
from __future__ import annotations
from sqlalchemy.orm import Session
from app.admin.repositories import audit_log as audit_repo
from app.models.admin import AdminUser
def write_audit(
db: Session,
admin: AdminUser,
*,
action: str,
target_type: str,
target_id: str | int | None = None,
detail: dict | None = None,
ip: str | None = None,
commit: bool = True,
) -> None:
audit_repo.add_audit_log(
db,
admin_id=admin.id,
admin_username=admin.username,
action=action,
target_type=target_type,
target_id=str(target_id) if target_id is not None else None,
detail=detail,
ip=ip,
commit=commit,
)
+84
View File
@@ -0,0 +1,84 @@
"""Admin API 共享依赖:DB session、当前 admin、角色守卫、客户端 IP。
仿 app/api/deps.py 的 get_current_user,但验的是独立的 admin JWT、查的是 admin_user 表。
"""
from __future__ import annotations
from typing import Annotated
from fastapi import Depends, HTTPException, Request, status
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from sqlalchemy.orm import Session
from app.admin.repositories import admin_user as admin_repo
from app.admin.security import AdminTokenError, decode_admin_token
from app.db.session import get_db
from app.models.admin import AdminUser
_bearer = HTTPBearer(auto_error=False, scheme_name="AdminBearer")
def get_current_admin(
credentials: Annotated[HTTPAuthorizationCredentials | None, Depends(_bearer)],
db: Annotated[Session, Depends(get_db)],
) -> AdminUser:
"""从 Authorization: Bearer <admin_token> 解出当前管理员。失败统一 401。"""
if credentials is None or credentials.scheme.lower() != "bearer":
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="missing bearer token",
headers={"WWW-Authenticate": "Bearer"},
)
try:
payload = decode_admin_token(credentials.credentials)
except AdminTokenError as e:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail=str(e),
headers={"WWW-Authenticate": "Bearer"},
) from e
admin = admin_repo.get_by_id(db, int(payload["sub"]))
if admin is None or admin.status != "active":
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="admin not found or disabled",
headers={"WWW-Authenticate": "Bearer"},
)
return admin
CurrentAdmin = Annotated[AdminUser, Depends(get_current_admin)]
AdminDb = Annotated[Session, Depends(get_db)]
def require_role(*roles: str):
"""角色守卫依赖工厂。super_admin 恒通过(全权)。
用法:在路由签名加 `_: Annotated[AdminUser, Depends(require_role("finance"))]`,
或 `dependencies=[Depends(require_role("finance"))]`(不需要拿 admin 时)。
"""
allowed = set(roles)
def _checker(admin: CurrentAdmin) -> AdminUser:
if admin.role != "super_admin" and admin.role not in allowed:
need = sorted(allowed | {"super_admin"})
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail=f"role '{admin.role}' not allowed (need one of {need})",
)
return admin
return _checker
def get_client_ip(request: Request) -> str:
"""取客户端 IP(审计日志用)。生产经 nginx 反代,优先 X-Forwarded-For 第一段;否则直连 IP。
⚠️ XFF 可被客户端伪造。nginx 必须用 `proxy_set_header X-Forwarded-For $remote_addr`
覆盖客户端传入值(见 M4 部署),否则审计里的 IP 可被伪造。审计 IP 仅作记录、不参与鉴权。
"""
xff = request.headers.get("x-forwarded-for")
if xff:
return xff.split(",")[0].strip()
return request.client.host if request.client else ""
+82
View File
@@ -0,0 +1,82 @@
"""Admin 后台 FastAPI app(独立进程)。
启动:uvicorn app.admin.main:admin_app --host 127.0.0.1 --port 8771
复用 App 的 DB/models/repositories/integrations;鉴权独立(admin JWT,见 app/admin/security.py)。
现有 app.main:app 不 import 本模块,两进程互不影响。
"""
from __future__ import annotations
import logging
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from app.admin.routers.admins import router as admins_router
from app.admin.routers.audit import router as audit_router
from app.admin.routers.auth import router as auth_router
from app.admin.routers.config import router as config_router
from app.admin.routers.dashboard import router as dashboard_router
from app.admin.routers.feedback import router as feedback_router
from app.admin.routers.users import router as users_router
from app.admin.routers.wallet import router as wallet_router
from app.admin.routers.withdraw import router as withdraw_router
from app.core.config import settings
from app.core.logging import setup_logging
setup_logging(debug=settings.APP_DEBUG)
logger = logging.getLogger("shagua.admin")
@asynccontextmanager
async def lifespan(_: FastAPI) -> AsyncIterator[None]:
logger.info(
"admin app started env=%s db=%s",
settings.APP_ENV,
settings.DATABASE_URL.split("://", 1)[0],
)
yield
logger.info("admin app shutting down")
admin_app = FastAPI(
title=f"{settings.APP_NAME} · Admin",
version="0.1.0",
docs_url="/admin/docs" if not settings.is_prod else None,
redoc_url=None,
lifespan=lifespan,
)
# admin 前端独立部署。生产同域(nginx)无需 CORS;本地 next dev 跨域需放行开发源。
_dev_origins = [
"http://localhost:3001",
"http://127.0.0.1:3001",
"http://localhost:3000",
"http://127.0.0.1:3000",
]
_origins = settings.cors_origins_list or ([] if settings.is_prod else _dev_origins)
if _origins:
admin_app.add_middleware(
CORSMiddleware,
allow_origins=_origins,
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
@admin_app.get("/admin/api/health", tags=["meta"])
def health() -> dict[str, str]:
return {"status": "ok", "service": "admin"}
admin_app.include_router(auth_router)
admin_app.include_router(dashboard_router)
admin_app.include_router(users_router)
admin_app.include_router(wallet_router)
admin_app.include_router(withdraw_router)
admin_app.include_router(feedback_router)
admin_app.include_router(admins_router)
admin_app.include_router(audit_router)
admin_app.include_router(config_router)
+1
View File
@@ -0,0 +1 @@
"""admin 专用数据访问层(跨用户查询 + admin 账号 + 审计 + 大盘聚合)。"""
+64
View File
@@ -0,0 +1,64 @@
"""admin_user 表 CRUD。"""
from __future__ import annotations
from datetime import datetime, timezone
from sqlalchemy import select
from sqlalchemy.orm import Session
from app.core.security import hash_password
from app.models.admin import AdminUser
def get_by_id(db: Session, admin_id: int) -> AdminUser | None:
return db.get(AdminUser, admin_id)
def get_by_username(db: Session, username: str) -> AdminUser | None:
stmt = select(AdminUser).where(AdminUser.username == username)
return db.execute(stmt).scalar_one_or_none()
def create_admin(
db: Session, *, username: str, password: str, role: str = "operator"
) -> AdminUser:
admin = AdminUser(
username=username,
password_hash=hash_password(password),
role=role,
)
db.add(admin)
db.commit()
db.refresh(admin)
return admin
def update_last_login(db: Session, admin: AdminUser) -> None:
admin.last_login_at = datetime.now(timezone.utc)
db.commit()
def list_admins(db: Session) -> list[AdminUser]:
stmt = select(AdminUser).order_by(AdminUser.id)
return list(db.execute(stmt).scalars().all())
def set_role(db: Session, admin: AdminUser, *, role: str) -> AdminUser:
admin.role = role
db.commit()
db.refresh(admin)
return admin
def set_status(db: Session, admin: AdminUser, *, status: str) -> AdminUser:
admin.status = status
db.commit()
db.refresh(admin)
return admin
def set_password(db: Session, admin: AdminUser, *, password: str) -> AdminUser:
admin.password_hash = hash_password(password)
db.commit()
db.refresh(admin)
return admin
+70
View File
@@ -0,0 +1,70 @@
"""admin_audit_log 写入 + 查询。
审计日志只增不改不删——任何写操作经 app.admin.audit.write_audit 落一条。
"""
from __future__ import annotations
from sqlalchemy import select
from sqlalchemy.orm import Session
from app.models.admin import AdminAuditLog
def add_audit_log(
db: Session,
*,
admin_id: int,
admin_username: str,
action: str,
target_type: str,
target_id: str | None = None,
detail: dict | None = None,
ip: str | None = None,
commit: bool = True,
) -> AdminAuditLog:
"""插一条审计。commit=False 时只 flush,让调用方把审计和业务写操作放同一事务。"""
log = AdminAuditLog(
admin_id=admin_id,
admin_username=admin_username,
action=action,
target_type=target_type,
target_id=target_id,
detail=detail,
ip=ip,
)
db.add(log)
if commit:
db.commit()
db.refresh(log)
else:
db.flush()
return log
def list_audit_logs(
db: Session,
*,
action: str | None = None,
target_type: str | None = None,
admin_id: int | None = None,
limit: int = 50,
cursor: int | None = None,
) -> tuple[list[AdminAuditLog], int | None]:
"""游标分页(id 倒序),与现有 list_* 约定一致。返回 (rows, next_cursor)。"""
stmt = select(AdminAuditLog)
if action:
stmt = stmt.where(AdminAuditLog.action == action)
if target_type:
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())
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
+36
View File
@@ -0,0 +1,36 @@
"""admin 写操作 repo(状态改写)。
涉钱的金币/提现复用 app.repositories.wallet(grant_coins / refresh_withdraw_status /
reconcile_pending_withdraws),不在这里重写——重写涉钱逻辑就是给自己埋雷。
set_user_status / update_feedback_status 支持 commit=False,让 router 把"业务写 + 审计写"
放进同一事务一起 commit(原子:改了就有审计、有审计就真改了,见 plan 风险点 1)。
"""
from __future__ import annotations
from sqlalchemy.orm import Session
from app.models.feedback import Feedback
from app.models.user import User
def set_user_status(db: Session, user: User, *, status: str, commit: bool = True) -> User:
user.status = status
if commit:
db.commit()
db.refresh(user)
else:
db.flush()
return user
def update_feedback_status(
db: Session, feedback: Feedback, *, status: str, commit: bool = True
) -> Feedback:
feedback.status = status
if commit:
db.commit()
db.refresh(feedback)
else:
db.flush()
return feedback
+153
View File
@@ -0,0 +1,153 @@
"""admin 跨用户查询(去掉现有 repo 的 user_id 强制过滤)+ 通用游标分页 helper + 用户概览。
现有 app/repositories/ 的 list_* 都强绑单个 user_id(C 端只看自己);admin 要看全量、按条件筛,
所以在这里另起一套。游标约定与现有一致:id 倒序,cursor=上页最后一条 id,返回 (items, next_cursor)。
"""
from __future__ import annotations
from sqlalchemy import Select, func, select
from sqlalchemy.orm import Session
from app.models.comparison import ComparisonRecord
from app.models.feedback import Feedback
from app.models.user import User
from app.models.wallet import CashTransaction, CoinAccount, CoinTransaction, WithdrawOrder
def cursor_paginate(
db: Session, stmt: Select, id_col, *, limit: int, cursor: int | None
) -> tuple[list, int | None]:
"""通用游标分页(id 倒序)。stmt 不要预先带 order_by/limit。
多取 1 条探测有没有下一页;next_cursor 取本页最后一条的 id(下一页查 id < 它),
绝不用第 limit+1 条的 id——那条既不在本页也不在下页,会每页边界丢一条(见 audit_log 同款修复)。
"""
if cursor is not None:
stmt = stmt.where(id_col < cursor)
stmt = stmt.order_by(id_col.desc()).limit(limit + 1)
rows = list(db.execute(stmt).scalars().all())
has_more = len(rows) > limit
items = rows[:limit]
next_cursor = items[-1].id if has_more else None
return items, next_cursor
def list_users(
db: Session,
*,
phone: str | None = None,
register_channel: str | None = None,
status: str | None = None,
limit: int = 20,
cursor: int | None = None,
) -> tuple[list[User], int | None]:
stmt = select(User)
if phone:
stmt = stmt.where(User.phone.like(f"{phone}%")) # 前缀匹配
if register_channel:
stmt = stmt.where(User.register_channel == register_channel)
if status:
stmt = stmt.where(User.status == status)
return cursor_paginate(db, stmt, User.id, limit=limit, cursor=cursor)
def list_all_coin_transactions(
db: Session,
*,
user_id: int | None = None,
biz_type: str | None = None,
limit: int = 20,
cursor: int | None = None,
) -> tuple[list[CoinTransaction], int | None]:
stmt = select(CoinTransaction)
if user_id is not None:
stmt = stmt.where(CoinTransaction.user_id == user_id)
if biz_type:
stmt = stmt.where(CoinTransaction.biz_type == biz_type)
return cursor_paginate(db, stmt, CoinTransaction.id, limit=limit, cursor=cursor)
def list_all_cash_transactions(
db: Session,
*,
user_id: int | None = None,
biz_type: str | None = None,
limit: int = 20,
cursor: int | None = None,
) -> tuple[list[CashTransaction], int | None]:
stmt = select(CashTransaction)
if user_id is not None:
stmt = stmt.where(CashTransaction.user_id == user_id)
if biz_type:
stmt = stmt.where(CashTransaction.biz_type == biz_type)
return cursor_paginate(db, stmt, CashTransaction.id, limit=limit, cursor=cursor)
def list_all_withdraw_orders(
db: Session,
*,
user_id: int | None = None,
status: str | None = None,
limit: int = 20,
cursor: int | None = None,
) -> tuple[list[WithdrawOrder], int | None]:
stmt = select(WithdrawOrder)
if user_id is not None:
stmt = stmt.where(WithdrawOrder.user_id == user_id)
if status:
stmt = stmt.where(WithdrawOrder.status == status)
return cursor_paginate(db, stmt, WithdrawOrder.id, limit=limit, cursor=cursor)
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(
select(WithdrawOrder).where(WithdrawOrder.out_bill_no == out_bill_no)
).scalar_one_or_none()
def get_user_overview(db: Session, user_id: int) -> dict | None:
"""用户 360 概览:基础资料 + 钱包余额 + 各项 count。历史明细走各自分页接口(带 user_id 过滤)。"""
user = db.get(User, user_id)
if user is None:
return None
acc = db.get(CoinAccount, user_id) # 可能为 None(从未发生过金币动作)
def _count(model, *conds) -> int:
return db.execute(select(func.count(model.id)).where(*conds)).scalar_one()
return {
"user": user,
"coin_balance": acc.coin_balance if acc else 0,
"cash_balance_cents": acc.cash_balance_cents if acc else 0,
"total_coin_earned": acc.total_coin_earned if acc else 0,
"comparison_total": _count(ComparisonRecord, ComparisonRecord.user_id == user_id),
"comparison_success": _count(
ComparisonRecord,
ComparisonRecord.user_id == user_id,
ComparisonRecord.status == "success",
),
"withdraw_total": _count(WithdrawOrder, WithdrawOrder.user_id == user_id),
"withdraw_success_cents": db.execute(
select(func.coalesce(func.sum(WithdrawOrder.amount_cents), 0)).where(
WithdrawOrder.user_id == user_id, WithdrawOrder.status == "success"
)
).scalar_one(),
"feedback_total": _count(Feedback, Feedback.user_id == user_id),
}
+91
View File
@@ -0,0 +1,91 @@
"""admin 大盘聚合查询(全局 count/sum/DAU/成功率)。全部只读、不改任何数据。
⚠️ 性能:这些是全表 count/sum,P0 数据量小够用;用户量上来后热点字段(user.created_at /
user.last_login_at / comparison_record.status / withdraw_order.status)要加索引,或改增量统计表。
"""
from __future__ import annotations
from datetime import datetime, timedelta, timezone
from sqlalchemy import func, select
from sqlalchemy.orm import Session
from app.models.comparison import ComparisonRecord
from app.models.feedback import Feedback
from app.models.user import User
from app.models.wallet import CoinTransaction, WithdrawOrder
_BEIJING = timezone(timedelta(hours=8))
def _beijing_today_start_utc() -> datetime:
"""北京时间今天 0 点对应的 UTC 时刻(DAU / 今日新增按北京时区切天)。"""
now_bj = datetime.now(_BEIJING)
start_bj = now_bj.replace(hour=0, minute=0, second=0, microsecond=0)
return start_bj.astimezone(timezone.utc)
def dashboard_overview(db: Session) -> dict:
today_start = _beijing_today_start_utc()
def _count(model, *conds) -> int:
stmt = select(func.count(model.id))
if conds:
stmt = stmt.where(*conds)
return db.execute(stmt).scalar_one()
def _sum(col, *conds) -> int:
stmt = select(func.coalesce(func.sum(col), 0))
if conds:
stmt = stmt.where(*conds)
return db.execute(stmt).scalar_one()
# ===== 用户 =====
by_status = dict(
db.execute(select(User.status, func.count(User.id)).group_by(User.status)).all()
)
# ===== 提现状态分布 =====
wd_by_status = dict(
db.execute(
select(WithdrawOrder.status, func.count(WithdrawOrder.id)).group_by(
WithdrawOrder.status
)
).all()
)
# ===== 比价 =====
comparison_total = _count(ComparisonRecord)
comparison_success = _count(ComparisonRecord, ComparisonRecord.status == "success")
success_rate = round(comparison_success / comparison_total, 4) if comparison_total else 0.0
return {
"users": {
"total": _count(User),
"active": by_status.get("active", 0),
"disabled": by_status.get("disabled", 0),
"deleted": by_status.get("deleted", 0),
"new_today": _count(User, User.created_at >= today_start),
"dau": _count(User, User.last_login_at >= today_start),
},
"coins": {
# 累计发放金币(coin_transaction 里所有 amount>0 之和;负数是兑换/扣减不计)
"granted_total": _sum(CoinTransaction.amount, CoinTransaction.amount > 0),
},
"cash": {
"withdraw_success_cents": _sum(
WithdrawOrder.amount_cents, WithdrawOrder.status == "success"
),
"withdraw_pending_count": wd_by_status.get("pending", 0),
"withdraw_success_count": wd_by_status.get("success", 0),
"withdraw_failed_count": wd_by_status.get("failed", 0),
},
"comparison": {
"total": comparison_total,
"success": comparison_success,
"success_rate": success_rate,
},
"feedback": {"new": _count(Feedback, Feedback.status == "new")},
# CPS 收入数据源未接(referral-link 只换链接,转化/佣金未回收)→ 前端显示"待接入"。
"cps": {"available": False, "note": "CPS 转化数据未接入(P2)"},
}
+1
View File
@@ -0,0 +1 @@
"""admin 路由(前缀统一 /admin/api)。"""
+69
View File
@@ -0,0 +1,69 @@
"""admin 账号管理(仅 super_admin):列表 / 创建 / 改角色启停重置密码。均写审计。"""
from __future__ import annotations
from fastapi import APIRouter, Depends, HTTPException, Request
from app.admin.audit import write_audit
from app.admin.deps import AdminDb, CurrentAdmin, get_client_ip, require_role
from app.admin.repositories import admin_user as admin_repo
from app.admin.schemas.admin import AdminCreateRequest, AdminUpdateRequest
from app.admin.schemas.auth import AdminOut
from app.core.security import hash_password
router = APIRouter(
prefix="/admin/api/admins",
tags=["admin-accounts"],
dependencies=[Depends(require_role())], # require_role() 无参 = 仅 super_admin 通过
)
@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)]
@router.post("", response_model=AdminOut, summary="创建管理员")
def create_admin(
body: AdminCreateRequest, request: Request, admin: CurrentAdmin, db: AdminDb
) -> AdminOut:
if admin_repo.get_by_username(db, body.username) is not None:
raise HTTPException(status_code=409, detail="用户名已存在")
new = admin_repo.create_admin(
db, username=body.username, password=body.password, role=body.role
)
write_audit(
db, admin, action="admin.create", target_type="admin", target_id=new.id,
detail={"username": new.username, "role": new.role}, ip=get_client_ip(request), commit=True,
)
return AdminOut.model_validate(new)
@router.patch("/{admin_id}", response_model=AdminOut, summary="改角色/启停/重置密码")
def update_admin(
admin_id: int, body: AdminUpdateRequest, request: Request, admin: CurrentAdmin, db: AdminDb
) -> AdminOut:
target = admin_repo.get_by_id(db, admin_id)
if target is None:
raise HTTPException(status_code=404, detail="管理员不存在")
if admin_id == admin.id and body.status == "disabled":
raise HTTPException(status_code=400, detail="不能禁用自己")
changes: dict = {}
if body.role is not None:
target.role = body.role
changes["role"] = body.role
if body.status is not None:
target.status = body.status
changes["status"] = body.status
if body.password is not None:
target.password_hash = hash_password(body.password)
changes["password"] = "reset"
if not changes:
raise HTTPException(status_code=400, detail="无任何变更字段")
db.commit()
db.refresh(target)
write_audit(
db, admin, action="admin.update", target_type="admin", target_id=admin_id,
detail=changes, ip=get_client_ip(request), commit=True,
)
return AdminOut.model_validate(target)
+34
View File
@@ -0,0 +1,34 @@
"""admin 操作审计日志查询(所有 admin 可看:谁在何时对什么做了什么)。"""
from __future__ import annotations
from typing import Annotated
from fastapi import APIRouter, Depends, Query
from app.admin.deps import AdminDb, get_current_admin
from app.admin.repositories import audit_log as audit_repo
from app.admin.schemas.admin import AdminAuditLogOut
from app.admin.schemas.common import CursorPage
router = APIRouter(
prefix="/admin/api/audit-logs",
tags=["admin-audit"],
dependencies=[Depends(get_current_admin)],
)
@router.get("", response_model=CursorPage[AdminAuditLogOut], summary="审计日志(谁改了什么)")
def list_audit_logs(
db: AdminDb,
action: Annotated[str | None, Query()] = None,
target_type: Annotated[str | None, Query()] = None,
admin_id: Annotated[int | None, Query()] = None,
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(
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,
)
+48
View File
@@ -0,0 +1,48 @@
"""Admin 认证:账号密码登录 → admin JWT(独立 secret)。"""
from __future__ import annotations
import logging
from fastapi import APIRouter, Depends, HTTPException
from app.admin.deps import AdminDb, CurrentAdmin
from app.admin.repositories import admin_user as admin_repo
from app.admin.schemas.auth import AdminLoginRequest, AdminLoginResponse, AdminOut
from app.admin.security import create_admin_token
from app.core.ratelimit import rate_limit
from app.core.security import verify_password
logger = logging.getLogger("shagua.admin.auth")
router = APIRouter(prefix="/admin/api/auth", tags=["admin-auth"])
@router.post(
"/login",
response_model=AdminLoginResponse,
summary="管理员登录",
dependencies=[Depends(rate_limit(10, 60, "admin-login"))], # 同 IP 每分钟≤10 次,防爆破
)
def login(req: AdminLoginRequest, db: AdminDb) -> AdminLoginResponse:
admin = admin_repo.get_by_username(db, req.username)
# 用户名不存在 / 密码错统一 401 同文案(防账号枚举)。
# disabled 账号单独 403"账号已禁用":admin 是内部少数已知账号、无枚举价值,
# 明确提示比防枚举更有运维价值(与 App 端 auth.py 对 disabled 用户的 403 一致)。
if admin is None or not verify_password(req.password, admin.password_hash):
raise HTTPException(status_code=401, detail="用户名或密码错误")
if admin.status != "active":
raise HTTPException(status_code=403, detail="账号已禁用")
token, expires_in = create_admin_token(admin_id=admin.id, role=admin.role)
admin_repo.update_last_login(db, admin)
logger.info("admin login ok id=%d username=%s role=%s", admin.id, admin.username, admin.role)
return AdminLoginResponse(
access_token=token,
expires_in=expires_in,
admin=AdminOut.model_validate(admin),
)
@router.get("/me", response_model=AdminOut, summary="当前管理员")
def me(admin: CurrentAdmin) -> AdminOut:
return AdminOut.model_validate(admin)
+81
View File
@@ -0,0 +1,81 @@
"""admin 运营配置:列出所有可配项 + 改某项(带审计 + 改值校验)。
配置项定义见 app.core.config_schema.CONFIG_DEFS;业务读配置 fallback 默认(见 app_config repo)。
权限:operator / finance 都可改(运营调奖励、财务调额度),super 恒可。
"""
from __future__ import annotations
from typing import Annotated, Any
from fastapi import APIRouter, Depends, HTTPException, Request
from app.admin.audit import write_audit
from app.admin.deps import AdminDb, get_client_ip, get_current_admin, require_role
from app.admin.schemas.config import ConfigItemOut, ConfigUpdateRequest
from app.core.config_schema import CONFIG_DEFS
from app.models.admin import AdminUser
from app.repositories import app_config
router = APIRouter(
prefix="/admin/api/config",
tags=["admin-config"],
dependencies=[Depends(get_current_admin)],
)
def _validate(key: str, value: Any) -> None:
"""按配置项 type 校验新值,不合法抛 ValueError(router 转 400)。"""
t = CONFIG_DEFS[key]["type"]
if t == "int":
if not isinstance(value, int) or isinstance(value, bool) or value < 0:
raise ValueError("需为非负整数")
elif t == "int_list":
if not isinstance(value, list) or not value:
raise ValueError("需为非空整数列表")
if not all(isinstance(x, int) and not isinstance(x, bool) and x >= 0 for x in value):
raise ValueError("列表元素需为非负整数")
if key == "signin_rewards" and len(value) != 7:
raise ValueError("签到档位必须正好 7 个(对应 7 天循环)")
elif t == "dict_str_int":
if not isinstance(value, dict) or not all(
isinstance(k, str) and isinstance(v, int) and not isinstance(v, bool)
for k, v in value.items()
):
raise ValueError("需为 {字符串: 整数} 映射")
def _item(db, key: str) -> ConfigItemOut:
for item in app_config.list_all(db):
if item["key"] == key:
return ConfigItemOut(**item)
raise HTTPException(status_code=404, detail="未知配置项")
@router.get("", response_model=list[ConfigItemOut], summary="所有可配项 + 当前值")
def list_config(db: AdminDb) -> list[ConfigItemOut]:
return [ConfigItemOut(**item) for item in app_config.list_all(db)]
@router.patch("/{key}", response_model=ConfigItemOut, summary="改某项配置(带审计)")
def update_config(
key: str,
body: ConfigUpdateRequest,
request: Request,
admin: Annotated[AdminUser, Depends(require_role("operator", "finance"))],
db: AdminDb,
) -> ConfigItemOut:
if key not in CONFIG_DEFS:
raise HTTPException(status_code=404, detail="未知配置项")
try:
_validate(key, body.value)
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e)) from e
before = app_config.get_value(db, key)
app_config.set_value(db, key, body.value, admin_id=admin.id, commit=False)
write_audit(
db, admin, action="config.set", target_type="config", target_id=key,
detail={"before": before, "after": body.value}, ip=get_client_ip(request), commit=False,
)
db.commit()
return _item(db, key)
+19
View File
@@ -0,0 +1,19 @@
"""admin 数据大盘(只读聚合)。"""
from __future__ import annotations
from fastapi import APIRouter, Depends
from app.admin.deps import AdminDb, get_current_admin
from app.admin.repositories import stats
from app.admin.schemas.dashboard import DashboardOverview
router = APIRouter(
prefix="/admin/api/stats",
tags=["admin-stats"],
dependencies=[Depends(get_current_admin)],
)
@router.get("/overview", response_model=DashboardOverview, summary="大盘核心指标")
def overview(db: AdminDb) -> DashboardOverview:
return DashboardOverview.model_validate(stats.dashboard_overview(db))
+56
View File
@@ -0,0 +1,56 @@
"""admin 反馈工单:列表(读)+ 标记已处理(写,带审计)。"""
from __future__ import annotations
from typing import Annotated
from fastapi import APIRouter, Depends, HTTPException, Query, Request
from app.admin.audit import write_audit
from app.admin.deps import AdminDb, get_client_ip, get_current_admin, require_role
from app.admin.repositories import mutations, queries
from app.admin.schemas.common import CursorPage, OkResponse
from app.admin.schemas.feedback import FeedbackOut
from app.models.admin import AdminUser
from app.models.feedback import Feedback
router = APIRouter(
prefix="/admin/api/feedbacks",
tags=["admin-feedback"],
dependencies=[Depends(get_current_admin)],
)
@router.get("", response_model=CursorPage[FeedbackOut], summary="反馈工单列表")
def list_feedbacks(
db: AdminDb,
status: Annotated[str | None, Query()] = None,
user_id: Annotated[int | None, Query()] = None,
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,
)
return CursorPage(
items=[FeedbackOut.model_validate(f) for f in items], next_cursor=next_cursor,
)
@router.post("/{feedback_id}/handle", response_model=OkResponse, summary="标记反馈已处理")
def handle_feedback(
feedback_id: int,
request: Request,
admin: Annotated[AdminUser, Depends(require_role("operator"))],
db: AdminDb,
) -> OkResponse:
fb = db.get(Feedback, feedback_id)
if fb is None:
raise HTTPException(status_code=404, detail="反馈不存在")
before = fb.status
mutations.update_feedback_status(db, fb, status="handled", commit=False)
write_audit(
db, admin, action="feedback.handle", target_type="feedback", target_id=feedback_id,
detail={"before": before, "after": "handled"}, ip=get_client_ip(request), commit=False,
)
db.commit()
return OkResponse()
+111
View File
@@ -0,0 +1,111 @@
"""admin 用户管理:列表 + 360 详情(读)+ 封禁/解封 + 手动调金币(写,带审计)。"""
from __future__ import annotations
from typing import Annotated
from fastapi import APIRouter, Depends, HTTPException, Query, Request
from app.admin.audit import write_audit
from app.admin.deps import AdminDb, get_client_ip, get_current_admin, require_role
from app.admin.repositories import mutations, queries
from app.admin.schemas.common import CursorPage, OkResponse
from app.admin.schemas.user import (
AdminUserListItem,
AdminUserOverview,
GrantCoinsRequest,
SetUserStatusRequest,
)
from app.models.admin import AdminUser
from app.repositories import user as user_repo
from app.repositories import wallet as wallet_repo
router = APIRouter(
prefix="/admin/api/users",
tags=["admin-users"],
dependencies=[Depends(get_current_admin)],
)
@router.get("", response_model=CursorPage[AdminUserListItem], summary="用户列表(筛选+分页)")
def list_users(
db: AdminDb,
phone: Annotated[str | None, Query()] = None,
register_channel: Annotated[str | None, Query()] = None,
status: Annotated[str | None, Query()] = None,
limit: Annotated[int, Query(ge=1, le=100)] = 20,
cursor: Annotated[int | None, Query()] = None,
) -> CursorPage[AdminUserListItem]:
items, next_cursor = queries.list_users(
db, phone=phone, register_channel=register_channel, status=status,
limit=limit, cursor=cursor,
)
return CursorPage(
items=[AdminUserListItem.model_validate(u) for u in items],
next_cursor=next_cursor,
)
@router.get("/{user_id}", response_model=AdminUserOverview, summary="用户 360 详情")
def get_user(user_id: int, db: AdminDb) -> AdminUserOverview:
overview = queries.get_user_overview(db, user_id)
if overview is None:
raise HTTPException(status_code=404, detail="用户不存在")
return AdminUserOverview.model_validate(overview)
@router.post("/{user_id}/status", response_model=OkResponse, summary="封禁/解封用户")
def set_user_status(
user_id: int,
body: SetUserStatusRequest,
request: Request,
admin: Annotated[AdminUser, Depends(require_role("operator"))],
db: AdminDb,
) -> OkResponse:
user = user_repo.get_user_by_id(db, user_id)
if user is None:
raise HTTPException(status_code=404, detail="用户不存在")
if user.status == "deleted":
raise HTTPException(status_code=400, detail="已注销账号不可改状态")
before = user.status
# 业务写 + 审计写同一事务(commit=False),最后一起 commit:改了就有痕、有痕就真改了
mutations.set_user_status(db, user, status=body.status, commit=False)
write_audit(
db, admin, action="user.status.set", target_type="user", target_id=user_id,
detail={"before": before, "after": body.status}, ip=get_client_ip(request), commit=False,
)
db.commit()
return OkResponse()
@router.post("/{user_id}/coins", response_model=OkResponse, summary="手动增减金币(带审计)")
def grant_user_coins(
user_id: int,
body: GrantCoinsRequest,
request: Request,
admin: Annotated[AdminUser, Depends(require_role("finance"))],
db: AdminDb,
) -> OkResponse:
if body.amount == 0:
raise HTTPException(status_code=400, detail="amount 不能为 0")
user = user_repo.get_user_by_id(db, user_id)
if user is None:
raise HTTPException(status_code=404, detail="用户不存在")
# 负数扣减时不允许扣成负余额(运营误操作保护)
if body.amount < 0:
acc_now = wallet_repo.get_or_create_account(db, user_id, commit=False)
if acc_now.coin_balance + body.amount < 0:
raise HTTPException(
status_code=400, detail=f"扣减后金币为负(当前余额 {acc_now.coin_balance})"
)
biz_type = "admin_grant" if body.amount > 0 else "admin_deduct"
# grant_coins 只 flush 不 commit;审计同 commit=False;最后一起 commit → 原子(改钱+留痕)
acc, _ = wallet_repo.grant_coins(
db, user_id, body.amount, biz_type=biz_type, remark=f"admin:{body.reason}"[:128],
)
write_audit(
db, admin, action="user.coins.grant", target_type="user", target_id=user_id,
detail={"amount": body.amount, "balance_after": acc.coin_balance, "reason": body.reason},
ip=get_client_ip(request), commit=False,
)
db.commit()
return OkResponse()
+49
View File
@@ -0,0 +1,49 @@
"""admin 钱包:金币流水 + 现金流水(跨用户,可按 user_id 过滤)。手动调金币见 M3。"""
from __future__ import annotations
from typing import Annotated
from fastapi import APIRouter, Depends, Query
from app.admin.deps import AdminDb, get_current_admin
from app.admin.repositories import queries
from app.admin.schemas.common import CursorPage
from app.admin.schemas.wallet import CashTxnOut, CoinTxnOut
router = APIRouter(
prefix="/admin/api/wallet",
tags=["admin-wallet"],
dependencies=[Depends(get_current_admin)],
)
@router.get("/coin-transactions", response_model=CursorPage[CoinTxnOut], summary="金币流水")
def coin_transactions(
db: AdminDb,
user_id: Annotated[int | None, Query()] = None,
biz_type: Annotated[str | None, Query()] = None,
limit: Annotated[int, Query(ge=1, le=100)] = 20,
cursor: Annotated[int | None, Query()] = None,
) -> CursorPage[CoinTxnOut]:
items, next_cursor = queries.list_all_coin_transactions(
db, user_id=user_id, biz_type=biz_type, limit=limit, cursor=cursor,
)
return CursorPage(
items=[CoinTxnOut.model_validate(t) for t in items], next_cursor=next_cursor,
)
@router.get("/cash-transactions", response_model=CursorPage[CashTxnOut], summary="现金流水")
def cash_transactions(
db: AdminDb,
user_id: Annotated[int | None, Query()] = None,
biz_type: Annotated[str | None, Query()] = None,
limit: Annotated[int, Query(ge=1, le=100)] = 20,
cursor: Annotated[int | None, Query()] = None,
) -> CursorPage[CashTxnOut]:
items, next_cursor = queries.list_all_cash_transactions(
db, user_id=user_id, biz_type=biz_type, limit=limit, cursor=cursor,
)
return CursorPage(
items=[CashTxnOut.model_validate(t) for t in items], next_cursor=next_cursor,
)
+140
View File
@@ -0,0 +1,140 @@
"""admin 提现:列表(读)+ 单笔重试查单 + 批量对账(写,带审计)。
提现的钱逻辑(查微信/退款/撤单/幂等)全部复用 app.repositories.wallet,admin 只触发 + 记审计。
这些 wallet 函数内部各自 commit(涉及微信调用),审计在其后单独 commit:操作本身幂等,
审计记录"谁触发的 + 结果",事务边界比"改金币"宽松是有意的(不能塞进 wallet 的自有事务)。
"""
from __future__ import annotations
from typing import Annotated
from fastapi import APIRouter, Depends, HTTPException, Query, Request
from app.admin.audit import write_audit
from app.admin.deps import AdminDb, get_client_ip, get_current_admin, require_role
from app.admin.repositories import queries
from app.admin.schemas.common import CursorPage
from app.admin.schemas.wallet import ReconcileResult, WithdrawOrderOut, WithdrawRejectRequest
from app.integrations import wxpay
from app.models.admin import AdminUser
from app.repositories import wallet as wallet_repo
router = APIRouter(
prefix="/admin/api/withdraws",
tags=["admin-withdraw"],
dependencies=[Depends(get_current_admin)],
)
@router.get("", response_model=CursorPage[WithdrawOrderOut], summary="提现单列表")
def list_withdraws(
db: AdminDb,
user_id: Annotated[int | None, Query()] = None,
status: Annotated[str | None, Query()] = None,
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(
db, user_id=user_id, status=status, limit=limit, cursor=cursor,
)
return CursorPage(
items=[WithdrawOrderOut.model_validate(o) for o in items], next_cursor=next_cursor,
)
# 注意:/reconcile 必须在 /{out_bill_no}/refresh 之前声明(静态路径优先于路径参数)
@router.post("/reconcile", response_model=ReconcileResult, summary="批量对账(扫超时 pending 单)")
def reconcile(
request: Request,
admin: Annotated[AdminUser, Depends(require_role("finance"))],
db: AdminDb,
older_than_minutes: Annotated[int, Query(ge=0)] = 15,
) -> ReconcileResult:
try:
result = wallet_repo.reconcile_pending_withdraws(db, older_than_minutes=older_than_minutes)
except wxpay.WxPayNotConfiguredError as e:
raise HTTPException(status_code=503, detail="微信支付未配置") from e
write_audit(
db, admin, action="withdraw.reconcile", target_type="withdraw", target_id=None,
detail=result, ip=get_client_ip(request), commit=True,
)
return ReconcileResult(**result)
@router.post("/{out_bill_no}/refresh", response_model=WithdrawOrderOut, summary="单笔重试查单")
def refresh_withdraw(
out_bill_no: str,
request: Request,
admin: Annotated[AdminUser, Depends(require_role("finance"))],
db: AdminDb,
) -> WithdrawOrderOut:
order = queries.get_withdraw_by_out_bill_no(db, out_bill_no)
if order is None:
raise HTTPException(status_code=404, detail="提现单不存在")
try:
refreshed = wallet_repo.refresh_withdraw_status(
db, order.user_id, out_bill_no, cancel_if_unconfirmed=True,
)
except wxpay.WxPayNotConfiguredError as e:
raise HTTPException(status_code=503, detail="微信支付未配置") from e
write_audit(
db, admin, action="withdraw.refresh", target_type="withdraw", target_id=out_bill_no,
detail={"status": refreshed.status, "wechat_state": refreshed.wechat_state},
ip=get_client_ip(request), commit=True,
)
return WithdrawOrderOut.model_validate(refreshed)
@router.post("/{out_bill_no}/approve", response_model=WithdrawOrderOut, summary="审核通过(发起微信打款)")
def approve_withdraw_order(
out_bill_no: str,
request: Request,
admin: Annotated[AdminUser, Depends(require_role("finance"))],
db: AdminDb,
) -> WithdrawOrderOut:
"""审核通过 → 调微信转账。仅 reviewing 单可通过(非 reviewing 返 409,防重复打款)。
打款的钱逻辑(转账/查单/退款/幂等)复用 wallet_repo.approve_withdraw → execute_withdraw_transfer;
返回的 order 可能是 pending(在途/待用户确认)/success/failed(打款失败已退),admin 前端按 status 展示。
"""
try:
order = wallet_repo.approve_withdraw(db, out_bill_no)
except wallet_repo.WithdrawOrderNotFound as e:
raise HTTPException(status_code=404, detail="提现单不存在") from e
except wallet_repo.WithdrawNotReviewable as e:
raise HTTPException(status_code=409, detail=str(e)) from e
except wxpay.WxPayNotConfiguredError as e:
raise HTTPException(status_code=503, detail="微信支付未配置") from e
write_audit(
db, admin, action="withdraw.approve", target_type="withdraw", target_id=out_bill_no,
detail={
"status": order.status,
"wechat_state": order.wechat_state,
"amount_cents": order.amount_cents,
},
ip=get_client_ip(request), commit=True,
)
return WithdrawOrderOut.model_validate(order)
@router.post("/{out_bill_no}/reject", response_model=WithdrawOrderOut, summary="审核拒绝(退回现金)")
def reject_withdraw_order(
out_bill_no: str,
body: WithdrawRejectRequest,
request: Request,
admin: Annotated[AdminUser, Depends(require_role("finance"))],
db: AdminDb,
) -> WithdrawOrderOut:
"""审核拒绝 → 退回用户现金 + 置 rejected,理由写入 fail_reason(用户可见)。仅 reviewing 单可拒。"""
try:
order = wallet_repo.reject_withdraw(db, out_bill_no, body.reason)
except wallet_repo.WithdrawOrderNotFound as e:
raise HTTPException(status_code=404, detail="提现单不存在") from e
except wallet_repo.WithdrawNotReviewable as e:
raise HTTPException(status_code=409, detail=str(e)) from e
write_audit(
db, admin, action="withdraw.reject", target_type="withdraw", target_id=out_bill_no,
detail={"reason": body.reason, "amount_cents": order.amount_cents},
ip=get_client_ip(request), commit=True,
)
return WithdrawOrderOut.model_validate(order)
+1
View File
@@ -0,0 +1 @@
"""admin 请求/响应 schemas(snake_case,与前端约定一致)。"""
+37
View File
@@ -0,0 +1,37 @@
"""admin 账号管理 + 审计日志 schemas。"""
from __future__ import annotations
from datetime import datetime
from typing import Literal
from pydantic import BaseModel, ConfigDict, Field
_Role = Literal["super_admin", "finance", "operator"]
class AdminCreateRequest(BaseModel):
username: str = Field(..., min_length=3, max_length=64)
password: str = Field(..., min_length=8, max_length=72) # bcrypt ≤72 字节
role: _Role = "operator"
class AdminUpdateRequest(BaseModel):
"""改角色 / 启用禁用 / 重置密码,字段都可选(只改传了的)。"""
role: _Role | None = None
status: Literal["active", "disabled"] | None = None
password: str | None = Field(None, min_length=8, max_length=72)
class AdminAuditLogOut(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: int
admin_id: int
admin_username: str
action: str
target_type: str
target_id: str | None = None
detail: dict | None = None
ip: str | None = None
created_at: datetime
+29
View File
@@ -0,0 +1,29 @@
"""Admin 认证 schemas。"""
from __future__ import annotations
from datetime import datetime
from pydantic import BaseModel, ConfigDict, Field
class AdminLoginRequest(BaseModel):
username: str = Field(..., min_length=1, max_length=64)
password: str = Field(..., min_length=1, max_length=128)
class AdminOut(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: int
username: str
role: str
status: str
created_at: datetime
last_login_at: datetime | None = None
class AdminLoginResponse(BaseModel):
access_token: str
token_type: str = "Bearer"
expires_in: int = Field(..., description="access_token 剩余秒数(过期重新登录)")
admin: AdminOut
+19
View File
@@ -0,0 +1,19 @@
"""通用 schema:游标分页响应 + 通用 ok 响应。"""
from __future__ import annotations
from typing import Generic, TypeVar
from pydantic import BaseModel
T = TypeVar("T")
class CursorPage(BaseModel, Generic[T]):
"""游标分页响应:items + 下一页游标(next_cursor=None 表示末页)。"""
items: list[T]
next_cursor: int | None = None
class OkResponse(BaseModel):
ok: bool = True
+22
View File
@@ -0,0 +1,22 @@
"""admin 运营配置 schemas。"""
from __future__ import annotations
from typing import Any
from pydantic import BaseModel
class ConfigItemOut(BaseModel):
key: str
label: str
group: str
type: str # int / int_list / dict_str_int
help: str | None = None
default: Any
value: Any
overridden: bool # 是否被运营覆盖过(false=用默认)
updated_at: str | None = None
class ConfigUpdateRequest(BaseModel):
value: Any
+48
View File
@@ -0,0 +1,48 @@
"""admin 大盘 schemas(对应 stats.dashboard_overview 的嵌套结构)。"""
from __future__ import annotations
from pydantic import BaseModel
class DashboardUsers(BaseModel):
total: int
active: int
disabled: int
deleted: int
new_today: int
dau: int
class DashboardCoins(BaseModel):
granted_total: int
class DashboardCash(BaseModel):
withdraw_success_cents: int
withdraw_pending_count: int
withdraw_success_count: int
withdraw_failed_count: int
class DashboardComparison(BaseModel):
total: int
success: int
success_rate: float
class DashboardFeedback(BaseModel):
new: int
class DashboardCps(BaseModel):
available: bool
note: str
class DashboardOverview(BaseModel):
users: DashboardUsers
coins: DashboardCoins
cash: DashboardCash
comparison: DashboardComparison
feedback: DashboardFeedback
cps: DashboardCps
+18
View File
@@ -0,0 +1,18 @@
"""admin 反馈工单 schemas。"""
from __future__ import annotations
from datetime import datetime
from pydantic import BaseModel, ConfigDict
class FeedbackOut(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: int
user_id: int
content: str
contact: str
images: list[str] | None = None
status: str
created_at: datetime
+47
View File
@@ -0,0 +1,47 @@
"""admin 用户管理 schemas。"""
from __future__ import annotations
from datetime import datetime
from typing import Literal
from pydantic import BaseModel, ConfigDict, Field
class AdminUserListItem(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: int
phone: str
nickname: str | None = None
register_channel: str
status: str
wechat_openid: str | None = None
created_at: datetime
last_login_at: datetime
class AdminUserOverview(BaseModel):
"""用户 360 概览:基础资料 + 钱包余额 + 各项 count(历史明细走各自分页接口)。"""
model_config = ConfigDict(from_attributes=True)
user: AdminUserListItem
coin_balance: int
cash_balance_cents: int
total_coin_earned: int
comparison_total: int
comparison_success: int
withdraw_total: int
withdraw_success_cents: int
feedback_total: int
class GrantCoinsRequest(BaseModel):
amount: int = Field(..., description="金币变动:正=增加,负=扣减(不可为 0)")
reason: str = Field(..., min_length=1, max_length=128, description="操作原因(必填,入审计)")
class SetUserStatusRequest(BaseModel):
status: Literal["active", "disabled"] = Field(
..., description="active=解封 / disabled=封禁(注销 deleted 不走此接口)"
)
+59
View File
@@ -0,0 +1,59 @@
"""admin 钱包(金币/现金流水 + 提现单)schemas。"""
from __future__ import annotations
from datetime import datetime
from pydantic import BaseModel, ConfigDict, Field
class CoinTxnOut(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: int
user_id: int
amount: int
balance_after: int
biz_type: str
ref_id: str | None = None
remark: str | None = None
created_at: datetime
class CashTxnOut(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: int
user_id: int
amount_cents: int
balance_after_cents: int
biz_type: str
ref_id: str | None = None
remark: str | None = None
created_at: datetime
class WithdrawOrderOut(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: int
user_id: int
out_bill_no: str
amount_cents: int
user_name: str | None = None # 提现实名(审核核对 + 打款用)
status: str
wechat_state: str | None = None
transfer_bill_no: str | None = None
fail_reason: str | None = None
created_at: datetime
updated_at: datetime
class ReconcileResult(BaseModel):
checked: int
resolved: int
class WithdrawRejectRequest(BaseModel):
reason: str = Field(
..., min_length=1, max_length=200, description="拒绝理由(写入 fail_reason,用户可见)"
)
+55
View File
@@ -0,0 +1,55 @@
"""Admin JWT:与 App 用户 token 完全隔离。
用独立 secret(settings.ADMIN_JWT_SECRET ≠ JWT_SECRET_KEY)+ payload typ="admin",
双重保证 App 用户的 access_token 无法当 admin token 用(secret 不同直接验签失败)。
admin 无 refresh:过期(默认 12h)重新登录,简单。
"""
from __future__ import annotations
from datetime import datetime, timedelta, timezone
from typing import Any
import jwt
from app.core.config import settings
_ALGO = "HS256"
class AdminTokenError(Exception):
"""admin token 解析/校验失败,api 层捕获后返回 401。"""
def _now() -> datetime:
return datetime.now(timezone.utc)
def create_admin_token(*, admin_id: int, role: str) -> tuple[str, int]:
"""签发 admin access token,返回 (token, expires_in_seconds)。"""
now = _now()
expire = now + timedelta(minutes=settings.ADMIN_JWT_EXPIRE_MINUTES)
payload: dict[str, Any] = {
"sub": str(admin_id),
"role": role,
"typ": "admin",
"iat": int(now.timestamp()),
"exp": int(expire.timestamp()),
}
token = jwt.encode(payload, settings.ADMIN_JWT_SECRET, algorithm=_ALGO)
return token, int((expire - now).total_seconds())
def decode_admin_token(token: str) -> dict[str, Any]:
"""解析校验 admin token。签名错/过期/typ 非 admin → AdminTokenError。"""
try:
payload = jwt.decode(token, settings.ADMIN_JWT_SECRET, algorithms=[_ALGO])
except jwt.ExpiredSignatureError as e:
raise AdminTokenError("token expired") from e
except jwt.InvalidTokenError as e:
raise AdminTokenError(f"invalid token: {e}") from e
if payload.get("typ") != "admin":
raise AdminTokenError("not an admin token")
if "sub" not in payload:
raise AdminTokenError("token missing sub")
return payload
+73 -4
View File
@@ -19,8 +19,18 @@ from app.core import rewards
from app.core.config import settings
from app.integrations import pangle
from app.core.ratelimit import rate_limit
from app.repositories import ad_ecpm as crud_ecpm
from app.repositories import ad_reward as crud_ad
from app.schemas.ad import AdRewardStatusOut, PangleCallbackOut, TestGrantOut
from app.repositories import ad_watch as crud_watch
from app.schemas.ad import (
AdRewardStatusOut,
EcpmReportIn,
EcpmReportOut,
PangleCallbackOut,
TestGrantOut,
WatchReportIn,
WatchReportOut,
)
logger = logging.getLogger("shagua.ad")
@@ -64,7 +74,7 @@ def pangle_callback(request: Request, db: DbSession) -> PangleCallbackOut:
return PangleCallbackOut(is_verify=False, reason=REASON_BAD_PARAMS)
user_id = int(raw_user_id)
coin = rewards.resolve_ad_reward_coin(params.get("reward_amount"))
coin = rewards.resolve_ad_reward_coin(db, params.get("reward_amount"))
raw = "&".join(f"{k}={v}" for k, v in sorted(params.items()) if k != "sign")
try:
rec = crud_ad.grant_ad_reward(
@@ -84,15 +94,71 @@ def pangle_callback(request: Request, db: DbSession) -> PangleCallbackOut:
@router.get("/reward-status", response_model=AdRewardStatusOut, summary="今日看广告发奖进度")
def reward_status(user: CurrentUser, db: DbSession) -> AdRewardStatusOut:
used, limit, coin_per = crud_ad.today_status(db, user.id)
(used, limit, coin_per, round_count, cooldown_until,
watched_sec, watch_limit) = crud_ad.today_status(db, user.id)
return AdRewardStatusOut(
used_today=used,
daily_limit=limit,
remaining=max(0, limit - used),
coin_per_ad=coin_per,
round_count=round_count,
cooldown_until=cooldown_until,
watched_seconds_today=watched_sec,
watch_seconds_limit=watch_limit,
watch_seconds_remaining=max(0, watch_limit - watched_sec),
)
@router.post(
"/watch-report",
response_model=WatchReportOut,
summary="上报激励视频观看时长(每日 50 分钟防刷计时)",
dependencies=[Depends(rate_limit(120, 60, "ad-watch-report"))],
)
def watch_report(payload: WatchReportIn, user: CurrentUser, db: DbSession) -> WatchReportOut:
"""客户端在激励视频关闭(onAdClose)后上报本次实际观看秒数,服务端累计到当日总时长。
Bearer 鉴权,user_id 取自 JWT(不信 body)。seconds 服务端夹 [0, MAX_SINGLE_WATCH_SECONDS]。
当日累计达 DAILY_AD_WATCH_SECONDS_LIMIT(50 分钟)后:① 本接口 / reward-status 返回 remaining=0,
客户端不再展示广告;② 后端发奖(S2S 回调)也会因时长闸记 capped 不发金币(双闸,前端绕不过)。
"""
total = crud_watch.add_watch_seconds(db, user.id, payload.seconds)
limit = rewards.DAILY_AD_WATCH_SECONDS_LIMIT
logger.info(
"ad watch report user_id=%d +%ds total=%d/%d",
user.id, payload.seconds, total, limit,
)
return WatchReportOut(
watched_seconds_today=total,
watch_seconds_limit=limit,
watch_seconds_remaining=max(0, limit - total),
)
@router.post(
"/ecpm-report",
response_model=EcpmReportOut,
summary="上报本次广告展示的 eCPM(内部收益统计)",
dependencies=[Depends(rate_limit(120, 60, "ad-ecpm-report"))],
)
def ecpm_report(payload: EcpmReportIn, user: CurrentUser, db: DbSession) -> EcpmReportOut:
"""客户端在广告展示后(onAdShow 读 getShowEcpm)上报 eCPM,落库做内部收益统计/对账。
Bearer 鉴权,user_id 取自 JWT(不信 body)。best-effort:落库即 ok,客户端 fire-and-forget,
丢一两条不影响业务(穿山甲后台报表是结算权威)。eCPM 与发奖(S2S)是两条独立流,不逐条关联。
"""
crud_ecpm.create_ecpm_record(
db, user.id,
ad_type=payload.ad_type, ecpm_raw=payload.ecpm,
adn=payload.adn, slot_id=payload.slot_id,
)
logger.info(
"ad ecpm report user_id=%d type=%s ecpm=%s adn=%s slot=%s",
user.id, payload.ad_type, payload.ecpm, payload.adn, payload.slot_id,
)
return EcpmReportOut(ok=True)
@router.post(
"/test-grant",
response_model=TestGrantOut,
@@ -118,7 +184,8 @@ def test_grant(user: CurrentUser, db: DbSession) -> TestGrantOut:
except crud_ad.UnknownUserError as e:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="user not found") from e
used, limit, coin_per = crud_ad.today_status(db, user.id)
(used, limit, coin_per, round_count, cooldown_until,
_watched, _watch_limit) = crud_ad.today_status(db, user.id)
logger.info("ad TEST grant user_id=%d status=%s coin=%d", user.id, rec.status, rec.coin)
return TestGrantOut(
granted=(rec.status == "granted"),
@@ -128,4 +195,6 @@ def test_grant(user: CurrentUser, db: DbSession) -> TestGrantOut:
daily_limit=limit,
remaining=max(0, limit - used),
coin_per_ad=coin_per,
round_count=round_count,
cooldown_until=cooldown_until,
)
+15 -4
View File
@@ -12,9 +12,10 @@ from __future__ import annotations
import logging
from fastapi import APIRouter, HTTPException, status
from fastapi import APIRouter, Depends, HTTPException, status
from app.api.deps import CurrentUser, DbSession
from app.core.ratelimit import rate_limit
from app.core.security import TokenError, decode_token, issue_token_pair
from app.integrations.jiguang import JiguangError, mask_phone, verify_and_get_phone
from app.integrations.sms import SmsError, send_code, verify_code
@@ -71,19 +72,29 @@ def jverify_login(req: JverifyLoginRequest, db: DbSession) -> TokenWithUser:
# ===================== 短信登录 =====================
@router.post("/sms/send", response_model=SmsSendResponse, summary="发送短信验证码 (mock)")
@router.post(
"/sms/send",
response_model=SmsSendResponse,
summary="发送短信验证码",
dependencies=[Depends(rate_limit(10, 60, "sms-send"))], # 同 IP 每分钟≤10 次(防一 IP 刷不同号)
)
def sms_send(req: SmsSendRequest) -> SmsSendResponse:
try:
cooldown = send_code(req.phone)
except SmsError as e:
raise HTTPException(status_code=429, detail=str(e)) from e
raise HTTPException(status_code=e.status_code, detail=str(e)) from e
from app.core.config import settings # 局部 import 避免循环
return SmsSendResponse(sent=True, mock=settings.SMS_MOCK, cooldown_sec=cooldown)
@router.post("/sms/login", response_model=TokenWithUser, summary="手机号+验证码登录")
@router.post(
"/sms/login",
response_model=TokenWithUser,
summary="手机号+验证码登录",
dependencies=[Depends(rate_limit(20, 60, "sms-login"))], # 防撞库爆破(另有单码失败次数上限)
)
def sms_login(req: SmsLoginRequest, db: DbSession) -> TokenWithUser:
if not verify_code(req.phone, req.code):
raise HTTPException(status_code=400, detail="invalid sms code")
+7
View File
@@ -83,6 +83,13 @@ async def intent_recognize(request: Request) -> dict[str, Any]:
return await _passthrough(request, "/api/intent/recognize")
@router.post("/intent/step", summary="外卖比价 Phase 1 多帧意图识别 (透传到 pricebot, 仅淘宝源)")
async def intent_step(request: Request) -> dict[str, Any]:
# 多帧版意图识别(展开+滚动采集→提取): 循环调用直到 done(done 帧顶层带
# result+calibration)。目前仅淘宝源走这条, 其它源走上面单次 /intent/recognize。
return await _passthrough(request, "/api/intent/step")
@router.post("/price/step", summary="外卖比价 Phase 2 步进 (透传到 pricebot)")
async def price_step(request: Request) -> dict[str, Any]:
return await _passthrough(request, "/api/price/step")
+76
View File
@@ -0,0 +1,76 @@
"""比价战绩里程碑 endpoint(福利页「记录比价战绩」)。
路由前缀 `/api/v1/compare`:
GET /milestones 进度与各档领取状态
POST /milestones/{milestone}/claim 领取某档奖励
**均需鉴权**。解锁进度 = 该用户 status='success' 的 comparison_record 条数;每档领一次。
"""
from __future__ import annotations
import logging
from fastapi import APIRouter, HTTPException, status
from app.api.deps import CurrentUser, DbSession
from app.repositories import comparison_milestone as crud_milestone
from app.schemas.compare_record import (
MilestoneClaimResultOut,
MilestoneStateOut,
MilestoneStatusOut,
)
logger = logging.getLogger("shagua.compare_milestone")
router = APIRouter(prefix="/api/v1/compare", tags=["compare-milestone"])
@router.get(
"/milestones",
response_model=MilestoneStatusOut,
summary="比价战绩里程碑进度",
)
def get_milestones(user: CurrentUser, db: DbSession) -> MilestoneStatusOut:
st = crud_milestone.get_status(db, user.id)
return MilestoneStatusOut(
success_count=st.success_count,
claimable_count=st.claimable_count,
milestones=[
MilestoneStateOut(milestone=m.milestone, coin=m.coin, state=m.state)
for m in st.milestones
],
)
@router.post(
"/milestones/{milestone}/claim",
response_model=MilestoneClaimResultOut,
summary="领取比价战绩里程碑奖励",
)
def claim_milestone(
milestone: int, user: CurrentUser, db: DbSession
) -> MilestoneClaimResultOut:
try:
coin, balance = crud_milestone.claim(db, user.id, milestone)
except crud_milestone.UnknownMilestoneError as e:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND, detail="unknown milestone"
) from e
except crud_milestone.MilestoneLockedError as e:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT, detail="milestone locked"
) from e
except crud_milestone.AlreadyClaimedError as e:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT, detail="milestone already claimed"
) from e
logger.info(
"compare milestone claimed user_id=%d milestone=%d coin=%d",
user.id,
milestone,
coin,
)
return MilestoneClaimResultOut(
milestone=milestone, coin_awarded=coin, coin_balance=balance
)
+86
View File
@@ -0,0 +1,86 @@
"""比价记录 endpoint(「我的比价记录」数据源)。
路由前缀 `/api/v1/compare`:
POST /record 上报一次比价结果(幂等:同 user+trace_id 覆盖)
GET /records 比价记录列表(游标分页)
GET /records/{id} 单条详情(含 raw_payload 全量)
**均需鉴权**(CurrentUser)——与同文件无关的不鉴权透传 `compare.py` 分开:那个是
转发壳(MVP 不鉴权),这里是按用户维度落库的业务接口,必须有 user_id。
注:本轮只做 server 端,客户端(android 仓)在 done 帧后调 POST /record 上报的改动
另起一轮(见 app-server docs/待办与技术债.md P1)。
"""
from __future__ import annotations
import logging
from fastapi import APIRouter, HTTPException, Query, status
from app.api.deps import CurrentUser, DbSession
from app.repositories import comparison as crud_compare
from app.schemas.compare_record import (
ComparisonRecordCreatedOut,
ComparisonRecordDetailOut,
ComparisonRecordIn,
ComparisonRecordPage,
ComparisonRecordOut,
)
logger = logging.getLogger("shagua.compare_record")
router = APIRouter(prefix="/api/v1/compare", tags=["compare-record"])
@router.post(
"/record",
response_model=ComparisonRecordCreatedOut,
summary="上报一次比价结果(幂等)",
)
def report_record(
payload: ComparisonRecordIn, user: CurrentUser, db: DbSession
) -> ComparisonRecordCreatedOut:
rec = crud_compare.upsert_record(db, user_id=user.id, payload=payload)
logger.info(
"compare record user=%s trace=%s biz=%s status=%s saved=%s",
user.id,
rec.trace_id,
rec.business_type,
rec.status,
rec.saved_amount_cents,
)
return ComparisonRecordCreatedOut(id=rec.id)
@router.get(
"/records",
response_model=ComparisonRecordPage,
summary="比价记录列表(游标分页)",
)
def list_records(
user: CurrentUser,
db: DbSession,
limit: int = Query(20, ge=1, le=100),
cursor: int | None = Query(None, description="上一页末条 id"),
) -> ComparisonRecordPage:
items, next_cursor = crud_compare.list_records(
db, user.id, limit=limit, cursor=cursor
)
return ComparisonRecordPage(
items=[ComparisonRecordOut.model_validate(it) for it in items],
next_cursor=next_cursor,
)
@router.get(
"/records/{record_id}",
response_model=ComparisonRecordDetailOut,
summary="比价记录详情(含 raw_payload)",
)
def get_record(
record_id: int, user: CurrentUser, db: DbSession
) -> ComparisonRecordDetailOut:
rec = crud_compare.get_record(db, user.id, record_id)
if rec is None:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="record not found")
return ComparisonRecordDetailOut.model_validate(rec)
+25
View File
@@ -0,0 +1,25 @@
from fastapi import APIRouter
from app.api.deps import CurrentUser, DbSession
from app.repositories import savings as crud_savings
from app.schemas.order import OrderReportOut, OrderReportRequest
router = APIRouter(prefix="/api/v1/order", tags=["order"])
@router.post(
"/report",
response_model=OrderReportOut,
summary="上报归因订单(比价后5分钟内点链接 + 支付金额与比价价相差≤1元)",
)
def report_order(req: OrderReportRequest, user: CurrentUser, db: DbSession) -> OrderReportOut:
# 记账唯一真相表是 savings_record(source='compare')。
rec, duplicated = crud_savings.create_from_report(db, user.id, req)
return OrderReportOut(
id=rec.id,
platform=rec.platform or req.platform,
pay_channel=rec.pay_channel or req.pay_channel,
compared_price_cents=rec.compared_price_cents or 0,
paid_amount_cents=rec.order_amount_cents,
duplicated=duplicated,
)
+145
View File
@@ -0,0 +1,145 @@
"""上报更低价 endpoint。
路由前缀 `/api/v1/report`,需 Bearer 鉴权(上报绑登录用户)。
POST / 提交上报(multipart:comparison_record_id / reported_platform_id /
reported_price(元) + images 1~4 张)
GET /records 上报记录列表(?status= pending/approved/rejected 可选筛选)
要点:
- 截图复用 [app.core.media] 落盘到 /media/price_report/。
- 原最低价由 comparison_record_id 反查比价记录 best_*(不信任客户端传的快照)。
- 校验(D):上报价必须 < 原最低价,否则 400。
- 提交一律 status=pending;通过发奖励是人工审核后台动作,不在此。
"""
from __future__ import annotations
import logging
from fastapi import APIRouter, File, Form, HTTPException, UploadFile
from app.api.deps import CurrentUser, DbSession
from app.core import media
from app.models.comparison import ComparisonRecord
from app.repositories import report as report_repo
from app.schemas.report import (
ReportRecordCounts,
ReportRecordOut,
ReportRecordsOut,
ReportSubmitOut,
)
logger = logging.getLogger("shagua.report")
router = APIRouter(prefix="/api/v1/report", tags=["report"])
_MAX_IMAGES = 4
_VALID_STATUS = ("pending", "approved", "rejected")
# 上报平台(canonical id → 展示名),与原型 fb-platform-chip data-pf 一致
_PLATFORM_NAMES = {
"meituan-waimai": "美团外卖",
"jd-waimai": "京东外卖",
"taobao-shanguang": "淘宝闪购",
}
def _dish_summary(items: list | None) -> str | None:
"""把比价记录 items[{name,qty,...}] 拼成菜品摘要文案。"""
if not items:
return None
names = [str(it.get("name", "")).strip() for it in items if isinstance(it, dict)]
names = [n for n in names if n]
return "".join(names) if names else None
@router.post("", response_model=ReportSubmitOut, summary="提交上报更低价")
async def submit_report(
user: CurrentUser,
db: DbSession,
comparison_record_id: int = Form(...),
reported_platform_id: str = Form(...),
reported_price: str = Form(..., description="用户填的更低价(元)"),
images: list[UploadFile] = File(default=[]),
) -> ReportSubmitOut:
# 平台
reported_platform_id = reported_platform_id.strip()
platform_name = _PLATFORM_NAMES.get(reported_platform_id)
if platform_name is None:
raise HTTPException(status_code=400, detail="不支持的上报平台")
# 价格(元 → 分)
try:
price_yuan = float(reported_price)
except (TypeError, ValueError):
raise HTTPException(status_code=400, detail="价格格式不正确") from None
if price_yuan <= 0:
raise HTTPException(status_code=400, detail="价格必须大于 0")
reported_price_cents = round(price_yuan * 100)
# 反查比价记录(必须属于本人)→ 取原最低价快照
rec = db.get(ComparisonRecord, comparison_record_id)
if rec is None or rec.user_id != user.id:
raise HTTPException(status_code=404, detail="比价记录不存在")
original_price_cents = rec.best_price_cents
# D:上报价必须低于原最低价
if original_price_cents is not None and reported_price_cents >= original_price_cents:
raise HTTPException(
status_code=400,
detail=f"上报价需低于原最低价 ¥{original_price_cents / 100:.2f}",
)
# 截图(至少 1 张,最多 4 张)
files = [f for f in (images or []) if f is not None and f.filename]
if not files:
raise HTTPException(status_code=400, detail="请至少上传一张截图证明")
if len(files) > _MAX_IMAGES:
raise HTTPException(status_code=400, detail=f"最多上传 {_MAX_IMAGES} 张图片")
urls: list[str] = []
for f in files:
data = await f.read()
try:
urls.append(media.save_report_image(user.id, data))
except media.MediaError as e:
raise HTTPException(status_code=400, detail=str(e)) from e
rep = report_repo.create_report(
db,
user_id=user.id,
comparison_record_id=comparison_record_id,
store_name=rec.store_name,
dish_summary=_dish_summary(rec.items),
original_platform_id=rec.best_platform_id,
original_platform_name=rec.best_platform_name,
original_price_cents=original_price_cents,
reported_platform_id=reported_platform_id,
reported_platform_name=platform_name,
reported_price_cents=reported_price_cents,
images=urls,
)
logger.info("price_report id=%d user_id=%d images=%d", rep.id, user.id, len(urls))
return ReportSubmitOut.model_validate(rep)
@router.get("/records", response_model=ReportRecordsOut, summary="上报记录列表")
def list_report_records(
user: CurrentUser,
db: DbSession,
status: str | None = None,
) -> ReportRecordsOut:
status = (status or "").strip() or None
if status and status not in _VALID_STATUS:
raise HTTPException(status_code=400, detail="无效的状态筛选")
rows = report_repo.list_reports(db, user.id, status)
# counts 始终基于全量(不受 status 筛选影响),供前端 chip 计数
all_rows = rows if status is None else report_repo.list_reports(db, user.id, None)
counts = ReportRecordCounts(
all=len(all_rows),
pending=sum(1 for r in all_rows if r.status == "pending"),
approved=sum(1 for r in all_rows if r.status == "approved"),
rejected=sum(1 for r in all_rows if r.status == "rejected"),
)
return ReportRecordsOut(
records=[ReportRecordOut.model_validate(r) for r in rows],
counts=counts,
)
+77 -6
View File
@@ -33,6 +33,8 @@ from app.schemas.welfare import (
ExchangeInfoOut,
ExchangeRequest,
ExchangeResultOut,
TransferAuthResultOut,
TransferAuthStatusOut,
UnbindWechatResultOut,
WithdrawInfoOut,
WithdrawOrderOut,
@@ -155,25 +157,29 @@ def unbind_wechat(user: CurrentUser, db: DbSession) -> UnbindWechatResultOut:
return UnbindWechatResultOut(bound=False)
@router.get("/withdraw-info", response_model=WithdrawInfoOut, summary="提现额度/绑定状态")
@router.get("/withdraw-info", response_model=WithdrawInfoOut, summary="提现额度/绑定状态/免确认开关")
def withdraw_info(user: CurrentUser, db: DbSession) -> WithdrawInfoOut:
u = db.get(User, user.id)
# 顺带同步免确认授权状态(捕获首单确认后已生效的授权 pending→active),让开关展示实时
auth = crud_wallet.sync_transfer_auth(db, user.id)
return WithdrawInfoOut(
min_cents=WITHDRAW_MIN_CENTS,
max_cents=WITHDRAW_MAX_CENTS,
wechat_bound=bool(u and u.wechat_openid),
wechat_nickname=u.wechat_nickname if u else None,
wechat_avatar_url=u.wechat_avatar_url if u else None,
transfer_auth_enabled=bool(auth and auth.state == "active"),
)
@router.post(
"/withdraw",
response_model=WithdrawResultOut,
summary="发起提现到微信零钱",
summary="发起提现(扣款建单,待人工审核;审核通过后才打款)",
dependencies=[Depends(rate_limit(20, 60, "withdraw"))], # #6 同 IP 每分钟≤20 次
)
def withdraw(req: WithdrawRequest, user: CurrentUser, db: DbSession) -> WithdrawResultOut:
# 提现发起本身不调微信(打款在审核通过后),但仍要求微信支付已配置——否则审核通过也打不了款,提前拦
if not settings.wxpay_configured:
raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="wechat pay not configured")
try:
@@ -189,12 +195,24 @@ def withdraw(req: WithdrawRequest, user: CurrentUser, db: DbSession) -> Withdraw
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="wechat not bound") from e
except crud_wallet.InsufficientCashError as e:
raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="insufficient cash balance") from e
except crud_wallet.WithdrawTransferError as e:
# 转账调用失败,余额已退回
raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail=f"wechat transfer failed: {e}") from e
# 调试直发(非生产):skip_review=true 时跳过人工审核,立即发起微信转账(等价 admin approve)。
# 双闸保护——客户端仅 debug 包下发此 flag,服务端仅非 prod 才认;任一道闸拦住即恢复正常审核,
# 生产绝不会被客户端 flag 绕过审核。用于本地联调"提现→微信转账/免确认到账"全链路。
if req.skip_review and not settings.is_prod and order.status == "reviewing":
logger.warning(
"withdraw skip_review(非prod调试直发,跳过人工审核立即打款) user_id=%d bill=%s",
user.id, order.out_bill_no,
)
order = crud_wallet.execute_withdraw_transfer(db, order)
acc = crud_wallet.get_or_create_account(db, user.id)
logger.info("withdraw user_id=%d cents=%d bill=%s state=%s", user.id, req.amount_cents, order.out_bill_no, order.wechat_state)
logger.info(
"withdraw submitted user_id=%d cents=%d bill=%s status=%s",
user.id, req.amount_cents, order.out_bill_no, order.status,
)
# 此刻 status=reviewing,尚未打款 → 无 package_info;App 据 status 提示"已提交,等待审核"。
# 审核通过后客户端轮询 /withdraw/status 拿到 package_info(若需确认)再拉起微信确认页。
return WithdrawResultOut(
out_bill_no=order.out_bill_no,
status=order.status,
@@ -223,6 +241,11 @@ def withdraw_status(
status=order.status,
wechat_state=order.wechat_state,
amount_cents=order.amount_cents,
fail_reason=order.fail_reason,
# 审核通过后 status=pending 且可能带 package_info(WAIT_USER_CONFIRM):供 App 拉起微信确认页
package_info=order.package_info,
mch_id=settings.WXPAY_MCH_ID if order.package_info else None,
app_id=settings.WECHAT_APP_ID if order.package_info else None,
)
@@ -238,3 +261,51 @@ def withdraw_orders(
items=[WithdrawOrderOut.model_validate(it) for it in items],
next_cursor=next_cursor,
)
# ===== 免确认收款授权(用户授权免确认模式)=====
# 开启一次后,后续提现走免确认转账直接到账,不再跳微信确认。绑定 openid 是前提。
@router.post(
"/transfer-auth",
response_model=TransferAuthResultOut,
summary="开启免确认到账(申请授权,返回拉起微信授权页的 package)",
dependencies=[Depends(rate_limit(10, 60, "transfer-auth"))],
)
def open_transfer_auth(user: CurrentUser, db: DbSession) -> TransferAuthResultOut:
if not settings.wxpay_auth_configured:
raise HTTPException(
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
detail="wechat transfer auth not configured",
)
try:
info = crud_wallet.apply_transfer_auth(db, user.id)
except crud_wallet.WechatNotBoundError as e:
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="wechat not bound") from e
except crud_wallet.WithdrawTransferError as e:
raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail=str(e)) from e
logger.info("open transfer-auth user_id=%d already_active=%s", user.id, info["already_active"])
return TransferAuthResultOut(**info)
@router.get(
"/transfer-auth/status",
response_model=TransferAuthStatusOut,
summary="查免确认授权状态(从微信授权页返回后轮询)",
)
def transfer_auth_status(user: CurrentUser, db: DbSession) -> TransferAuthStatusOut:
auth = crud_wallet.sync_transfer_auth(db, user.id)
state = auth.state if auth else "none"
return TransferAuthStatusOut(state=state, enabled=(state == "active"))
@router.post(
"/transfer-auth/close",
response_model=TransferAuthStatusOut,
summary="关闭免确认到账(解除授权)",
)
def close_transfer_auth_endpoint(user: CurrentUser, db: DbSession) -> TransferAuthStatusOut:
crud_wallet.close_transfer_auth(db, user.id)
logger.info("close transfer-auth user_id=%d", user.id)
return TransferAuthStatusOut(state="closed", enabled=False)
+30
View File
@@ -0,0 +1,30 @@
"""微信支付相关回调(一期:免确认收款授权结果通知 stub)。
⚠️ 一期不处理回调内容、不验签:授权状态以主动查询(query_transfer_authorization)为准。
本端点仅向微信回 200 避免重试风暴,**绝不依据回调内容改账**。二期接入时必须先补
V3 平台证书/公钥验签(Wechatpay-Signature) + APIv3 密钥 AEAD 解密,验签通过后方可信任并处理。
授权回调地址通过 settings.WXPAY_AUTH_NOTIFY_URL 配置,需指向本端点的公网地址。
"""
from __future__ import annotations
import logging
from fastapi import APIRouter, Request
logger = logging.getLogger("shagua.wxpay")
router = APIRouter(prefix="/api/v1/wxpay", tags=["wxpay"])
@router.post("/transfer-auth-notify", summary="免确认收款授权结果通知(一期 stub:仅应答,不处理)")
async def transfer_auth_notify(request: Request) -> dict:
# 一期:不验签、不解密、不改账。仅记录 + 应答成功;真实授权状态靠 /transfer-auth/status 查询兜底。
try:
body = await request.json()
logger.info(
"transfer-auth notify id=%s type=%s", body.get("id"), body.get("event_type")
)
except Exception: # noqa: BLE001 — body 解析失败也照常应答 200,避免微信重试
logger.info("transfer-auth notify (unparseable body)")
return {"code": "SUCCESS", "message": "OK"}
+54
View File
@@ -0,0 +1,54 @@
"""看激励视频冷却策略 —— 与发奖记录查询解耦的纯计算。
当前策略:**每 N 次一轮,看满一轮后强制冷却若干秒**(N / 秒数 取自 [rewards] 常量)。
[repositories.ad_reward.today_status] 只负责取数据(今日 granted 的 created_at 列表),
"本轮已看几次 + 冷却到几点"的策略判断委托到这里。
⚠️ 这是临时策略,后续要调。换策略(间隔式 / 每日配额式 / 指数退避 …)**只改本文件**,
repository 不碰——这就是把它独立出来的目的。保持 [compute_cooldown] 签名稳定即可。
"""
from __future__ import annotations
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from app.core.rewards import VIDEO_ROUND_COOLDOWN_SECONDS, VIDEO_ROUND_REQUIRED_COUNT
@dataclass(frozen=True)
class CooldownState:
"""冷却策略的输出。"""
round_count: int # 本轮已看次数 0..N-1(展示用)
cooldown_until: datetime | None # 本轮冷却结束时间(UTC);None = 不在冷却
def compute_cooldown(
granted_times_desc: list[datetime],
now: datetime,
*,
round_size: int = VIDEO_ROUND_REQUIRED_COUNT,
cooldown_seconds: int = VIDEO_ROUND_COOLDOWN_SECONDS,
) -> CooldownState:
""""每 round_size 次一轮、看满一轮后冷却 cooldown_seconds 秒"算本轮进度 + 冷却结束时间。
:param granted_times_desc: 今日 status=granted 记录的 created_at,**按时间倒序**(最新在前)。
:param now: 当前时间(UTC,带 tzinfo),用于判断冷却是否已过。
:param round_size / cooldown_seconds: 策略参数,默认取 rewards 常量,可注入便于测试/调参。
纯函数,不碰 DB。冷却派生算法:把今日 granted 倒序,跳过当前未完成轮的 round_count 条,
下一条即"上一轮最后一次"的时间,+ cooldown_seconds 仍 > now 则在冷却中。
SQLite 上 created_at 可能是 naive,按 UTC 解读再比较。
"""
used = len(granted_times_desc)
round_count = used % round_size
cooldown_until: datetime | None = None
if used >= round_size:
# round_count 必 < round_size <= used,索引合法
last_round_end = granted_times_desc[round_count]
if last_round_end.tzinfo is None:
last_round_end = last_round_end.replace(tzinfo=timezone.utc)
cd_end = last_round_end + timedelta(seconds=cooldown_seconds)
if cd_end > now:
cooldown_until = cd_end
return CooldownState(round_count=round_count, cooldown_until=cooldown_until)
+31
View File
@@ -40,6 +40,20 @@ class Settings(BaseSettings):
JWT_ACCESS_TOKEN_EXPIRE_MINUTES: int = 120
JWT_REFRESH_TOKEN_EXPIRE_DAYS: int = 30
# ===== Admin 后台 =====
# admin 用独立 JWT secret(≠ JWT_SECRET_KEY),App 用户 token 无法越权访问后台。
# 生产必须改成高熵随机串(同 JWT_SECRET_KEY 的要求)。
ADMIN_JWT_SECRET: str = "change-me-admin"
ADMIN_JWT_EXPIRE_MINUTES: int = 720 # admin 登录态 12 小时(无 refresh,过期重登)
# 可选 IP 白名单(逗号分隔),为空=应用层不限制(靠 nginx allow/deny 兜底)。
ADMIN_IP_ALLOWLIST: str = ""
@property
def admin_ip_allowlist(self) -> list[str]:
if not self.ADMIN_IP_ALLOWLIST.strip():
return []
return [ip.strip() for ip in self.ADMIN_IP_ALLOWLIST.split(",") if ip.strip()]
# ===== 极光 =====
JG_APP_KEY: str = ""
JG_MASTER_SECRET: str = ""
@@ -51,6 +65,14 @@ class Settings(BaseSettings):
SMS_MOCK: bool = True
SMS_CODE_TTL_SEC: int = 300
SMS_SEND_INTERVAL_SEC: int = 60
# 真实发送走极光短信 REST(自定义验证码模式:本服务生成 code,极光只负责发)。
# 复用极光一键登录的 JG_APP_KEY / JG_MASTER_SECRET(同一个极光应用)+ JG_REQUEST_TIMEOUT_SEC。
SMS_SEND_ENDPOINT: str = "https://api.sms.jpush.cn/v1/messages"
SMS_SIGN_ID: int = 31729 # 极光短信签名 ID(非机密,可被 .env 覆盖)
SMS_TEMPLATE_ID: int = 1 # 极光短信模板 ID(变量名 code,有效期 5 分钟)
SMS_CODE_LENGTH: int = 6 # 验证码位数(本服务生成;前端 code 字段 4-8 位兼容)
SMS_DAILY_LIMIT_PER_PHONE: int = 10 # 单手机号每日发送上限(防刷 + 控费)
SMS_MAX_VERIFY_ATTEMPTS: int = 5 # 单个验证码最多校验失败次数,超过即作废(防爆破)
# ===== 美团联盟 CPS =====
# 未配置时所有 /api/v1/meituan/* 接口 200 返空(优雅降级),不影响登录/领券等其他业务。
@@ -77,6 +99,10 @@ class Settings(BaseSettings):
WXPAY_PUBLIC_KEY_PATH: str = "./secrets/pub_key.pem" # 微信支付平台公钥
WXPAY_TRANSFER_SCENE_ID: str = "1000" # 转账场景 ID(1000=现金营销)
WXPAY_REQUEST_TIMEOUT_SEC: int = 10
# 免确认收款授权(用户授权免确认模式)的授权结果回调地址,必须公网可访问 HTTPS、不带参数。
# 发起授权 / 首单顺带授权时作为 authorization_notify_url 传给微信。一期不处理回调内容
# (授权状态靠 query 查询兜底),但微信要求该字段非空,故启用免确认前必须配置;留空时免确认相关接口返回未配置。
WXPAY_AUTH_NOTIFY_URL: str = ""
@property
def wxpay_configured(self) -> bool:
@@ -88,6 +114,11 @@ class Settings(BaseSettings):
and self.WXPAY_PUBLIC_KEY_ID
)
@property
def wxpay_auth_configured(self) -> bool:
"""免确认收款授权可用 = 微信支付凭证齐全 + 授权回调地址已配。"""
return bool(self.wxpay_configured and self.WXPAY_AUTH_NOTIFY_URL)
# ===== 穿山甲激励视频(服务端发奖回调)=====
# 看完激励视频后穿山甲服务器回调本服务发金币(S2S,客户端被破解也刷不到)。
# PANGLE_REWARD_SECRET 是穿山甲后台配置的"奖励校验密钥",验签用,从后台取到后填 .env。
+63
View File
@@ -0,0 +1,63 @@
"""运营可配置项注册表:默认值(= 原 rewards 常量)+ 元信息(label/group/type)。
配置项的 single source of truth:
- 业务读配置时 fallback 到这里的 default(空 DB = 原行为不变)。
- admin 配置页用它展示"有哪些可配项 + 当前值 + 默认值 + 说明"
新增可配项 = 这里加一条 + 业务处改用 app_config.get_value(db, key) 读。
"""
from __future__ import annotations
from typing import Any
from app.core import rewards as r
# type 约定(给前端渲染编辑控件用):int / int_list / dict_str_int
CONFIG_DEFS: dict[str, dict[str, Any]] = {
"signin_rewards": {
"default": list(r.SIGNIN_REWARDS), "label": "签到 7 天金币档位",
"group": "签到", "type": "int_list",
"help": "第 1~7 天每天签到发的金币;断签重置回第 1 天。长度需为 7。",
},
"min_exchange_coin": {
"default": r.MIN_EXCHANGE_COIN, "label": "最低兑换金币",
"group": "钱包", "type": "int", "help": "金币兑现金的单次最低金币(10000 金币=1 元)。",
},
"withdraw_min_cents": {
"default": r.WITHDRAW_MIN_CENTS, "label": "提现最低额(分)",
"group": "钱包", "type": "int", "help": "单次提现最低;微信商家转账最低 10 分(0.1 元)。",
},
"withdraw_max_cents": {
"default": r.WITHDRAW_MAX_CENTS, "label": "提现最高额(分)",
"group": "钱包", "type": "int",
},
"task_rewards": {
"default": dict(r.TASK_REWARDS), "label": "一次性任务奖励",
"group": "任务", "type": "dict_str_int", "help": "task_key → 金币。",
},
"record_milestones": {
"default": list(r.RECORD_MILESTONES), "label": "比价里程碑金币档位",
"group": "里程碑", "type": "int_list",
"help": "累计成功比价第 1~N 档解锁发的金币。",
},
"ad_reward_coin": {
"default": r.AD_REWARD_COIN, "label": "看广告单次金币",
"group": "看广告", "type": "int",
"help": "看完一个激励视频发的金币;须与穿山甲后台代码位'奖励数量'保持一致。",
},
"ad_daily_limit": {
"default": r.DAILY_AD_REWARD_LIMIT, "label": "看广告每日上限(次)",
"group": "看广告", "type": "int",
},
"ad_max_coin": {
"default": r.MAX_AD_REWARD_COIN, "label": "看广告单次金币上限",
"group": "看广告", "type": "int", "help": "夹紧穿山甲回调里异常的 reward_amount,防刷爆。",
},
"ad_round_count": {
"default": r.VIDEO_ROUND_REQUIRED_COUNT, "label": "每轮看广告次数",
"group": "看广告", "type": "int", "help": "看完一轮触发冷却。",
},
"ad_cooldown_sec": {
"default": r.VIDEO_ROUND_COOLDOWN_SECONDS, "label": "每轮冷却(秒)",
"group": "看广告", "type": "int",
},
}
+5
View File
@@ -62,6 +62,11 @@ def save_feedback_image(user_id: int, data: bytes) -> str:
return _save_image("feedback", user_id, data)
def save_report_image(user_id: int, data: bytes) -> str:
"""保存上报更低价截图,返回相对 URL(`/media/price_report/<file>`)。"""
return _save_image("price_report", user_id, data)
def delete_avatar(url: str | None) -> None:
"""删除本服务托管的旧头像文件;外部 URL(如微信头像)或空值不处理。"""
prefix = f"{settings.MEDIA_URL_PREFIX}/avatars/"
+99 -13
View File
@@ -53,35 +53,121 @@ WITHDRAW_MAX_CENTS: int = 5_000_000 # 5 万元
TASK_ENABLE_NOTIFICATION = "enable_notification"
# task_key -> 奖励金币
# 打开消息提醒: 1000 金币(=¥0.1, 客户端原型展示口径; 量级与签到/里程碑相称)。
# 注意: 已不再 = 兑换下限(MIN_EXCHANGE_COIN=10000), test_exchange_flow 改走 grant_coins 直接供款。
TASK_REWARDS: dict[str, int] = {
TASK_ENABLE_NOTIFICATION: 10000,
TASK_ENABLE_NOTIFICATION: 1000,
}
# ===== 比价战绩里程碑(累计成功比价 N 次,逐档解锁领金币)=====
# 第 1→6 次的金币奖励(1-based:第 N 次比价解锁第 N 档)。值沿用客户端原型档位。
# 数据源是 comparison_record 里 status='success' 的条数;每档领一次,
# comparison_milestone_claim 去重(仿一次性任务)。要调档位/金额直接改这里。
RECORD_MILESTONES: tuple[int, ...] = (120, 180, 300, 500, 800, 1200)
RECORD_MILESTONE_COUNT: int = len(RECORD_MILESTONES)
def record_milestone_reward(milestone: int) -> int:
"""第 milestone 档(1..RECORD_MILESTONE_COUNT)的金币。越界抛 IndexError。"""
return RECORD_MILESTONES[milestone - 1]
# ===== 看激励视频发金币(穿山甲 S2S 服务端回调发奖)=====
# 看完一个激励视频发的金币(≈¥0.01,汇率 10000 金币=1 元)。
# 作用:① 回调缺/坏 reward_amount 时的回退值;② 客户端进度接口展示的"单次预告金币"
# 看完一个激励视频发的金币(666 金币 ≈¥0.0666,汇率 10000 金币=1 元)。
# 作用:① 回调缺/坏 reward_amount 时的回退值;② 客户端进度接口展示的"单次预告金币";
# ③ test-grant 本地联调的发奖额。
# 真实发放以穿山甲回调带回的 reward_amount 为准(见 resolve_ad_reward_coin),后台应把
# 代码位"奖励数量"配成与本值一致(如 100),保证"广告内展示 / 进度预告 / 实际到账"三者一致。
AD_REWARD_COIN: int = 100
# 代码位"奖励数量"配成与本值一致(=666),保证"广告内展示 / 进度预告 / 实际到账"三者一致。
AD_REWARD_COIN: int = 666
# 单次发奖金币上限:夹紧穿山甲回调里异常的 reward_amount(如后台多打一个 0),防刷爆余额。
MAX_AD_REWARD_COIN: int = 1000
# 每用户每日发奖次数上限,防刷 + 控成本
# ⚠️ 占位值:正式上线前按产品策略 + 广告 eCPM 实测收益重新核定,改这里即可。
DAILY_AD_REWARD_LIMIT: int = 10
# 每用户每日发奖次数上限——现作为"看广告时长上限"的**次数兜底**(防前端少报观看时长绕过时长闸)
# 主闸是 DAILY_AD_WATCH_SECONDS_LIMIT(50 分钟);次数兜底要 ≥ 时长闸内的"合理最大次数",
# 否则会比主闸先掐、误伤看大量短广告的正常用户:50 分钟里若广告各 ~15s,理论可看 ~200 次,
# 故取 200(≈时长闸的次数等价),正常用户先撞 50 分钟时长闸,仅前端时长全报 0 等异常时本闸才兜住。
# 不要为"更严"而下调到 120 之类(会在 ~30 分钟就掐短广告用户);要更严应在客户端如实上报时长。
DAILY_AD_REWARD_LIMIT: int = 200
# ===== 看激励视频每日总时长上限(防刷主闸)=====
# 用户每天最多看 50 分钟激励视频,超了不发奖 / 不给看。时长由前端 onAdClose 上报真实观看
# 秒数累计(POST /api/v1/ad/watch-report → ad_watch_log 表 SUM)。前端不可信,故配合上面
# DAILY_AD_REWARD_LIMIT 次数兜底一起防刷。要调上限直接改这里。
DAILY_AD_WATCH_SECONDS_LIMIT: int = 50 * 60 # 3000 秒 = 50 分钟
# 单次上报观看时长的合理上限(夹紧异常值):激励视频一般 15~60 秒,超 120 秒按 120 记,防刷大值。
MAX_SINGLE_WATCH_SECONDS: int = 120
# 一轮看 N 次激励视频,看完一轮强制 10 分钟冷却(reward-status 派生 cooldown_until 给客户端,
# 客户端任务行 CTA 在冷却期间显示"MM:SS"且不可点)。冷却是 UX 约束,后端发奖只看 daily_limit,
# 冷却期间穿山甲回调仍照常发奖,不另设拒发分支。
VIDEO_ROUND_REQUIRED_COUNT: int = 3
VIDEO_ROUND_COOLDOWN_SECONDS: int = 600
def resolve_ad_reward_coin(reward_amount: str | int | None) -> int:
def resolve_ad_reward_coin(db, reward_amount: str | int | None) -> int: # noqa: ANN001
"""把穿山甲回调的 reward_amount(后台代码位配的"奖励数量")解析成本次发放金币。
缺失 / 非数字 / ≤0 → 回退 AD_REWARD_COIN;超过 MAX_AD_REWARD_COIN → 夹紧
缺失 / 非数字 / ≤0 → 回退配置单次金币;超过配置单次上限 → 夹紧(均从 app_config 读)
注:reward_amount 不参与穿山甲 sign(只签 trans_id),但伪造回调需先伪造合法 sign
(要 SecurityKey),故能过验签的 reward_amount 即视为可信,这里只防后台配错。
"""
default_coin = get_ad_reward_coin(db)
try:
coin = int(reward_amount) # type: ignore[arg-type]
except (TypeError, ValueError):
return AD_REWARD_COIN
return default_coin
if coin <= 0:
return AD_REWARD_COIN
return min(coin, MAX_AD_REWARD_COIN)
return default_coin
return min(coin, get_ad_max_coin(db))
# ===== 配置读取(运营后台可覆盖;app_config 表空时返回上面的常量默认 = 现状不变)=====
# 业务处用这些 getter(传 db)取值,而非直接用常量,这样 admin 改 app_config 即生效。
# 函数内延迟 import,避免 rewards ← app_config ← config_schema ← rewards 的模块级循环。
def _cfg(db, key: str): # noqa: ANN001
from app.repositories import app_config
return app_config.get_value(db, key)
def get_signin_rewards(db) -> list[int]: # noqa: ANN001
return list(_cfg(db, "signin_rewards"))
def get_min_exchange_coin(db) -> int: # noqa: ANN001
return int(_cfg(db, "min_exchange_coin"))
def get_withdraw_min_cents(db) -> int: # noqa: ANN001
return int(_cfg(db, "withdraw_min_cents"))
def get_withdraw_max_cents(db) -> int: # noqa: ANN001
return int(_cfg(db, "withdraw_max_cents"))
def get_task_rewards(db) -> dict[str, int]: # noqa: ANN001
return dict(_cfg(db, "task_rewards"))
def get_record_milestones(db) -> list[int]: # noqa: ANN001
return list(_cfg(db, "record_milestones"))
def get_ad_reward_coin(db) -> int: # noqa: ANN001
return int(_cfg(db, "ad_reward_coin"))
def get_ad_daily_limit(db) -> int: # noqa: ANN001
return int(_cfg(db, "ad_daily_limit"))
def get_ad_max_coin(db) -> int: # noqa: ANN001
return int(_cfg(db, "ad_max_coin"))
def get_ad_round_count(db) -> int: # noqa: ANN001
return int(_cfg(db, "ad_round_count"))
def get_ad_cooldown_sec(db) -> int: # noqa: ANN001
return int(_cfg(db, "ad_cooldown_sec"))
+22
View File
@@ -14,6 +14,7 @@ from __future__ import annotations
from datetime import datetime, timedelta, timezone
from typing import Any, Literal
import bcrypt
import jwt
from app.core.config import settings
@@ -84,3 +85,24 @@ def issue_token_pair(user_id: int) -> dict[str, Any]:
"expires_in": int((access_exp - _now()).total_seconds()),
"refresh_expires_in": int((refresh_exp - _now()).total_seconds()),
}
# ===================== 密码 hash(admin 后台账号用)=====================
# 用户侧是手机号+验证码登录,不存密码;仅 admin 账号用 username+password 登录。
def hash_password(plain: str) -> str:
"""bcrypt 加盐 hash,返回可入库的字符串。
bcrypt 限制明文 ≤72 字节,超出抛 ValueError(实测 bcrypt 5.0:25 个中文=75 字节即触发)。
统一截断到 72 字节(bcrypt 标准做法)——72 字节仍是强密码,verify 同样截断保证一致。
"""
return bcrypt.hashpw(plain.encode("utf-8")[:72], bcrypt.gensalt()).decode("utf-8")
def verify_password(plain: str, hashed: str) -> bool:
"""校验明文与 bcrypt hash。明文同样截断 72 字节(与 hash_password 一致,否则超长密码永远不匹配);
hash 串损坏/格式非法时返回 False,不抛异常。"""
try:
return bcrypt.checkpw(plain.encode("utf-8")[:72], hashed.encode("utf-8"))
except (ValueError, TypeError):
return False
+1 -1
View File
@@ -4,7 +4,7 @@
"""
from __future__ import annotations
from sqlalchemy.orm import DeclarativeBase, MappedAsDataclass
from sqlalchemy.orm import DeclarativeBase
class Base(DeclarativeBase):
+175 -32
View File
@@ -1,68 +1,211 @@
"""短信验证码服务。
当前实现:**mock 模式**。
- send_code(): 不真发短信,仅 log。记录 phone → 发送时间(防 60s 内重复发)
- verify_code(): 任意 6 位数字均通过(配合前台 demo 行为)
两种运行模式由 `SMS_MOCK` 切换:
- **mock**(开发/测试,默认):不真发短信,验证码打到日志;校验**放行任意 N 位数字**
(测试/开发便利)。真实校验逻辑(比对存码 / 一次性 / 防爆破)由 real 分支 + 单测覆盖。
- **real**(生产 `SMS_MOCK=false`):本服务生成 N 位验证码 → 调极光短信 REST
`/v1/messages` 发送(自定义验证码模式,极光只负责发,code 由本服务生成/保管/
校验)→ 鉴权复用极光一键登录的 `JG_APP_KEY`/`JG_MASTER_SECRET`(同一极光应用)。
后续接真供应商(阿里云 / 腾讯云)时:
- send_code() 改为调供应商 API,把生成的 6 位码存到 cache(redis / sqlite)
- verify_code() 比对 cache 里的码,且验过即作废
验证码存储:**进程内存**(单 worker uvicorn 够用)。重启丢失(用户重发即可)。多
worker / 多机时内存不共享 → 冷却、每日上限、校验都会失效,届时迁移到 DB/Redis。
见 docs/待办与技术债.md。
进程内存方案占坑期够用;切真实供应商时一并切 redis。
防刷三层(短信花钱 + `/sms/send` 在登录前无法 JWT 鉴权):
1. 单号 `SMS_SEND_INTERVAL_SEC` 冷却(本文件)
2. 单号每日 `SMS_DAILY_LIMIT_PER_PHONE` 条上限(本文件)
3. 单 IP 频控(api 层 rate_limit 依赖)+ 极光控制台 IP 白名单/防轰炸(运维侧)
另:单码校验失败 `SMS_MAX_VERIFY_ATTEMPTS` 次即作废(防爆破),验过即作废(一次性)。
"""
from __future__ import annotations
import base64
import logging
import secrets
import time
from dataclasses import dataclass
from datetime import datetime
from threading import Lock
import httpx
from app.core.config import settings
logger = logging.getLogger("shagua.sms")
# 进程内最近发送时间表:{phone: epoch_seconds},用于 60s 内拒发
_last_sent: dict[str, float] = {}
_lock = Lock()
class SmsError(Exception):
"""业务异常:发送过频 / 验证码不对。api 层 catch 后翻成 4xx。"""
"""业务异常。`status_code` 决定 api 层翻成哪个 HTTP 码:
过频/每日超限 = 429(客户端等会再来),供应商不可用 = 503,手机号无效 = 400。
"""
def __init__(self, message: str, status_code: int = 429) -> None:
super().__init__(message)
self.status_code = status_code
@dataclass
class _CodeRecord:
code: str
expires_at: float
attempts: int = 0
# 进程内存(单 worker 有效;多 worker 不共享,见模块 docstring)
_codes: dict[str, _CodeRecord] = {} # phone -> 当前有效验证码
_last_sent: dict[str, float] = {} # phone -> 上次发送 epoch(冷却)
_daily_count: dict[str, tuple[str, int]] = {} # phone -> (date_str, 当日发送数)
_lock = Lock()
_GC_THRESHOLD = 10000 # 任一内存 dict 超此阈值,send 时顺手清过期项(防无限增长,仿 ratelimit)
def _today() -> str:
return datetime.now().strftime("%Y-%m-%d")
def _gen_code() -> str:
"""生成 N 位数字验证码(用 secrets 而非 random;允许前导 0)。"""
return "".join(secrets.choice("0123456789") for _ in range(settings.SMS_CODE_LENGTH))
def _gc(now: float) -> None:
"""顺手清理过期内存项,防三个 dict 无限增长。仅在持锁时调用,且某 dict 超
_GC_THRESHOLD 才扫它(低频,开销可忽略)。"""
if len(_codes) > _GC_THRESHOLD:
for p in [p for p, r in _codes.items() if now > r.expires_at]:
_codes.pop(p, None)
if len(_last_sent) > _GC_THRESHOLD:
cutoff = now - settings.SMS_SEND_INTERVAL_SEC
for p in [p for p, ts in _last_sent.items() if ts < cutoff]:
_last_sent.pop(p, None)
if len(_daily_count) > _GC_THRESHOLD:
today = _today()
for p in [p for p, (d, _c) in _daily_count.items() if d != today]:
_daily_count.pop(p, None)
def send_code(phone: str) -> int:
"""发送(或 mock 发送)验证码。
"""发送验证码。
Returns: 距下次可发的秒数(=0 表示刚刚已发,需等 SMS_SEND_INTERVAL_SEC)
Raises: SmsError 如果上次发送在 SMS_SEND_INTERVAL_SEC 内
Returns: 距下次可发的秒数(= SMS_SEND_INTERVAL_SEC)
Raises: SmsError(过频 429 / 当日超限 429 / 供应商失败 503 / 手机号无效 400)
"""
now = time.time()
# --- lock 内:防刷检查 + 预占(防并发重复发烧钱)---
with _lock:
last = _last_sent.get(phone, 0.0)
elapsed = now - last
_gc(now) # 顺手清过期内存(超阈值才扫)
elapsed = now - _last_sent.get(phone, 0.0)
if elapsed < settings.SMS_SEND_INTERVAL_SEC:
remain = int(settings.SMS_SEND_INTERVAL_SEC - elapsed)
raise SmsError(f"send too frequent, retry in {remain}s")
_last_sent[phone] = now
raise SmsError(f"发送过于频繁,请 {remain}s 后再试")
if settings.SMS_MOCK:
logger.info("[SMS-MOCK] send to %s****, code=any-6-digits", phone[:3])
else:
# TODO: 接真实短信供应商
logger.warning("[SMS] real provider not implemented yet, phone=%s****", phone[:3])
raise SmsError("sms provider not configured")
today = _today()
day, cnt = _daily_count.get(phone, ("", 0))
if day != today:
cnt = 0
if cnt >= settings.SMS_DAILY_LIMIT_PER_PHONE:
raise SmsError("今日验证码发送次数已达上限,请明天再试")
code = _gen_code()
# 预占:先记冷却/计数/存码,释放锁后再发网络(发失败保留冷却+计数,见下)
_last_sent[phone] = now
_daily_count[phone] = (today, cnt + 1)
_codes[phone] = _CodeRecord(code=code, expires_at=now + settings.SMS_CODE_TTL_SEC)
# --- lock 外:真正发送(网络 IO 不持锁)---
try:
if settings.SMS_MOCK:
logger.info("[SMS-MOCK] to %s**** code=%s (不真发)", phone[:3], code)
else:
_send_via_jiguang(phone, code)
logger.info("[SMS] sent to %s****", phone[:3])
except Exception as e:
# 发送失败:**保留冷却 + 每日计数**(失败也限速,挡住余额不足/签名失效时
# 前端重试狂打极光),只清掉没发出去的码(用户收不到,留着无意义且占内存)。
with _lock:
_codes.pop(phone, None)
if isinstance(e, SmsError):
raise
logger.exception("[SMS] send failed phone=%s****", phone[:3])
raise SmsError("验证码发送失败,请稍后重试", status_code=503) from e
return settings.SMS_SEND_INTERVAL_SEC
def verify_code(phone: str, code: str) -> bool:
"""校验验证码。mock 模式下任意 6 位数字均通过。"""
"""校验验证码。
- **mock 模式**:放行任意 N 位数字(测试/开发便利,不真校验)。
- **real 模式**:比对本服务存的码,匹配即作废(一次性);失败累计到上限也作废(防爆破)。
"""
if settings.SMS_MOCK:
if len(code) == 6 and code.isdigit():
logger.info("[SMS-MOCK] verify ok for %s****", phone[:3])
ok = len(code) == settings.SMS_CODE_LENGTH and code.isdigit()
logger.info("[SMS-MOCK] verify %s for %s****", "ok" if ok else "fail", phone[:3])
return ok
with _lock:
rec = _codes.get(phone)
if rec is None:
return False
if time.time() > rec.expires_at:
_codes.pop(phone, None)
return False
if rec.attempts >= settings.SMS_MAX_VERIFY_ATTEMPTS:
_codes.pop(phone, None) # 试错过多,作废
return False
if secrets.compare_digest(code.encode("utf-8"), rec.code.encode("utf-8")):
_codes.pop(phone, None) # 验过即作废
return True
logger.info("[SMS-MOCK] verify fail (need 6 digits) for %s****", phone[:3])
rec.attempts += 1
return False
# TODO: 比对供应商发出的真实验证码
logger.warning("[SMS] real provider not implemented yet, phone=%s****", phone[:3])
return False
def _send_via_jiguang(phone: str, code: str) -> None:
"""调极光短信 REST /v1/messages 发送(自定义验证码模式)。失败抛 SmsError。"""
if not settings.JG_APP_KEY or not settings.JG_MASTER_SECRET:
raise SmsError("短信服务未配置(缺 JG_APP_KEY/JG_MASTER_SECRET)", status_code=503)
if not settings.SMS_SIGN_ID or not settings.SMS_TEMPLATE_ID:
raise SmsError("短信服务未配置(缺 SMS_SIGN_ID/SMS_TEMPLATE_ID)", status_code=503)
auth_b64 = base64.b64encode(
f"{settings.JG_APP_KEY}:{settings.JG_MASTER_SECRET}".encode()
).decode()
body = {
"mobile": phone,
"sign_id": settings.SMS_SIGN_ID,
"temp_id": settings.SMS_TEMPLATE_ID,
"temp_para": {"code": code},
}
try:
resp = httpx.post(
settings.SMS_SEND_ENDPOINT,
json=body,
headers={
"Content-Type": "application/json",
"Authorization": f"Basic {auth_b64}",
},
timeout=settings.JG_REQUEST_TIMEOUT_SEC,
)
except httpx.HTTPError as e:
raise SmsError(f"短信网关网络错误: {e}", status_code=503) from e
if resp.status_code == 200:
return # {"msg_id": ...}
# 极光错误码映射(节选常见;完整码表见 docs/integrations/sms.md)
try:
err = resp.json().get("error", {})
ecode, emsg = err.get("code"), err.get("message", "")
except Exception:
ecode, emsg = None, resp.text[:200]
logger.error("[SMS] jiguang error http=%s code=%s msg=%s", resp.status_code, ecode, emsg)
if ecode == 50014: # 余额不足:运维要立即告警充值
logger.critical("[SMS] 极光短信余额不足(50014),需充值!")
raise SmsError("短信服务暂不可用,请稍后重试", status_code=503)
if ecode == 50009: # 极光侧超频
raise SmsError("发送过于频繁,请稍后再试", status_code=429)
if ecode == 50006: # 手机号无效(schema 已挡格式,这里多是空号/停机)
raise SmsError("手机号无效", status_code=400)
raise SmsError(f"短信发送失败(code={ecode})", status_code=503)
+165 -18
View File
@@ -24,6 +24,10 @@ from app.core.config import settings
_API_HOST = "https://api.mch.weixin.qq.com"
_TRANSFER_PATH = "/v3/fund-app/mch-transfer/transfer-bills"
# 免确认收款授权(用户授权免确认模式)
_AUTH_PATH = "/v3/fund-app/mch-transfer/user-confirm-authorization"
_PRE_TRANSFER_AUTH_PATH = "/v3/fund-app/mch-transfer/transfer-bills/pre-transfer-with-authorization"
_TRANSFER_WITH_AUTH_PATH = "/v3/fund-app/mch-transfer/transfer-bills/transfer"
# 懒加载缓存
_private_key: RSAPrivateKey | None = None
@@ -162,24 +166,6 @@ def cancel_transfer(out_bill_no: str) -> dict:
return {"status_code": resp.status_code, "data": resp.json()}
def code_to_openid(code: str) -> str:
"""用微信授权 code 换 openid(开放平台移动应用 sns/oauth2)。失败抛 ValueError。"""
resp = httpx.get(
"https://api.weixin.qq.com/sns/oauth2/access_token",
params={
"appid": settings.WECHAT_APP_ID,
"secret": settings.WECHAT_APP_SECRET,
"code": code,
"grant_type": "authorization_code",
},
timeout=settings.WXPAY_REQUEST_TIMEOUT_SEC,
)
data = resp.json()
if "openid" not in data:
raise ValueError(f"微信授权失败: {data.get('errmsg', data)}")
return data["openid"]
def code_to_userinfo(code: str) -> dict:
"""code 换 access_token+openid,再拉 sns/userinfo 取昵称头像。
返回 {openid, nickname, avatar_url, raw}。失败抛 ValueError。
@@ -217,3 +203,164 @@ def code_to_userinfo(code: str) -> dict:
pass
return {"openid": openid, "nickname": nickname, "avatar_url": avatar_url, "raw": raw}
# ===== 免确认收款授权(用户授权免确认模式)=====
# 用户授权一次后,后续转账走 transfer_with_authorization 免逐笔确认直接到账。
# 复用上面同一套 V3 签名(_build_authorization)、敏感字段加密(encrypt_sensitive)、懒加载密钥。
def _transfer_report_infos() -> list[dict]:
"""现金营销(scene 1000)转账场景报备信息,与 create_transfer 同口径。"""
return [
{"info_type": "活动名称", "info_content": "比价返现"},
{"info_type": "奖励说明", "info_content": "现金提现到微信零钱"},
]
def apply_transfer_authorization(
out_authorization_no: str,
openid: str,
user_display_name: str,
notify_url: str,
*,
scene_info: dict | None = None,
) -> dict:
"""发起免确认收款授权(方式二:不转账,仅申请授权)。返回 {status_code, data}。
成功(200 + state=WAIT_USER_CONFIRM)时 data 带 package_info,供 App 拉起微信授权页。"""
body: dict = {
"out_authorization_no": out_authorization_no,
"appid": settings.WECHAT_APP_ID,
"openid": openid,
"transfer_scene_id": settings.WXPAY_TRANSFER_SCENE_ID,
"user_display_name": user_display_name,
"authorization_notify_url": notify_url,
}
if scene_info:
body["scene_info"] = scene_info
body_str = json.dumps(body, ensure_ascii=False)
headers = {
"Authorization": _build_authorization("POST", _AUTH_PATH, body_str),
"Accept": "application/json",
"Content-Type": "application/json",
}
with httpx.Client() as client:
resp = client.post(
f"{_API_HOST}{_AUTH_PATH}",
content=body_str,
headers=headers,
timeout=settings.WXPAY_REQUEST_TIMEOUT_SEC,
)
return {"status_code": resp.status_code, "data": resp.json()}
def query_transfer_authorization(out_authorization_no: str) -> dict:
"""按商户授权单号查授权结果。返回 {status_code, data}。
data.state: WAIT_USER_CONFIRM / TAKING_EFFECT(已生效) / CLOSED;TAKING_EFFECT 时带 authorization_id。"""
path = f"{_AUTH_PATH}/out-authorization-no/{out_authorization_no}"
headers = {
"Authorization": _build_authorization("GET", path, ""),
"Accept": "application/json",
}
with httpx.Client() as client:
resp = client.get(
f"{_API_HOST}{path}", headers=headers, timeout=settings.WXPAY_REQUEST_TIMEOUT_SEC
)
return {"status_code": resp.status_code, "data": resp.json()}
def close_transfer_authorization(out_authorization_no: str) -> dict:
"""解除免确认收款授权。返回 {status_code, data}(成功 data.state=CLOSED)。"""
path = f"{_AUTH_PATH}/out-authorization-no/{out_authorization_no}/close"
headers = {
"Authorization": _build_authorization("POST", path, ""),
"Accept": "application/json",
"Content-Type": "application/json",
}
with httpx.Client() as client:
resp = client.post(
f"{_API_HOST}{path}", headers=headers, timeout=settings.WXPAY_REQUEST_TIMEOUT_SEC
)
return {"status_code": resp.status_code, "data": resp.json()}
def pre_transfer_with_authorization(
openid: str,
amount_fen: int,
out_bill_no: str,
out_authorization_no: str,
user_display_name: str,
notify_url: str,
user_name: str | None = None,
) -> dict:
"""方式一:发起转账并同时申请免确认收款授权(用户在确认这笔收款时一并完成授权)。
返回 {status_code, data};成功 state=WAIT_USER_CONFIRM 时带 package_info(拉确认+授权页)。"""
body: dict = {
"appid": settings.WECHAT_APP_ID,
"out_bill_no": out_bill_no,
"transfer_scene_id": settings.WXPAY_TRANSFER_SCENE_ID,
"openid": openid,
"transfer_amount": amount_fen,
"transfer_remark": "金币提现",
"transfer_scene_report_infos": _transfer_report_infos(),
"authorization_info": {
"user_display_name": user_display_name,
"out_authorization_no": out_authorization_no,
"authorization_notify_url": notify_url,
},
}
if user_name and amount_fen >= 30:
body["user_name"] = encrypt_sensitive(user_name)
body_str = json.dumps(body, ensure_ascii=False)
headers = {
"Authorization": _build_authorization("POST", _PRE_TRANSFER_AUTH_PATH, body_str),
"Accept": "application/json",
"Content-Type": "application/json",
"Wechatpay-Serial": settings.WXPAY_PUBLIC_KEY_ID,
}
with httpx.Client() as client:
resp = client.post(
f"{_API_HOST}{_PRE_TRANSFER_AUTH_PATH}",
content=body_str,
headers=headers,
timeout=settings.WXPAY_REQUEST_TIMEOUT_SEC,
)
return {"status_code": resp.status_code, "data": resp.json()}
def transfer_with_authorization(
authorization_id: str,
amount_fen: int,
out_bill_no: str,
user_name: str | None = None,
) -> dict:
"""方式二 / 二次起:用户已授权后免确认转账(无需用户逐笔确认,直接到账)。返回 {status_code, data}。
成功 state ∈ ACCEPTED/PROCESSING/TRANSFERING/SUCCESS(无 WAIT_USER_CONFIRM、无 package_info)。"""
body: dict = {
"appid": settings.WECHAT_APP_ID,
"out_bill_no": out_bill_no,
"transfer_scene_id": settings.WXPAY_TRANSFER_SCENE_ID,
"transfer_amount": amount_fen,
"transfer_remark": "金币提现",
"transfer_scene_report_infos": _transfer_report_infos(),
"authorization_id": authorization_id,
}
if user_name and amount_fen >= 30:
body["user_name"] = encrypt_sensitive(user_name)
body_str = json.dumps(body, ensure_ascii=False)
headers = {
"Authorization": _build_authorization("POST", _TRANSFER_WITH_AUTH_PATH, body_str),
"Accept": "application/json",
"Content-Type": "application/json",
"Wechatpay-Serial": settings.WXPAY_PUBLIC_KEY_ID,
}
with httpx.Client() as client:
resp = client.post(
f"{_API_HOST}{_TRANSFER_WITH_AUTH_PATH}",
content=body_str,
headers=headers,
timeout=settings.WXPAY_REQUEST_TIMEOUT_SEC,
)
return {"status_code": resp.status_code, "data": resp.json()}
+10
View File
@@ -17,14 +17,19 @@ from fastapi.staticfiles import StaticFiles
from app.api.v1.ad import router as ad_router
from app.api.v1.auth import router as auth_router
from app.api.v1.compare import router as compare_router
from app.api.v1.compare_milestone import router as compare_milestone_router
from app.api.v1.compare_record import router as compare_record_router
from app.api.v1.coupon import router as coupon_router
from app.api.v1.feedback import router as feedback_router
from app.api.v1.meituan import router as meituan_router
from app.api.v1.order import router as order_router
from app.api.v1.report import router as report_router
from app.api.v1.savings import router as savings_router
from app.api.v1.signin import router as signin_router
from app.api.v1.tasks import router as tasks_router
from app.api.v1.user import router as user_router
from app.api.v1.wallet import router as wallet_router
from app.api.v1.wxpay import router as wxpay_router
from app.core.config import settings
from app.core.logging import setup_logging
@@ -74,12 +79,17 @@ app.include_router(user_router)
app.include_router(feedback_router)
app.include_router(coupon_router)
app.include_router(compare_router)
app.include_router(compare_record_router)
app.include_router(compare_milestone_router)
app.include_router(meituan_router)
app.include_router(wallet_router)
app.include_router(wxpay_router)
app.include_router(signin_router)
app.include_router(tasks_router)
app.include_router(savings_router)
app.include_router(ad_router)
app.include_router(order_router)
app.include_router(report_router)
# 用户上传文件(头像)静态服务。生产可改由 nginx 直接 serve MEDIA_ROOT。
_media_root = Path(settings.MEDIA_ROOT)
+7
View File
@@ -1,6 +1,13 @@
"""所有 ORM model 必须在这里 import 一次,Alembic / metadata 才能扫到。"""
from app.models.ad_ecpm import AdEcpmRecord # noqa: F401
from app.models.ad_reward import AdRewardRecord # noqa: F401
from app.models.ad_watch_log import AdWatchLog # noqa: F401
from app.models.admin import AdminAuditLog, AdminUser # noqa: F401
from app.models.app_config import AppConfig # noqa: F401
from app.models.comparison import ComparisonRecord # noqa: F401
from app.models.comparison_milestone import ComparisonMilestoneClaim # noqa: F401
from app.models.feedback import Feedback # noqa: F401
from app.models.price_report import PriceReport # noqa: F401
from app.models.savings import SavingsRecord # noqa: F401
from app.models.signin import SigninRecord # noqa: F401
from app.models.task import UserTask # noqa: F401
+49
View File
@@ -0,0 +1,49 @@
"""广告展示 eCPM 上报记录(内部收益统计/对账)。
每条 = 客户端一次广告展示(`onAdShow`)后读到的 eCPM 信息。和发奖记录
[ad_reward.AdRewardRecord] 是**两条独立的数据流**:
- 发奖走穿山甲 S2S 回调(后端 → 有 trans_id、无 ecpm);
- eCPM 走客户端上报(客户端 → 有 ecpm、无 trans_id)。
两者没有公共键,无法逐条一一对应,所以本表用于**按用户/按天聚合收益**口径的对账,
不做"这条发奖 = 这条 ecpm"的精确关联。穿山甲后台报表才是结算权威,本表是细粒度补充。
⚠️ `ecpm_raw` 原样存客户端上报的字符串——eCPM 单位(分 / 元)截至 2026-05-31 尚未最终确认,
确认后再加一列解析好的数值;在此之前对账按"待定单位"处理。
"""
from __future__ import annotations
from datetime import datetime
from sqlalchemy import DateTime, ForeignKey, Integer, String, func
from sqlalchemy.orm import Mapped, mapped_column
from app.db.base import Base
class AdEcpmRecord(Base):
__tablename__ = "ad_ecpm_record"
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
user_id: Mapped[int] = mapped_column(
Integer, ForeignKey("user.id"), index=True, nullable=False
)
# 广告类型:reward_video(激励视频) / draw(Draw 信息流) 等;不强行统一代码位,各类型各自上报
ad_type: Mapped[str] = mapped_column(String(32), nullable=False)
# 实际投放的 ADN(穿山甲 getShowEcpm().getSdkName(),如 pangle / gdt)
adn: Mapped[str | None] = mapped_column(String(32), nullable=True)
# 实际展示用的代码位(底层 mediation rit,非客户端配置位)
slot_id: Mapped[str | None] = mapped_column(String(64), nullable=True)
# 客户端上报的 eCPM 原始字符串(单位待确认,原样存)
ecpm_raw: Mapped[str] = mapped_column(String(32), nullable=False)
# 北京时间日期串 'YYYY-MM-DD',按它等值做"按天聚合"(不在 SQL 里做跨时区 date 比较)
report_date: Mapped[str] = mapped_column(String(10), index=True, nullable=False)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), server_default=func.now(), index=True, nullable=False
)
def __repr__(self) -> str: # pragma: no cover
return (
f"<AdEcpmRecord user_id={self.user_id} {self.ad_type} "
f"ecpm={self.ecpm_raw} adn={self.adn}>"
)
+41
View File
@@ -0,0 +1,41 @@
"""看激励视频观看时长记录(每日总时长防刷)。
每条 = 客户端 onAdClose 上报的一次激励视频实际观看秒数。按 (user_id, watch_date) 聚合
SUM(watch_seconds) 得当日总观看时长,用于"每天最多看 50 分钟激励视频"硬上限(超了不再发奖 /
不给看,见 core.rewards.DAILY_AD_WATCH_SECONDS_LIMIT)。
这是看广告的第三条数据流,与另两条并列、互不关联:
- ad_reward(AdRewardRecord):穿山甲 S2S 回调发金币(后端,有 trans_id);
- ad_ecpm(AdEcpmRecord):客户端上报展示收益(前端,有 ecpm);
- ad_watch_log(本表):客户端上报观看时长(前端,有 watch_seconds)。
前端上报的时长不可信 → 时长是防刷"主闸",配合 ad_reward 的每日次数上限(DAILY_AD_REWARD_LIMIT)
"次数兜底",前端少报时长也刷不过次数闸。
"""
from __future__ import annotations
from datetime import datetime
from sqlalchemy import DateTime, ForeignKey, Integer, String, func
from sqlalchemy.orm import Mapped, mapped_column
from app.db.base import Base
class AdWatchLog(Base):
__tablename__ = "ad_watch_log"
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
user_id: Mapped[int] = mapped_column(
Integer, ForeignKey("user.id"), index=True, nullable=False
)
# 本次激励视频实际观看秒数(前端 onAdShow→onAdClose);服务端已夹 [0, MAX_SINGLE_WATCH_SECONDS]
watch_seconds: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
# 北京时间日期串 'YYYY-MM-DD',按它等值做"当日总时长"聚合(不在 SQL 里跨时区比较)
watch_date: Mapped[str] = mapped_column(String(10), index=True, nullable=False)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), server_default=func.now(), index=True, nullable=False
)
def __repr__(self) -> str: # pragma: no cover
return f"<AdWatchLog user_id={self.user_id} sec={self.watch_seconds} {self.watch_date}>"
+70
View File
@@ -0,0 +1,70 @@
"""Admin 后台:管理员账号 + 操作审计日志。
与 App 用户(user 表)完全隔离:admin 走独立 JWT secret、独立鉴权链(见 app/admin/)。
- admin_user:账号密码(bcrypt)登录,带角色(super_admin / finance / operator)。
- admin_audit_log:每个写操作落一条,记前后值,不可删,用于追溯"谁在何时改了谁的钱/状态"
admin_username / target_id 冗余存字符串,即使关联对象被删/改名也能追溯。
"""
from __future__ import annotations
from datetime import datetime
from sqlalchemy import JSON, DateTime, ForeignKey, Integer, String, func
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.orm import Mapped, mapped_column
from app.db.base import Base
# PG 用 JSONB,SQLite(本地/测试)退化为通用 JSON(同 comparison_record.raw_payload)。
_JSON = JSON().with_variant(JSONB(), "postgresql")
class AdminUser(Base):
__tablename__ = "admin_user"
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
username: Mapped[str] = mapped_column(String(64), unique=True, index=True, nullable=False)
password_hash: Mapped[str] = mapped_column(String(255), nullable=False)
# super_admin(全权+管账号)/ finance(钱:提现+金币)/ operator(用户+反馈+大盘)
role: Mapped[str] = mapped_column(String(20), nullable=False, default="operator")
# active / disabled
status: Mapped[str] = mapped_column(String(20), nullable=False, default="active")
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), server_default=func.now(), nullable=False
)
last_login_at: Mapped[datetime | None] = mapped_column(
DateTime(timezone=True), nullable=True
)
def __repr__(self) -> str: # pragma: no cover
return f"<AdminUser id={self.id} username={self.username} role={self.role}>"
class AdminAuditLog(Base):
__tablename__ = "admin_audit_log"
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
admin_id: Mapped[int] = mapped_column(
Integer, ForeignKey("admin_user.id"), index=True, nullable=False
)
# 冗余存操作者用户名:admin 改名/禁用后仍可追溯是谁干的
admin_username: Mapped[str] = mapped_column(String(64), nullable=False)
# 操作类型,如 user.coins.grant / user.status.set / withdraw.refresh / feedback.handle
action: Mapped[str] = mapped_column(String(64), index=True, nullable=False)
# 被操作对象类型 + id(id 用字符串以兼容 out_bill_no 等非整型主键)
target_type: Mapped[str] = mapped_column(String(32), nullable=False)
target_id: Mapped[str | None] = mapped_column(String(64), nullable=True)
# 上下文 + 前后值,如 {"amount": 1000, "reason": "...", "before": {...}, "after": {...}}
detail: Mapped[dict | None] = mapped_column(_JSON, nullable=True)
ip: Mapped[str | None] = mapped_column(String(64), nullable=True)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), server_default=func.now(), index=True, nullable=False
)
def __repr__(self) -> str: # pragma: no cover
return (
f"<AdminAuditLog id={self.id} admin={self.admin_username} "
f"action={self.action} target={self.target_type}:{self.target_id}>"
)
+32
View File
@@ -0,0 +1,32 @@
"""运营可配置项(把 rewards.py 等硬编码规则挪到 DB,运营后台可改)。
key 是配置标识(见 app.core.config_schema.CONFIG_DEFS),value 用 JSON 存任意值(int/list/dict)。
表里没有的 key,业务读配置时 fallback 到 CONFIG_DEFS 默认值(= 原 rewards 常量),
所以**空表 = 现有行为完全不变**。admin 改某项 → 写一行 → 业务下次读到新值(跨进程一致)。
"""
from __future__ import annotations
from datetime import datetime
from typing import Any
from sqlalchemy import JSON, DateTime, Integer, String, func
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.orm import Mapped, mapped_column
from app.db.base import Base
_JSON = JSON().with_variant(JSONB(), "postgresql")
class AppConfig(Base):
__tablename__ = "app_config"
key: Mapped[str] = mapped_column(String(64), primary_key=True)
value: Mapped[Any] = mapped_column(_JSON, nullable=False)
updated_by_admin_id: Mapped[int | None] = mapped_column(Integer, nullable=True)
updated_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), server_default=func.now(), onupdate=func.now(), nullable=False
)
def __repr__(self) -> str: # pragma: no cover
return f"<AppConfig key={self.key}>"
+104
View File
@@ -0,0 +1,104 @@
"""比价记录表。
每完成一次比价(外卖/电商/领券),客户端在 done 帧后用带 JWT 的通道上报一条,落这里。
是未来「我的比价记录」页的数据源,也沉淀用户级行为画像(哪个用户在哪两家之间比了什么)。
与 savings_record 的区别:savings_record 是「省了多少钱」的视角(只有省到才有意义,当前由
demo seeder 灌),本表是「每一次比价的完整明细」——不省钱、甚至失败的比价也照记一条。
两表独立,互不影响。
「越详细越好」的落地:结构化列给查询/排序/聚合用,raw_payload(JSONB)把客户端上报的
原始 calibration + done.params 原样存一份,未来前端要展示什么都能拿到、不丢信息。
"""
from __future__ import annotations
from datetime import datetime
from sqlalchemy import (
JSON,
Boolean,
DateTime,
ForeignKey,
Integer,
String,
UniqueConstraint,
func,
)
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.orm import Mapped, mapped_column
from app.db.base import Base
# PG 上用 JSONB(可建 GIN 索引),SQLite(本地/测试)退化为通用 JSON——
# 否则 SQLite 无 JSONB,Base.metadata.create_all 编译报错(同 savings_record.dishes)。
_JSON = JSON().with_variant(JSONB(), "postgresql")
class ComparisonRecord(Base):
__tablename__ = "comparison_record"
__table_args__ = (
# 同一用户同一次比价(trace_id)只存一条:客户端重试/误点重复上报时幂等覆盖。
UniqueConstraint("user_id", "trace_id", name="uq_comparison_user_trace"),
)
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
user_id: Mapped[int] = mapped_column(
Integer, ForeignKey("user.id"), index=True, nullable=False
)
# 仍记录设备号(同一用户多设备的行为区分 / 与不鉴权期 device_id 数据对账)
device_id: Mapped[str | None] = mapped_column(String(64), nullable=True)
# 业务类型:food(外卖,当前唯一接通)/ ecom(电商)/ coupon(领券)。预留扩展。
business_type: Mapped[str] = mapped_column(
String(16), nullable=False, default="food", index=True
)
# pricebot 侧 trace_id:关联调试落盘 + 幂等去重键
trace_id: Mapped[str] = mapped_column(String(64), nullable=False)
# ===== 源平台(发起比价的那家)=====
source_platform_id: Mapped[str | None] = mapped_column(String(32), nullable=True)
source_platform_name: Mapped[str | None] = mapped_column(String(32), nullable=True)
source_package: Mapped[str | None] = mapped_column(String(128), nullable=True)
source_price_cents: Mapped[int | None] = mapped_column(Integer, nullable=True)
# ===== 最优结果(全平台最便宜的一家,= comparison_results 里 rank=1)=====
best_platform_id: Mapped[str | None] = mapped_column(String(32), nullable=True)
best_platform_name: Mapped[str | None] = mapped_column(String(32), nullable=True)
best_price_cents: Mapped[int | None] = mapped_column(Integer, nullable=True)
# 最优平台的商家/商品深链(客户端比价时从剪贴板采到 = collectedLinks[best_index]);
# 「再次比价」写剪贴板 + launch 该平台 App 直达。仅 2026-06 起新比价有,旧记录为 None。
best_deeplink: Mapped[str | None] = mapped_column(String(1024), nullable=True)
# 源价 - 最优价(可为 0 / 负:源平台本来就最便宜时没省到)
saved_amount_cents: Mapped[int | None] = mapped_column(Integer, nullable=True)
# 源平台就是最便宜的一家(= 这次没省到钱)
is_source_best: Mapped[bool | None] = mapped_column(Boolean, nullable=True)
# ===== 订单概要 =====
store_name: Mapped[str | None] = mapped_column(String(128), nullable=True)
total_dish_count: Mapped[int | None] = mapped_column(Integer, nullable=True)
skipped_dish_count: Mapped[int | None] = mapped_column(Integer, nullable=True)
# success(拿到有效对比)/ failed(出错或没采到目标价)
status: Mapped[str] = mapped_column(String(16), nullable=False, default="success")
# done 帧 information 文案。成功:"在美团找到同店,到手价 ¥X…";
# 失败:具体原因(如"美团、京东外卖均未找到该商品")。前端在比价失败时当原因展示。
information: Mapped[str | None] = mapped_column(String(256), nullable=True)
# ===== 明细(JSON,越详细越好)=====
# 下单菜品 [{name, qty, specs?}]
items: Mapped[list] = mapped_column(_JSON, nullable=False, default=list)
# 逐平台对比 [{platform_id, platform_name, package, price, is_source, rank, coupon_saved}](price/coupon_saved 单位:元,原样存)
comparison_results: Mapped[list] = mapped_column(_JSON, nullable=False, default=list)
# 目标平台未找到、跳过的菜名
skipped_dish_names: Mapped[list] = mapped_column(_JSON, nullable=False, default=list)
# 客户端上报的原始 payload(calibration + done.params 全量),未来取数兜底
raw_payload: Mapped[dict | None] = mapped_column(_JSON, nullable=True)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), server_default=func.now(), index=True, nullable=False
)
def __repr__(self) -> str: # pragma: no cover
return (
f"<ComparisonRecord id={self.id} user_id={self.user_id} "
f"trace_id={self.trace_id} status={self.status}>"
)
+36
View File
@@ -0,0 +1,36 @@
"""比价战绩里程碑领取记录表。
「记录比价战绩」每档(第 1~6 次)只能领一次,领取后写一行,(user_id, milestone)
唯一,防止重复领奖。解锁进度由 comparison_record 里 status='success' 的条数决定,
不存进度本身——只在这里记"哪几档已领"。仿 user_task 的一次性领取模型。
"""
from __future__ import annotations
from datetime import datetime
from sqlalchemy import DateTime, ForeignKey, Integer, UniqueConstraint, func
from sqlalchemy.orm import Mapped, mapped_column
from app.db.base import Base
class ComparisonMilestoneClaim(Base):
__tablename__ = "comparison_milestone_claim"
__table_args__ = (
UniqueConstraint("user_id", "milestone", name="uq_compare_milestone_user"),
)
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
user_id: Mapped[int] = mapped_column(
Integer, ForeignKey("user.id"), index=True, nullable=False
)
# 档位序号(1-based),见 app.core.rewards.RECORD_MILESTONES
milestone: Mapped[int] = mapped_column(Integer, nullable=False)
coin_awarded: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
claimed_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), server_default=func.now(), nullable=False
)
def __repr__(self) -> str: # pragma: no cover
return f"<ComparisonMilestoneClaim user_id={self.user_id} m={self.milestone}>"
+67
View File
@@ -0,0 +1,67 @@
"""上报更低价记录表(price_report)。
用户在「比价记录」里选一条记录,上报「某平台有比我们算出的最低价更低的价格」,带截图证明,
人工审核通过奖励金币。与 feedback(自由文本反馈)不同:本表结构化——关联具体比价记录、
原最低价 vs 上报价、审核状态、奖励。
- 原最低价快照(original_*)由后端按 comparison_record_id 反查比价记录的 best_* 冗余填入,
避免被关联记录后续删/改后对不上。
- 价格统一存「分」(cents),与 comparison_record / savings_record 一致。
- status: pending(审核中)/ approved(已通过)/ rejected(未通过),对齐前端筛选与状态徽标。
- images: 截图相对 URL 列表(/media/price_report/...,复用 app.core.media)。
- 奖励发放(通过 → 钱包 +reward_coins)是人工审核后台动作,提交时一律 pending,本表只留字段。
"""
from __future__ import annotations
from datetime import datetime
from sqlalchemy import JSON, DateTime, ForeignKey, Integer, String, func
from sqlalchemy.orm import Mapped, mapped_column
from app.db.base import Base
class PriceReport(Base):
__tablename__ = "price_report"
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
user_id: Mapped[int] = mapped_column(
Integer, ForeignKey("user.id"), index=True, nullable=False
)
# 选中的那条比价记录(可空:记录被删后仍保留上报历史)
comparison_record_id: Mapped[int | None] = mapped_column(
Integer, ForeignKey("comparison_record.id"), index=True, nullable=True
)
# ===== 选中比价记录的快照(后端反查 comparison_record 填入,冗余存) =====
store_name: Mapped[str | None] = mapped_column(String(128), nullable=True)
dish_summary: Mapped[str | None] = mapped_column(String(256), nullable=True)
original_platform_id: Mapped[str | None] = mapped_column(String(32), nullable=True)
original_platform_name: Mapped[str | None] = mapped_column(String(32), nullable=True)
original_price_cents: Mapped[int | None] = mapped_column(Integer, nullable=True)
# ===== 用户上报的更低价 =====
reported_platform_id: Mapped[str] = mapped_column(String(32), nullable=False)
reported_platform_name: Mapped[str] = mapped_column(String(32), nullable=False)
reported_price_cents: Mapped[int] = mapped_column(Integer, nullable=False)
# 截图证明 URL 列表(相对路径,如 ["/media/price_report/u1_ab.jpg"])
images: Mapped[list[str]] = mapped_column(JSON, nullable=False, default=list)
# ===== 审核(人工后台) =====
status: Mapped[str] = mapped_column(
String(16), nullable=False, default="pending", index=True
)
reject_reason: Mapped[str | None] = mapped_column(String(256), nullable=True)
reward_coins: Mapped[int | None] = mapped_column(Integer, nullable=True)
reviewed_at: Mapped[datetime | None] = mapped_column(
DateTime(timezone=True), nullable=True
)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), server_default=func.now(), index=True, nullable=False
)
def __repr__(self) -> str: # pragma: no cover
return (
f"<PriceReport id={self.id} user_id={self.user_id} status={self.status}>"
)
+27 -3
View File
@@ -8,7 +8,7 @@ from __future__ import annotations
from datetime import datetime
from sqlalchemy import DateTime, ForeignKey, Integer, String, func
from sqlalchemy import JSON, DateTime, ForeignKey, Integer, String, UniqueConstraint, func
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.orm import Mapped, mapped_column
@@ -17,6 +17,11 @@ from app.db.base import Base
class SavingsRecord(Base):
__tablename__ = "savings_record"
__table_args__ = (
# 真实上报(source='compare')按 (user, client_event_id) 幂等;
# demo 行 client_event_id 为 NULL,不参与唯一性冲突(SQLite/PG 均允许多 NULL)
UniqueConstraint("user_id", "client_event_id", name="uq_savings_user_event"),
)
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
user_id: Mapped[int] = mapped_column(
@@ -29,11 +34,30 @@ class SavingsRecord(Base):
title: Mapped[str | None] = mapped_column(String(128), nullable=True)
# 店铺名(订单明细卡标题,如「窑鸡王(王府井店)」)
shop_name: Mapped[str | None] = mapped_column(String(128), nullable=True)
# 菜品名列表(JSONB),PG 上可建 GIN 索引,前 2 道直接展示,其余收进「还有 N 道菜」展开
dishes: Mapped[list[str]] = mapped_column(JSONB, nullable=False, default=list)
# 菜品名列表:PG 用 JSONB(可建 GIN 索引),SQLite 用 JSON(TEXT)。前 2 道直接展示,其余收进「还有 N 道菜」。
dishes: Mapped[list[str]] = mapped_column(
JSON().with_variant(JSONB(), "postgresql"), nullable=False, default=list
)
# 来源:demo(演示) / compare(真实比价上报)
source: Mapped[str] = mapped_column(String(16), nullable=False, default="compare")
# ===== 真实比价上报(source='compare')新增字段;demo 行这些均为 NULL =====
# 源平台原价(用户原本要付的钱,分);省额 saved_amount_cents = original order_amount(实付)
original_price_cents: Mapped[int | None] = mapped_column(Integer, nullable=True)
# 我们当时给出的比价价(分),审计/备用展示
compared_price_cents: Mapped[int | None] = mapped_column(Integer, nullable=True)
# 支付渠道:wechat / alipay
pay_channel: Mapped[str | None] = mapped_column(String(16), nullable=True)
# 实际下单平台包名
platform_package: Mapped[str | None] = mapped_column(String(128), nullable=True)
# 源平台展示名(如「美团」),用于「原价 ¥36.8(美团)」展示
source_platform_name: Mapped[str | None] = mapped_column(String(32), nullable=True)
# 源平台重进链接(预留:后续「重新进店/重复下单」用,本期只存不展示)
source_deeplink: Mapped[str | None] = mapped_column(String(512), nullable=True)
# 客户端幂等键(UUID);demo 行为 NULL
client_event_id: Mapped[str | None] = mapped_column(String(64), nullable=True)
device_id: Mapped[str | None] = mapped_column(String(128), nullable=True)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), server_default=func.now(), index=True, nullable=False
)
+48 -6
View File
@@ -65,11 +65,14 @@ class CoinTransaction(Base):
class WithdrawOrder(Base):
"""提现单(现金 → 微信零钱)。状态机:pending → success / failed。
"""提现单(现金 → 微信零钱)。状态机:
reviewing →(管理员通过)→ pending → success / failed
reviewing →(管理员拒绝)→ rejected(已退款)
扣现金 + 写 cash_transaction(withdraw) 与建本单在同一事务;失败/取消时退回现金
并写 cash_transaction(withdraw_refund)。wechat_state 存微信侧原始状态(WAIT_USER_CONFIRM /
SUCCESS / FAIL / CANCELLED ...),status 是我们归一化后的三态
扣现金 + 写 cash_transaction(withdraw) 与建本单在同一事务,初始 reviewing(待人工审核,
**不打款**);管理员审核通过才调微信转账(execute_withdraw_transfer)→ pending,拒绝则退回
现金 + 置 rejected。微信侧失败/取消时退回现金并写 cash_transaction(withdraw_refund) → failed
wechat_state 存微信侧原始状态(WAIT_USER_CONFIRM / SUCCESS / FAIL / CANCELLED ...)。
"""
__tablename__ = "withdraw_order"
@@ -81,8 +84,10 @@ class WithdrawOrder(Base):
# 商户单号(我们生成,微信查单的 out_bill_no),唯一
out_bill_no: Mapped[str] = mapped_column(String(64), unique=True, index=True, nullable=False)
amount_cents: Mapped[int] = mapped_column(Integer, nullable=False)
# 归一化状态:pending / success / failed
status: Mapped[str] = mapped_column(String(16), nullable=False, default="pending")
# 提现实名(微信达额转账要求):审核后异步打款时要用,发起提现时存下,可空
user_name: Mapped[str | None] = mapped_column(String(64), nullable=True)
# 归一化状态:reviewing(待审核) / pending(打款在途) / success / failed(打款失败已退) / rejected(审核拒绝已退)
status: Mapped[str] = mapped_column(String(16), nullable=False, default="reviewing")
# 微信侧原始状态
wechat_state: Mapped[str | None] = mapped_column(String(32), nullable=True)
# 微信转账单号
@@ -102,6 +107,43 @@ class WithdrawOrder(Base):
return f"<WithdrawOrder id={self.id} user_id={self.user_id} cents={self.amount_cents} {self.status}>"
class WechatTransferAuthorization(Base):
"""微信商家转账「免确认收款授权」(用户授权免确认模式)。一个用户一条(user_id 主键)。
状态机:
(无) →(申请 / 首单转账顺带申请)→ pending(微信 WAIT_USER_CONFIRM,待用户在微信确认授权)
pending →(用户确认授权)→ active(微信 TAKING_EFFECT,此后转账免用户逐笔确认)
pending / active →(用户在微信关 / 商户解除 / 风控)→ closed(终态,需重新开启)
out_authorization_no 我方生成(查授权 / 发起授权用,一个用户一条稳定值,重开时换新);
authorization_id 微信在 active 后返回,免确认转账(transfer_with_authorization)时必传。
"""
__tablename__ = "wechat_transfer_authorization"
user_id: Mapped[int] = mapped_column(
Integer, ForeignKey("user.id"), primary_key=True
)
openid: Mapped[str] = mapped_column(String(64), nullable=False)
# 商户侧授权单号(我方生成),唯一
out_authorization_no: Mapped[str] = mapped_column(
String(64), unique=True, index=True, nullable=False
)
# 微信侧授权单号(active 后返回,免确认转账要用)
authorization_id: Mapped[str | None] = mapped_column(String(64), nullable=True)
# pending(待用户确认) / active(已生效可免确认) / closed(已关闭需重开)
state: Mapped[str] = mapped_column(String(16), nullable=False, default="pending")
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), server_default=func.now(), nullable=False
)
updated_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), server_default=func.now(), onupdate=func.now(), nullable=False
)
def __repr__(self) -> str: # pragma: no cover
return f"<WechatTransferAuthorization user_id={self.user_id} state={self.state}>"
class CashTransaction(Base):
"""现金流水(单位:分)。金币兑现金、提现都记在这里。"""
+49
View File
@@ -0,0 +1,49 @@
"""广告 eCPM 上报 CRUD(内部收益统计/对账)。
客户端在广告展示后(onAdShow)读到 eCPM,经鉴权接口上报,这里落库。鉴权接口已确保
user 存在(JWT),故不做 UnknownUser 校验。best-effort 上报:丢一两条不影响业务,
穿山甲后台报表是结算权威兜底。
"""
from __future__ import annotations
from sqlalchemy import func, select
from sqlalchemy.orm import Session
from app.core.rewards import cn_today
from app.models.ad_ecpm import AdEcpmRecord
def create_ecpm_record(
db: Session,
user_id: int,
*,
ad_type: str,
ecpm_raw: str,
adn: str | None = None,
slot_id: str | None = None,
) -> AdEcpmRecord:
"""落一条 eCPM 上报记录。report_date 用北京时间当天,供按天聚合。"""
rec = AdEcpmRecord(
user_id=user_id,
ad_type=ad_type,
adn=adn,
slot_id=slot_id,
ecpm_raw=ecpm_raw,
report_date=cn_today().isoformat(),
)
db.add(rec)
db.commit()
db.refresh(rec)
return rec
def count_today(db: Session, user_id: int) -> int:
"""该用户今日(北京时间)上报的 eCPM 条数,排查/对账辅助用。"""
return db.execute(
select(func.count())
.select_from(AdEcpmRecord)
.where(
AdEcpmRecord.user_id == user_id,
AdEcpmRecord.report_date == cn_today().isoformat(),
)
).scalar_one()
+63 -12
View File
@@ -6,18 +6,23 @@
3. 当日发奖次数 ≥ 上限 → 记一条 status='capped' 但不发金币。
发金币复用 `wallet.grant_coins`(biz_type='ad_reward', ref_id=trans_id),与发奖记录同事务,
保证"记一笔 + 加金币"原子化。
保证"记一笔 + 加金币"原子化。单次金币 / 每日上限 / 每轮冷却 都从 app_config 读(运营后台可改)。
"""
from __future__ import annotations
from datetime import datetime, timezone
from sqlalchemy import func, select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session
from app.core.rewards import AD_REWARD_COIN, DAILY_AD_REWARD_LIMIT, cn_today
from app.repositories import wallet as crud_wallet
from app.core import rewards
from app.core.ad_cooldown import compute_cooldown
from app.core.rewards import DAILY_AD_WATCH_SECONDS_LIMIT, cn_today
from app.models.ad_reward import AdRewardRecord
from app.models.user import User
from app.repositories import wallet as crud_wallet
from app.repositories.ad_watch import watched_seconds_today
class UnknownUserError(Exception):
@@ -47,14 +52,14 @@ def grant_ad_reward(
user_id: int,
trans_id: str,
*,
coin: int = AD_REWARD_COIN,
coin: int | None = None,
reward_name: str | None = None,
raw: str | None = None,
) -> AdRewardRecord:
"""看广告发奖(幂等 + 每日限额)。返回发奖记录(status=granted/capped)。
coin 为本次发放金币(由调用方按穿山甲回调 reward_amount 解析,见
rewards.resolve_ad_reward_coin);默认 AD_REWARD_COIN 供 test-grant / 缺省场景
rewards.resolve_ad_reward_coin);None → 读配置 get_ad_reward_coin(test-grant / 缺省场景)
user_id 不存在抛 UnknownUserError(不建账户,防伪造 user_id 刷出脏数据)。
"""
# #2 幂等:同 trans_id 已处理过 → 原样返回,不重复发
@@ -67,18 +72,25 @@ def grant_ad_reward(
today = cn_today().isoformat()
# #3 每日上限:超了记一条 capped(不发金币),让审计能看到"今天到顶了"
if _granted_today(db, user_id, today) >= DAILY_AD_REWARD_LIMIT:
# #3 每日上限:观看时长(主闸,50 分钟) 或 发奖次数(兜底) 任一到顶 → 记 capped 不发金币,
# 让审计能看到"今天到顶了"。时长由前端 watch-report 累计(防刷主闸),次数防前端少报时长
# 绕过;次数上限走 app_config(运营后台可改,默认 rewards.DAILY_AD_REWARD_LIMIT)。
over_time = watched_seconds_today(db, user_id, today=today) >= DAILY_AD_WATCH_SECONDS_LIMIT
over_count = _granted_today(db, user_id, today) >= rewards.get_ad_daily_limit(db)
if over_time or over_count:
rec = AdRewardRecord(
trans_id=trans_id, user_id=user_id, coin=0, status="capped",
reward_date=today, reward_name=reward_name, raw=raw,
)
return _commit_record(db, rec, trans_id)
if coin is None:
coin = rewards.get_ad_reward_coin(db)
# 发金币 + 记一笔,同事务
crud_wallet.grant_coins(
db, user_id, coin,
biz_type="ad_reward", ref_id=trans_id, remark="广告奖励",
biz_type="ad_reward", ref_id=trans_id, remark="视频奖励金币",
)
rec = AdRewardRecord(
trans_id=trans_id, user_id=user_id, coin=coin, status="granted",
@@ -102,7 +114,46 @@ def _commit_record(db: Session, rec: AdRewardRecord, trans_id: str) -> AdRewardR
return rec
def today_status(db: Session, user_id: int) -> tuple[int, int, int]:
"""客户端查"今日看广告发奖"进度:返回 (今日已发次数, 每日上限, 单次金币)"""
used = _granted_today(db, user_id, cn_today().isoformat())
return used, DAILY_AD_REWARD_LIMIT, AD_REWARD_COIN
def _granted_times_today_desc(db: Session, user_id: int, reward_date: str) -> list[datetime]:
"""当日 status=granted 记录的 created_at,按时间倒序(最新在前)——冷却策略的输入数据"""
return list(
db.execute(
select(AdRewardRecord.created_at)
.where(
AdRewardRecord.user_id == user_id,
AdRewardRecord.reward_date == reward_date,
AdRewardRecord.status == "granted",
)
.order_by(AdRewardRecord.created_at.desc())
).scalars()
)
def today_status(
db: Session, user_id: int
) -> tuple[int, int, int, int, datetime | None, int, int]:
"""客户端查"今日看广告发奖"进度。
返回 (今日已发次数, 每日次数上限, 单次金币, 本轮已看次数, 本轮冷却结束时间(UTC),
今日已观看总秒数, 每日观看总时长上限(秒))。
次数上限/单次金币/每轮次数/冷却秒 均从 app_config 读(运营后台可改);次数维度的"本轮已看
几次 + 冷却到几点"委托 [app.core.ad_cooldown.compute_cooldown](纯函数);时长维度(50 分钟
主闸)查 ad_watch_log 当日累计——前端据此展示"今日已看 X/50 分钟"+ 满则不给看。
"""
today = cn_today().isoformat()
granted_desc = _granted_times_today_desc(db, user_id, today)
state = compute_cooldown(
granted_desc,
datetime.now(timezone.utc),
round_size=rewards.get_ad_round_count(db),
cooldown_seconds=rewards.get_ad_cooldown_sec(db),
)
return (
len(granted_desc),
rewards.get_ad_daily_limit(db),
rewards.get_ad_reward_coin(db),
state.round_count,
state.cooldown_until,
watched_seconds_today(db, user_id, today=today),
DAILY_AD_WATCH_SECONDS_LIMIT,
)
+36
View File
@@ -0,0 +1,36 @@
"""看激励视频观看时长 CRUD(每日总时长防刷)。
前端 onAdClose 上报本次观看秒数 → add_watch_seconds 落库 + 返回当日累计;
watched_seconds_today 给"时长闸"(发奖 / 展示前)判断当天是否已达 50 分钟上限。
鉴权接口已确保 user 存在(JWT),故不做 UnknownUser 校验。
"""
from __future__ import annotations
from sqlalchemy import func, select
from sqlalchemy.orm import Session
from app.core.rewards import MAX_SINGLE_WATCH_SECONDS, cn_today
from app.models.ad_watch_log import AdWatchLog
def watched_seconds_today(db: Session, user_id: int, *, today: str | None = None) -> int:
"""该用户今日(北京时间)累计观看秒数。coalesce 防无记录时 SUM 返 None。"""
d = today or cn_today().isoformat()
return db.execute(
select(func.coalesce(func.sum(AdWatchLog.watch_seconds), 0)).where(
AdWatchLog.user_id == user_id,
AdWatchLog.watch_date == d,
)
).scalar_one()
def add_watch_seconds(db: Session, user_id: int, seconds: int) -> int:
"""记一次观看时长,返回当日累计秒数。
seconds 夹 [0, MAX_SINGLE_WATCH_SECONDS] 防前端报异常大值(刷时长无意义,但夹紧更稳)。
"""
clamped = max(0, min(int(seconds), MAX_SINGLE_WATCH_SECONDS))
today = cn_today().isoformat()
db.add(AdWatchLog(user_id=user_id, watch_seconds=clamped, watch_date=today))
db.commit()
return watched_seconds_today(db, user_id, today=today)
+65
View File
@@ -0,0 +1,65 @@
"""运营配置读写。
get_value:业务读配置,DB 没有则 fallback 到 CONFIG_DEFS 默认(= 原 rewards 常量,空表行为不变)。
不缓存:配置读频率低(每次福利操作读一次,key 主键查极快),admin 改了立即生效、跨进程一致。
"""
from __future__ import annotations
from typing import Any
from sqlalchemy import select
from sqlalchemy.orm import Session
from app.core.config_schema import CONFIG_DEFS
from app.models.app_config import AppConfig
def get_value(db: Session, key: str) -> Any:
"""读配置值。DB 有用 DB,没有 fallback 到 CONFIG_DEFS 默认。未知 key 抛 KeyError。"""
row = db.get(AppConfig, key)
if row is not None:
return row.value
return CONFIG_DEFS[key]["default"]
def set_value(
db: Session, key: str, value: Any, *, admin_id: int, commit: bool = True
) -> AppConfig:
"""写配置(admin 用)。未知 key 抛 KeyError。commit=False 让调用方与审计同事务。"""
if key not in CONFIG_DEFS:
raise KeyError(f"unknown config key: {key}")
row = db.get(AppConfig, key)
if row is None:
row = AppConfig(key=key, value=value, updated_by_admin_id=admin_id)
db.add(row)
else:
row.value = value
row.updated_by_admin_id = admin_id
if commit:
db.commit()
db.refresh(row)
else:
db.flush()
return row
def list_all(db: Session) -> list[dict]:
"""所有可配项:default + 当前值(DB 有则用 DB) + 元信息。给 admin 配置页渲染。"""
overrides = {r.key: r for r in db.execute(select(AppConfig)).scalars().all()}
out: list[dict] = []
for key, defn in CONFIG_DEFS.items():
row = overrides.get(key)
out.append(
{
"key": key,
"label": defn["label"],
"group": defn["group"],
"type": defn["type"],
"help": defn.get("help"),
"default": defn["default"],
"value": row.value if row is not None else defn["default"],
"overridden": row is not None,
"updated_at": row.updated_at.isoformat() if row is not None else None,
}
)
return out
+173
View File
@@ -0,0 +1,173 @@
"""比价记录 CRUD:上报 upsert(按 user_id+trace_id 幂等) + 派生字段 + 明细分页。
派生逻辑:best_* / saved_amount_cents / is_source_best 全部从 comparison_results 算出
(协议保证已按 price 升序、rank=1 最便宜),客户端不用自己算、也不可信它算。
"""
from __future__ import annotations
from sqlalchemy import func, select
from sqlalchemy.orm import Session
from app.models.comparison import ComparisonRecord
from app.models.savings import SavingsRecord
from app.schemas.compare_record import ComparisonRecordIn
def _yuan_to_cents(yuan: float | None) -> int | None:
"""元(float)→ 分(int)。None 透传。"""
if yuan is None:
return None
return round(yuan * 100)
def _derive(payload: ComparisonRecordIn) -> dict:
"""从上报 payload 派生结构化列(best/saved/is_source_best/status)。"""
results = payload.comparison_results
# 最优 = rank 最小的一条;协议已升序,但不信顺序,显式按 rank/price 兜底取最小价。
best = None
priced = [r for r in results if r.price is not None]
if priced:
best = min(
priced,
key=lambda r: (r.rank if r.rank is not None else 10**9, r.price),
)
source_price_cents = _yuan_to_cents(payload.source_price)
if source_price_cents is None:
# 源价没单独给,从 comparison_results 里的 is_source 行兜底
src_row = next((r for r in results if r.is_source and r.price is not None), None)
if src_row is not None:
source_price_cents = _yuan_to_cents(src_row.price)
best_price_cents = _yuan_to_cents(best.price) if best else None
saved_amount_cents = None
if source_price_cents is not None and best_price_cents is not None:
saved_amount_cents = source_price_cents - best_price_cents
is_source_best = best.is_source if best is not None else None
# status:客户端显式给了就用;否则有"非源且有价"的结果=success,否则 failed
status = payload.status
if status is None:
has_valid_target = any(
(not r.is_source) and r.price is not None for r in results
)
status = "success" if has_valid_target else "failed"
return {
"source_price_cents": source_price_cents,
"best_platform_id": best.platform_id if best else None,
"best_platform_name": best.platform_name if best else None,
"best_price_cents": best_price_cents,
"saved_amount_cents": saved_amount_cents,
"is_source_best": is_source_best,
"status": status,
}
def upsert_record(
db: Session, *, user_id: int, payload: ComparisonRecordIn
) -> ComparisonRecord:
"""按 (user_id, trace_id) 幂等写入:已存在则覆盖(更完整的重试上报胜出),否则新建。"""
derived = _derive(payload)
fields = dict(
device_id=payload.device_id,
business_type=payload.business_type,
store_name=payload.store_name,
source_platform_id=payload.source_platform_id,
source_platform_name=payload.source_platform_name,
source_package=payload.source_package,
information=payload.information,
best_deeplink=payload.best_deeplink,
total_dish_count=payload.total_dish_count,
skipped_dish_count=payload.skipped_dish_count,
items=[it.model_dump(exclude_none=True) for it in payload.items],
comparison_results=[r.model_dump() for r in payload.comparison_results],
skipped_dish_names=list(payload.skipped_dish_names),
raw_payload=payload.model_dump(),
**derived,
)
existing = db.execute(
select(ComparisonRecord).where(
ComparisonRecord.user_id == user_id,
ComparisonRecord.trace_id == payload.trace_id,
)
).scalar_one_or_none()
if existing is not None:
for k, v in fields.items():
setattr(existing, k, v)
db.commit()
db.refresh(existing)
return existing
rec = ComparisonRecord(user_id=user_id, trace_id=payload.trace_id, **fields)
db.add(rec)
db.commit()
db.refresh(rec)
return rec
def _ordered_shop_names(db: Session, user_id: int) -> set[str]:
"""该用户「真实下单」(source='compare')覆盖到的店名集合,用来给比价记录打「已下单」。
只认 compare(归因命中后真实上报),demo 演示数据不算。下单上报不带 trace_id,
只能按店名对齐——两边店名同源(都来自比价意图识别阶段的门店名 query),精确相等即视为同店。
语义=店级:同一家店比价过多次,这些记录会一并标「已下单」。
"""
rows = db.execute(
select(SavingsRecord.shop_name).where(
SavingsRecord.user_id == user_id,
SavingsRecord.source == "compare",
SavingsRecord.shop_name.is_not(None),
)
).scalars().all()
return {s for s in rows if s}
def list_records(
db: Session,
user_id: int,
*,
limit: int = 20,
cursor: int | None = None,
) -> tuple[list[ComparisonRecord], int | None]:
"""比价记录分页(按创建时间倒序、id 兜底,游标式)。附「已下单」店级标记(瞬态,不写库)。"""
stmt = select(ComparisonRecord).where(ComparisonRecord.user_id == user_id)
if cursor is not None:
stmt = stmt.where(ComparisonRecord.id < cursor)
stmt = stmt.order_by(ComparisonRecord.created_at.desc(), ComparisonRecord.id.desc()).limit(limit)
items = list(db.execute(stmt).scalars().all())
next_cursor = items[-1].id if len(items) == limit else None
# 「已下单」标记:本页记录的 store_name 若落在该用户真实下单的店名集合里即 True。
# ordered 非 ORM 列,仅挂实例上供 ComparisonRecordOut(from_attributes) 读出,不持久化。
ordered_shops = _ordered_shop_names(db, user_id)
for it in items:
it.ordered = bool(it.store_name and it.store_name in ordered_shops)
return items, next_cursor
def count_success(db: Session, user_id: int) -> int:
"""该用户成功比价(status='success')的条数。比价战绩里程碑的解锁进度源。"""
return db.execute(
select(func.count(ComparisonRecord.id)).where(
ComparisonRecord.user_id == user_id,
ComparisonRecord.status == "success",
)
).scalar_one()
def get_record(db: Session, user_id: int, record_id: int) -> ComparisonRecord | None:
"""取单条(限本人,避免越权读他人记录)。"""
return db.execute(
select(ComparisonRecord).where(
ComparisonRecord.id == record_id,
ComparisonRecord.user_id == user_id,
)
).scalar_one_or_none()
+114
View File
@@ -0,0 +1,114 @@
"""比价战绩里程碑 CRUD:进度查询 + 逐档领奖。
进度 = comparison_record 里 status='success' 的条数(crud_compare.count_success)。
第 N 档在"成功次数 >= N"时解锁;每档领一次,写 comparison_milestone_claim 去重(标记已领)。
档位金额来自 rewards.get_record_milestones(运营后台可改)。
⚠️ 当前产品定暂不真发金币(后续整体删除该功能):claim 只写领取记录,不调 grant_coins、
不写 coin_transaction,coin_awarded 恒为 0,余额不变。仿一次性任务 (task.py)。
"""
from __future__ import annotations
from dataclasses import dataclass
from sqlalchemy import select
from sqlalchemy.orm import Session
from app.core import rewards
from app.models.comparison_milestone import ComparisonMilestoneClaim
from app.repositories import comparison as crud_compare
from app.repositories import wallet as crud_wallet
class UnknownMilestoneError(Exception):
"""档位序号越界(不在 1..档位数)。"""
class MilestoneLockedError(Exception):
"""该档还没解锁(成功比价次数不够)。"""
class AlreadyClaimedError(Exception):
"""该档已经领过了。"""
@dataclass
class MilestoneState:
milestone: int # 1-based 档位序号
coin: int
state: str # claimed / active / locked
@dataclass
class MilestoneStatus:
success_count: int # 累计成功比价次数(解锁进度)
claimable_count: int # 当前可领(active)档数
milestones: list[MilestoneState]
def _claimed_set(db: Session, user_id: int) -> set[int]:
rows = db.execute(
select(ComparisonMilestoneClaim.milestone).where(
ComparisonMilestoneClaim.user_id == user_id
)
).scalars().all()
return set(rows)
def get_status(db: Session, user_id: int) -> MilestoneStatus:
"""各档领取状态:已领=claimed,已解锁未领=active(可领),未解锁=locked。"""
success_count = crud_compare.count_success(db, user_id)
claimed = _claimed_set(db, user_id)
milestones_def = rewards.get_record_milestones(db)
milestones: list[MilestoneState] = []
claimable = 0
for m in range(1, len(milestones_def) + 1):
if m in claimed:
state = "claimed"
elif success_count >= m:
state = "active"
claimable += 1
else:
state = "locked"
milestones.append(
MilestoneState(milestone=m, coin=milestones_def[m - 1], state=state)
)
return MilestoneStatus(
success_count=success_count,
claimable_count=claimable,
milestones=milestones,
)
def claim(db: Session, user_id: int, milestone: int) -> tuple[int, int]:
"""领取第 milestone 档奖励。返回 (发放金币, 领奖后余额)。
越界抛 UnknownMilestoneError;未解锁抛 MilestoneLockedError;重复领抛 AlreadyClaimedError。
"""
if milestone < 1 or milestone > len(rewards.get_record_milestones(db)):
raise UnknownMilestoneError
existing = db.execute(
select(ComparisonMilestoneClaim).where(
ComparisonMilestoneClaim.user_id == user_id,
ComparisonMilestoneClaim.milestone == milestone,
)
).scalar_one_or_none()
if existing is not None:
raise AlreadyClaimedError
# 解锁校验:成功比价次数必须 >= 档位序号
if crud_compare.count_success(db, user_id) < milestone:
raise MilestoneLockedError
# ⚠️ 比价战绩里程碑暂不真发金币(产品定,后续整体删除该功能):仍记一条领取(去重/标记已领),
# 但不调 grant_coins、不写 coin_transaction,余额不变,返回发放 0 金币 + 当前余额。
db.add(
ComparisonMilestoneClaim(
user_id=user_id, milestone=milestone, coin_awarded=0
)
)
acc = crud_wallet.get_or_create_account(db, user_id, commit=False)
db.commit()
return 0, acc.coin_balance
+53
View File
@@ -0,0 +1,53 @@
"""price_report 表读写。"""
from __future__ import annotations
from sqlalchemy import select
from sqlalchemy.orm import Session
from app.models.price_report import PriceReport
def create_report(
db: Session,
*,
user_id: int,
comparison_record_id: int | None,
store_name: str | None,
dish_summary: str | None,
original_platform_id: str | None,
original_platform_name: str | None,
original_price_cents: int | None,
reported_platform_id: str,
reported_platform_name: str,
reported_price_cents: int,
images: list[str],
) -> PriceReport:
rep = PriceReport(
user_id=user_id,
comparison_record_id=comparison_record_id,
store_name=store_name,
dish_summary=dish_summary,
original_platform_id=original_platform_id,
original_platform_name=original_platform_name,
original_price_cents=original_price_cents,
reported_platform_id=reported_platform_id,
reported_platform_name=reported_platform_name,
reported_price_cents=reported_price_cents,
images=images,
status="pending",
)
db.add(rep)
db.commit()
db.refresh(rep)
return rep
def list_reports(
db: Session, user_id: int, status: str | None = None
) -> list[PriceReport]:
"""该用户的上报记录,按时间倒序;status 非空时按状态筛选。"""
stmt = select(PriceReport).where(PriceReport.user_id == user_id)
if status:
stmt = stmt.where(PriceReport.status == status)
stmt = stmt.order_by(PriceReport.created_at.desc(), PriceReport.id.desc())
return list(db.execute(stmt).scalars().all())
+85 -9
View File
@@ -17,6 +17,7 @@ from sqlalchemy.orm import Session
from app.core.rewards import CN_TZ, cn_today
from app.models.savings import SavingsRecord
from app.schemas.order import OrderReportRequest
# 演示数据规模:最近 N 天每天 1 单(撑起"连续省钱"和"本周"),再补若干历史单
_DEMO_STREAK_DAYS = 9
@@ -65,6 +66,26 @@ def _all_records(db: Session, user_id: int) -> list[SavingsRecord]:
return list(db.execute(stmt).scalars().all())
def _has_real(db: Session, user_id: int) -> bool:
"""该用户是否已有真实比价上报(source='compare')记录。"""
return db.execute(
select(SavingsRecord.id)
.where(SavingsRecord.user_id == user_id, SavingsRecord.source == "compare")
.limit(1)
).first() is not None
def _effective_records(db: Session, user_id: int) -> list[SavingsRecord]:
"""统计口径:有真实(compare)记录就只用真实的;否则 seed 一批 demo 并用 demo 兜底。"""
if _has_real(db, user_id):
stmt = select(SavingsRecord).where(
SavingsRecord.user_id == user_id, SavingsRecord.source == "compare"
)
return list(db.execute(stmt).scalars().all())
ensure_seeded(db, user_id)
return _all_records(db, user_id)
def ensure_seeded(db: Session, user_id: int) -> None:
"""该用户没有任何省钱记录时,幂等灌一批 demo 数据。"""
exists = db.execute(
@@ -108,8 +129,7 @@ def ensure_seeded(db: Session, user_id: int) -> None:
def get_summary(db: Session, user_id: int) -> SavingsSummary:
ensure_seeded(db, user_id)
records = _all_records(db, user_id)
records = _effective_records(db, user_id)
total = sum(r.saved_amount_cents for r in records)
count = len(records)
avg = total // count if count else 0
@@ -129,8 +149,7 @@ def _streak_days(dates: set) -> int:
def get_battle(db: Session, user_id: int) -> SavingsBattle:
ensure_seeded(db, user_id)
records = _all_records(db, user_id)
records = _effective_records(db, user_id)
today = cn_today()
week_start = today - timedelta(days=today.weekday()) # 本周一
@@ -155,10 +174,20 @@ def _compute_beat_percent(db: Session, user_id: int) -> int:
全部省得比我少 → 100;只有自己一个用户 → 0(无可比)。
"""
rows = db.execute(
select(SavingsRecord.user_id, func.sum(SavingsRecord.saved_amount_cents))
.group_by(SavingsRecord.user_id)
select(
SavingsRecord.user_id,
SavingsRecord.source,
func.sum(SavingsRecord.saved_amount_cents),
).group_by(SavingsRecord.user_id, SavingsRecord.source)
).all()
totals = {uid: (total or 0) for uid, total in rows}
# 每用户口径与展示一致:有 compare 用 compare 之和,否则退回 demo 之和
by_user: dict[int, dict[str, int]] = {}
for uid, source, total in rows:
by_user.setdefault(uid, {})[source] = total or 0
totals = {
uid: (d["compare"] if "compare" in d else d.get("demo", 0))
for uid, d in by_user.items()
}
others = {uid: t for uid, t in totals.items() if uid != user_id}
if not others:
return 0
@@ -174,9 +203,12 @@ def list_records(
limit: int = 20,
cursor: int | None = None,
) -> tuple[list[SavingsRecord], int | None]:
"""省钱明细分页(按 id 倒序,游标式)。"""
ensure_seeded(db, user_id)
"""省钱明细分页(按 id 倒序,游标式)。有真实(compare)记录只列真实,否则 demo 兜底。"""
stmt = select(SavingsRecord).where(SavingsRecord.user_id == user_id)
if _has_real(db, user_id):
stmt = stmt.where(SavingsRecord.source == "compare")
else:
ensure_seeded(db, user_id)
if cursor is not None:
stmt = stmt.where(SavingsRecord.id < cursor)
stmt = stmt.order_by(SavingsRecord.id.desc()).limit(limit)
@@ -184,3 +216,47 @@ def list_records(
items = list(db.execute(stmt).scalars().all())
next_cursor = items[-1].id if len(items) == limit else None
return items, next_cursor
def create_from_report(
db: Session, user_id: int, req: OrderReportRequest
) -> tuple[SavingsRecord, bool]:
"""真实比价上报 → 写入一条 savings_record(source='compare')。返回 (记录, 是否重复上报)。
省额 = 源平台原价 − 实付(下限 0);原价缺失则记 0。
幂等:同 (user_id, client_event_id) 已存在则直接返回旧记录,duplicated=True,不新增。
"""
existing = db.execute(
select(SavingsRecord).where(
SavingsRecord.user_id == user_id,
SavingsRecord.client_event_id == req.client_event_id,
)
).scalar_one_or_none()
if existing is not None:
return existing, True
paid = req.paid_amount_cents
original = req.original_price_cents
saved = max(0, original - paid) if original is not None else 0
rec = SavingsRecord(
user_id=user_id,
order_amount_cents=paid,
saved_amount_cents=saved,
original_price_cents=original,
compared_price_cents=req.compared_price_cents,
platform=req.platform,
title=req.shop_name, # 明细卡标题用门店名
shop_name=req.shop_name,
dishes=req.dishes or [],
pay_channel=req.pay_channel,
platform_package=req.platform_package,
source_platform_name=req.source_platform_name,
source_deeplink=req.source_deeplink,
client_event_id=req.client_event_id,
device_id=req.device_id,
source="compare",
)
db.add(rec)
db.commit()
db.refresh(rec)
return rec, False
+10 -7
View File
@@ -3,7 +3,7 @@
7 天循环规则:
- 昨天签过 → 今天 cycle_day = 昨天 % 7 + 1,streak += 1(连续)
- 否则(首签 / 断签) → cycle_day = 1,streak = 1(重置)
今天的 cycle_day 决定发多少金币(见 app.core.rewards.SIGNIN_REWARDS)。
今天的 cycle_day 决定发多少金币(档位来自 rewards.get_signin_rewards,运营后台可改)。
"""
from __future__ import annotations
@@ -13,9 +13,10 @@ from datetime import timedelta
from sqlalchemy import select
from sqlalchemy.orm import Session
from app.core.rewards import SIGNIN_CYCLE_LEN, SIGNIN_REWARDS, cn_today, signin_reward
from app.repositories import wallet as crud_wallet
from app.core import rewards
from app.core.rewards import SIGNIN_CYCLE_LEN, cn_today
from app.models.signin import SigninRecord
from app.repositories import wallet as crud_wallet
class AlreadySignedError(Exception):
@@ -58,6 +59,7 @@ def _next_cycle_day(last: SigninRecord | None, today) -> int:
def get_status(db: Session, user_id: int) -> SigninStatus:
today = cn_today()
rewards_list = rewards.get_signin_rewards(db) # 配置 or 默认(长度=SIGNIN_CYCLE_LEN)
last = _latest_record(db, user_id)
today_signed = last is not None and last.signin_date == today
@@ -83,13 +85,13 @@ def get_status(db: Session, user_id: int) -> SigninStatus:
status = "today"
else:
status = "locked"
steps.append(SigninStep(day=day, coin=SIGNIN_REWARDS[day - 1], status=status))
steps.append(SigninStep(day=day, coin=rewards_list[day - 1], status=status))
return SigninStatus(
today_signed=today_signed,
consecutive_days=consecutive_days,
today_cycle_day=today_cycle_day,
today_coin=signin_reward(today_cycle_day),
today_coin=rewards_list[today_cycle_day - 1],
can_claim=not today_signed,
steps=steps,
)
@@ -109,7 +111,7 @@ def do_signin(db: Session, user_id: int) -> tuple[SigninRecord, int]:
cycle_day = 1
streak = 1
coin = signin_reward(cycle_day)
coin = rewards.get_signin_rewards(db)[cycle_day - 1]
record = SigninRecord(
user_id=user_id,
signin_date=today,
@@ -119,7 +121,8 @@ def do_signin(db: Session, user_id: int) -> tuple[SigninRecord, int]:
)
db.add(record)
acc, _ = crud_wallet.grant_coins(
db, user_id, coin, biz_type="signin", ref_id=today.isoformat()
db, user_id, coin, biz_type="signin", ref_id=today.isoformat(),
remark=f"每日签到 第{cycle_day}",
)
db.commit()
db.refresh(record)
+8 -5
View File
@@ -1,6 +1,7 @@
"""一次性任务 CRUD:列表状态 + 领奖。
领奖原子化:写 user_task(唯一约束防重复)+ 发金币,同一事务 commit。
任务列表 + 奖励来自 rewards.get_task_rewards(运营后台可改)。
"""
from __future__ import annotations
@@ -9,9 +10,9 @@ from dataclasses import dataclass
from sqlalchemy import select
from sqlalchemy.orm import Session
from app.core.rewards import TASK_REWARDS
from app.repositories import wallet as crud_wallet
from app.core import rewards
from app.models.task import UserTask
from app.repositories import wallet as crud_wallet
class UnknownTaskError(Exception):
@@ -31,11 +32,12 @@ class TaskState:
def list_tasks(db: Session, user_id: int) -> list[TaskState]:
"""返回所有已知一次性任务及其领取状态。"""
task_rewards = rewards.get_task_rewards(db)
stmt = select(UserTask.task_key).where(UserTask.user_id == user_id)
claimed_keys = set(db.execute(stmt).scalars().all())
return [
TaskState(task_key=key, coin=coin, claimed=key in claimed_keys)
for key, coin in TASK_REWARDS.items()
for key, coin in task_rewards.items()
]
@@ -44,7 +46,8 @@ def claim_task(db: Session, user_id: int, task_key: str) -> tuple[int, int]:
未知任务抛 UnknownTaskError;重复领取抛 AlreadyClaimedError。
"""
if task_key not in TASK_REWARDS:
task_rewards = rewards.get_task_rewards(db)
if task_key not in task_rewards:
raise UnknownTaskError
existing = db.execute(
@@ -55,7 +58,7 @@ def claim_task(db: Session, user_id: int, task_key: str) -> tuple[int, int]:
if existing is not None:
raise AlreadyClaimedError
coin = TASK_REWARDS[task_key]
coin = task_rewards[task_key]
db.add(
UserTask(
user_id=user_id,
+304 -42
View File
@@ -7,27 +7,33 @@
from __future__ import annotations
import re
import unicodedata
import uuid
from datetime import datetime, timedelta, timezone
from sqlalchemy import select, update
from sqlalchemy.orm import Session
from app.core.rewards import (
COIN_PER_CENT,
MIN_EXCHANGE_COIN,
WITHDRAW_MAX_CENTS,
WITHDRAW_MIN_CENTS,
coins_to_cents,
)
from app.core import rewards
from app.core.config import settings
from app.core.rewards import COIN_PER_CENT, coins_to_cents
from app.integrations import wxpay
from app.models.user import User
from app.models.wallet import CashTransaction, CoinAccount, CoinTransaction, WithdrawOrder
from app.models.wallet import (
CashTransaction,
CoinAccount,
CoinTransaction,
WechatTransferAuthorization,
WithdrawOrder,
)
# 微信转账终态:成功 / 失败(失败/取消/关闭都退款)
_WX_STATE_SUCCESS = "SUCCESS"
_WX_STATE_FAILED = {"FAIL", "CANCELLED", "CLOSED"}
_WX_STATE_WAIT_CONFIRM = "WAIT_USER_CONFIRM" # 用户还没在微信确认页确认
# 免确认收款授权状态
_WX_AUTH_ACTIVE = "TAKING_EFFECT" # 已生效,可免确认转账
_WX_AUTH_CLOSED = "CLOSED" # 已关闭(用户/商户/风控),需重新开启
class InvalidExchangeAmountError(Exception):
@@ -62,6 +68,10 @@ class WithdrawOrderNotFound(Exception):
"""提现单不存在。"""
class WithdrawNotReviewable(Exception):
"""提现单当前状态不可审核(非 reviewing,可能已被处理过)。"""
def get_or_create_account(db: Session, user_id: int, *, commit: bool = True) -> CoinAccount:
"""取用户金币账户,不存在则建一个空账户。"""
acc = db.get(CoinAccount, user_id)
@@ -142,7 +152,7 @@ def exchange_coins_to_cash(
校验:金额 >= MIN_EXCHANGE_COIN 且为整分倍数(InvalidExchangeAmountError),
余额充足(InsufficientCoinError)扣金币 + 加现金,两条流水同事务 commit
"""
if coin_amount < MIN_EXCHANGE_COIN or coin_amount % COIN_PER_CENT != 0:
if coin_amount < rewards.get_min_exchange_coin(db) or coin_amount % COIN_PER_CENT != 0:
raise InvalidExchangeAmountError
acc = get_or_create_account(db, user_id, commit=False)
@@ -259,10 +269,18 @@ def _add_cash(db: Session, user_id: int, amount_cents: int) -> int:
return bal
def _refund_withdraw(db: Session, order: WithdrawOrder, reason: str) -> None:
"""转账失败/取消:退回现金 + 写 withdraw_refund 流水 + 单子置 failed。幂等:已 failed 不重复退。"""
if order.status == "failed":
return # 防重复退款(并发/对账与查单同时触发)
def _refund_withdraw(
db: Session, order: WithdrawOrder, reason: str, *, final_status: str = "failed"
) -> None:
"""退回现金 + 写 withdraw_refund 流水 + 单子置终态。幂等:已是退款终态不重复退。
final_status:
- "failed" (默认):微信转账失败/取消 自动退款
- "rejected":管理员审核拒绝 退款( reject_withdraw)
两种都已退款, in 判定防重复退(并发/对账/拒绝重复点)
"""
if order.status in ("failed", "rejected"):
return # 防重复退款(并发/对账与查单/重复拒绝同时触发)
bal = _add_cash(db, order.user_id, order.amount_cents)
db.add(
CashTransaction(
@@ -271,10 +289,11 @@ def _refund_withdraw(db: Session, order: WithdrawOrder, reason: str) -> None:
balance_after_cents=bal,
biz_type="withdraw_refund",
ref_id=order.out_bill_no,
remark="提现未成功,金额已退回", # 用户可见;技术原因记在 order.fail_reason
# 用户可见文案区分"未成功(自动退)"vs"审核未通过";技术原因记在 order.fail_reason
remark="提现审核未通过,金额已退回" if final_status == "rejected" else "提现未成功,金额已退回",
)
)
order.status = "failed"
order.status = final_status
order.fail_reason = reason[:256]
db.commit()
@@ -332,22 +351,26 @@ def create_withdraw(
user_name: str | None = None,
out_bill_no: str | None = None,
) -> WithdrawOrder:
"""发起提现(原子扣款 + 幂等 + 模糊失败查单)
"""发起提现:原子扣款 + 建单 reviewing(待人工审核),**不打款**
#1 用原子条件 UPDATE 扣款,杜绝并发超额。
#2 out_bill_no 由客户端提供作幂等键:同号重试不会重复转账,直接返回该单当前状态。
#3 微信调用结果不明时先查单再决定,绝不盲目退款。
返回提现(可能含 package_info App 拉起确认页)
打款推迟到管理员在 admin 后台审核通过(approve_withdraw execute_withdraw_transfer)才发起;
拒绝则退回现金(reject_withdraw)
#1 用原子条件 UPDATE 扣款,杜绝并发超额——且"待审核期间钱已扣减",防止用户拿同一笔余额
重复发起多笔提现(审核拒绝再退回)
#2 out_bill_no 客户端幂等键:同号重试返回该单现状(reviewing 等审核),不重复扣款建单。
实名 user_name 在此存下(WithdrawOrder.user_name),供异步审核打款时传给微信(达额需实名)
"""
if amount_cents < WITHDRAW_MIN_CENTS or amount_cents > WITHDRAW_MAX_CENTS:
if amount_cents < rewards.get_withdraw_min_cents(db) or amount_cents > rewards.get_withdraw_max_cents(db):
raise InvalidWithdrawAmountError
# 提现即要求已绑微信:否则审核通过也打不了款,提前拦更友好
user = db.get(User, user_id)
openid = user.wechat_openid if user else None
if not openid:
raise WechatNotBoundError
# #2 幂等:客户端传了合法 out_bill_no 且该单已存在 → 返回现状(pending 的顺手查一下),不重复发起
# #2 幂等:该单已存在 → 直接返回现状(reviewing 等审核 / 已处理态原样返回),不重复扣款。
# 不再像旧版对 pending 查微信:本函数不再涉及微信,查单交给 /withdraw/status 与对账任务。
if out_bill_no and _OUT_BILL_NO_RE.match(out_bill_no):
existing = db.execute(
select(WithdrawOrder).where(
@@ -355,8 +378,6 @@ def create_withdraw(
)
).scalar_one_or_none()
if existing is not None:
if existing.status == "pending":
return refresh_withdraw_status(db, user_id, out_bill_no)
return existing
else:
out_bill_no = uuid.uuid4().hex
@@ -379,38 +400,173 @@ def create_withdraw(
balance_after_cents=bal,
biz_type="withdraw",
ref_id=out_bill_no,
remark="提现到微信零钱",
remark="提现到微信零钱(待审核)",
)
)
order = WithdrawOrder(
user_id=user_id, out_bill_no=out_bill_no, amount_cents=amount_cents, status="pending"
user_id=user_id,
out_bill_no=out_bill_no,
amount_cents=amount_cents,
user_name=user_name,
status="reviewing",
)
db.add(order)
db.commit()
db.refresh(order)
return order # 待管理员审核;**不在此处打款**
# 2) 调微信转账;结果不明先查单再决定(#3)
# ===== 免确认收款授权(用户授权免确认模式)=====
# 用户授权一次后,后续提现走 transfer_with_authorization 免逐笔确认直接到账。
# WechatTransferAuthorization 一个用户一条;out_authorization_no 我方生成,authorization_id 微信生效后返回。
def _auth_display_name(user: User | None) -> str:
"""微信授权页展示的"开通账号"昵称(≤32,utf8)。
微信对 user_display_name 校验极严:**连空格和标点(. )都算"控制字符"拒收**实测
'wonderable ai'(带空格) 400 'wonderableai'/'周周'/'傻瓜比价用户8888' 才过
故只保留 文字(Unicode L*,中英文/CJK)与数字(N*),其余(emoji/符号/空格/标点/控制符/
零宽连接符/变体选择符/星形面字符)一律剔除;清空则兜底手机号尾号
"""
raw = ((user.nickname if user else None) or (user.wechat_nickname if user else None) or "").strip()
kept = [
ch for ch in raw
if ord(ch) <= 0xFFFF and unicodedata.category(ch)[0] in ("L", "N")
]
name = "".join(kept).strip()
if not name and user and user.phone:
tail = user.phone[-4:] if len(user.phone) >= 4 else user.phone
name = f"傻瓜比价用户{tail}"
return (name or "傻瓜比价用户")[:32]
def _refresh_active_auth(db: Session, user_id: int) -> None:
"""免确认转账失败后回查授权有效性(权威判定,不靠猜错误码):
微信侧明确非生效(用户在微信关闭 / 风控 / 无此单) closed,下次提现自动回退方式一重新授权;
仍生效(失败实为商户余额不足等与授权无关的原因) 保持 active,下次免确认重试
仅对本地 active 记录查询;查询本身失败(5xx/网络)不改状态,下次再核,避免误关有效授权"""
auth = db.get(WechatTransferAuthorization, user_id)
if auth is None or auth.state != "active" or not settings.wxpay_configured:
return
try:
result = wxpay.create_transfer(openid, amount_cents, out_bill_no, user_name=user_name)
except Exception as e: # noqa: BLE001 — 超时/网络异常≠失败,查单确认
_settle_after_ambiguous(db, order, reason=f"转账调用异常: {e}")
db.refresh(order)
if order.status == "failed":
raise WithdrawTransferError(str(e)) from e
return order
res = wxpay.query_transfer_authorization(auth.out_authorization_no)
except Exception: # noqa: BLE001 — 查询失败保持 active,下次再核
return
if res["status_code"] == 200:
if res["data"].get("state") != _WX_AUTH_ACTIVE: # CLOSED / 其他非生效 → 失效
auth.state = "closed"
db.commit()
elif _wx_not_found(res): # 明确无此授权单 → 失效
auth.state = "closed"
db.commit()
# 其他查询错误(5xx 等):不改状态,保持 active,下次再核
def sync_transfer_auth(db: Session, user_id: int) -> WechatTransferAuthorization | None:
"""同步待确认授权最新状态(仅 pending 时查微信):
TAKING_EFFECT active + authorization_id;CLOSED/超期无此单 closed返回最新记录( None)
pending(active/closed)直接返回不查;查询异常或缺凭证则保持原状"""
auth = db.get(WechatTransferAuthorization, user_id)
if auth is None or auth.state != "pending" or not settings.wxpay_configured:
return auth
try:
res = wxpay.query_transfer_authorization(auth.out_authorization_no)
except Exception: # noqa: BLE001 — 查询失败保持 pending,下次再同步
return auth
if res["status_code"] != 200:
if _wx_not_found(res): # 超期未确认被关闭 → closed;其他错误保持 pending
auth.state = "closed"
db.commit()
return auth
state = res["data"].get("state", "")
if state == _WX_AUTH_ACTIVE:
auth.state = "active"
auth.authorization_id = res["data"].get("authorization_id") or auth.authorization_id
db.commit()
elif state == _WX_AUTH_CLOSED:
auth.state = "closed"
db.commit()
# WAIT_USER_CONFIRM:保持 pending,等用户确认
return auth
def _ensure_pending_auth_no(db: Session, user_id: int, openid: str) -> str:
""""方式一(转账顺带授权)"准备待确认授权单号:
已有 pending 复用其 out_authorization_no(避免重复申请刷满"同场景≤5"上限);
/ closed 生成新单号并 upsert pending返回 out_authorization_no"""
auth = db.get(WechatTransferAuthorization, user_id)
if auth is not None and auth.state == "pending":
if auth.openid != openid: # 换绑过 → 刷新快照
auth.openid = openid
db.commit()
return auth.out_authorization_no
new_no = uuid.uuid4().hex # 32 位,符合 [0-9A-Za-z_-]{8,32}
if auth is None:
db.add(
WechatTransferAuthorization(
user_id=user_id, openid=openid, out_authorization_no=new_no,
authorization_id=None, state="pending",
)
)
else: # closed → 重新开启,换新单号
auth.openid = openid
auth.out_authorization_no = new_no
auth.authorization_id = None
auth.state = "pending"
db.commit()
return new_no
def apply_transfer_auth(db: Session, user_id: int) -> dict:
"""方式二:显式开启免确认收款(申请授权,不转账)。
返回 {already_active, package_info, mch_id, app_id};already_active=True 表示已开启无需再授权
未绑微信抛 WechatNotBoundError;微信返回非 200 WithdrawTransferError"""
user = db.get(User, user_id)
openid = user.wechat_openid if user else None
if not openid:
raise WechatNotBoundError
auth = db.get(WechatTransferAuthorization, user_id)
if auth is not None and auth.state == "active" and auth.authorization_id:
return {
"already_active": True, "package_info": None,
"mch_id": settings.WXPAY_MCH_ID, "app_id": settings.WECHAT_APP_ID,
}
out_auth_no = _ensure_pending_auth_no(db, user_id, openid)
result = wxpay.apply_transfer_authorization(
out_auth_no, openid, _auth_display_name(user), settings.WXPAY_AUTH_NOTIFY_URL
)
if result["status_code"] != 200:
msg = str(result["data"].get("message") or result["data"])
_settle_after_ambiguous(db, order, reason=msg)
db.refresh(order)
if order.status == "failed":
raise WithdrawTransferError(msg)
return order
raise WithdrawTransferError(str(result["data"].get("message") or result["data"]))
return {
"already_active": False,
"package_info": result["data"].get("package_info"),
"mch_id": settings.WXPAY_MCH_ID,
"app_id": settings.WECHAT_APP_ID,
}
data = result["data"]
def close_transfer_auth(db: Session, user_id: int) -> None:
"""关闭免确认收款(解除授权)。best-effort 调微信解除 + 本地置 closed。"""
auth = db.get(WechatTransferAuthorization, user_id)
if auth is None:
return
if auth.state != "closed" and settings.wxpay_configured:
try:
wxpay.close_transfer_authorization(auth.out_authorization_no)
except Exception: # noqa: BLE001 — 微信解除失败不阻塞本地置 closed(用户也可在微信侧自行关闭)
pass
auth.state = "closed"
db.commit()
def _apply_transfer_result(db: Session, order: WithdrawOrder, data: dict) -> WithdrawOrder:
"""落微信转账应答到提现单:记 state/转账单号/package_info,SUCCESS 即 success。"""
order.wechat_state = data.get("state")
order.transfer_bill_no = data.get("transfer_bill_no")
order.package_info = data.get("package_info")
order.package_info = data.get("package_info") # 免确认转账无此字段(None);确认模式带它供拉确认页
if data.get("state") == _WX_STATE_SUCCESS:
order.status = "success"
db.commit()
@@ -418,6 +574,112 @@ def create_withdraw(
return order
def execute_withdraw_transfer(db: Session, order: WithdrawOrder) -> WithdrawOrder:
"""审核通过后真正发起微信转账(复用原"转账 + 模糊失败查单"逻辑,绝不盲目退款)。**不抛异常**。
免确认收款(用户授权免确认模式)分叉:
已有生效授权(active) transfer_with_authorization 免确认转账,直接到账, package_info
无生效授权:
- 已配 WXPAY_AUTH_NOTIFY_URL pre_transfer_with_authorization(方式一):转账 + 顺带申请授权,
返回 WAIT_USER_CONFIRM + package_info,用户在确认这笔收款时一并授权,下次起免确认
- 未配回调地址(未启用免确认) 退化为原 create_transfer 确认模式,行为同改造前
免确认转账失败 _settle_after_ambiguous 保金额安全, _refresh_active_auth 回查授权,失效则标 closed(下次回退方式一)
用户审核期间解绑微信 退款 failed结果不明(超时/非200) 先查单再决定,绝不盲目退款
"""
user = db.get(User, order.user_id)
openid = user.wechat_openid if user else None
if not openid:
_refund_withdraw(db, order, reason="用户已解绑微信,无法打款")
db.refresh(order)
return order
# 先同步待确认授权:捕获首单确认后微信已生效的 authorization_id(pending→active),或失效(→closed)
auth = sync_transfer_auth(db, order.user_id)
order.status = "pending" # 进入打款在途(转账调用前崩溃留 pending,交对账兜底)
db.commit()
# ① 有生效授权 → 免确认转账(无需用户确认,直接到账)
if auth is not None and auth.state == "active" and auth.authorization_id:
try:
result = wxpay.transfer_with_authorization(
auth.authorization_id, order.amount_cents, order.out_bill_no, user_name=order.user_name
)
except Exception as e: # noqa: BLE001 — 超时/网络异常≠失败,查单确认
_settle_after_ambiguous(db, order, reason=f"免确认转账调用异常: {e}")
db.refresh(order)
return order
if result["status_code"] != 200:
# 金额安全:查转账单后定夺,绝不盲退(未创建→退款,已创建→按真实状态)
_settle_after_ambiguous(db, order, reason=str(result["data"].get("message") or result["data"]))
# 授权有效性:回查授权单,微信侧已失效(用户关闭/风控)→标 closed,下次提现自动回退方式一重新授权
_refresh_active_auth(db, order.user_id)
db.refresh(order)
return order
return _apply_transfer_result(db, order, result["data"])
# ② 无生效授权 → 启用免确认则转账+顺带授权(方式一),否则退化为原确认模式
try:
if settings.WXPAY_AUTH_NOTIFY_URL:
out_auth_no = _ensure_pending_auth_no(db, order.user_id, openid)
result = wxpay.pre_transfer_with_authorization(
openid, order.amount_cents, order.out_bill_no,
out_authorization_no=out_auth_no,
user_display_name=_auth_display_name(user),
notify_url=settings.WXPAY_AUTH_NOTIFY_URL,
user_name=order.user_name,
)
else:
result = wxpay.create_transfer(
openid, order.amount_cents, order.out_bill_no, user_name=order.user_name
)
except Exception as e: # noqa: BLE001 — 超时/网络异常≠失败,查单确认
_settle_after_ambiguous(db, order, reason=f"转账调用异常: {e}")
db.refresh(order)
return order
if result["status_code"] != 200:
_settle_after_ambiguous(db, order, reason=str(result["data"].get("message") or result["data"]))
db.refresh(order)
return order
return _apply_transfer_result(db, order, result["data"])
def _get_withdraw_or_raise(db: Session, out_bill_no: str) -> WithdrawOrder:
"""按 out_bill_no 取提现单(admin 跨用户,不限 user_id)。不存在抛 WithdrawOrderNotFound。"""
order = db.execute(
select(WithdrawOrder).where(WithdrawOrder.out_bill_no == out_bill_no)
).scalar_one_or_none()
if order is None:
raise WithdrawOrderNotFound
return order
def approve_withdraw(db: Session, out_bill_no: str) -> WithdrawOrder:
"""管理员审核通过:校验 reviewing → 发起微信转账。返回最终 order(pending/success/failed)。
幂等防误操作: reviewing(已通过/已拒绝/已打款) WithdrawNotReviewable,不会重复打款
"""
order = _get_withdraw_or_raise(db, out_bill_no)
if order.status != "reviewing":
raise WithdrawNotReviewable(f"提现单状态为 {order.status},非待审核,不能通过")
return execute_withdraw_transfer(db, order)
def reject_withdraw(db: Session, out_bill_no: str, reason: str) -> WithdrawOrder:
"""管理员审核拒绝:校验 reviewing → 退回现金 + 置 rejected。
reviewing WithdrawNotReviewable(防对已打款单误退导致重复退款)
"""
order = _get_withdraw_or_raise(db, out_bill_no)
if order.status != "reviewing":
raise WithdrawNotReviewable(f"提现单状态为 {order.status},非待审核,不能拒绝")
_refund_withdraw(db, order, reason=reason or "审核未通过", final_status="rejected")
db.refresh(order)
return order
def refresh_withdraw_status(
db: Session, user_id: int, out_bill_no: str, *, cancel_if_unconfirmed: bool = False
) -> WithdrawOrder:
+53
View File
@@ -4,6 +4,8 @@
"""
from __future__ import annotations
from datetime import datetime
from pydantic import BaseModel, Field
@@ -27,6 +29,53 @@ class AdRewardStatusOut(BaseModel):
daily_limit: int = Field(..., description="每日发奖次数上限")
remaining: int = Field(..., description="今日剩余可领次数")
coin_per_ad: int = Field(..., description="看完一个激励视频发的金币")
round_count: int = Field(
..., description="本轮(3 次一组)已看次数,0..N-1。客户端可由它判断'刚刚看完一轮'"
)
cooldown_until: datetime | None = Field(
None,
description="本轮看完后 10 分钟冷却结束时间(UTC ISO),null 表示不在冷却。"
"冷却是 UX 约束,后端发奖不受影响,客户端 CTA 据此显示 MM:SS 倒计时且不可点",
)
# ===== 看广告每日总时长(50 分钟防刷主闸):客户端据此展示"今日已看 X/50 分钟"、满则不给看 =====
watched_seconds_today: int = Field(0, description="今日已观看激励视频总秒数(前端累计上报)")
watch_seconds_limit: int = Field(0, description="每日观看总时长上限(秒,默认 3000=50 分钟)")
watch_seconds_remaining: int = Field(0, description="今日剩余可观看秒数(<=0 时客户端不应再展示广告)")
class EcpmReportIn(BaseModel):
"""客户端上报一次广告展示的 eCPM(内部收益统计/对账)。
user_id 不在 body JWT (Bearer),防伪造ecpm 原样上报字符串(单位待确认)
"""
ad_type: str = Field(..., description="广告类型:reward_video(激励视频) / draw(Draw 信息流) 等")
ecpm: str = Field(..., description="穿山甲 getShowEcpm().getEcpm() 原始字符串,单位待确认,原样上报")
adn: str | None = Field(None, description="实际投放 ADN(getSdkName),如 pangle")
slot_id: str | None = Field(None, description="实际展示代码位(底层 mediation rit)")
class EcpmReportOut(BaseModel):
"""eCPM 上报结果。best-effort,落库即 ok。"""
ok: bool = True
class WatchReportIn(BaseModel):
"""客户端上报一次激励视频的实际观看时长(onAdClose 时,onAdShow→onAdClose 墙钟秒数)。
user_id JWT (Bearer),不在 bodyseconds 服务端会夹 [0, MAX_SINGLE_WATCH_SECONDS]
"""
seconds: int = Field(..., ge=0, description="本次观看秒数")
class WatchReportOut(BaseModel):
"""上报后返回当日累计 + 剩余,客户端即时更新"今日还能看多久""""
watched_seconds_today: int = Field(..., description="今日累计观看秒数")
watch_seconds_limit: int = Field(..., description="每日上限(秒)")
watch_seconds_remaining: int = Field(..., description="今日剩余可观看秒数")
class TestGrantOut(BaseModel):
@@ -39,3 +88,7 @@ class TestGrantOut(BaseModel):
daily_limit: int = Field(..., description="每日发奖次数上限")
remaining: int = Field(..., description="今日剩余可领次数")
coin_per_ad: int = Field(..., description="看完一个激励视频发的金币")
round_count: int = Field(..., description="本轮已看次数,详见 AdRewardStatusOut")
cooldown_until: datetime | None = Field(
None, description="本轮冷却结束时间(UTC),详见 AdRewardStatusOut"
)
+135
View File
@@ -0,0 +1,135 @@
"""比价记录上报 / 读取 schemas。
约定同 welfare:字段 snake_case金额存整数()时间 ISO 8601
上报请求(ComparisonRecordIn)的字段刻意对齐 pricebot 协议
(docs/main/02_api_protocol.md calibration + done.params.comparison_results),
让客户端把 Phase1 calibration done 帧的 params 字段**零翻译**直接映射上来,
server 端负责拆成结构化列(best_*/saved/is_source_best comparison_results 派生)
"""
from __future__ import annotations
from datetime import datetime
from pydantic import BaseModel, ConfigDict, Field
# ===== 上报请求 =====
class ComparisonItemIn(BaseModel):
"""下单菜品(来自 calibration.items)。"""
name: str
qty: int = 1
specs: list[str] | None = None
class ComparisonResultIn(BaseModel):
"""逐平台对比项(来自 done.params.comparison_results)。price 单位:元。"""
platform_id: str | None = None
platform_name: str | None = None
package: str | None = None
price: float | None = None
is_source: bool = False
rank: int | None = None
# 该平台本单用红包省的**纯红包优惠额**(元, 不含配送费/代金券/满减)。None=没用/没抠到。
# 必须显式声明: 落库走 model_dump(), pydantic 默认丢未知字段, 不声明这行会被悄悄吞掉。
# 各平台抠到红包即带值(2026-06 起源平台 Phase1 意图识别也抠, 当前仅淘宝源)。见 pricebot 侧 比价红包额留痕-实现方案.md。
coupon_saved: float | None = None
class ComparisonRecordIn(BaseModel):
# 幂等键:同一用户同一 trace_id 重复上报只保留一条(覆盖)
trace_id: str = Field(..., min_length=1, description="pricebot 侧 trace_id")
business_type: str = Field("food", description="food / ecom / coupon")
device_id: str | None = None
# 源平台(来自 calibration)
store_name: str | None = Field(None, description="店铺名(外卖)")
source_platform_id: str | None = None
source_platform_name: str | None = None
source_package: str | None = None
source_price: float | None = Field(None, description="源平台到手价(元)")
# 明细
items: list[ComparisonItemIn] = Field(default_factory=list)
comparison_results: list[ComparisonResultIn] = Field(default_factory=list)
skipped_dish_count: int | None = None
skipped_dish_names: list[str] = Field(default_factory=list)
total_dish_count: int | None = None
information: str | None = Field(None, description="done 帧文案,留存备查")
# 不传则服务端按 comparison_results 派生(有非源有效价=success,否则 failed)
status: str | None = Field(None, description="success / failed,可不传由服务端派生")
# 最优平台商家/商品深链(客户端从 collectedLinks[best_index] 取);「再次比价」直达用
best_deeplink: str | None = Field(None, description="最优平台深链,再次比价直达")
# ===== 读取出参 =====
class ComparisonRecordOut(BaseModel):
"""列表项:结构化概要字段(不含 raw_payload,减小列表 payload)。"""
model_config = ConfigDict(from_attributes=True)
id: int
business_type: str
trace_id: str
source_platform_id: str | None = None
source_platform_name: str | None = None
source_package: str | None = None
source_price_cents: int | None = None
best_platform_id: str | None = None
best_platform_name: str | None = None
best_price_cents: int | None = None
best_deeplink: str | None = None
saved_amount_cents: int | None = None
is_source_best: bool | None = None
store_name: str | None = None
total_dish_count: int | None = None
skipped_dish_count: int | None = None
status: str
information: str | None = None
items: list = []
comparison_results: list = []
skipped_dish_names: list = []
# 「已下单」(店级):该店名在该用户真实下单(source='compare')里出现过即 True。
# 由 list_records 动态算出挂在 ORM 实例上(非 DB 列),from_attributes 读出;缺省 False。
ordered: bool = False
created_at: datetime
class ComparisonRecordDetailOut(ComparisonRecordOut):
"""详情:在概要之上额外带 raw_payload 全量。"""
raw_payload: dict | None = None
class ComparisonRecordPage(BaseModel):
items: list[ComparisonRecordOut]
next_cursor: int | None = Field(None, description="下一页游标(末条 id);为空表示到底")
class ComparisonRecordCreatedOut(BaseModel):
id: int = Field(..., description="写入(或已存在)的记录 id")
# ===== 比价战绩里程碑(福利页「记录比价战绩」)=====
class MilestoneStateOut(BaseModel):
milestone: int = Field(..., description="档位序号 1-based(= 解锁所需的成功比价次数)")
coin: int = Field(..., description="该档应发金币额(产品规则值);当前领取暂不真发, 见 compare-milestone-claim 文档")
state: str = Field(..., description="claimed(已领) / active(可领) / locked(未解锁)")
class MilestoneStatusOut(BaseModel):
success_count: int = Field(..., description="累计成功比价次数(解锁进度)")
claimable_count: int = Field(..., description="当前可领(active)档数")
milestones: list[MilestoneStateOut]
class MilestoneClaimResultOut(BaseModel):
milestone: int = Field(..., description="本次领取的档位序号")
coin_awarded: int = Field(..., description="本次发放金币")
coin_balance: int = Field(..., description="领奖后金币余额")
+30
View File
@@ -0,0 +1,30 @@
from pydantic import BaseModel, ConfigDict, Field
class OrderReportRequest(BaseModel):
"""客户端归因成功后的上报体。金额一律用「分」(int) 传,避免浮点误差。"""
client_event_id: str = Field(..., max_length=64, description="客户端幂等键(UUID)")
platform: str = Field(..., max_length=32, description="平台展示名,如 美团")
platform_package: str | None = Field(None, max_length=128, description="平台包名")
pay_channel: str = Field(..., max_length=16, description="支付渠道 wechat/alipay")
compared_price_cents: int = Field(..., ge=0, description="我们给出的比价价(分)")
paid_amount_cents: int = Field(..., ge=0, description="实际支付金额(分)")
device_id: str | None = Field(None, max_length=128)
# ===== 比价时携带的记账信息(客户端从意图识别阶段缓存而来;旧版客户端可能不传,故全部可空)=====
shop_name: str | None = Field(None, max_length=128, description="门店名,如 肯德基宅急送(天北路店)")
dishes: list[str] = Field(default_factory=list, description="菜品名列表")
original_price_cents: int | None = Field(None, ge=0, description="源平台原价(分),省额=原价−实付")
source_platform_name: str | None = Field(None, max_length=32, description="源平台展示名,如 美团")
source_deeplink: str | None = Field(None, max_length=512, description="源平台重进链接(预留,本期只存不展示)")
class OrderReportOut(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: int
platform: str
pay_channel: str
compared_price_cents: int
paid_amount_cents: int
duplicated: bool = False
+57
View File
@@ -0,0 +1,57 @@
"""上报更低价 响应 schema。
请求是 multipart 表单(comparison_record_id / reported_platform_id / reported_price / images),
在路由里直接校验,不单独建请求 schema( feedback)价格对外统一(cents)
"""
from __future__ import annotations
from datetime import datetime
from pydantic import BaseModel, ConfigDict, Field
class ReportSubmitOut(BaseModel):
"""提交上报后的回执。"""
model_config = ConfigDict(from_attributes=True)
id: int
status: str
created_at: datetime
class ReportRecordOut(BaseModel):
"""一条上报记录(「上报记录」列表用)。价格单位:分。"""
model_config = ConfigDict(from_attributes=True)
id: int
store_name: str | None = None
dish_summary: str | None = None
# 原最低价(上报前系统给出的最低,反查比价记录 best_*)
original_platform_id: str | None = None
original_platform_name: str | None = None
original_price_cents: int | None = None
# 用户上报的更低价
reported_platform_id: str
reported_platform_name: str
reported_price_cents: int
images: list[str] = Field(default_factory=list)
status: str # pending / approved / rejected
reject_reason: str | None = None
reward_coins: int | None = None
created_at: datetime
class ReportRecordCounts(BaseModel):
"""四态计数,供前端筛选 chip 显示(全部/审核中/已通过/未通过)。"""
all: int = 0
pending: int = 0
approved: int = 0
rejected: int = 0
class ReportRecordsOut(BaseModel):
records: list[ReportRecordOut]
counts: ReportRecordCounts
+37 -5
View File
@@ -80,6 +80,23 @@ class WithdrawInfoOut(BaseModel):
wechat_bound: bool = Field(..., description="当前用户是否已绑定微信")
wechat_nickname: str | None = Field(None, description="微信昵称(可能为空/脱敏)")
wechat_avatar_url: str | None = Field(None, description="微信头像 URL(可能为空)")
transfer_auth_enabled: bool = Field(
False, description="是否已开启免确认到账(开启后提现免跳微信确认,直接到账)"
)
# ===== 免确认收款授权(用户授权免确认模式)=====
class TransferAuthResultOut(BaseModel):
already_active: bool = Field(False, description="是否已是开启状态(无需再授权)")
package_info: str | None = Field(None, description="拉起微信授权页的 package(已开启时为空)")
mch_id: str | None = None
app_id: str | None = None
class TransferAuthStatusOut(BaseModel):
state: str = Field(..., description="none(未开启)/pending(待确认)/active(已开启)/closed(已关闭)")
enabled: bool = Field(..., description="是否已开启免确认到账")
class BindWechatRequest(BaseModel):
@@ -102,11 +119,18 @@ class WithdrawRequest(BaseModel):
out_bill_no: str | None = Field(
None, description="客户端幂等键(商户单号):同号重试不重复转账。不传则服务端生成"
)
skip_review: bool = Field(
False,
description=(
"调试直发:跳过人工审核立即打款。仅非生产(APP_ENV!=prod)生效,"
"客户端仅 debug 包下发(开发设置开关)。生产恒走人工审核。"
),
)
class WithdrawResultOut(BaseModel):
out_bill_no: str = Field(..., description="商户提现单号(查单用)")
status: str = Field(..., description="pending / success / failed")
status: str = Field(..., description="reviewing(待审核) / pending / success / failed / rejected")
wechat_state: str | None = Field(None, description="微信侧原始状态")
amount_cents: int
cash_balance_cents: int = Field(..., description="提现后现金余额(分)")
@@ -118,9 +142,14 @@ class WithdrawResultOut(BaseModel):
class WithdrawStatusOut(BaseModel):
out_bill_no: str
status: str = Field(..., description="pending / success / failed")
status: str = Field(..., description="reviewing(待审核) / pending / success / failed / rejected")
wechat_state: str | None = None
amount_cents: int
fail_reason: str | None = Field(None, description="failed/rejected 时的原因(用户可读)")
# 审核通过进入打款、且微信要求用户确认(WAIT_USER_CONFIRM)时,带回这三项供 App 拉起微信确认页
package_info: str | None = None
mch_id: str | None = None
app_id: str | None = None
class WithdrawOrderOut(BaseModel):
@@ -129,7 +158,7 @@ class WithdrawOrderOut(BaseModel):
id: int
out_bill_no: str
amount_cents: int
status: str
status: str = Field(..., description="reviewing(待审核) / pending / success / failed / rejected")
wechat_state: str | None = None
fail_reason: str | None = None
created_at: datetime
@@ -200,12 +229,15 @@ class SavingsRecordOut(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: int
order_amount_cents: int
saved_amount_cents: int
order_amount_cents: int # 实付金额(分)
saved_amount_cents: int # 省下(分)
original_price_cents: int | None = None # 源平台原价(分);demo 行为空
platform: str | None = None
title: str | None = None
shop_name: str | None = None
dishes: list[str] = []
pay_channel: str | None = None # wechat/alipay;demo 行为空
source_platform_name: str | None = None # 源平台名,如 美团
created_at: datetime
+23
View File
@@ -0,0 +1,23 @@
[Unit]
Description=Shaguabijia Admin Backend (FastAPI / uvicorn, 独立进程,复用 app-server codebase)
After=network.target
[Service]
Type=simple
User=root
WorkingDirectory=/opt/shaguabijia-app-server
Environment="PATH=/opt/shaguabijia-app-server/.venv/bin:/usr/bin:/bin"
EnvironmentFile=/opt/shaguabijia-app-server/.env
# 与主 App 后端(8770)复用同一 codebase / .env / .venv / DB,独立进程跑 admin app
ExecStart=/opt/shaguabijia-app-server/.venv/bin/uvicorn app.admin.main:admin_app --host 127.0.0.1 --port 8771 --workers 1 --log-level info
Restart=on-failure
RestartSec=3
NoNewPrivileges=true
PrivateTmp=true
ProtectSystem=strict
ReadWritePaths=/opt/shaguabijia-app-server
ProtectHome=true
[Install]
WantedBy=multi-user.target
+36 -4
View File
@@ -3,7 +3,7 @@
> Base URL:生产 `https://app-api.shaguabijia.com`;本地联调 `http://<开发机>:8770`
> 协议:HTTP / JSON,请求与响应体均 `application/json`,字段统一 **snake_case**
> 鉴权:需鉴权的接口在请求头带 `Authorization: Bearer <access_token>`
> 最后更新:2026-05-27
> 最后更新:2026-05-31+ 比价战绩里程碑 12d/12e
> 架构:`app/api/v1/` 只放很轻的接口层;穿山甲/微信支付/极光/短信/美团等 SDK 集成的重逻辑在 `app/integrations/`,实现细节见 [docs/integrations/](../integrations/README.md)。
---
@@ -26,6 +26,13 @@
| **比价透传**(前缀 `/api/v1`,外卖 MVP;与 `coupon/step` 同为透传 pricebot-backend |||
| 12 | `POST /api/v1/intent/recognize` | 无 | [详情](./compare-intent-recognize.md) |
| 13 | `POST /api/v1/price/step` | 无 | [详情](./compare-price-step.md) |
| **比价记录**(前缀 `/api/v1/compare`;按用户落库,**鉴权**,区别于上面不鉴权的透传) |||
| 12a | `POST /api/v1/compare/record` | Bearer | [详情](./compare-record-report.md) |
| 12b | `GET /api/v1/compare/records` | Bearer | [详情](./compare-records.md) |
| 12c | `GET /api/v1/compare/records/{id}` | Bearer | [详情](./compare-record-detail.md) |
| **比价战绩里程碑**(前缀 `/api/v1/compare`;福利页「记录比价战绩」,按成功比价数解锁逐档发金币) |||
| 12d | `GET /api/v1/compare/milestones` | Bearer | [详情](./compare-milestones.md) |
| 12e | `POST /api/v1/compare/milestones/{milestone}/claim` | Bearer | [详情](./compare-milestone-claim.md) |
| **钱包 / 我的资产**(前缀 `/api/v1/wallet` |||
| 14 | `GET /api/v1/wallet/account` | Bearer | [详情](./wallet-account.md) |
| 15 | `GET /api/v1/wallet/coin-transactions` | Bearer | [详情](./wallet-coin-transactions.md) |
@@ -52,6 +59,16 @@
| 32 | `GET /api/v1/ad/pangle-callback` | 验签 | [详情](./ad-pangle-callback.md) |
| 33 | `GET /api/v1/ad/reward-status` | Bearer | [详情](./ad-reward-status.md) |
| 34 | `POST /api/v1/ad/test-grant` | Bearer | [详情](./ad-test-grant.md) |
| 35 | `POST /api/v1/ad/ecpm-report` | Bearer | [详情](./ad-ecpm-report.md) |
| **用户资料**(前缀 `/api/v1/user` |||
| 35 | `PATCH /api/v1/user/profile` | Bearer | [详情](./user-profile.md) |
| 36 | `POST /api/v1/user/avatar` | Bearer | [详情](./user-avatar.md) |
| 37 | `DELETE /api/v1/user` | Bearer | [详情](./user-delete.md) |
| **帮助与反馈**(前缀 `/api/v1/feedback` |||
| 38 | `POST /api/v1/feedback` | Bearer | [详情](./feedback.md) |
| **静态资源**StaticFiles 挂载,见下方 `/media` 静态服务) |||
| - | `GET /media/avatars/<file>` | 无 | 用户头像;返回二进制图片 |
| - | `GET /media/feedback/<file>` | 无 | 反馈截图;返回二进制图片 |
> ⚠️ 美团三个接口当前**无鉴权**,且 `referral-link``sid` 允许客户端传值覆盖默认渠道——见各接口"备注"。
> `coupon/step` 及外卖比价的 `intent/recognize` / `price/step` 都透传到 pricebot-backend,**MVP 阶段均不鉴权**(device_id 透传,待补 JWT——见各接口详情)。
@@ -99,9 +116,9 @@
| 字段 | 类型 | 说明 |
|---|---|---|
| `id` | int | 用户主键 |
| `phone` | string | 手机号 |
| `nickname` | string \| null | 昵称(当前无接口可改,恒为 null |
| `avatar_url` | string \| null | 头像(同上) |
| `phone` | string | 手机号(注销账号后变 `deleted_<id>` 占位释放唯一约束) |
| `nickname` | string \| null | 昵称,经 [`PATCH /api/v1/user/profile`](./user-profile.md) 修改 |
| `avatar_url` | string \| null | 头像相对 URL(`/media/avatars/...`),经 [`POST /api/v1/user/avatar`](./user-avatar.md) 上传 |
| `register_channel` | string | 注册渠道:`jverify` / `sms` |
| `status` | string | `active` / `disabled` / `deleted` |
| `created_at` | datetime | 注册时间 |
@@ -135,6 +152,21 @@
---
## /media 静态服务
用户上传文件(头像/反馈截图)落盘到 `settings.MEDIA_ROOT`(默认 `./data/media/`),由 FastAPI 的 `StaticFiles` 挂在 `settings.MEDIA_URL_PREFIX`(默认 `/media`)对外暴露:
- `GET /media/avatars/<file>` — 用户头像(JPEG/PNG/WebP)
- `GET /media/feedback/<file>` — 反馈截图(同上)
**生产建议**:由 nginx 直接 serve `MEDIA_ROOT` 目录,绕过应用进程减少压力。
**URL 格式**:服务端返回**相对路径**(如 `/media/avatars/u1_a4f2b3c8e9d2e0a1.jpg`),客户端按自己的 `BASE_URL` 拼绝对地址——dev 下可能是 `10.0.2.2`/LAN IP/`127.0.0.1`,服务端不知道客户端怎么访问到自己。
**文件名服务端随机生成** `u<user_id>_<16 位 hex>.<ext>`,杜绝路径穿越与覆盖。
---
## 附:鉴权与刷新机制
- **签发**:登录成功后签发 access(HS256,2h) + refresh(30d),payload 含 `sub`(user_id)、`typ`(access/refresh)、`iat``exp`
+36
View File
@@ -0,0 +1,36 @@
# POST /api/v1/ad/ecpm-report — 上报本次广告展示的 eCPM(内部收益统计)
> 所属:Ad 组(前缀 `/api/v1/ad` | 鉴权:Bearer | [← 返回 API 索引](./README.md)
## 入参
请求体:`EcpmReportIn`
| 字段 | 类型 | 必填 | 说明 |
|---|---|---|---|
| `ad_type` | str | 是 | 广告类型:`reward_video`(激励视频) / `draw`(Draw 信息流) 等 |
| `ecpm` | str | 是 | 穿山甲 `getShowEcpm().getEcpm()` 原始字符串,**单位待确认(分/元)**,原样上报 |
| `adn` | str\|null | 否 | 实际投放 ADN(`getSdkName`),如 `pangle` |
| `slot_id` | str\|null | 否 | 实际展示代码位(底层 mediation rit,非客户端配置位) |
`user_id` 不在 body 里——由 JWT 取(Bearer),防伪造。
## 出参
响应 `200`:`EcpmReportOut`
| 字段 | 类型 | 说明 |
|---|---|---|
| `ok` | bool | 落库即 `true` |
## 说明
客户端在广告**展示后**(`onAdShow``getShowEcpm()`)调用,把本次展示的 eCPM 落库做**内部收益统计/对账**。
- **best-effort**:客户端 fire-and-forget,丢一两条不影响业务;穿山甲后台报表是结算权威兜底。
- 落 `ad_ecpm_record` 表,`report_date` 用北京时间当天,供「按用户/按天聚合」对账。
- **与发奖是两条独立流**:发奖走 [ad-pangle-callback](./ad-pangle-callback.md)(穿山甲 S2S,有 `trans_id`、无 ecpm),本接口客户端上报(有 ecpm、无 `trans_id`)。两者无公共键,**不逐条一一对应**,只做按用户/按天聚合口径。
- eCPM 是**每千次展示预估**,单条展示预估收益 ≈ `ecpm / 1000`,且为客户端预估口径,非最终结算。
- ⚠️ eCPM **单位(分/元)截至 2026-05-31 未最终确认**,故 `ecpm_raw` 原样存字符串;确认后再加一列解析好的数值。
- 限流:同 IP 120 次/分钟。
## 相关
- [ad-pangle-callback](./ad-pangle-callback.md) — 穿山甲 S2S 发奖回调
- [ad-reward-status](./ad-reward-status.md) — 今日看广告发奖进度
+10 -1
View File
@@ -14,6 +14,15 @@
| `daily_limit` | int | 每日发奖次数上限 |
| `remaining` | int | 今日剩余可领次数 |
| `coin_per_ad` | int | 看完一个激励视频发的金币 |
| `round_count` | int | 本轮(3 次一组)已看次数,0..N-1。客户端可据「拿到 round_count==0 && cooldown_until!=null」判定刚刚看完一轮 |
| `cooldown_until` | datetime\|null | 本轮 10 分钟冷却结束时间(UTC ISO);null 表示不在冷却 |
## 说明
福利页「看视频赚金币」用:展示「今日还能看 N 次」「看一次得 M 金币」。真正发奖走 [ad-pangle-callback](./ad-pangle-callback.md)(穿山甲 S2S)。
福利页「看视频赚金币」用:
- 展示「今日还能看 N 次」「看一次得 M 金币」(`remaining`/`coin_per_ad`)
- 任务行 CTA 4 态推导:`adLoading` → Loading;`remaining==0` → Capped("明天再来");`cooldown_until` 在未来 → CoolingDown(显示 MM:SS 倒计时,不可点);否则 Normal("去赚取")
- 弹窗 limit note:`used_today >= daily_limit` 显示「今日视频已到限额,明天再来」;`round_count==0 && cooldown_until!=null` 显示「本轮视频已看完,10分钟后再来」
冷却派生算法:取当日 `status='granted'` 记录里**最近一个已完成轮**(每 N=3 条算一轮)末尾那次的 `created_at`,加 10 分钟。若仍 > now → 返回;否则返回 null。**冷却是 UX 约束**,后端发奖(`/api/v1/ad/pangle-callback`)只看 daily_limit,冷却期间穿山甲回调照常发奖。
真正发奖走 [ad-pangle-callback](./ad-pangle-callback.md)(穿山甲 S2S)。
+34
View File
@@ -0,0 +1,34 @@
# POST /api/v1/compare/milestones/{milestone}/claim — 领取比价战绩里程碑奖励
> 所属:比价记录组(前缀 `/api/v1/compare` | 鉴权:Bearer | [← 返回 API 索引](./README.md)
领取某一档(第 `milestone` 次)。⚠️ **当前不真发金币**(产品定,后续整体删除该功能):仍写
`comparison_milestone_claim`((user_id, milestone) 唯一)标记该档已领、**每档只能领一次**,但不调
`grant_coins`、不写 `coin_transaction`,余额不变,`coin_awarded` 恒为 0。进度口径见
[milestones 进度接口](./compare-milestones.md)。
## 入参
- 路径参数 `milestone`(int):档位序号,1..6(= `RECORD_MILESTONES` 长度)。
## 出参
响应 `200`
| 字段 | 类型 | 说明 |
|---|---|---|
| `milestone` | int | 本次领取的档位序号 |
| `coin_awarded` | int | 本次发放金币(当前恒为 `0`,暂不真发) |
| `coin_balance` | int | 领奖后金币余额(当前不变) |
```json
{ "milestone": 1, "coin_awarded": 0, "coin_balance": 29087 }
```
## 错误
- `401` 未鉴权
- `404` 档位越界(`unknown milestone`,milestone < 1 或 > 档位总数)
- `409` 该档还没解锁(`milestone locked`,成功比价次数 < milestone)
- `409` 该档已领过(`milestone already claimed`)
## 说明
- 当前不发金币,客户端领取后只需把本档状态刷成 claimed 即可;若后续恢复发奖,再按 `coin_awarded` 刷新钱包余额([`GET /api/v1/wallet/account`](./wallet-account.md))。
- 幂等:重复领同一档返回 409(唯一约束 + 领取前查重)。
+53
View File
@@ -0,0 +1,53 @@
# GET /api/v1/compare/milestones — 比价战绩里程碑进度
> 所属:比价记录组(前缀 `/api/v1/compare` | 鉴权:Bearer | [← 返回 API 索引](./README.md)
福利页「记录比价战绩」的数据源。返回各档(第 1~6 次)解锁/领取状态。
**解锁进度** = 当前用户 `comparison_record``status='success'` 的条数(只算成功比价,失败的不计入)。第 N 档在「成功比价次数 ≥ N」时解锁;每档领一次(领取见 [claim 接口](./compare-milestone-claim.md))。
档位金额是**产品规则**,定义在后端 `app/core/rewards.py``RECORD_MILESTONES`(当前 `120/180/300/500/800/1200`),客户端**不要写死**,以本接口返回为准。
> ⚠️ **当前领取暂不真发金币**(产品定,后续整体删除该功能):`coin` 仍返回产品规则值仅供展示,但 [claim 接口](./compare-milestone-claim.md) 实际 `coin_awarded` 恒为 0、余额不变。前端展示须与之对齐,勿让用户误以为领取可到账。
## 入参
无(用户身份取自 Bearer token)。
## 出参
响应 `200`
| 字段 | 类型 | 说明 |
|---|---|---|
| `success_count` | int | 累计成功比价次数(解锁进度) |
| `claimable_count` | int | 当前可领(state=active)的档数 |
| `milestones` | Milestone[] | 各档状态,按 milestone 升序 |
**Milestone**
| 字段 | 类型 | 说明 |
|---|---|---|
| `milestone` | int | 档位序号(1-based),= 解锁所需的成功比价次数 |
| `coin` | int | 该档**应发**金币额(产品规则值);⚠️ 当前领取不真发,见上方说明 |
| `state` | string | `claimed`(已领) / `active`(已解锁可领) / `locked`(未解锁) |
```json
{
"success_count": 2,
"claimable_count": 1,
"milestones": [
{"milestone": 1, "coin": 120, "state": "claimed"},
{"milestone": 2, "coin": 180, "state": "active"},
{"milestone": 3, "coin": 300, "state": "locked"},
{"milestone": 4, "coin": 500, "state": "locked"},
{"milestone": 5, "coin": 800, "state": "locked"},
{"milestone": 6, "coin": 1200, "state": "locked"}
]
}
```
## 错误
- `401` 未鉴权
## 说明
- 同时可有多档处于 `active`(如累计 3 次却一档没领,则前 3 档都可领),逐档调 claim。
- 进度只增不减:领取不消耗成功次数,只是把对应档从 active→claimed。
+25
View File
@@ -0,0 +1,25 @@
# GET /api/v1/compare/records/{record_id} — 比价记录详情
> 所属:比价记录组(前缀 `/api/v1/compare` | 鉴权:Bearer | [← 返回 API 索引](./README.md)
单条比价记录详情,在列表项基础上额外带 `raw_payload`(客户端上报的原始全量),供未来 UI 展示任意细节。
## 入参(path
| 字段 | 类型 | 说明 |
|---|---|---|
| `record_id` | int | 记录 id |
## 出参
响应 `200`**ComparisonRecordDetailOut** = [ComparisonRecordOut](./compare-records.md#出参) 全部字段 + 下列:
| 字段 | 类型 | 说明 |
|---|---|---|
| `raw_payload` | object \| null | 客户端上报的原始 body 全量(calibration + done.params |
## 错误
- `401` 未鉴权
- `404` 记录不存在,或不属于当前用户(限本人,避免越权读他人记录)
## 说明
`404` 同时覆盖「id 不存在」和「id 属于他人」两种情况——不区分以免泄露他人记录是否存在。
+59
View File
@@ -0,0 +1,59 @@
# POST /api/v1/compare/record — 上报一次比价结果(幂等)
> 所属:比价记录组(前缀 `/api/v1/compare` | 鉴权:Bearer | [← 返回 API 索引](./README.md)
比价 `done` 帧后,客户端用**带 JWT 的通道**上报一条比价结果,落 `comparison_record` 表,作为「我的比价记录」数据源 + 用户级行为画像。
> ⚠️ 与不鉴权的透传端点 [`/api/v1/price/step`](./compare-price-step.md) 不同:那是转发壳,本接口按用户维度落库,**必须鉴权**。
> 本轮只做 server 端;客户端在 done 帧后调本接口的改动另起一轮(见 [待办与技术债.md](../待办与技术债.md) P1)。
## 入参(JSON body
字段刻意对齐 pricebot 协议([02_api_protocol.md](../../../pricebot-backend/docs/main/02_api_protocol.md) 的 `calibration` + `done.params`),客户端把 Phase1 的 calibration 和 done 帧 params **零翻译**映射上来即可。
| 字段 | 类型 | 必填 | 默认 | 说明 |
|---|---|---|---|---|
| `trace_id` | string | ✅ | — | pricebot 侧 trace_id。**幂等键**:同用户同 trace_id 重复上报覆盖、返回同一 id |
| `business_type` | string | ❌ | `food` | `food`(外卖,当前唯一接通) / `ecom`(电商) / `coupon`(领券) |
| `device_id` | string \| null | ❌ | null | 设备号 |
| `store_name` | string \| null | ❌ | null | 店铺名(外卖,来自 calibration.result |
| `source_platform_id` | string \| null | ❌ | null | 源平台代号,如 `taobao_flash` |
| `source_platform_name` | string \| null | ❌ | null | 源平台中文名 |
| `source_package` | string \| null | ❌ | null | 源平台 Android 包名 |
| `source_price` | float \| null | ❌ | null | 源平台到手价(**元** |
| `items` | Item[] | ❌ | [] | 下单菜品,`{name, qty, specs?}` |
| `comparison_results` | Result[] | ❌ | [] | 逐平台对比,见下表(price 单位**元**) |
| `skipped_dish_count` | int \| null | ❌ | null | 目标平台未找到、跳过的菜品数 |
| `skipped_dish_names` | string[] | ❌ | [] | 被跳过的菜名 |
| `total_dish_count` | int \| null | ❌ | null | 原始菜品总数 |
| `information` | string \| null | ❌ | null | done 帧文案(成功摘要 / 失败具体原因),持久化为列并在列表/详情返回,前端失败时当原因展示 |
| `status` | string \| null | ❌ | null | `success` / `failed`。**不传则服务端派生** |
**Item**`{ name: string, qty: int=1, specs: string[]\|null }`
**Resultcomparison_results 元素)**`{ platform_id, platform_name, package, price(元,float\|null), is_source(bool), rank(int\|null), coupon_saved(元,float\|null) }`
- `coupon_saved`:该平台本单**平台主优惠额**(元)——美团红包 / 淘宝平台红包 / 京东优惠券·百亿补贴,**只取那一笔**,不含配送费减免/共减总额。仅外卖目标平台带值,源平台/没用为 `null`,前端记录页据此展示**「已优惠 ¥X」**(null 不展示)。
## 服务端派生(客户端不用算)
`comparison_results` 派生并落结构化列:
- `best_*`:取最便宜的一家(按 `rank` 升序、再按 `price`),写 `best_platform_id/name/best_price_cents`
- `source_price_cents`:优先用 `source_price`,缺失则取 `comparison_results``is_source=true` 行的价
- `saved_amount_cents` = `源价 最优价`(可为 0 / 负:源平台本来就最便宜)
- `is_source_best`:最便宜的一家是否为源平台(= 这次没省到)
- `status`(未显式传时):存在「非源且有价」的结果 → `success`,否则 → `failed`
- 金额一律 `round(元 × 100)` 存**分**
`comparison_results` / `items` / `skipped_dish_names` 原样存 JSON;完整上报 body 另存 `raw_payload`(详情接口可取回)。
## 出参
响应 `200``{ "id": int }`(写入或已存在记录的 id
## 错误
- `401` 未鉴权
- `422``trace_id` 或字段类型不符
## 说明
失败的比价(done 帧只有源/空 comparison_results)也会落一条 `status=failed`——「越详细越好」,是否在 UI 展示由前端过滤。
+47
View File
@@ -0,0 +1,47 @@
# GET /api/v1/compare/records — 比价记录列表(游标分页)
> 所属:比价记录组(前缀 `/api/v1/compare` | 鉴权:Bearer | [← 返回 API 索引](./README.md)
「我的比价记录」列表页数据源。按 `id` 倒序(最新在前)。
## 入参(query
| 字段 | 类型 | 必填 | 默认 | 说明 |
|---|---|---|---|---|
| `limit` | int | ❌ | 20 | 1100 |
| `cursor` | int | ❌ | null | 上一页末条 `id`,首页不传 |
## 出参
响应 `200``{ items: ComparisonRecordOut[], next_cursor: int|null }`(分页见 [索引#游标分页约定](./README.md#游标分页约定)
**ComparisonRecordOut**(列表项,不含 `raw_payload`,减小 payload
| 字段 | 类型 | 说明 |
|---|---|---|
| `id` | int | 记录 id(也是游标) |
| `business_type` | string | `food` / `ecom` / `coupon` |
| `trace_id` | string | pricebot trace_id |
| `source_platform_id` | string \| null | 源平台代号 |
| `source_platform_name` | string \| null | 源平台中文名 |
| `source_package` | string \| null | 源平台包名 |
| `source_price_cents` | int \| null | 源平台到手价(分) |
| `best_platform_id` | string \| null | 最优平台代号 |
| `best_platform_name` | string \| null | 最优平台中文名 |
| `best_price_cents` | int \| null | 最优价(分) |
| `saved_amount_cents` | int \| null | 省下(分,可 0/负) |
| `is_source_best` | bool \| null | 源平台是否最便宜(= 没省到) |
| `store_name` | string \| null | 店铺名 |
| `total_dish_count` | int \| null | 菜品总数 |
| `skipped_dish_count` | int \| null | 跳过菜品数 |
| `status` | string | `success` / `failed` |
| `information` | string \| null | done 帧文案。成功:"在美团找到同店,到手价 ¥X…";失败:具体原因(如"美团、京东外卖均未找到该商品"),前端在 `status=failed` 时当原因展示 |
| `items` | object[] | 下单菜品 `{name, qty, specs?}` |
| `comparison_results` | object[] | 逐平台对比(price 单位元,已按 rank 升序) |
| `skipped_dish_names` | string[] | 被跳过的菜名 |
| `created_at` | datetime | 时间 |
## 错误
- `401` 未鉴权
## 说明
只返回当前登录用户自己的记录。需要单条全量(含 `raw_payload`)走 [详情接口](./compare-record-detail.md)。
+40
View File
@@ -0,0 +1,40 @@
# POST /api/v1/feedback — 提交反馈
> 所属:Feedback 组(前缀 `/api/v1/feedback` | 鉴权:Bearer access_token | [← 返回 API 索引](./README.md)
## 入参
**multipart/form-data**:
| 字段 | 类型 | 必填 | 说明 |
|---|---|---|---|
| `content` | string | ✓ | 反馈正文,**1-2000 字**(strip 后) |
| `contact` | string | ✓ | 联系方式(微信/QQ/手机号),**1-128 字**,便于回访 |
| `images` | file[] | ✗ | 截图,**最多 4 张**,每张走头像同款校验(JPEG/PNG/WebP,≤ 5 MB,魔数嗅探) |
## 出参
响应 `200`:
```json
{
"id": 123,
"status": "new",
"created_at": "2026-05-29T12:34:56Z"
}
```
| 字段 | 类型 | 说明 |
|---|---|---|
| `id` | int | 反馈单 ID |
| `status` | string | `new`(待处理)/ `handled`(已处理) |
| `created_at` | datetime | 提交时间(UTC) |
> 不返回上传的 image URL——这是给运营后台看的,客户端通常不需要。
## 错误码
- `400` 内容为空 / 内容超 2000 字 / 联系方式为空 / 联系方式超 128 字 / 图片超 4 张 / 单图非法(空/过大/格式不对)
- `401` 未带 token / token 无效或过期 / 用户被禁用
- `422``content``contact` 字段
## 说明
- **反馈绑用户**:`feedback.user_id = current_user.id`,便于回访
- **截图落盘**:存到 `settings.MEDIA_ROOT/feedback/`,URL 写进 `feedback.images`(JSON 列,SQLite 是 TEXT、PG 是 JSON/JSONB——为不影响现有迁移用 JSON 而非 JSONB)
- **截图可读**:`GET /media/feedback/<file>` 静态服务暴露,详见 [API 索引 — /media 静态服务](./README.md#media-静态服务)
- 文件名随机 `u<user_id>_<rand>.<ext>`,杜绝路径穿越
+3 -2
View File
@@ -23,7 +23,8 @@
响应 `200`:`{ items: CouponCard[], has_next: bool, search_id: string|null }``CouponCard` 结构见 [API 索引](./README.md#复用数据结构)。
## 错误码
- `502` 美团接口失败
- `502` 美团接口失败(仅在**已配置 `MT_CPS_APP_KEY/APP_SECRET` 但调用失败**时;未配凭证场景见下)
## 说明
当前 Android 客户端**未调用**此接口(搜索功能尚未实现),仅后端实现完整
- 当前 Android 客户端**未调用**此接口(搜索功能尚未实现),仅后端实现完整
- **未配置 MT_CPS 凭证时降级**:`settings.mt_cps_configured == false` 时直接返 `{ items: [], has_next: false, search_id: null }`,**不报 502**。设计目的是让新开发机首屏不炸——后端业务层先不依赖 CPS。已配凭证但调美团失败时才走 502

Some files were not shown because too many files have changed in this diff Show More