Compare commits

..

19 Commits

Author SHA1 Message Date
xianze fcfac7ad07 feat(report): 上报更低价后端——price_report 表 + 提交/列表接口
- 新表 price_report:关联比价记录,存原最低价快照(反查 best_*)/上报平台+价/截图URL/审核状态(pending/approved/rejected)/奖励字段
- POST /api/v1/report (multipart):反查比价记录填原最低价,校验上报价 < 原最低价(D),截图复用 core.media 落 /media/price_report/,提交一律 pending
- GET /api/v1/report/records?status= :按状态筛选 + 四态 counts,供「上报记录」页
- core.media 加 save_report_image;main 挂 report 路由;models 登记 PriceReport
- alembic 迁移仅建 price_report(autogenerate 误检的「删 order_record」已手动剔除——那是降级为审计的历史表)

奖励发放(通过→钱包+1000金币)是人工审核后台动作,本期不含。

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-05 10:22:21 +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
126 changed files with 6621 additions and 220 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')
@@ -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,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')
+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 ""
+80
View File
@@ -0,0 +1,80 @@
"""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.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)
+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)
+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,
)
+85
View File
@@ -0,0 +1,85 @@
"""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
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)
+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
+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 不走此接口)"
)
+52
View File
@@ -0,0 +1,52 @@
"""admin 钱包(金币/现金流水 + 提现单)schemas。"""
from __future__ import annotations
from datetime import datetime
from pydantic import BaseModel, ConfigDict
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
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
+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
+38 -3
View File
@@ -19,8 +19,15 @@ 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.schemas.ad import (
AdRewardStatusOut,
EcpmReportIn,
EcpmReportOut,
PangleCallbackOut,
TestGrantOut,
)
logger = logging.getLogger("shagua.ad")
@@ -84,15 +91,41 @@ 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 = 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,
)
@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 +151,7 @@ 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 = 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 +161,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")
+14 -20
View File
@@ -1,20 +1,17 @@
"""比价业务透传端点(外卖 + 电商两套)
"""外卖比价业务透传端点。
把客户端的 POST /api/v1/{intent/recognize, price/step, ecom/intent/recognize, ecom/step}
请求原样转发给 pricebot-backend 的对应路径。MVP 阶段**不鉴权**(同 coupon/step,
见 docs/待办与技术债.md P1:device_id 透传区分设备,待补 JWT + device_id↔user_id
绑定后才能做用户级画像)。
把客户端的 POST /api/v1/intent/recognize 和 /api/v1/price/step 请求原样转发给
pricebot-backend 的 /api/intent/recognize 和 /api/price/step。MVP 阶段**不鉴权**
(同 coupon/step,见 docs/待办与技术债.md P1:device_id 透传区分设备,待补 JWT +
device_id↔user_id 绑定后才能做用户级画像)。
外卖(food):
- Phase 1 /intent/recognize:从源平台(淘宝闪购/美团/京东外卖)购物车页识别店名+菜品+
价格,一次性。
- Phase 2 /price/step:多轮循环,在目标平台复现订单读到手价,直到 done。
电商(ecom):
- Phase 1 /ecom/intent/recognize:从源平台(淘宝/抖音/拼多多)SKU+详情页识别商品,一次性。
- Phase 2 /ecom/step:多轮循环,在目标平台搜索商品并采价,直到 done。
真正的目标驱动比价逻辑在 pricebot-backend(GoalEngine,另一个 repo),本接口只是透传壳。
电商(ecom)那两个端点 food MVP 暂不需要,以后放开 scene 时再加 /ecom/intent/recognize
和 /ecom/step 两行即可。
pricebot 协议文档:
pricebot-backend/docs/main/02_api_protocol.md
@@ -86,16 +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")
@router.post("/ecom/intent/recognize", summary="电商比价 Phase 1 意图识别 (透传到 pricebot)")
async def ecom_intent_recognize(request: Request) -> dict[str, Any]:
return await _passthrough(request, "/api/ecom/intent/recognize")
@router.post("/ecom/step", summary="电商比价 Phase 2 步进 (透传到 pricebot)")
async def ecom_step(request: Request) -> dict[str, Any]:
return await _passthrough(request, "/api/ecom/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,
)
+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 一致。
+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/"
+28 -7
View File
@@ -53,22 +53,43 @@ WITHDRAW_MAX_CENTS: int = 5_000_000 # 5 万元
TASK_ENABLE_NOTIFICATION = "enable_notification"
# task_key -> 奖励金币
# 打开消息提醒: 1000 金币(=¥0.1, 客户端原型展示口径; 量级与签到/里程碑相称)。
# 注意: 已不再 = 兑换下限(MIN_EXCHANGE_COIN=10000), test_exchange_flow 改走 grant_coins 直接供款。
TASK_REWARDS: dict[str, int] = {
TASK_ENABLE_NOTIFICATION: 10000,
TASK_ENABLE_NOTIFICATION: 1000,
}
# ===== 比价战绩里程碑(累计成功比价 N 次,逐档解锁领金币)=====
# 第 1→6 次的金币奖励(1-based:第 N 次比价解锁第 N 档)。值沿用客户端原型档位。
# 数据源是 comparison_record 里 status='success' 的条数;每档领一次,
# comparison_milestone_claim 去重(仿一次性任务)。要调档位/金额直接改这里。
RECORD_MILESTONES: tuple[int, ...] = (120, 180, 300, 500, 800, 1200)
RECORD_MILESTONE_COUNT: int = len(RECORD_MILESTONES)
def record_milestone_reward(milestone: int) -> int:
"""第 milestone 档(1..RECORD_MILESTONE_COUNT)的金币。越界抛 IndexError。"""
return RECORD_MILESTONES[milestone - 1]
# ===== 看激励视频发金币(穿山甲 S2S 服务端回调发奖)=====
# 看完一个激励视频发的金币(≈¥0.01,汇率 10000 金币=1 元)。
# 作用:① 回调缺/坏 reward_amount 时的回退值;② 客户端进度接口展示的"单次预告金币"
# 看完一个激励视频发的金币(666 金币 ≈¥0.0666,汇率 10000 金币=1 元)。
# 作用:① 回调缺/坏 reward_amount 时的回退值;② 客户端进度接口展示的"单次预告金币";
# ③ test-grant 本地联调的发奖额。
# 真实发放以穿山甲回调带回的 reward_amount 为准(见 resolve_ad_reward_coin),后台应把
# 代码位"奖励数量"配成与本值一致(如 100),保证"广告内展示 / 进度预告 / 实际到账"三者一致。
AD_REWARD_COIN: int = 100
# 代码位"奖励数量"配成与本值一致(=666),保证"广告内展示 / 进度预告 / 实际到账"三者一致。
AD_REWARD_COIN: int = 666
# 单次发奖金币上限:夹紧穿山甲回调里异常的 reward_amount(如后台多打一个 0),防刷爆余额。
MAX_AD_REWARD_COIN: int = 1000
# 每用户每日发奖次数上限,防刷 + 控成本。
# ⚠️ 占位值:正式上线前按产品策略 + 广告 eCPM 实测收益重新核定,改这里即可。
DAILY_AD_REWARD_LIMIT: int = 10
# ⚠️ 测试阶段值=20:正式上线前按产品策略 + 广告 eCPM 实测收益重新核定,改这里即可。
DAILY_AD_REWARD_LIMIT: int = 20
# 一轮看 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:
+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)
+5
View File
@@ -1,6 +1,11 @@
"""所有 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.admin import AdminAuditLog, AdminUser # 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}>"
)
+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}>"
)
+101
View File
@@ -0,0 +1,101 @@
"""比价记录表。
每完成一次比价(外卖/电商/领券),客户端在 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)
# 源价 - 最优价(可为 0 / 负:源平台本来就最便宜时没省到)
saved_amount_cents: Mapped[int | None] = mapped_column(Integer, nullable=True)
# 源平台就是最便宜的一家(= 这次没省到钱)
is_source_best: Mapped[bool | None] = mapped_column(Boolean, nullable=True)
# ===== 订单概要 =====
store_name: Mapped[str | None] = mapped_column(String(128), nullable=True)
total_dish_count: Mapped[int | None] = mapped_column(Integer, nullable=True)
skipped_dish_count: Mapped[int | None] = mapped_column(Integer, nullable=True)
# success(拿到有效对比)/ failed(出错或没采到目标价)
status: Mapped[str] = mapped_column(String(16), nullable=False, default="success")
# done 帧 information 文案。成功:"在美团找到同店,到手价 ¥X…";
# 失败:具体原因(如"美团、京东外卖均未找到该商品")。前端在比价失败时当原因展示。
information: Mapped[str | None] = mapped_column(String(256), nullable=True)
# ===== 明细(JSON,越详细越好)=====
# 下单菜品 [{name, qty, specs?}]
items: Mapped[list] = mapped_column(_JSON, nullable=False, default=list)
# 逐平台对比 [{platform_id, platform_name, package, price, is_source, rank, coupon_saved}](price/coupon_saved 单位:元,原样存)
comparison_results: Mapped[list] = mapped_column(_JSON, nullable=False, default=list)
# 目标平台未找到、跳过的菜名
skipped_dish_names: Mapped[list] = mapped_column(_JSON, nullable=False, default=list)
# 客户端上报的原始 payload(calibration + done.params 全量),未来取数兜底
raw_payload: Mapped[dict | None] = mapped_column(_JSON, nullable=True)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), server_default=func.now(), index=True, nullable=False
)
def __repr__(self) -> str: # pragma: no cover
return (
f"<ComparisonRecord id={self.id} user_id={self.user_id} "
f"trace_id={self.trace_id} status={self.status}>"
)
+36
View File
@@ -0,0 +1,36 @@
"""比价战绩里程碑领取记录表。
「记录比价战绩」每档(第 1~6 次)只能领一次,领取后写一行,(user_id, milestone)
唯一,防止重复领奖。解锁进度由 comparison_record 里 status='success' 的条数决定,
不存进度本身——只在这里记"哪几档已领"。仿 user_task 的一次性领取模型。
"""
from __future__ import annotations
from datetime import datetime
from sqlalchemy import DateTime, ForeignKey, Integer, UniqueConstraint, func
from sqlalchemy.orm import Mapped, mapped_column
from app.db.base import Base
class ComparisonMilestoneClaim(Base):
__tablename__ = "comparison_milestone_claim"
__table_args__ = (
UniqueConstraint("user_id", "milestone", name="uq_compare_milestone_user"),
)
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
user_id: Mapped[int] = mapped_column(
Integer, ForeignKey("user.id"), index=True, nullable=False
)
# 档位序号(1-based),见 app.core.rewards.RECORD_MILESTONES
milestone: Mapped[int] = mapped_column(Integer, nullable=False)
coin_awarded: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
claimed_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), server_default=func.now(), nullable=False
)
def __repr__(self) -> str: # pragma: no cover
return f"<ComparisonMilestoneClaim user_id={self.user_id} m={self.milestone}>"
+67
View File
@@ -0,0 +1,67 @@
"""上报更低价记录表(price_report)。
用户在「比价记录」里选一条记录,上报「某平台有比我们算出的最低价更低的价格」,带截图证明,
人工审核通过奖励金币。与 feedback(自由文本反馈)不同:本表结构化——关联具体比价记录、
原最低价 vs 上报价、审核状态、奖励。
- 原最低价快照(original_*)由后端按 comparison_record_id 反查比价记录的 best_* 冗余填入,
避免被关联记录后续删/改后对不上。
- 价格统一存「分」(cents),与 comparison_record / savings_record 一致。
- status: pending(审核中)/ approved(已通过)/ rejected(未通过),对齐前端筛选与状态徽标。
- images: 截图相对 URL 列表(/media/price_report/...,复用 app.core.media)。
- 奖励发放(通过 → 钱包 +reward_coins)是人工审核后台动作,提交时一律 pending,本表只留字段。
"""
from __future__ import annotations
from datetime import datetime
from sqlalchemy import JSON, DateTime, ForeignKey, Integer, String, func
from sqlalchemy.orm import Mapped, mapped_column
from app.db.base import Base
class PriceReport(Base):
__tablename__ = "price_report"
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
user_id: Mapped[int] = mapped_column(
Integer, ForeignKey("user.id"), index=True, nullable=False
)
# 选中的那条比价记录(可空:记录被删后仍保留上报历史)
comparison_record_id: Mapped[int | None] = mapped_column(
Integer, ForeignKey("comparison_record.id"), index=True, nullable=True
)
# ===== 选中比价记录的快照(后端反查 comparison_record 填入,冗余存) =====
store_name: Mapped[str | None] = mapped_column(String(128), nullable=True)
dish_summary: Mapped[str | None] = mapped_column(String(256), nullable=True)
original_platform_id: Mapped[str | None] = mapped_column(String(32), nullable=True)
original_platform_name: Mapped[str | None] = mapped_column(String(32), nullable=True)
original_price_cents: Mapped[int | None] = mapped_column(Integer, nullable=True)
# ===== 用户上报的更低价 =====
reported_platform_id: Mapped[str] = mapped_column(String(32), nullable=False)
reported_platform_name: Mapped[str] = mapped_column(String(32), nullable=False)
reported_price_cents: Mapped[int] = mapped_column(Integer, nullable=False)
# 截图证明 URL 列表(相对路径,如 ["/media/price_report/u1_ab.jpg"])
images: Mapped[list[str]] = mapped_column(JSON, nullable=False, default=list)
# ===== 审核(人工后台) =====
status: Mapped[str] = mapped_column(
String(16), nullable=False, default="pending", index=True
)
reject_reason: Mapped[str | None] = mapped_column(String(256), nullable=True)
reward_coins: Mapped[int | None] = mapped_column(Integer, nullable=True)
reviewed_at: Mapped[datetime | None] = mapped_column(
DateTime(timezone=True), nullable=True
)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), server_default=func.now(), index=True, nullable=False
)
def __repr__(self) -> str: # pragma: no cover
return (
f"<PriceReport id={self.id} user_id={self.user_id} status={self.status}>"
)
+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
)
+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()
+43 -6
View File
@@ -10,11 +10,18 @@
"""
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.core.ad_cooldown import compute_cooldown
from app.core.rewards import (
AD_REWARD_COIN,
DAILY_AD_REWARD_LIMIT,
cn_today,
)
from app.repositories import wallet as crud_wallet
from app.models.ad_reward import AdRewardRecord
from app.models.user import User
@@ -78,7 +85,7 @@ def grant_ad_reward(
# 发金币 + 记一笔,同事务
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 +109,37 @@ 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]:
"""客户端查"今日看广告发奖"进度。
返回 (今日已发次数, 每日上限, 单次金币, 本轮已看次数, 本轮冷却结束时间(UTC))。
本函数只**取数据**(今日 granted 的 created_at 倒序),把"本轮已看几次 + 冷却到几点"
**策略判断**委托给 [app.core.ad_cooldown.compute_cooldown](纯函数)——换冷却策略只动那个模块。
"""
today = cn_today().isoformat()
granted_desc = _granted_times_today_desc(db, user_id, today)
state = compute_cooldown(granted_desc, datetime.now(timezone.utc))
return (
len(granted_desc),
DAILY_AD_REWARD_LIMIT,
AD_REWARD_COIN,
state.round_count,
state.cooldown_until,
)
+147
View File
@@ -0,0 +1,147 @@
"""比价记录 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.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,
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 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.id.desc()).limit(limit)
items = list(db.execute(stmt).scalars().all())
next_cursor = items[-1].id if len(items) == limit else None
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()
+115
View File
@@ -0,0 +1,115 @@
"""比价战绩里程碑 CRUD:进度查询 + 逐档领奖。
进度 = comparison_record 里 status='success' 的条数(crud_compare.count_success)。
第 N 档在"成功次数 >= N"时解锁;每档领一次,写 comparison_milestone_claim 去重(标记已领)。
⚠️ 当前产品定暂不真发金币(后续整体删除该功能):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.rewards import (
RECORD_MILESTONE_COUNT,
RECORD_MILESTONES,
)
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..RECORD_MILESTONE_COUNT)。"""
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: list[MilestoneState] = []
claimable = 0
for m in range(1, RECORD_MILESTONE_COUNT + 1):
if m in claimed:
state = "claimed"
elif success_count >= m:
state = "active"
claimable += 1
else:
state = "locked"
milestones.append(
MilestoneState(milestone=m, coin=RECORD_MILESTONES[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 > RECORD_MILESTONE_COUNT:
raise UnknownMilestoneError
existing = db.execute(
select(ComparisonMilestoneClaim).where(
ComparisonMilestoneClaim.user_id == user_id,
ComparisonMilestoneClaim.milestone == milestone,
)
).scalar_one_or_none()
if existing is not None:
raise AlreadyClaimedError
# 解锁校验:成功比价次数必须 >= 档位序号
if crud_compare.count_success(db, user_id) < milestone:
raise MilestoneLockedError
# ⚠️ 比价战绩里程碑暂不真发金币(产品定,后续整体删除该功能):仍记一条领取(去重/标记已领),
# 但不调 grant_coins、不写 coin_transaction,余额不变,返回发放 0 金币 + 当前余额。
db.add(
ComparisonMilestoneClaim(
user_id=user_id, milestone=milestone, coin_awarded=0
)
)
acc = crud_wallet.get_or_create_account(db, user_id, commit=False)
db.commit()
return 0, acc.coin_balance
+53
View File
@@ -0,0 +1,53 @@
"""price_report 表读写。"""
from __future__ import annotations
from sqlalchemy import select
from sqlalchemy.orm import Session
from app.models.price_report import PriceReport
def create_report(
db: Session,
*,
user_id: int,
comparison_record_id: int | None,
store_name: str | None,
dish_summary: str | None,
original_platform_id: str | None,
original_platform_name: str | None,
original_price_cents: int | None,
reported_platform_id: str,
reported_platform_name: str,
reported_price_cents: int,
images: list[str],
) -> PriceReport:
rep = PriceReport(
user_id=user_id,
comparison_record_id=comparison_record_id,
store_name=store_name,
dish_summary=dish_summary,
original_platform_id=original_platform_id,
original_platform_name=original_platform_name,
original_price_cents=original_price_cents,
reported_platform_id=reported_platform_id,
reported_platform_name=reported_platform_name,
reported_price_cents=reported_price_cents,
images=images,
status="pending",
)
db.add(rep)
db.commit()
db.refresh(rep)
return rep
def list_reports(
db: Session, user_id: int, status: str | None = None
) -> list[PriceReport]:
"""该用户的上报记录,按时间倒序;status 非空时按状态筛选。"""
stmt = select(PriceReport).where(PriceReport.user_id == user_id)
if status:
stmt = stmt.where(PriceReport.status == status)
stmt = stmt.order_by(PriceReport.created_at.desc(), PriceReport.id.desc())
return list(db.execute(stmt).scalars().all())
+85 -9
View File
@@ -17,6 +17,7 @@ from sqlalchemy.orm import Session
from app.core.rewards import CN_TZ, cn_today
from app.models.savings import SavingsRecord
from app.schemas.order import OrderReportRequest
# 演示数据规模:最近 N 天每天 1 单(撑起"连续省钱"和"本周"),再补若干历史单
_DEMO_STREAK_DAYS = 9
@@ -65,6 +66,26 @@ def _all_records(db: Session, user_id: int) -> list[SavingsRecord]:
return list(db.execute(stmt).scalars().all())
def _has_real(db: Session, user_id: int) -> bool:
"""该用户是否已有真实比价上报(source='compare')记录。"""
return db.execute(
select(SavingsRecord.id)
.where(SavingsRecord.user_id == user_id, SavingsRecord.source == "compare")
.limit(1)
).first() is not None
def _effective_records(db: Session, user_id: int) -> list[SavingsRecord]:
"""统计口径:有真实(compare)记录就只用真实的;否则 seed 一批 demo 并用 demo 兜底。"""
if _has_real(db, user_id):
stmt = select(SavingsRecord).where(
SavingsRecord.user_id == user_id, SavingsRecord.source == "compare"
)
return list(db.execute(stmt).scalars().all())
ensure_seeded(db, user_id)
return _all_records(db, user_id)
def ensure_seeded(db: Session, user_id: int) -> None:
"""该用户没有任何省钱记录时,幂等灌一批 demo 数据。"""
exists = db.execute(
@@ -108,8 +129,7 @@ def ensure_seeded(db: Session, user_id: int) -> None:
def get_summary(db: Session, user_id: int) -> SavingsSummary:
ensure_seeded(db, user_id)
records = _all_records(db, user_id)
records = _effective_records(db, user_id)
total = sum(r.saved_amount_cents for r in records)
count = len(records)
avg = total // count if count else 0
@@ -129,8 +149,7 @@ def _streak_days(dates: set) -> int:
def get_battle(db: Session, user_id: int) -> SavingsBattle:
ensure_seeded(db, user_id)
records = _all_records(db, user_id)
records = _effective_records(db, user_id)
today = cn_today()
week_start = today - timedelta(days=today.weekday()) # 本周一
@@ -155,10 +174,20 @@ def _compute_beat_percent(db: Session, user_id: int) -> int:
全部省得比我少 → 100;只有自己一个用户 → 0(无可比)。
"""
rows = db.execute(
select(SavingsRecord.user_id, func.sum(SavingsRecord.saved_amount_cents))
.group_by(SavingsRecord.user_id)
select(
SavingsRecord.user_id,
SavingsRecord.source,
func.sum(SavingsRecord.saved_amount_cents),
).group_by(SavingsRecord.user_id, SavingsRecord.source)
).all()
totals = {uid: (total or 0) for uid, total in rows}
# 每用户口径与展示一致:有 compare 用 compare 之和,否则退回 demo 之和
by_user: dict[int, dict[str, int]] = {}
for uid, source, total in rows:
by_user.setdefault(uid, {})[source] = total or 0
totals = {
uid: (d["compare"] if "compare" in d else d.get("demo", 0))
for uid, d in by_user.items()
}
others = {uid: t for uid, t in totals.items() if uid != user_id}
if not others:
return 0
@@ -174,9 +203,12 @@ def list_records(
limit: int = 20,
cursor: int | None = None,
) -> tuple[list[SavingsRecord], int | None]:
"""省钱明细分页(按 id 倒序,游标式)。"""
ensure_seeded(db, user_id)
"""省钱明细分页(按 id 倒序,游标式)。有真实(compare)记录只列真实,否则 demo 兜底。"""
stmt = select(SavingsRecord).where(SavingsRecord.user_id == user_id)
if _has_real(db, user_id):
stmt = stmt.where(SavingsRecord.source == "compare")
else:
ensure_seeded(db, user_id)
if cursor is not None:
stmt = stmt.where(SavingsRecord.id < cursor)
stmt = stmt.order_by(SavingsRecord.id.desc()).limit(limit)
@@ -184,3 +216,47 @@ def list_records(
items = list(db.execute(stmt).scalars().all())
next_cursor = items[-1].id if len(items) == limit else None
return items, next_cursor
def create_from_report(
db: Session, user_id: int, req: OrderReportRequest
) -> tuple[SavingsRecord, bool]:
"""真实比价上报 → 写入一条 savings_record(source='compare')。返回 (记录, 是否重复上报)。
省额 = 源平台原价 − 实付(下限 0);原价缺失则记 0。
幂等:同 (user_id, client_event_id) 已存在则直接返回旧记录,duplicated=True,不新增。
"""
existing = db.execute(
select(SavingsRecord).where(
SavingsRecord.user_id == user_id,
SavingsRecord.client_event_id == req.client_event_id,
)
).scalar_one_or_none()
if existing is not None:
return existing, True
paid = req.paid_amount_cents
original = req.original_price_cents
saved = max(0, original - paid) if original is not None else 0
rec = SavingsRecord(
user_id=user_id,
order_amount_cents=paid,
saved_amount_cents=saved,
original_price_cents=original,
compared_price_cents=req.compared_price_cents,
platform=req.platform,
title=req.shop_name, # 明细卡标题用门店名
shop_name=req.shop_name,
dishes=req.dishes or [],
pay_channel=req.pay_channel,
platform_package=req.platform_package,
source_platform_name=req.source_platform_name,
source_deeplink=req.source_deeplink,
client_event_id=req.client_event_id,
device_id=req.device_id,
source="compare",
)
db.add(rec)
db.commit()
db.refresh(rec)
return rec, False
+2 -1
View File
@@ -119,7 +119,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)
+32
View File
@@ -4,6 +4,8 @@
"""
from __future__ import annotations
from datetime import datetime
from pydantic import BaseModel, Field
@@ -27,6 +29,32 @@ 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 倒计时且不可点",
)
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 TestGrantOut(BaseModel):
@@ -39,3 +67,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"
)
+129
View File
@@ -0,0 +1,129 @@
"""比价记录上报 / 读取 schemas。
约定同 welfare:字段 snake_case、金额存整数(分)、时间 ISO 8601。
上报请求(ComparisonRecordIn)的字段刻意对齐 pricebot 协议
(docs/main/02_api_protocol.md 的 calibration + done.params.comparison_results),
让客户端把 Phase1 的 calibration 和 done 帧的 params 字段**零翻译**直接映射上来,
server 端负责拆成结构化列(best_*/saved/is_source_best 由 comparison_results 派生)。
"""
from __future__ import annotations
from datetime import datetime
from pydantic import BaseModel, ConfigDict, Field
# ===== 上报请求 =====
class ComparisonItemIn(BaseModel):
"""下单菜品(来自 calibration.items)。"""
name: str
qty: int = 1
specs: list[str] | None = None
class ComparisonResultIn(BaseModel):
"""逐平台对比项(来自 done.params.comparison_results)。price 单位:元。"""
platform_id: str | None = None
platform_name: str | None = None
package: str | None = None
price: float | None = None
is_source: bool = False
rank: int | None = None
# 该平台本单用红包省的**纯红包优惠额**(元, 不含配送费/代金券/满减)。None=没用/没抠到。
# 必须显式声明: 落库走 model_dump(), pydantic 默认丢未知字段, 不声明这行会被悄悄吞掉。
# 各平台抠到红包即带值(2026-06 起源平台 Phase1 意图识别也抠, 当前仅淘宝源)。见 pricebot 侧 比价红包额留痕-实现方案.md。
coupon_saved: float | None = None
class ComparisonRecordIn(BaseModel):
# 幂等键:同一用户同一 trace_id 重复上报只保留一条(覆盖)
trace_id: str = Field(..., min_length=1, description="pricebot 侧 trace_id")
business_type: str = Field("food", description="food / ecom / coupon")
device_id: str | None = None
# 源平台(来自 calibration)
store_name: str | None = Field(None, description="店铺名(外卖)")
source_platform_id: str | None = None
source_platform_name: str | None = None
source_package: str | None = None
source_price: float | None = Field(None, description="源平台到手价(元)")
# 明细
items: list[ComparisonItemIn] = Field(default_factory=list)
comparison_results: list[ComparisonResultIn] = Field(default_factory=list)
skipped_dish_count: int | None = None
skipped_dish_names: list[str] = Field(default_factory=list)
total_dish_count: int | None = None
information: str | None = Field(None, description="done 帧文案,留存备查")
# 不传则服务端按 comparison_results 派生(有非源有效价=success,否则 failed)
status: str | None = Field(None, description="success / failed,可不传由服务端派生")
# ===== 读取出参 =====
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
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 = []
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
+5 -2
View File
@@ -200,12 +200,15 @@ class SavingsRecordOut(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: int
order_amount_cents: int
saved_amount_cents: int
order_amount_cents: int # 实付金额(分)
saved_amount_cents: int # 省下(分)
original_price_cents: int | None = None # 源平台原价(分);demo 行为空
platform: str | None = None
title: str | None = None
shop_name: str | None = None
dishes: list[str] = []
pay_channel: str | None = None # wechat/alipay;demo 行为空
source_platform_name: str | None = None # 源平台名,如 美团
created_at: datetime
+23
View File
@@ -0,0 +1,23 @@
[Unit]
Description=Shaguabijia Admin Backend (FastAPI / uvicorn, 独立进程,复用 app-server codebase)
After=network.target
[Service]
Type=simple
User=root
WorkingDirectory=/opt/shaguabijia-app-server
Environment="PATH=/opt/shaguabijia-app-server/.venv/bin:/usr/bin:/bin"
EnvironmentFile=/opt/shaguabijia-app-server/.env
# 与主 App 后端(8770)复用同一 codebase / .env / .venv / DB,独立进程跑 admin app
ExecStart=/opt/shaguabijia-app-server/.venv/bin/uvicorn app.admin.main:admin_app --host 127.0.0.1 --port 8771 --workers 1 --log-level info
Restart=on-failure
RestartSec=3
NoNewPrivileges=true
PrivateTmp=true
ProtectSystem=strict
ReadWritePaths=/opt/shaguabijia-app-server
ProtectHome=true
[Install]
WantedBy=multi-user.target
+36 -4
View File
@@ -3,7 +3,7 @@
> Base URL:生产 `https://app-api.shaguabijia.com`;本地联调 `http://<开发机>:8770`
> 协议:HTTP / JSON,请求与响应体均 `application/json`,字段统一 **snake_case**
> 鉴权:需鉴权的接口在请求头带 `Authorization: Bearer <access_token>`
> 最后更新:2026-05-27
> 最后更新:2026-05-31+ 比价战绩里程碑 12d/12e
> 架构:`app/api/v1/` 只放很轻的接口层;穿山甲/微信支付/极光/短信/美团等 SDK 集成的重逻辑在 `app/integrations/`,实现细节见 [docs/integrations/](../integrations/README.md)。
---
@@ -26,6 +26,13 @@
| **比价透传**(前缀 `/api/v1`,外卖 MVP;与 `coupon/step` 同为透传 pricebot-backend |||
| 12 | `POST /api/v1/intent/recognize` | 无 | [详情](./compare-intent-recognize.md) |
| 13 | `POST /api/v1/price/step` | 无 | [详情](./compare-price-step.md) |
| **比价记录**(前缀 `/api/v1/compare`;按用户落库,**鉴权**,区别于上面不鉴权的透传) |||
| 12a | `POST /api/v1/compare/record` | Bearer | [详情](./compare-record-report.md) |
| 12b | `GET /api/v1/compare/records` | Bearer | [详情](./compare-records.md) |
| 12c | `GET /api/v1/compare/records/{id}` | Bearer | [详情](./compare-record-detail.md) |
| **比价战绩里程碑**(前缀 `/api/v1/compare`;福利页「记录比价战绩」,按成功比价数解锁逐档发金币) |||
| 12d | `GET /api/v1/compare/milestones` | Bearer | [详情](./compare-milestones.md) |
| 12e | `POST /api/v1/compare/milestones/{milestone}/claim` | Bearer | [详情](./compare-milestone-claim.md) |
| **钱包 / 我的资产**(前缀 `/api/v1/wallet` |||
| 14 | `GET /api/v1/wallet/account` | Bearer | [详情](./wallet-account.md) |
| 15 | `GET /api/v1/wallet/coin-transactions` | Bearer | [详情](./wallet-coin-transactions.md) |
@@ -52,6 +59,16 @@
| 32 | `GET /api/v1/ad/pangle-callback` | 验签 | [详情](./ad-pangle-callback.md) |
| 33 | `GET /api/v1/ad/reward-status` | Bearer | [详情](./ad-reward-status.md) |
| 34 | `POST /api/v1/ad/test-grant` | Bearer | [详情](./ad-test-grant.md) |
| 35 | `POST /api/v1/ad/ecpm-report` | Bearer | [详情](./ad-ecpm-report.md) |
| **用户资料**(前缀 `/api/v1/user` |||
| 35 | `PATCH /api/v1/user/profile` | Bearer | [详情](./user-profile.md) |
| 36 | `POST /api/v1/user/avatar` | Bearer | [详情](./user-avatar.md) |
| 37 | `DELETE /api/v1/user` | Bearer | [详情](./user-delete.md) |
| **帮助与反馈**(前缀 `/api/v1/feedback` |||
| 38 | `POST /api/v1/feedback` | Bearer | [详情](./feedback.md) |
| **静态资源**StaticFiles 挂载,见下方 `/media` 静态服务) |||
| - | `GET /media/avatars/<file>` | 无 | 用户头像;返回二进制图片 |
| - | `GET /media/feedback/<file>` | 无 | 反馈截图;返回二进制图片 |
> ⚠️ 美团三个接口当前**无鉴权**,且 `referral-link` 的 `sid` 允许客户端传值覆盖默认渠道——见各接口"备注"。
> `coupon/step` 及外卖比价的 `intent/recognize` / `price/step` 都透传到 pricebot-backend,**MVP 阶段均不鉴权**(device_id 透传,待补 JWT——见各接口详情)。
@@ -99,9 +116,9 @@
| 字段 | 类型 | 说明 |
|---|---|---|
| `id` | int | 用户主键 |
| `phone` | string | 手机号 |
| `nickname` | string \| null | 昵称(当前无接口可改,恒为 null |
| `avatar_url` | string \| null | 头像(同上) |
| `phone` | string | 手机号(注销账号后变 `deleted_<id>` 占位释放唯一约束) |
| `nickname` | string \| null | 昵称,经 [`PATCH /api/v1/user/profile`](./user-profile.md) 修改 |
| `avatar_url` | string \| null | 头像相对 URL(`/media/avatars/...`),经 [`POST /api/v1/user/avatar`](./user-avatar.md) 上传 |
| `register_channel` | string | 注册渠道:`jverify` / `sms` |
| `status` | string | `active` / `disabled` / `deleted` |
| `created_at` | datetime | 注册时间 |
@@ -135,6 +152,21 @@
---
## /media 静态服务
用户上传文件(头像/反馈截图)落盘到 `settings.MEDIA_ROOT`(默认 `./data/media/`),由 FastAPI 的 `StaticFiles` 挂在 `settings.MEDIA_URL_PREFIX`(默认 `/media`)对外暴露:
- `GET /media/avatars/<file>` — 用户头像(JPEG/PNG/WebP)
- `GET /media/feedback/<file>` — 反馈截图(同上)
**生产建议**:由 nginx 直接 serve `MEDIA_ROOT` 目录,绕过应用进程减少压力。
**URL 格式**:服务端返回**相对路径**(如 `/media/avatars/u1_a4f2b3c8e9d2e0a1.jpg`),客户端按自己的 `BASE_URL` 拼绝对地址——dev 下可能是 `10.0.2.2`/LAN IP/`127.0.0.1`,服务端不知道客户端怎么访问到自己。
**文件名服务端随机生成** `u<user_id>_<16 位 hex>.<ext>`,杜绝路径穿越与覆盖。
---
## 附:鉴权与刷新机制
- **签发**:登录成功后签发 access(HS256,2h) + refresh(30d),payload 含 `sub`(user_id)、`typ`(access/refresh)、`iat``exp`
+36
View File
@@ -0,0 +1,36 @@
# POST /api/v1/ad/ecpm-report — 上报本次广告展示的 eCPM(内部收益统计)
> 所属:Ad 组(前缀 `/api/v1/ad` | 鉴权:Bearer | [← 返回 API 索引](./README.md)
## 入参
请求体:`EcpmReportIn`
| 字段 | 类型 | 必填 | 说明 |
|---|---|---|---|
| `ad_type` | str | 是 | 广告类型:`reward_video`(激励视频) / `draw`(Draw 信息流) 等 |
| `ecpm` | str | 是 | 穿山甲 `getShowEcpm().getEcpm()` 原始字符串,**单位待确认(分/元)**,原样上报 |
| `adn` | str\|null | 否 | 实际投放 ADN(`getSdkName`),如 `pangle` |
| `slot_id` | str\|null | 否 | 实际展示代码位(底层 mediation rit,非客户端配置位) |
`user_id` 不在 body 里——由 JWT 取(Bearer),防伪造。
## 出参
响应 `200`:`EcpmReportOut`
| 字段 | 类型 | 说明 |
|---|---|---|
| `ok` | bool | 落库即 `true` |
## 说明
客户端在广告**展示后**(`onAdShow``getShowEcpm()`)调用,把本次展示的 eCPM 落库做**内部收益统计/对账**。
- **best-effort**:客户端 fire-and-forget,丢一两条不影响业务;穿山甲后台报表是结算权威兜底。
-`ad_ecpm_record` 表,`report_date` 用北京时间当天,供「按用户/按天聚合」对账。
- **与发奖是两条独立流**:发奖走 [ad-pangle-callback](./ad-pangle-callback.md)(穿山甲 S2S,有 `trans_id`、无 ecpm),本接口客户端上报(有 ecpm、无 `trans_id`)。两者无公共键,**不逐条一一对应**,只做按用户/按天聚合口径。
- eCPM 是**每千次展示预估**,单条展示预估收益 ≈ `ecpm / 1000`,且为客户端预估口径,非最终结算。
- ⚠️ eCPM **单位(分/元)截至 2026-05-31 未最终确认**,故 `ecpm_raw` 原样存字符串;确认后再加一列解析好的数值。
- 限流:同 IP 120 次/分钟。
## 相关
- [ad-pangle-callback](./ad-pangle-callback.md) — 穿山甲 S2S 发奖回调
- [ad-reward-status](./ad-reward-status.md) — 今日看广告发奖进度
+10 -1
View File
@@ -14,6 +14,15 @@
| `daily_limit` | int | 每日发奖次数上限 |
| `remaining` | int | 今日剩余可领次数 |
| `coin_per_ad` | int | 看完一个激励视频发的金币 |
| `round_count` | int | 本轮(3 次一组)已看次数,0..N-1。客户端可据「拿到 round_count==0 && cooldown_until!=null」判定刚刚看完一轮 |
| `cooldown_until` | datetime\|null | 本轮 10 分钟冷却结束时间(UTC ISO);null 表示不在冷却 |
## 说明
福利页「看视频赚金币」用:展示「今日还能看 N 次」「看一次得 M 金币」。真正发奖走 [ad-pangle-callback](./ad-pangle-callback.md)(穿山甲 S2S)。
福利页「看视频赚金币」用:
- 展示「今日还能看 N 次」「看一次得 M 金币」(`remaining`/`coin_per_ad`)
- 任务行 CTA 4 态推导:`adLoading` → Loading;`remaining==0` → Capped("明天再来");`cooldown_until` 在未来 → CoolingDown(显示 MM:SS 倒计时,不可点);否则 Normal("去赚取")
- 弹窗 limit note:`used_today >= daily_limit` 显示「今日视频已到限额,明天再来」;`round_count==0 && cooldown_until!=null` 显示「本轮视频已看完,10分钟后再来」
冷却派生算法:取当日 `status='granted'` 记录里**最近一个已完成轮**(每 N=3 条算一轮)末尾那次的 `created_at`,加 10 分钟。若仍 > now → 返回;否则返回 null。**冷却是 UX 约束**,后端发奖(`/api/v1/ad/pangle-callback`)只看 daily_limit,冷却期间穿山甲回调照常发奖。
真正发奖走 [ad-pangle-callback](./ad-pangle-callback.md)(穿山甲 S2S)。
+34
View File
@@ -0,0 +1,34 @@
# POST /api/v1/compare/milestones/{milestone}/claim — 领取比价战绩里程碑奖励
> 所属:比价记录组(前缀 `/api/v1/compare` | 鉴权:Bearer | [← 返回 API 索引](./README.md)
领取某一档(第 `milestone` 次)。⚠️ **当前不真发金币**(产品定,后续整体删除该功能):仍写
`comparison_milestone_claim`((user_id, milestone) 唯一)标记该档已领、**每档只能领一次**,但不调
`grant_coins`、不写 `coin_transaction`,余额不变,`coin_awarded` 恒为 0。进度口径见
[milestones 进度接口](./compare-milestones.md)。
## 入参
- 路径参数 `milestone`(int):档位序号,1..6(= `RECORD_MILESTONES` 长度)。
## 出参
响应 `200`
| 字段 | 类型 | 说明 |
|---|---|---|
| `milestone` | int | 本次领取的档位序号 |
| `coin_awarded` | int | 本次发放金币(当前恒为 `0`,暂不真发) |
| `coin_balance` | int | 领奖后金币余额(当前不变) |
```json
{ "milestone": 1, "coin_awarded": 0, "coin_balance": 29087 }
```
## 错误
- `401` 未鉴权
- `404` 档位越界(`unknown milestone`,milestone < 1 或 > 档位总数)
- `409` 该档还没解锁(`milestone locked`,成功比价次数 < milestone)
- `409` 该档已领过(`milestone already claimed`)
## 说明
- 当前不发金币,客户端领取后只需把本档状态刷成 claimed 即可;若后续恢复发奖,再按 `coin_awarded` 刷新钱包余额([`GET /api/v1/wallet/account`](./wallet-account.md))。
- 幂等:重复领同一档返回 409(唯一约束 + 领取前查重)。
+53
View File
@@ -0,0 +1,53 @@
# GET /api/v1/compare/milestones — 比价战绩里程碑进度
> 所属:比价记录组(前缀 `/api/v1/compare` | 鉴权:Bearer | [← 返回 API 索引](./README.md)
福利页「记录比价战绩」的数据源。返回各档(第 1~6 次)解锁/领取状态。
**解锁进度** = 当前用户 `comparison_record``status='success'` 的条数(只算成功比价,失败的不计入)。第 N 档在「成功比价次数 ≥ N」时解锁;每档领一次(领取见 [claim 接口](./compare-milestone-claim.md))。
档位金额是**产品规则**,定义在后端 `app/core/rewards.py``RECORD_MILESTONES`(当前 `120/180/300/500/800/1200`),客户端**不要写死**,以本接口返回为准。
> ⚠️ **当前领取暂不真发金币**(产品定,后续整体删除该功能):`coin` 仍返回产品规则值仅供展示,但 [claim 接口](./compare-milestone-claim.md) 实际 `coin_awarded` 恒为 0、余额不变。前端展示须与之对齐,勿让用户误以为领取可到账。
## 入参
无(用户身份取自 Bearer token)。
## 出参
响应 `200`
| 字段 | 类型 | 说明 |
|---|---|---|
| `success_count` | int | 累计成功比价次数(解锁进度) |
| `claimable_count` | int | 当前可领(state=active)的档数 |
| `milestones` | Milestone[] | 各档状态,按 milestone 升序 |
**Milestone**
| 字段 | 类型 | 说明 |
|---|---|---|
| `milestone` | int | 档位序号(1-based),= 解锁所需的成功比价次数 |
| `coin` | int | 该档**应发**金币额(产品规则值);⚠️ 当前领取不真发,见上方说明 |
| `state` | string | `claimed`(已领) / `active`(已解锁可领) / `locked`(未解锁) |
```json
{
"success_count": 2,
"claimable_count": 1,
"milestones": [
{"milestone": 1, "coin": 120, "state": "claimed"},
{"milestone": 2, "coin": 180, "state": "active"},
{"milestone": 3, "coin": 300, "state": "locked"},
{"milestone": 4, "coin": 500, "state": "locked"},
{"milestone": 5, "coin": 800, "state": "locked"},
{"milestone": 6, "coin": 1200, "state": "locked"}
]
}
```
## 错误
- `401` 未鉴权
## 说明
- 同时可有多档处于 `active`(如累计 3 次却一档没领,则前 3 档都可领),逐档调 claim。
- 进度只增不减:领取不消耗成功次数,只是把对应档从 active→claimed。
+25
View File
@@ -0,0 +1,25 @@
# GET /api/v1/compare/records/{record_id} — 比价记录详情
> 所属:比价记录组(前缀 `/api/v1/compare` | 鉴权:Bearer | [← 返回 API 索引](./README.md)
单条比价记录详情,在列表项基础上额外带 `raw_payload`(客户端上报的原始全量),供未来 UI 展示任意细节。
## 入参(path
| 字段 | 类型 | 说明 |
|---|---|---|
| `record_id` | int | 记录 id |
## 出参
响应 `200`**ComparisonRecordDetailOut** = [ComparisonRecordOut](./compare-records.md#出参) 全部字段 + 下列:
| 字段 | 类型 | 说明 |
|---|---|---|
| `raw_payload` | object \| null | 客户端上报的原始 body 全量(calibration + done.params |
## 错误
- `401` 未鉴权
- `404` 记录不存在,或不属于当前用户(限本人,避免越权读他人记录)
## 说明
`404` 同时覆盖「id 不存在」和「id 属于他人」两种情况——不区分以免泄露他人记录是否存在。
+59
View File
@@ -0,0 +1,59 @@
# POST /api/v1/compare/record — 上报一次比价结果(幂等)
> 所属:比价记录组(前缀 `/api/v1/compare` | 鉴权:Bearer | [← 返回 API 索引](./README.md)
比价 `done` 帧后,客户端用**带 JWT 的通道**上报一条比价结果,落 `comparison_record` 表,作为「我的比价记录」数据源 + 用户级行为画像。
> ⚠️ 与不鉴权的透传端点 [`/api/v1/price/step`](./compare-price-step.md) 不同:那是转发壳,本接口按用户维度落库,**必须鉴权**。
> 本轮只做 server 端;客户端在 done 帧后调本接口的改动另起一轮(见 [待办与技术债.md](../待办与技术债.md) P1)。
## 入参(JSON body
字段刻意对齐 pricebot 协议([02_api_protocol.md](../../../pricebot-backend/docs/main/02_api_protocol.md) 的 `calibration` + `done.params`),客户端把 Phase1 的 calibration 和 done 帧 params **零翻译**映射上来即可。
| 字段 | 类型 | 必填 | 默认 | 说明 |
|---|---|---|---|---|
| `trace_id` | string | ✅ | — | pricebot 侧 trace_id。**幂等键**:同用户同 trace_id 重复上报覆盖、返回同一 id |
| `business_type` | string | ❌ | `food` | `food`(外卖,当前唯一接通) / `ecom`(电商) / `coupon`(领券) |
| `device_id` | string \| null | ❌ | null | 设备号 |
| `store_name` | string \| null | ❌ | null | 店铺名(外卖,来自 calibration.result |
| `source_platform_id` | string \| null | ❌ | null | 源平台代号,如 `taobao_flash` |
| `source_platform_name` | string \| null | ❌ | null | 源平台中文名 |
| `source_package` | string \| null | ❌ | null | 源平台 Android 包名 |
| `source_price` | float \| null | ❌ | null | 源平台到手价(**元** |
| `items` | Item[] | ❌ | [] | 下单菜品,`{name, qty, specs?}` |
| `comparison_results` | Result[] | ❌ | [] | 逐平台对比,见下表(price 单位**元**) |
| `skipped_dish_count` | int \| null | ❌ | null | 目标平台未找到、跳过的菜品数 |
| `skipped_dish_names` | string[] | ❌ | [] | 被跳过的菜名 |
| `total_dish_count` | int \| null | ❌ | null | 原始菜品总数 |
| `information` | string \| null | ❌ | null | done 帧文案(成功摘要 / 失败具体原因),持久化为列并在列表/详情返回,前端失败时当原因展示 |
| `status` | string \| null | ❌ | null | `success` / `failed`。**不传则服务端派生** |
**Item**`{ name: string, qty: int=1, specs: string[]\|null }`
**Resultcomparison_results 元素)**`{ platform_id, platform_name, package, price(元,float\|null), is_source(bool), rank(int\|null), coupon_saved(元,float\|null) }`
- `coupon_saved`:该平台本单**平台主优惠额**(元)——美团红包 / 淘宝平台红包 / 京东优惠券·百亿补贴,**只取那一笔**,不含配送费减免/共减总额。仅外卖目标平台带值,源平台/没用为 `null`,前端记录页据此展示**「已优惠 ¥X」**(null 不展示)。
## 服务端派生(客户端不用算)
`comparison_results` 派生并落结构化列:
- `best_*`:取最便宜的一家(按 `rank` 升序、再按 `price`),写 `best_platform_id/name/best_price_cents`
- `source_price_cents`:优先用 `source_price`,缺失则取 `comparison_results``is_source=true` 行的价
- `saved_amount_cents` = `源价 最优价`(可为 0 / 负:源平台本来就最便宜)
- `is_source_best`:最便宜的一家是否为源平台(= 这次没省到)
- `status`(未显式传时):存在「非源且有价」的结果 → `success`,否则 → `failed`
- 金额一律 `round(元 × 100)` 存**分**
`comparison_results` / `items` / `skipped_dish_names` 原样存 JSON;完整上报 body 另存 `raw_payload`(详情接口可取回)。
## 出参
响应 `200``{ "id": int }`(写入或已存在记录的 id
## 错误
- `401` 未鉴权
- `422``trace_id` 或字段类型不符
## 说明
失败的比价(done 帧只有源/空 comparison_results)也会落一条 `status=failed`——「越详细越好」,是否在 UI 展示由前端过滤。
+47
View File
@@ -0,0 +1,47 @@
# GET /api/v1/compare/records — 比价记录列表(游标分页)
> 所属:比价记录组(前缀 `/api/v1/compare` | 鉴权:Bearer | [← 返回 API 索引](./README.md)
「我的比价记录」列表页数据源。按 `id` 倒序(最新在前)。
## 入参(query
| 字段 | 类型 | 必填 | 默认 | 说明 |
|---|---|---|---|---|
| `limit` | int | ❌ | 20 | 1100 |
| `cursor` | int | ❌ | null | 上一页末条 `id`,首页不传 |
## 出参
响应 `200``{ items: ComparisonRecordOut[], next_cursor: int|null }`(分页见 [索引#游标分页约定](./README.md#游标分页约定)
**ComparisonRecordOut**(列表项,不含 `raw_payload`,减小 payload
| 字段 | 类型 | 说明 |
|---|---|---|
| `id` | int | 记录 id(也是游标) |
| `business_type` | string | `food` / `ecom` / `coupon` |
| `trace_id` | string | pricebot trace_id |
| `source_platform_id` | string \| null | 源平台代号 |
| `source_platform_name` | string \| null | 源平台中文名 |
| `source_package` | string \| null | 源平台包名 |
| `source_price_cents` | int \| null | 源平台到手价(分) |
| `best_platform_id` | string \| null | 最优平台代号 |
| `best_platform_name` | string \| null | 最优平台中文名 |
| `best_price_cents` | int \| null | 最优价(分) |
| `saved_amount_cents` | int \| null | 省下(分,可 0/负) |
| `is_source_best` | bool \| null | 源平台是否最便宜(= 没省到) |
| `store_name` | string \| null | 店铺名 |
| `total_dish_count` | int \| null | 菜品总数 |
| `skipped_dish_count` | int \| null | 跳过菜品数 |
| `status` | string | `success` / `failed` |
| `information` | string \| null | done 帧文案。成功:"在美团找到同店,到手价 ¥X…";失败:具体原因(如"美团、京东外卖均未找到该商品"),前端在 `status=failed` 时当原因展示 |
| `items` | object[] | 下单菜品 `{name, qty, specs?}` |
| `comparison_results` | object[] | 逐平台对比(price 单位元,已按 rank 升序) |
| `skipped_dish_names` | string[] | 被跳过的菜名 |
| `created_at` | datetime | 时间 |
## 错误
- `401` 未鉴权
## 说明
只返回当前登录用户自己的记录。需要单条全量(含 `raw_payload`)走 [详情接口](./compare-record-detail.md)。
+40
View File
@@ -0,0 +1,40 @@
# POST /api/v1/feedback — 提交反馈
> 所属:Feedback 组(前缀 `/api/v1/feedback` | 鉴权:Bearer access_token | [← 返回 API 索引](./README.md)
## 入参
**multipart/form-data**:
| 字段 | 类型 | 必填 | 说明 |
|---|---|---|---|
| `content` | string | ✓ | 反馈正文,**1-2000 字**(strip 后) |
| `contact` | string | ✓ | 联系方式(微信/QQ/手机号),**1-128 字**,便于回访 |
| `images` | file[] | ✗ | 截图,**最多 4 张**,每张走头像同款校验(JPEG/PNG/WebP,≤ 5 MB,魔数嗅探) |
## 出参
响应 `200`:
```json
{
"id": 123,
"status": "new",
"created_at": "2026-05-29T12:34:56Z"
}
```
| 字段 | 类型 | 说明 |
|---|---|---|
| `id` | int | 反馈单 ID |
| `status` | string | `new`(待处理)/ `handled`(已处理) |
| `created_at` | datetime | 提交时间(UTC) |
> 不返回上传的 image URL——这是给运营后台看的,客户端通常不需要。
## 错误码
- `400` 内容为空 / 内容超 2000 字 / 联系方式为空 / 联系方式超 128 字 / 图片超 4 张 / 单图非法(空/过大/格式不对)
- `401` 未带 token / token 无效或过期 / 用户被禁用
- `422``content``contact` 字段
## 说明
- **反馈绑用户**:`feedback.user_id = current_user.id`,便于回访
- **截图落盘**:存到 `settings.MEDIA_ROOT/feedback/`,URL 写进 `feedback.images`(JSON 列,SQLite 是 TEXT、PG 是 JSON/JSONB——为不影响现有迁移用 JSON 而非 JSONB)
- **截图可读**:`GET /media/feedback/<file>` 静态服务暴露,详见 [API 索引 — /media 静态服务](./README.md#media-静态服务)
- 文件名随机 `u<user_id>_<rand>.<ext>`,杜绝路径穿越
+3 -2
View File
@@ -23,7 +23,8 @@
响应 `200`:`{ items: CouponCard[], has_next: bool, search_id: string|null }``CouponCard` 结构见 [API 索引](./README.md#复用数据结构)。
## 错误码
- `502` 美团接口失败
- `502` 美团接口失败(仅在**已配置 `MT_CPS_APP_KEY/APP_SECRET` 但调用失败**时;未配凭证场景见下)
## 说明
当前 Android 客户端**未调用**此接口(搜索功能尚未实现),仅后端实现完整
- 当前 Android 客户端**未调用**此接口(搜索功能尚未实现),仅后端实现完整
- **未配置 MT_CPS 凭证时降级**:`settings.mt_cps_configured == false` 时直接返 `{ items: [], has_next: false, search_id: null }`,**不报 502**。设计目的是让新开发机首屏不炸——后端业务层先不依赖 CPS。已配凭证但调美团失败时才走 502
+8 -2
View File
@@ -17,9 +17,15 @@
响应 `200`:`{ items: CouponCard[], has_next: bool, page: int }``CouponCard` 结构见 [API 索引](./README.md#复用数据结构)。
## 错误码
见"说明"中的可观测性盲区
(任何失败场景都返 200 + 空 items,见"说明"中的可观测性盲区)
## 说明
后端在 `coupons` 之上封装的"伪推荐"——每页并发拉一次外卖、一次到店(到餐)榜单,按 **2 外卖 + 1 到店** 交叉去重。榜单组合写死 3 页(第1页爆款、第2页今日必推、第3页外卖精选+到店限时),**第 3 页起 `has_next=false`、第 4 页返空**。
⚠️ **可观测性盲区**:`feed` 内部对美团调用异常是静默吞掉返回空列表(不报 502、不打日志)。若 `items` 为空但无错误,优先用 `coupons` 接口逼出真实错误(它会以 502 暴露,如"MT_CPS_APP_KEY not configured")。
**两种空结果路径**:
- **未配置 `MT_CPS_APP_KEY/APP_SECRET`**:开头 `settings.mt_cps_configured` 判定后**直接返空**,完全不调美团
- **已配置但美团调用失败**:`_fetch_topic``except MeituanCpsError` **静默返空**(不报 502、不打日志)
⚠️ **可观测性盲区**:两种路径在响应上**无法区分**。若 `items` 为空想区分:
-`settings.mt_cps_configured`(`/health` 接口有暴露此字段)→ 区分"配置缺失" vs "调用失败"
- 改调 `coupons` 接口——它在"已配凭证但调用失败"时会 502 暴露错误细节(未配凭证它也降级返空,跟 feed 一致)
+4 -4
View File
@@ -23,9 +23,9 @@
| `link_map` | object | 各类型链接 `{ "linkType": "url" }`,如 `{"1": "<H5>", "3": "<deeplink>"}` |
## 错误码
- `502` 美团接口失败
- `502` 美团接口失败(仅在**已配置 `MT_CPS_APP_KEY/APP_SECRET` 但调用失败**时;未配凭证场景见下)
## 说明
客户端实际优先用 `link_map["3"]`deeplink)拉起美团 App,失败降级 `link_map["1"]`H5
⚠️ **安全**:`sid` 允许客户端传值覆盖服务端默认渠道,理论上他人可借本接口刷自己渠道的分佣;建议服务端锁定 `sid`、忽略客户端传值
- 客户端实际优先用 `link_map["3"]`deeplink)拉起美团 App,失败降级 `link_map["1"]`H5
- **未配置 MT_CPS 凭证时降级**:`settings.mt_cps_configured == false` 时直接返 `{ link: "", link_map: {} }`,**不报 502**。客户端拿到空 link 时跳转会失败——客户端应做兜底(toast "暂无可用链接"或不显示「抢」按钮)
- ⚠️ **安全**:`sid` 允许客户端传值覆盖服务端默认渠道,理论上他人可借本接口刷自己渠道的分佣;建议服务端锁定 `sid`、忽略客户端传值
+26
View File
@@ -0,0 +1,26 @@
# POST /api/v1/user/avatar — 上传头像
> 所属:User 组(前缀 `/api/v1/user` | 鉴权:Bearer access_token | [← 返回 API 索引](./README.md)
## 入参
**multipart/form-data**:
| 字段 | 类型 | 说明 |
|---|---|---|
| `file` | file | 头像图片,**仅 JPEG / PNG / WebP**,**≤ 5 MB** |
> 服务端**按文件头魔数嗅探真实类型**(`FF D8 FF` / `89 50 4E 47` / `RIFF…WEBP`),不信任 Content-Type 与文件名;不在三种之内的一律拒。
## 出参
响应 `200`:`UserOut`(`avatar_url` 字段已更新为新地址,形如 `/media/avatars/u<uid>_<rand>.jpg`)。结构见 [API 索引](./README.md#复用数据结构)。
> 返回**相对路径**——客户端按自己的 `BASE_URL` 拼绝对地址(dev 下 `10.0.2.2`/LAN IP 都可能,服务端不知道客户端怎么访问到自己)。读图走 [/media 静态服务](./README.md#media-静态服务) 的 `GET /media/...`。
## 错误码
- `400` 空文件 / 超过 5 MB / 非 JPEG/PNG/WebP
- `401` 未带 token / token 无效或过期 / 用户被禁用
- `422``file` 字段
## 说明
- **旧头像清理**:落库成功后才删旧文件,避免"删了新的没存上"的丢图风险;非本服务托管的 URL(如微信头像)不动
- **文件名服务端随机生成** `u<user_id>_<16 位 hex>.<ext>`,杜绝路径穿越与覆盖
- 存储目录:`settings.MEDIA_ROOT/avatars/`,默认 `data/media/avatars/`(生产 nginx 注意把它放共享卷或对象存储,见 [后端技术实现.md](../后端技术实现.md))
+18
View File
@@ -0,0 +1,18 @@
# DELETE /api/v1/user — 注销账号
> 所属:User 组(前缀 `/api/v1/user` | 鉴权:Bearer access_token | [← 返回 API 索引](./README.md)
## 入参
无(身份取自 Header token)
## 出参
响应 `200`:`{ "ok": true }`
## 错误码
- `401` 未带 token / token 无效或过期 / 用户被禁用(已注销账号再次调返 401,因 `status=deleted` 不再 active)
## 说明
- **软删除 + 匿名化**:`status="deleted"` + `phone="deleted_<id>"`(释放 phone 唯一约束,允许同号码重新注册成全新账号)+ `nickname`/`avatar_url` 清空。行保留用于审计 / 关联表外键完整性
- **头像文件物理删除**:本服务托管的头像文件(`/media/avatars/...`)同时删盘;微信头像等外部 URL 不动
- **后续登录**:`status=deleted` 的行登录接口校验不过,所以"注销账号"后旧 token 仍能用直到自然过期(2h),但一过期就再也换不出新 token。客户端应在调本接口后立刻清本地 token
- **未来若加 jti 黑名单**:本接口应同时把当前 access/refresh 加入黑名单实现"立即吊销"
+20
View File
@@ -0,0 +1,20 @@
# PATCH /api/v1/user/profile — 修改昵称
> 所属:User 组(前缀 `/api/v1/user` | 鉴权:Bearer access_token | [← 返回 API 索引](./README.md)
## 入参
请求体 JSON:
| 字段 | 类型 | 说明 |
|---|---|---|
| `nickname` | string | 昵称,**1-16 字**(前后空格自动 strip,strip 后为空判 400) |
## 出参
响应 `200`:`UserOut`(返回更新后的完整用户对象,`nickname` 字段已更新)。结构见 [API 索引](./README.md#复用数据结构)。
## 错误码
- `401` 未带 token / token 无效或过期 / 用户被禁用
- `422` 字段校验失败(`nickname` 缺失 / 长度超 16 字 / 全空白)
## 说明
- **服务端持久化**:写入 `user.nickname` 列。这是昵称的唯一数据源——客户端之前的本地缓存退登即丢,改完后下次登录走 [`/auth/me`](./auth-me.md) 与登录响应回传仍生效
- **strip 行为**:服务端会先去除前后空白再校验长度,所以 `" bob "` 等价于 `"bob"`
+37
View File
@@ -0,0 +1,37 @@
# 傻瓜比价 App 后端 — 数据库表文档(索引)
> 数据库:SQLite 起步(`data/app.db`),生产可切 PostgreSQL(改 `DATABASE_URL`)。
> ORM:SQLAlchemy 2.0(`app/models/`),迁移:Alembic(`alembic/versions/`,`render_as_batch` 兼容 SQLite)。
> 金额字段一律存**整数**:金币=个数,现金=**分**(`*_cents`)。时间列 `DateTime(timezone=True)`。
> 最后更新:2026-05-31
---
## 表总览
| 表 | 用途 | 模型 | 关联模块 | 文档 |
|---|---|---|---|---|
| `user` | 用户(登录主体) | `models/user.py` | 登录/鉴权 | [详情](./user.md) |
| `coin_account` | 金币+现金余额快照(一用户一行) | `models/wallet.py` | 钱包 | [详情](./coin_account.md) |
| `coin_transaction` | 金币流水账本 | `models/wallet.py` | 钱包 | [详情](./coin_transaction.md) |
| `cash_transaction` | 现金流水账本(分) | `models/wallet.py` | 钱包/提现 | [详情](./cash_transaction.md) |
| `withdraw_order` | 提现单(现金→微信零钱) | `models/wallet.py` | 提现 | [详情](./withdraw_order.md) |
| `signin_record` | 签到记录(7 天循环) | `models/signin.py` | 签到 | [详情](./signin_record.md) |
| `user_task` | 一次性任务领取去重 | `models/task.py` | 任务 | [详情](./user_task.md) |
| `savings_record` | 省钱记录(profile 省钱战绩源) | `models/savings.py` | 省钱 | [详情](./savings_record.md) |
| `ad_reward_record` | 看激励视频发奖记录(S2S 回调) | `models/ad_reward.py` | 看广告发奖 | [详情](./ad_reward_record.md) |
| `ad_ecpm_record` | 广告展示 eCPM 上报(收益对账) | `models/ad_ecpm.py` | 看广告 | [详情](./ad_ecpm_record.md) |
| `feedback` | 用户帮助与反馈 | `models/feedback.py` | 反馈 | [详情](./feedback.md) |
| `comparison_record` | 比价记录(每次比价完整明细) | `models/comparison.py` | 比价记录 | [详情](./comparison_record.md) |
| `comparison_milestone_claim` | 比价战绩里程碑领取记录 | `models/comparison_milestone.py` | 比价记录/福利 | [详情](./comparison_milestone_claim.md) |
---
## 通用约定
- **主键**:`id` Integer autoincrement(`coin_account` 例外:`user_id` 既是主键也是外键,一用户一行)。
- **外键**:所有用户维度表 `user_id``user.id`,且建 index。
- **金额**:整数;金币计数,现金/价格存「分」(`*_cents`)。
- **时间**:`created_at` 等用 `DateTime(timezone=True)` + `server_default=func.now()`;业务"今天"按**北京时间**(见 `core/rewards.cn_today`),跨天计数用「日期串」列(如 `reward_date`/`report_date`)等值查,不在 SQL 里做跨时区 date 比较(SQLite 不可靠)。
- **JSON 列**:用 `JSON().with_variant(JSONB(), "postgresql")` —— PG 上 JSONB(可建 GIN 索引),SQLite 退化为通用 JSON(否则 `create_all` 编译报错)。
- **迁移**:改表必写 alembic 迁移并保持单 head;改表/建表同时更新本目录对应文档(一表一文件)。
+26
View File
@@ -0,0 +1,26 @@
# ad_ecpm_record — 广告展示 eCPM 上报(收益对账)
> 模型 `app/models/ad_ecpm.py` | 关联接口 [ad-ecpm-report](../api/ad-ecpm-report.md) | [← 表索引](./README.md)
每条 = 客户端一次广告展示(`onAdShow`)后读到的 eCPM。与发奖记录 `ad_reward_record` 是**两条独立数据流**(发奖走 S2S 有 trans_id 无 ecpm;eCPM 走客户端有 ecpm 无 trans_id),无公共键,只用于**按用户/按天聚合**收益对账,不做逐条精确关联。穿山甲后台报表才是结算权威。
## 字段
| 列 | 类型 | 约束 / 默认 | 说明 |
|---|---|---|---|
| `id` | Integer | PK, autoincrement | |
| `user_id` | Integer | FK→user.id, index, NOT NULL | 归属用户 |
| `ad_type` | String(32) | NOT NULL | 广告类型:`reward_video`(激励视频)/ `draw`(Draw 信息流)等 |
| `adn` | String(32) | nullable | 实际投放 ADN(`getSdkName`,如 pangle/gdt) |
| `slot_id` | String(64) | nullable | 实际展示代码位(底层 mediation rit) |
| `ecpm_raw` | String(32) | NOT NULL | 客户端上报的 eCPM 原始串(单位待确认,原样存) |
| `report_date` | String(10) | index, NOT NULL | 北京时间日期串 `YYYY-MM-DD`,按它做按天聚合 |
| `created_at` | DateTime(tz) | server_default now(), index | 时间 |
## 索引与约束
- PK: `id`;index: `user_id``report_date``created_at`
## 关系
- `user_id``user.id`(多对一)
## 说明
- ⚠️ `ecpm_raw` 单位(分/元)截至 2026-05-31 未最终确认;确认后再加一列解析好的数值,在此之前对账按"待定单位"处理。
+28
View File
@@ -0,0 +1,28 @@
# ad_reward_record — 看激励视频发奖记录(S2S 回调)
> 模型 `app/models/ad_reward.py` | 关联接口 [ad-pangle-callback](../api/ad-pangle-callback.md) / [ad-reward-status](../api/ad-reward-status.md) | [← 表索引](./README.md)
每条 = 穿山甲一次发奖回调。`trans_id` 唯一做幂等键(穿山甲会重试,同号只发一次)。`reward_date`(北京时间日期串)给"每日上限"计数用,按日期串等值查。
## 字段
| 列 | 类型 | 约束 / 默认 | 说明 |
|---|---|---|---|
| `id` | Integer | PK, autoincrement | |
| `trans_id` | String(64) | UNIQUE, index, NOT NULL | 穿山甲交易号(幂等键) |
| `user_id` | Integer | FK→user.id, index, NOT NULL | 归属用户(回调 media_extra 带回) |
| `coin` | Integer | NOT NULL, default 0 | 实发金币(capped 时为 0) |
| `status` | String(16) | NOT NULL, default `granted` | `granted`(已发)/ `capped`(当日超限未发) |
| `reward_date` | String(10) | index, NOT NULL | 北京时间日期串 `YYYY-MM-DD`,按它统计当日次数 |
| `reward_name` | String(64) | nullable | 穿山甲上报奖励名(参考,不作发奖依据) |
| `raw` | String(1024) | nullable | 回调原始参数(审计排查) |
| `created_at` | DateTime(tz) | server_default now(), index | 时间 |
## 索引与约束
- PK: `id`;UNIQUE + index: `trans_id`;index: `user_id``reward_date``created_at`
## 关系
- `user_id``user.id`(多对一)
## 说明
- 三道闸:验签不过 403 → `trans_id` 唯一幂等(并发 catch IntegrityError)→ 当日次数 ≥ `DAILY_AD_REWARD_LIMIT``capped` 不发币。
- 发币复用 `grant_coins(biz_type='ad_reward', ref_id=trans_id)`
+26
View File
@@ -0,0 +1,26 @@
# cash_transaction — 现金流水账本(分)
> 模型 `app/models/wallet.py` | 关联接口 [wallet-cash-transactions](../api/wallet-cash-transactions.md) | [← 表索引](./README.md)
现金每次变动一笔流水(单位:分)。金币兑现金、提现、提现退款都记这里。
## 字段
| 列 | 类型 | 约束 / 默认 | 说明 |
|---|---|---|---|
| `id` | Integer | PK, autoincrement | |
| `user_id` | Integer | FK→user.id, index, NOT NULL | 归属用户 |
| `amount_cents` | Integer | NOT NULL | 正=入账(兑入),负=出账(提现) |
| `balance_after_cents` | Integer | NOT NULL | 本笔变动后现金余额(分) |
| `biz_type` | String(32) | NOT NULL | `exchange_in`(兑入)/ `withdraw`(提现出账)/ `withdraw_refund`(提现退回) |
| `ref_id` | String(64) | nullable | 关联业务 id(如提现 `out_bill_no`) |
| `remark` | String(128) | nullable | 用户可见备注(如"提现未成功,金额已退回") |
| `created_at` | DateTime(tz) | server_default now(), index | 时间 |
## 索引与约束
- PK: `id`;index: `user_id``created_at`
## 关系
- `user_id``user.id`(多对一)
## 说明
- 提现失败/取消退款写 `withdraw_refund`(+X);退款流水 `remark` 是用户可见文案,技术原因记在 `withdraw_order.fail_reason`
+24
View File
@@ -0,0 +1,24 @@
# coin_account — 金币 + 现金余额快照
> 模型 `app/models/wallet.py` | 关联接口 [wallet-account](../api/wallet-account.md) | [← 表索引](./README.md)
一个用户一行的余额快照,供读取展示用;每次余额变动都另写流水(`coin_transaction` / `cash_transaction`)并记 `balance_after`,出问题逐笔回溯。`user_id` 既是主键也是外键。
## 字段
| 列 | 类型 | 约束 / 默认 | 说明 |
|---|---|---|---|
| `user_id` | Integer | PK, FK→user.id | 用户(一用户一行) |
| `coin_balance` | Integer | NOT NULL, default 0 | 当前金币余额 |
| `cash_balance_cents` | Integer | NOT NULL, default 0 | 当前现金余额(分) |
| `total_coin_earned` | Integer | NOT NULL, default 0 | 累计赚取金币(只增不减) |
| `updated_at` | DateTime(tz) | server_default now(), onupdate now() | 最后更新时间 |
## 索引与约束
- PK: `user_id`(同时是 FK→user.id)
## 关系
- `user_id``user.id`(一对一)
## 说明
- **唯一发金币入口** `wallet.grant_coins` 更新本表余额快照 + 写 `coin_transaction`,不 commit,由调用方同事务提交。
- 提现扣现金用带条件 `UPDATE ... WHERE cash_balance_cents >= amount` 原子扣减,防并发超额。
+27
View File
@@ -0,0 +1,27 @@
# coin_transaction — 金币流水账本
> 模型 `app/models/wallet.py` | 关联接口 [wallet-coin-transactions](../api/wallet-coin-transactions.md) | [← 表索引](./README.md)
金币每次变动一笔流水,记变动后余额,用于对账与「金币明细」展示。
## 字段
| 列 | 类型 | 约束 / 默认 | 说明 |
|---|---|---|---|
| `id` | Integer | PK, autoincrement | |
| `user_id` | Integer | FK→user.id, index, NOT NULL | 归属用户 |
| `amount` | Integer | NOT NULL | 正=入账(赚),负=出账(花/兑换) |
| `balance_after` | Integer | NOT NULL | 本笔变动后金币余额(对账用) |
| `biz_type` | String(32) | NOT NULL | 业务类型:`signin` / `task_<key>` / `exchange_out` / `ad_reward` / `compare_milestone` … |
| `ref_id` | String(64) | nullable | 关联业务 id(签到日期 / 任务 key / trans_id / 里程碑序号等) |
| `remark` | String(128) | nullable | 备注 |
| `created_at` | DateTime(tz) | server_default now(), index | 时间 |
## 索引与约束
- PK: `id`;index: `user_id``created_at`
## 关系
- `user_id``user.id`(多对一)
## 说明
- 明细接口按 `id` 倒序游标分页。
-`biz_type` 取值由各业务写入(签到/任务/兑换/看广告/比价里程碑),无独立枚举约束,靠写入方约定。
@@ -0,0 +1,26 @@
# comparison_milestone_claim — 比价战绩里程碑领取记录
> 模型 `app/models/comparison_milestone.py` | 关联接口 [compare-milestones](../api/compare-milestones.md) / [compare-milestone-claim](../api/compare-milestone-claim.md) | [← 表索引](./README.md)
「记录比价战绩」每档(第 1~6 次)只能领一次,领取后写一行;解锁进度由 `comparison_record``status='success'` 的条数决定,本表只记"哪几档已领"。仿 `user_task` 的一次性领取模型。
## 字段
| 列 | 类型 | 约束 / 默认 | 说明 |
|---|---|---|---|
| `id` | Integer | PK, autoincrement | |
| `user_id` | Integer | FK→user.id, index, NOT NULL | 归属用户 |
| `milestone` | Integer | NOT NULL | 档位序号(1-based),见 `rewards.RECORD_MILESTONES` |
| `coin_awarded` | Integer | NOT NULL, default 0 | 该档应发金币额;⚠️ 当前产品定暂不真发,实际恒为 0 |
| `claimed_at` | DateTime(tz) | server_default now() | 领取时间 |
## 索引与约束
- PK: `id`;index: `user_id`
- UNIQUE(`user_id`, `milestone`) = `uq_compare_milestone_user`(防同档重复领)
## 关系
- `user_id``user.id`(多对一)
- 解锁进度依赖 `comparison_record`(`status='success'` 计数),本表不存进度本身。
## 说明
- ⚠️ 当前 claim 仅写本表标记已领(去重),**不调 `grant_coins`、不写 `coin_transaction`,`coin_awarded` 恒为 0、余额不变**(产品定暂不真发金币,后续整体删除该功能)。
- 领取校验:档位越界 404、未解锁(成功数 < milestone)409、已领 409。
+47
View File
@@ -0,0 +1,47 @@
# comparison_record — 比价记录(每次比价完整明细)
> 模型 `app/models/comparison.py` | 关联接口 [compare-record-report](../api/compare-record-report.md) / [compare-records](../api/compare-records.md) / [compare-record-detail](../api/compare-record-detail.md) | [← 表索引](./README.md)
每完成一次比价(外卖/电商/领券),客户端 done 帧后用带 JWT 通道上报一条。「我的比价记录」页数据源,也是比价战绩里程碑解锁进度的计数源(`status='success'` 条数)。
> 与 `savings_record` 的区别:本表是「每一次**比价行为**的完整明细」(不省钱、甚至失败也记);`savings_record` 是「真正**下单成交**省了多少」。两表独立、互不喂数据。
## 字段
| 列 | 类型 | 约束 / 默认 | 说明 |
|---|---|---|---|
| `id` | Integer | PK, autoincrement | |
| `user_id` | Integer | FK→user.id, index, NOT NULL | 归属用户 |
| `device_id` | String(64) | nullable | 设备号 |
| `business_type` | String(16) | NOT NULL, default `food`, index | `food`(当前唯一接通)/ `ecom` / `coupon` |
| `trace_id` | String(64) | NOT NULL | pricebot 侧 trace_id(关联调试落盘 + 幂等键) |
| `source_platform_id` | String(32) | nullable | 源平台代号 |
| `source_platform_name` | String(32) | nullable | 源平台中文名 |
| `source_package` | String(128) | nullable | 源平台 Android 包名 |
| `source_price_cents` | Integer | nullable | 源平台到手价(分) |
| `best_platform_id` | String(32) | nullable | 最优平台代号(= rank=1) |
| `best_platform_name` | String(32) | nullable | 最优平台中文名 |
| `best_price_cents` | Integer | nullable | 最优价(分) |
| `saved_amount_cents` | Integer | nullable | 源价 最优价(可 0/负) |
| `is_source_best` | Boolean | nullable | 源平台就是最便宜(= 没省到) |
| `store_name` | String(128) | nullable | 店铺名 |
| `total_dish_count` | Integer | nullable | 菜品总数 |
| `skipped_dish_count` | Integer | nullable | 跳过(没找到)菜品数 |
| `status` | String(16) | NOT NULL, default `success` | `success`(拿到有效对比)/ `failed`(出错/没采到目标价) |
| `information` | String(256) | nullable | done 帧文案;成功=摘要,失败=具体原因(前端失败时当原因展示) |
| `items` | JSON(PG: JSONB) | NOT NULL, default [] | 下单菜品 `[{name, qty, specs?}]` |
| `comparison_results` | JSON(PG: JSONB) | NOT NULL, default [] | 逐平台对比 `[{platform_id,platform_name,package,price(元),is_source,rank,coupon_saved}]`;`coupon_saved`=该平台主优惠额(元,美团红包/淘宝平台红包/京东优惠券·百亿补贴,只取那一笔,不含配送费/共减总额),各平台抠到红包即带值(2026-06 起源平台 Phase1 意图识别也抠,当前仅淘宝源),没用/没抠到为 null,前端展示「已优惠 ¥X」 |
| `skipped_dish_names` | JSON(PG: JSONB) | NOT NULL, default [] | 被跳过的菜名 |
| `raw_payload` | JSON(PG: JSONB) | nullable | 客户端原始上报(calibration + done.params 全量) |
| `created_at` | DateTime(tz) | server_default now(), index | 时间 |
## 索引与约束
- PK: `id`;index: `user_id``business_type``created_at`
- UNIQUE(`user_id`, `trace_id`) = `uq_comparison_user_trace`(同次比价重试/重复上报幂等覆盖)
## 关系
- `user_id``user.id`(多对一)
-`comparison_milestone_claim` 间接依赖:里程碑解锁进度 = 本表 `status='success'` 条数。
## 说明
- `best_*` / `saved_amount_cents` / `is_source_best` / `status``repositories/comparison.py:_derive``comparison_results` 派生,客户端不用自己算。
- 4 个 JSON 列用 `JSON().with_variant(JSONB(),"postgresql")`(SQLite 退化 JSON)。金额结构化列存「分」,`comparison_results.price` 原样存「元」。
+25
View File
@@ -0,0 +1,25 @@
# feedback — 用户帮助与反馈
> 模型 `app/models/feedback.py` | 关联接口 [feedback](../api/feedback.md) | [← 表索引](./README.md)
每条 = 用户一次反馈提交。`content``contact` 必填,`images` 为可选截图 URL 列表。
## 字段
| 列 | 类型 | 约束 / 默认 | 说明 |
|---|---|---|---|
| `id` | Integer | PK, autoincrement | |
| `user_id` | Integer | FK→user.id, index, NOT NULL | 提交用户 |
| `content` | Text | NOT NULL | 反馈正文 |
| `contact` | String(128) | NOT NULL | 联系方式(微信/QQ/手机,便于回访) |
| `images` | JSON | nullable | 截图 URL 列表(相对路径 `/media/feedback/...`);无图为 NULL |
| `status` | String(16) | NOT NULL, default `new` | `new`(待处理)/ `handled`(已处理) |
| `created_at` | DateTime(tz) | server_default now(), index | 提交时间 |
## 索引与约束
- PK: `id`;index: `user_id``created_at`
## 关系
- `user_id``user.id`(多对一)
## 说明
- `images` 用通用 `JSON`(本表未用 JSONB variant);截图先经 `/media/feedback/` 上传拿到相对路径再随反馈提交。

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