Compare commits

...

26 Commits

Author SHA1 Message Date
OuYingJun1024 f37eef75f2 Merge branch 'main' into feat/admin-api-db-docs 2026-06-06 10:33:33 +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
OuYingJun1024 8fea6d2b5d fix(compare): coupon_name 端到端透传 + 兑换下限正向测试
代码评审(server#12 / pricebot#38)修复:
- ComparisonResultIn / comparison 模型 / DB·API 文档补 coupon_name(优惠来源名),
  否则 pydantic 在上报边界静默丢弃,记录页来源 chip 落不了地;扩展 passthrough 测试覆盖
- 新增最小额兑换正向用例,钉死 MIN_EXCHANGE_COIN=100(可兑 1 分)新下限

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-05 15:59:43 +08:00
OuYingJun1024 1893c01f54 feat(welfare): savings/battle 返回 compare_count(累计完成比价次数)
- SavingsBattle 仓储/schema/接口新增 compare_count(= 有效记录数,与 order_count 同源)
- 更新 docs/api/savings-battle.md 与 tests/test_welfare.py 断言

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-05 10:45:21 +08:00
OuYingJun1024 0b8037c725 fix(wallet): 兑换下限 MIN_EXCHANGE_COIN 10000→100(可兑 1 分)
- MIN_EXCHANGE_COIN 从 COIN_PER_YUAN(=10000,1元)改为 COIN_PER_CENT(=100,1分),换现金小额即可用(也便于联调验证资产卡飞金币动画)
- 同步测试:test_exchange_flow 种子改 COIN_PER_YUAN;test_exchange_insufficient_and_invalid「低于下限」用例改 MIN_EXCHANGE_COIN-1(min=step=100 后差值为 0)

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-04 20:31:17 +08:00
OuYingJun1024 1a893d8d96 docs(admin): 补充管理后台 API 与数据库文档
新增 admin 模块接口文档(auth login/me、admins CRUD、audit-logs、feedback 处理、stats 概览、user coins/detail/status、wallet 流水、withdraw 对账/刷新/列表) + admin_user/admin_audit_log 两张表文档,更新 api/database README 索引。

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-04 15:43:07 +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
chenshuobo b3d3fda744 feat: 新增 PG 初始化脚本 scripts/init_postgres.py
新机器一键拉起数据库, 替代手动跑 psql 建用户建库的 6 行命令.

脚本特性:
- 跨平台 Python (Windows/macOS/Linux 通吃, 不依赖 psql CLI 在 PATH 里)
- 直接用 psycopg 连 PG, 不 shell out
- 幂等: 用户已存在则重置密码, 库已存在则跳过, 重复跑不出错
- 业务密码自动 secrets.token_urlsafe(32) 强随机, 不让同事自己想
- 自动从 .env.example 复制出 .env 并写入 DATABASE_URL
- .env 里已有 DATABASE_URL 会询问是否覆盖 (不无脑覆盖)
- 跑完自动 alembic upgrade head, 一条命令到可用状态
- 全交互项支持环境变量提前指定 (PG_SUPER_PASS / APP_DB_PASS 等), CI 友好

同事新机器初始化流程从 6 步简化为 3 步:
  pip install -e .
  python scripts/init_postgres.py    # 唯一手动输入项: postgres 超级用户密码
  ./run.sh

CREATE/ALTER USER 注意点: PG 不支持参数化密码绑定, 用 sql.Literal 内联,
psycopg 自带的转义会处理引号和注入.

docs/数据库迁移.md 拆出 PG / SQLite 两套初始化流程, PG 走脚本,
SQLite 保留旧步骤 (本地临时回退用).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-29 15:21:04 +08:00
chenshuobo b76e5bd515 feat: SQLite → PostgreSQL 迁移代码改动
按 docs/数据库迁移.md 第 3 节扫雷清单落地代码层改动:

- pyproject.toml: 新增 psycopg[binary]>=3.1 依赖 (psycopg3, SQLAlchemy 2.0 时代默认, 不要再装 psycopg2)

- app/db/session.py: 非 SQLite 自动加连接池参数
  - pool_size=10, max_overflow=20, pool_recycle=3600
  - SQLite 单文件不池化, _is_sqlite 判断分支保留以便本地 dev 临时回退

- app/models/savings.py: dishes 列从 JSON 改为 JSONB
  - PG 上能用 GIN 索引和 jsonb 操作符
  - SQLite 上 JSON 实际是 TEXT, 切到 PG 后用 JSON 类型存的是 json 不是 jsonb

- alembic/versions/ef96beb47b1e_*.py: 新增迁移把 savings_record.dishes 从 json 转 jsonb
  - 用 op.get_bind().dialect.name 判断 PG 才执行 ALTER COLUMN
  - SQLite 上是 no-op (SQLite 没有 jsonb 类型)
  - 旧的 savings_shop_dishes.py 迁移保持 sa.JSON() 不动 (按文档"不改老迁移"原则)

本地切 PG 验证: 10 张表全部建出, dishes 字段为 jsonb 类型,
登录/me/美团 feed 三个真接口全 200, user 数据正确写入 PG.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-29 15:20:22 +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
marco 4d444b7c43 Merge pull request 'feat(user): 用户资料与反馈接口(昵称/头像/注销 + 帮助反馈)' (#4) from feat/user-profile-feedback-api into main
Reviewed-on: #4
2026-05-29 01:04:10 +08:00
marco abd8eabf9b feat(meituan): 未配置 MT_CPS_APP_KEY 时接口降级返空,不再 502
之前未配美团 CPS 凭证时,/coupons 和 /referral-link 调用会抛 MeituanCpsError
被翻成 502,前端首页一进就拉 /feed → 看到错误(虽然 /feed 因 _fetch_topic
吞异常已经返空)。新开发机没填美团 key 时体验差。

本次让美团 3 个接口都在入口处早返空响应:
- Settings 新增 mt_cps_configured property(同 wxpay_configured 套路)
- /coupons / /feed / /referral-link 未配置时分别返空 list / 空 link
- 客户端首页 feed 看到空列表,其他业务(登录/领券/广告/提现)完全不受影响

副作用:/feed 返空现在同时表示"配置缺失"和"美团真无券"两种状态,排查时
看 settings.mt_cps_configured 区分。

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-28 15:54:51 +08:00
165 changed files with 8539 additions and 291 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,36 @@
"""convert savings_record.dishes from json to jsonb
PG only. SQLite 上 JSON/JSONB 都是 TEXT, 此迁移是 no-op。
Revision ID: ef96beb47b1e
Revises: c8d9e0f1a2b3
Create Date: 2026-05-29 10:57:21.471774
"""
from typing import Sequence, Union
from alembic import op
from sqlalchemy.dialects.postgresql import JSONB
revision: str = 'ef96beb47b1e'
down_revision: Union[str, Sequence[str], None] = 'c8d9e0f1a2b3'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
if op.get_bind().dialect.name != "postgresql":
return
op.execute(
"ALTER TABLE savings_record "
"ALTER COLUMN dishes TYPE JSONB USING dishes::JSONB"
)
def downgrade() -> None:
if op.get_bind().dialect.name != "postgresql":
return
op.execute(
"ALTER TABLE savings_record "
"ALTER COLUMN dishes TYPE JSON USING dishes::JSON"
)
@@ -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,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)
+7
View File
@@ -9,6 +9,7 @@ from concurrent.futures import ThreadPoolExecutor, as_completed
from fastapi import APIRouter, HTTPException
from app.core.config import settings
from app.integrations.meituan import MeituanCpsError, get_referral_link, query_coupon
from app.schemas.meituan import (
CouponCard,
@@ -27,6 +28,8 @@ router = APIRouter(prefix="/api/v1/meituan", tags=["meituan-cps"])
@router.post("/coupons", response_model=CouponListResponse, summary="券列表(为您推荐)")
def list_coupons(req: CouponListRequest) -> CouponListResponse:
if not settings.mt_cps_configured:
return CouponListResponse(items=[], has_next=False, search_id=None)
logger.info("[coupons] lon=%.6f lat=%.6f topic=%s", req.longitude, req.latitude, req.list_topic_id)
try:
raw = query_coupon(
@@ -78,6 +81,8 @@ def _interleave(waimai: list[dict], daodian: list[dict]) -> list[CouponCard]:
@router.post("/feed", response_model=FeedResponse, summary="混合feed(外卖+到店交叉, 无限流)")
def feed(req: FeedRequest) -> FeedResponse:
if not settings.mt_cps_configured:
return FeedResponse(items=[], has_next=False, page=req.page)
page_idx = req.page - 1
lon, lat = req.longitude, req.latitude
logger.info("[feed] page=%s lon=%.6f lat=%.6f", req.page, lon, lat)
@@ -108,6 +113,8 @@ def feed(req: FeedRequest) -> FeedResponse:
@router.post("/referral-link", response_model=ReferralLinkResponse, summary="换取推广链接(点抢时调)")
def referral_link(req: ReferralLinkRequest) -> ReferralLinkResponse:
if not settings.mt_cps_configured:
return ReferralLinkResponse(link="", link_map={})
try:
raw = get_referral_link(
product_view_sign=req.product_view_sign,
+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,
)
+1
View File
@@ -42,6 +42,7 @@ def battle(user: CurrentUser, db: DbSession) -> SavingsBattleOut:
week_saved_cents=b.week_saved_cents,
beat_percent=b.beat_percent,
streak_days=b.streak_days,
compare_count=b.compare_count,
)
+13 -5
View File
@@ -170,10 +170,11 @@ def withdraw_info(user: CurrentUser, db: DbSession) -> WithdrawInfoOut:
@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 +190,14 @@ 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
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 +226,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,
)
+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)
+29
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,13 +65,28 @@ 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 返空(优雅降级),不影响登录/领券等其他业务。
MT_CPS_APP_KEY: str = ""
MT_CPS_APP_SECRET: str = ""
MT_CPS_HOST: str = "https://media.meituan.com"
MT_CPS_TIMEOUT_SEC: int = 15
MT_CPS_DEFAULT_SID: str = "sgbjia"
@property
def mt_cps_configured(self) -> bool:
"""美团 CPS 凭证齐全(缺则接口返空,而非 502)。"""
return bool(self.MT_CPS_APP_KEY and self.MT_CPS_APP_SECRET)
# ===== 微信支付(商家转账到零钱 / 提现)=====
# 真实凭证放 .env(已 gitignore),证书 .pem 放 secrets/。WECHAT_APP_ID 同时用于
# 微信登录(code 换 openid)与转账,必须与 App 端开放平台 appid 一致。
+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/"
+102 -15
View File
@@ -34,8 +34,9 @@ COIN_PER_YUAN: int = 10000
CENTS_PER_YUAN: int = 100
# 1 分对应多少金币 = 100。兑换金币数必须是它的整数倍,否则会出现不足 1 分的零头。
COIN_PER_CENT: int = COIN_PER_YUAN // CENTS_PER_YUAN
# 单次兑换最少 1 元(避免大量 1 分级碎兑)
MIN_EXCHANGE_COIN: int = COIN_PER_YUAN
# 单次兑换最少 1 分(=COIN_PER_CENT;产品定:可兑 1 分起,与 step 同粒度)。早先为 1 元(=COIN_PER_YUAN),
# 用户 2026-06 改为 1 分:让"换现金"小额即可用(也方便联调验证资产卡飞金币动画)。
MIN_EXCHANGE_COIN: int = COIN_PER_CENT
def coins_to_cents(coin_amount: int) -> int:
@@ -53,35 +54,121 @@ WITHDRAW_MAX_CENTS: int = 5_000_000 # 5 万元
TASK_ENABLE_NOTIFICATION = "enable_notification"
# task_key -> 奖励金币
# 打开消息提醒: 1000 金币(=¥0.1, 客户端原型展示口径; 量级与签到/里程碑相称)。
# 注意: 不再 = 兑换下限(下限已降到 MIN_EXCHANGE_COIN=COIN_PER_CENT=100), 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):
+14 -8
View File
@@ -27,19 +27,25 @@ def _ensure_sqlite_dir(url: str) -> None:
_ensure_sqlite_dir(settings.DATABASE_URL)
_is_sqlite = settings.DATABASE_URL.startswith("sqlite")
# SQLite 跨线程访问要 check_same_thread=False;PG/MySQL 不需要这个参数
_connect_args: dict = {}
if settings.DATABASE_URL.startswith("sqlite"):
if _is_sqlite:
_connect_args["check_same_thread"] = False
engine = create_engine(
settings.DATABASE_URL,
connect_args=_connect_args,
_engine_kwargs: dict = {
"connect_args": _connect_args,
# echo 在 dev 下打 SQL,生产关掉
echo=settings.APP_DEBUG and not settings.is_prod,
future=True,
pool_pre_ping=True,
)
"echo": settings.APP_DEBUG and not settings.is_prod,
"future": True,
"pool_pre_ping": True,
}
# SQLite 用单文件不需要池;PG/MySQL 必须显式池化 + recycle 防 idle 断连
if not _is_sqlite:
_engine_kwargs.update(pool_size=10, max_overflow=20, pool_recycle=3600)
engine = create_engine(settings.DATABASE_URL, **_engine_kwargs)
SessionLocal = sessionmaker(bind=engine, autocommit=False, autoflush=False, expire_on_commit=False)
+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)
-18
View File
@@ -162,24 +162,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。
+8
View File
@@ -17,9 +17,13 @@ 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
@@ -74,12 +78,16 @@ 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(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, coupon_name}](price/coupon_saved 单位:元,原样存;coupon_name=优惠来源名)
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}>"
)
+28 -3
View File
@@ -8,7 +8,8 @@ from __future__ import annotations
from datetime import datetime
from sqlalchemy import JSON, 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
from app.db.base import Base
@@ -16,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(
@@ -28,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)
# 菜品名列表(JSON),前 2 道直接展示,其余收进「还有 N 道菜」展开。SQLite 存为 TEXT
dishes: Mapped[list[str]] = mapped_column(JSON, 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
)
+11 -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)
# 微信转账单号
+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())
+90 -10
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
@@ -51,6 +52,7 @@ class SavingsBattle:
week_saved_cents: int # 本周(周一起)已省
beat_percent: int # 超过百分之多少用户
streak_days: int # 连续省钱天数
compare_count: int # 累计完成比价次数(= 有效记录数:真实用 compare 记录、否则 demo 兜底)
def _local_date(dt: datetime):
@@ -65,6 +67,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 +130,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 +150,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()) # 本周一
@@ -144,7 +164,10 @@ def get_battle(db: Session, user_id: int) -> SavingsBattle:
beat_percent = _compute_beat_percent(db, user_id)
return SavingsBattle(
week_saved_cents=week_saved, beat_percent=beat_percent, streak_days=streak
week_saved_cents=week_saved,
beat_percent=beat_percent,
streak_days=streak,
compare_count=len(records),
)
@@ -155,10 +178,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 +207,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 +220,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,
+96 -31
View File
@@ -13,13 +13,8 @@ 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.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
@@ -62,6 +57,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 +141,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 +258,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 +278,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 +340,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 +367,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,32 +389,53 @@ 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 # 待管理员审核;**不在此处打款**
def execute_withdraw_transfer(db: Session, order: WithdrawOrder) -> WithdrawOrder:
"""审核通过后真正发起微信转账(复用原"转账 + 模糊失败查单"逻辑,绝不盲目退款)。
流程:reviewing → 置 pending → 调微信转账 → SUCCESS=success / 失败/取消=退款 failed /
结果不明=先查单再决定。**不抛异常**(admin 调用方按 order.status 判断结果)。
用户在审核期间解绑微信 → 退款 failed(打不了款)。
可能返回 package_info(WAIT_USER_CONFIRM 场景:需用户在 App 端确认页确认后才真正到账)。
"""
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
order.status = "pending" # 进入打款在途(转账调用前崩溃留 pending,交对账兜底)
db.commit()
# 2) 调微信转账;结果不明先查单再决定(#3)
try:
result = wxpay.create_transfer(openid, amount_cents, out_bill_no, user_name=user_name)
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)
if order.status == "failed":
raise WithdrawTransferError(str(e)) from e
return order
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
data = result["data"]
@@ -418,6 +449,40 @@ def create_withdraw(
return order
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),不在 body。seconds 服务端会夹 [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"
)
+138
View File
@@ -0,0 +1,138 @@
"""比价记录上报 / 读取 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
# 优惠**来源名**(展示用, best-effort): 美团"外卖大额神券"/京东"百亿补贴"/淘宝"平台红包"。
# None=没抠到 → 前端走通用"红包"。同样必须显式声明否则上报边界被 pydantic 静默丢弃(pricebot#38 引入)。
coupon_name: str | 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
+14 -5
View File
@@ -106,7 +106,7 @@ class WithdrawRequest(BaseModel):
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 +118,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 +134,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
@@ -194,18 +199,22 @@ class SavingsBattleOut(BaseModel):
week_saved_cents: int = Field(..., description="本周已省(分)")
beat_percent: int = Field(..., description="超过百分之多少用户")
streak_days: int = Field(..., description="连续省钱天数")
compare_count: int = Field(..., description="累计完成比价次数(= 比价上报记录数)")
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
+56 -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-06-04+ 运营后台 Admin 子应用 A1–A18,见下方「运营后台 Admin」组)
> 架构:`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,36 @@
| 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>` | 无 | 反馈截图;返回二进制图片 |
| **运营后台 Admin**(独立子应用 `app/admin/`,前缀 `/admin/api`,独立进程 + 独立 admin JWT。鉴权列:`admin`=任意已登录管理员,`operator`/`finance`/`super_admin`=需对应角色(`super_admin` 恒通过)) |||
| A1 | `POST /admin/api/auth/login` | 无 | [详情](./admin-auth-login.md) |
| A2 | `GET /admin/api/auth/me` | admin | [详情](./admin-auth-me.md) |
| A3 | `GET /admin/api/stats/overview` | admin | [详情](./admin-stats-overview.md) |
| A4 | `GET /admin/api/users` | admin | [详情](./admin-users-list.md) |
| A5 | `GET /admin/api/users/{user_id}` | admin | [详情](./admin-user-detail.md) |
| A6 | `POST /admin/api/users/{user_id}/status` | operator | [详情](./admin-user-status.md) |
| A7 | `POST /admin/api/users/{user_id}/coins` | finance | [详情](./admin-user-coins.md) |
| A8 | `GET /admin/api/wallet/coin-transactions` | admin | [详情](./admin-wallet-coin-transactions.md) |
| A9 | `GET /admin/api/wallet/cash-transactions` | admin | [详情](./admin-wallet-cash-transactions.md) |
| A10 | `GET /admin/api/withdraws` | admin | [详情](./admin-withdraws-list.md) |
| A11 | `POST /admin/api/withdraws/reconcile` | finance | [详情](./admin-withdraw-reconcile.md) |
| A12 | `POST /admin/api/withdraws/{out_bill_no}/refresh` | finance | [详情](./admin-withdraw-refresh.md) |
| A13 | `GET /admin/api/feedbacks` | admin | [详情](./admin-feedbacks-list.md) |
| A14 | `POST /admin/api/feedbacks/{feedback_id}/handle` | operator | [详情](./admin-feedback-handle.md) |
| A15 | `GET /admin/api/admins` | super_admin | [详情](./admin-admins-list.md) |
| A16 | `POST /admin/api/admins` | super_admin | [详情](./admin-admin-create.md) |
| A17 | `PATCH /admin/api/admins/{admin_id}` | super_admin | [详情](./admin-admin-update.md) |
| A18 | `GET /admin/api/audit-logs` | admin | [详情](./admin-audit-logs.md) |
| - | `GET /admin/api/health` | 无 | admin 健康检查(无单独文档) |
> ⚠️ 美团三个接口当前**无鉴权**,且 `referral-link` 的 `sid` 允许客户端传值覆盖默认渠道——见各接口"备注"。
> `coupon/step` 及外卖比价的 `intent/recognize` / `price/step` 都透传到 pricebot-backend,**MVP 阶段均不鉴权**(device_id 透传,待补 JWT——见各接口详情)。
@@ -99,9 +136,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 +172,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)。
+33
View File
@@ -0,0 +1,33 @@
# POST /admin/api/admins — 创建管理员
> 所属:Admin·Accounts 组(前缀 `/admin/api/admins` | 鉴权:Bearer admin_token(角色:super_admin | [← 返回 API 索引](./README.md)
## 入参
**application/json**:
| 字段 | 类型 | 必填 | 默认 | 说明 |
|---|---|---|---|---|
| `username` | string | ✓ | — | 账号,364 字 |
| `password` | string | ✓ | — | 初始密码,872 字(bcrypt ≤72 字节) |
| `role` | string | ✗ | `operator` | 角色,枚举:`super_admin` / `finance` / `operator` |
## 出参
响应 `200`:`AdminOut`(新建的管理员)
| 字段 | 类型 | 说明 |
|---|---|---|
| `id` | int | 管理员 id |
| `username` | string | 账号 |
| `role` | string | 角色 |
| `status` | string | 状态(新建默认 `active`) |
| `created_at` | datetime | 创建时间(UTC) |
| `last_login_at` | datetime \| null | 上次登录时间(新建为 null) |
## 错误码
- `401` 未带 admin token / token 无效或过期 / 管理员被禁用
- `403` 角色不足(仅 super_admin)
- `409` 用户名已存在
- `422` 缺字段 / `username` 长度不在 364 / `password` 长度不在 872 / `role` 非法枚举
## 说明
- 创建成功后写一条审计:`action=admin.create``target_type=admin``target_id=新管理员 id``detail={username, role}`。见 [admin_audit_log](../database/admin_audit_log.md)。
- 数据表见 [admin_user](../database/admin_user.md)。
+39
View File
@@ -0,0 +1,39 @@
# PATCH /admin/api/admins/{admin_id} — 改角色/启停/重置密码
> 所属:Admin·Accounts 组(前缀 `/admin/api/admins` | 鉴权:Bearer admin_token(角色:super_admin | [← 返回 API 索引](./README.md)
## 入参
**路径参数**:
| 字段 | 类型 | 必填 | 说明 |
|---|---|---|---|
| `admin_id` | int | ✓ | 目标管理员 id |
**application/json**(三字段都可选,只改传了的;至少传一个):
| 字段 | 类型 | 必填 | 说明 |
|---|---|---|---|
| `role` | string | ✗ | 改角色,枚举:`super_admin` / `finance` / `operator` |
| `status` | string | ✗ | 启停,枚举:`active`(启用)/ `disabled`(禁用) |
| `password` | string | ✗ | 重置密码,872 字(传则覆盖原密码) |
## 出参
响应 `200`:`AdminOut`(更新后的管理员)
| 字段 | 类型 | 说明 |
|---|---|---|
| `id` | int | 管理员 id |
| `username` | string | 账号 |
| `role` | string | 角色 |
| `status` | string | 状态 |
| `created_at` | datetime | 创建时间(UTC) |
| `last_login_at` | datetime \| null | 上次登录时间 |
## 错误码
- `400` 不能禁用自己(`admin_id == 当前 admin.id``status=disabled`) / 无任何变更字段(三字段全空)
- `401` 未带 admin token / token 无效或过期 / 管理员被禁用
- `403` 角色不足(仅 super_admin)
- `404` 管理员不存在
- `422` `role`/`status` 非法枚举 / `password` 长度不在 872
## 说明
- 更新成功后写一条审计:`action=admin.update``target_type=admin``target_id=admin_id``detail` 为本次实际变更字段(如 `{"role": "...", "status": "...", "password": "reset"}`,密码只记 `reset` 不记明文)。见 [admin_audit_log](../database/admin_audit_log.md)。
- 数据表见 [admin_user](../database/admin_user.md)。
+26
View File
@@ -0,0 +1,26 @@
# GET /admin/api/admins — 管理员列表
> 所属:Admin·Accounts 组(前缀 `/admin/api/admins` | 鉴权:Bearer admin_token(角色:super_admin | [← 返回 API 索引](./README.md)
## 入参
无(按 `id` 升序返回全部,无分页)
## 出参
响应 `200`:`AdminOut[]`
**AdminOut**
| 字段 | 类型 | 说明 |
|---|---|---|
| `id` | int | 管理员 id |
| `username` | string | 账号 |
| `role` | string | 角色:`super_admin` / `finance` / `operator` |
| `status` | string | 状态:`active` / `disabled` |
| `created_at` | datetime | 创建时间(UTC) |
| `last_login_at` | datetime \| null | 上次登录时间,从未登录为 null |
## 错误码
- `401` 未带 admin token / token 无效或过期 / 管理员被禁用
- `403` 角色不足(仅 super_admin 可访问,detail 形如 `role 'operator' not allowed (need one of [...])`)
## 说明
账号管理整组(`/admin/api/admins`)的角色守卫为 `require_role()` 无参,即仅 `super_admin` 通过。数据表见 [admin_user](../database/admin_user.md)。
+35
View File
@@ -0,0 +1,35 @@
# GET /admin/api/audit-logs — 审计日志(谁改了什么,游标分页)
> 所属:Admin·Audit 组(前缀 `/admin/api/audit-logs` | 鉴权:Bearer admin_token(角色:任意已登录 admin | [← 返回 API 索引](./README.md)
## 入参(query
| 字段 | 类型 | 必填 | 默认 | 说明 |
|---|---|---|---|---|
| `action` | string | ❌ | null | 按操作类型过滤,如 `admin.create` / `admin.update` |
| `target_type` | string | ❌ | null | 按目标对象类型过滤,如 `admin` |
| `admin_id` | int | ❌ | null | 按操作人(管理员 id)过滤 |
| `limit` | int | ❌ | 50 | 1100 |
| `cursor` | int | ❌ | null | 上一页末条 `id`,首页不传 |
## 出参
响应 `200`:`{ items: AdminAuditLogOut[], next_cursor: int|null }`(按 `id` 倒序;分页见 [索引#游标分页约定](./README.md#游标分页约定)
**AdminAuditLogOut**
| 字段 | 类型 | 说明 |
|---|---|---|
| `id` | int | 审计 id(也是游标) |
| `admin_id` | int | 操作人管理员 id |
| `admin_username` | string | 操作人账号(写入时快照) |
| `action` | string | 操作类型,如 `admin.create` / `admin.update` |
| `target_type` | string | 目标对象类型,如 `admin` |
| `target_id` | string \| null | 目标对象 id(字符串) |
| `detail` | object \| null | 操作详情(JSON,如变更字段) |
| `ip` | string \| null | 操作来源 IP(取 XFF 首段或直连 IP,仅记录不鉴权) |
| `created_at` | datetime | 操作时间(UTC) |
## 错误码
- `401` 未带 admin token / token 无效或过期 / 管理员被禁用
## 说明
- 整组(`/admin/api/audit-logs`)守卫为 `get_current_admin`,任意已登录 admin 均可查看,无角色限制。
- 审计日志只增不改不删,任何写操作经 `write_audit` 落一条。数据表见 [admin_audit_log](../database/admin_audit_log.md)。
+40
View File
@@ -0,0 +1,40 @@
# POST /admin/api/auth/login — 管理员登录
> 所属:Admin·Auth 组(前缀 `/admin/api/auth` | 鉴权:无 | [← 返回 API 索引](./README.md)
## 入参
**application/json**:
| 字段 | 类型 | 必填 | 说明 |
|---|---|---|---|
| `username` | string | ✓ | 管理员账号,164 字 |
| `password` | string | ✓ | 密码,1128 字 |
## 出参
响应 `200`:`AdminLoginResponse`
| 字段 | 类型 | 说明 |
|---|---|---|
| `access_token` | string | admin JWT(独立 `ADMIN_JWT_SECRET`、payload `typ=admin`、默认 12h、无 refresh) |
| `token_type` | string | 固定 `Bearer` |
| `expires_in` | int | access_token 剩余秒数(过期需重新登录) |
| `admin` | AdminOut | 当前管理员信息,字段见下 |
**AdminOut**
| 字段 | 类型 | 说明 |
|---|---|---|
| `id` | int | 管理员 id |
| `username` | string | 账号 |
| `role` | string | 角色:`super_admin` / `finance` / `operator` |
| `status` | string | 状态:`active` / `disabled` |
| `created_at` | datetime | 创建时间(UTC) |
| `last_login_at` | datetime \| null | 上次登录时间,从未登录为 null |
## 错误码
- `401` 用户名或密码错误(用户名不存在与密码错误统一同文案,防账号枚举)
- `403` 账号已禁用(`status != active`)
- `422``username``password` / 长度超限
- `429` 同 IP 每分钟登录超过 10 次(限流防爆破)
## 说明
- 登录成功后回写 `last_login_at` 为当前时间。
- 后续所有 admin 接口须带 `Authorization: Bearer <access_token>`;admin token 与 App 用户 token 完全隔离。

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