Compare commits

..

1 Commits

Author SHA1 Message Date
marco d46a5cd80f chore: alembic merge revision + PG 迁移指南文档
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>
2026-05-30 18:08:51 +08:00
237 changed files with 495 additions and 15379 deletions
+2 -22
View File
@@ -40,29 +40,13 @@ MT_CPS_APP_KEY=
MT_CPS_APP_SECRET=
# 默认渠道追踪标识(sid),用于区分不同 app 的 CPS 数据
MT_CPS_DEFAULT_SID=sgbjia
# 美团调用走的代理。⚠️ 本机/内网开发直连美团会 SSL EOF,必须填本地代理(如 http://127.0.0.1:7897);
# 线上国内服务器留空(=直连)。留空且本机直连失败时 /feed、/coupons、/top-sales 会返回空。
MT_CPS_PROXY=
# ===== Pricebot 上游 (领券/比价业务透传目标) =====
# 客户端调本服务的 /api/v1/coupon/step,我们透传到 pricebot-backend。
# ===== Pricebot 上游 (领券业务透传目标) =====
# 客户端调本服务的 /api/v1/coupon/step,我们透传到 pricebot-backend 的 /api/coupon/step
# 本地开发用 localhost:8000。生产部署改成内网地址(如 http://pricebot.internal:8000)。
PRICEBOT_BASE_URL=http://localhost:8000
# 【多实例】单机多进程部署时,填逗号分隔的实例列表(端口与 pricebot 集群对齐),透传层按
# trace_id 一致性 hash 选实例 → 同一比价所有帧落同一进程(进程内维护 state,无需 Redis)。
# 留空 = 单实例(用上面的 PRICEBOT_BASE_URL)。详见 pricebot-backend/docs/并发部署设计.md
# PRICEBOT_INSTANCES=http://127.0.0.1:8001,http://127.0.0.1:8002,http://127.0.0.1:8003,http://127.0.0.1:8004,http://127.0.0.1:8005,http://127.0.0.1:8006
# 领券单帧最多 wait 6s,加网络往返,30s 兜底
PRICEBOT_REQUEST_TIMEOUT_SEC=30
# 比价(intent/recognize + price/step)透传超时:大上下文 LLM + 逐帧 LLM,给 60s
PRICEBOT_COMPARE_TIMEOUT_SEC=60
# ===== 内部端点 (server→server, pricebot 上报价格观测) =====
# pricebot 比价 done 后把各平台到手价 POST 到本服务 /internal/price-observation 落库
# (price_observation 表,比价资产沉淀层),靠共享密钥头 X-Internal-Secret 校验。
# 必须与 pricebot 侧的 INTERNAL_API_SECRET **同值**;留空 = 内部写端点关闭(返 503)。
# 启用前两边都填同一高熵串:python -c "import secrets; print(secrets.token_urlsafe(48))"
INTERNAL_API_SECRET=
# ===== CORS =====
# 逗号分隔,生产留空(只让 app 调,不开放 web)。本地开发可加 http://localhost:5173 之类
@@ -79,10 +63,6 @@ WXPAY_MCH_PRIVATE_KEY_PATH=./secrets/apiclient_key.pem
WXPAY_PUBLIC_KEY_ID=PUB_KEY_ID_xxxxxxxx
WXPAY_PUBLIC_KEY_PATH=./secrets/pub_key.pem
WXPAY_TRANSFER_SCENE_ID=1000
WXPAY_AUTH_NOTIFY_URL=
WITHDRAW_AUTO_RECONCILE_ENABLED=false
WITHDRAW_AUTO_RECONCILE_INTERVAL_SEC=300
WITHDRAW_AUTO_RECONCILE_OLDER_THAN_MINUTES=15
# ===== 穿山甲激励视频(服务端发奖回调)=====
# 看完激励视频后穿山甲服务器 S2S 回调本服务发金币(客户端不参与发奖)。
+1 -8
View File
@@ -23,13 +23,7 @@ dist/
*.db-journal
*.db-shm
*.db-wal
# data/ 整体不入库(运行时数据/上传文件/大二进制)。例外:邀请落地页 dl.html——
# 它既是生产落地页又是本地测试资产,纳入 git 便于同事一致测试
# (见 docs/邀请功能-实现原理与本地测试.md)。其余(avatars/ / *.apk / app.db 等)仍忽略。
data/*
!data/media/
data/media/*
!data/media/dl.html
data/
secrets/*
!secrets/.gitkeep
@@ -43,4 +37,3 @@ secrets/*
# 运行日志(run.sh 输出, 不入库)
*.log
logs/
@@ -1,26 +0,0 @@
"""merge coupon_state and invite heads
Revision ID: 0cf18d590b1d
Revises: coupon_state_tables, invite_code_and_relation
Create Date: 2026-06-08 05:49:29.604571
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision: str = '0cf18d590b1d'
down_revision: Union[str, Sequence[str], None] = ('coupon_state_tables', 'invite_code_and_relation')
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
pass
def downgrade() -> None:
pass
@@ -1,26 +0,0 @@
"""merge meituan_coupon 与 ops/price_observation 迁移头
Revision ID: 11a1d08c6f55
Revises: meituan_coupon_table, opsrename01
Create Date: 2026-06-08 00:06:01.633796
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision: str = '11a1d08c6f55'
down_revision: Union[str, Sequence[str], None] = ('meituan_coupon_table', 'opsrename01')
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
pass
def downgrade() -> None:
pass
@@ -1,26 +0,0 @@
"""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
@@ -1,65 +0,0 @@
"""新增 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
@@ -1,47 +0,0 @@
"""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
@@ -1,65 +0,0 @@
"""admin_user + admin_audit_log tables (运营 admin 后台:账号 + 操作审计)
Revision ID: ad60a1b2c3d4
Revises: 8ac524a8ea02
Create Date: 2026-06-03 10:00:00.000000
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision: str = 'ad60a1b2c3d4'
down_revision: Union[str, Sequence[str], None] = '8ac524a8ea02'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
op.create_table(
'admin_user',
sa.Column('id', sa.Integer(), autoincrement=True, nullable=False),
sa.Column('username', sa.String(length=64), nullable=False),
sa.Column('password_hash', sa.String(length=255), nullable=False),
sa.Column('role', sa.String(length=20), nullable=False, server_default='operator'),
sa.Column('status', sa.String(length=20), nullable=False, server_default='active'),
sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('(CURRENT_TIMESTAMP)'), nullable=False),
sa.Column('last_login_at', sa.DateTime(timezone=True), nullable=True),
sa.PrimaryKeyConstraint('id'),
)
with op.batch_alter_table('admin_user', schema=None) as batch_op:
batch_op.create_index(batch_op.f('ix_admin_user_username'), ['username'], unique=True)
op.create_table(
'admin_audit_log',
sa.Column('id', sa.Integer(), autoincrement=True, nullable=False),
sa.Column('admin_id', sa.Integer(), nullable=False),
sa.Column('admin_username', sa.String(length=64), nullable=False),
sa.Column('action', sa.String(length=64), nullable=False),
sa.Column('target_type', sa.String(length=32), nullable=False),
sa.Column('target_id', sa.String(length=64), nullable=True),
sa.Column('detail', sa.JSON().with_variant(sa.dialects.postgresql.JSONB(), 'postgresql'), nullable=True),
sa.Column('ip', sa.String(length=64), nullable=True),
sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('(CURRENT_TIMESTAMP)'), nullable=False),
sa.ForeignKeyConstraint(['admin_id'], ['admin_user.id'], ),
sa.PrimaryKeyConstraint('id'),
)
with op.batch_alter_table('admin_audit_log', schema=None) as batch_op:
batch_op.create_index(batch_op.f('ix_admin_audit_log_admin_id'), ['admin_id'], unique=False)
batch_op.create_index(batch_op.f('ix_admin_audit_log_action'), ['action'], unique=False)
batch_op.create_index(batch_op.f('ix_admin_audit_log_created_at'), ['created_at'], unique=False)
def downgrade() -> None:
with op.batch_alter_table('admin_audit_log', schema=None) as batch_op:
batch_op.drop_index(batch_op.f('ix_admin_audit_log_created_at'))
batch_op.drop_index(batch_op.f('ix_admin_audit_log_action'))
batch_op.drop_index(batch_op.f('ix_admin_audit_log_admin_id'))
op.drop_table('admin_audit_log')
with op.batch_alter_table('admin_user', schema=None) as batch_op:
batch_op.drop_index(batch_op.f('ix_admin_user_username'))
op.drop_table('admin_user')
-33
View File
@@ -1,33 +0,0 @@
"""app_config table (运营可配置项:rewards 常量等后台化)
Revision ID: cfg1a2b3c4d5
Revises: ad60a1b2c3d4
Create Date: 2026-06-04 12:00:00.000000
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision: str = 'cfg1a2b3c4d5'
down_revision: Union[str, Sequence[str], None] = 'ad60a1b2c3d4'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
op.create_table(
'app_config',
sa.Column('key', sa.String(length=64), nullable=False),
sa.Column('value', sa.JSON().with_variant(sa.dialects.postgresql.JSONB(), 'postgresql'), nullable=False),
sa.Column('updated_by_admin_id', sa.Integer(), nullable=True),
sa.Column('updated_at', sa.DateTime(timezone=True), server_default=sa.text('(CURRENT_TIMESTAMP)'), nullable=False),
sa.PrimaryKeyConstraint('key'),
)
def downgrade() -> None:
op.drop_table('app_config')
@@ -1,32 +0,0 @@
"""comparison_record 加 best_deeplink(再次比价直达商家深链)
Revision ID: b7e2c1a9f4d3
Revises: 9258bddde4ea
Create Date: 2026-06-05
「再次比价」要直达上次最低价平台的商家/商品页。深链(link)在比价时客户端已从剪贴板采到
(collectedLinks[best_index]),本列把它落库;再次比价时写剪贴板 + launch 该平台 App 直达。
旧记录无此列值(None) → 前端降级为打开对应 App 首页。
手写迁移(只加一列):autogenerate 会误报"删 order_record"(降级为审计的历史表,model 已移除、
库表保留),手写规避该噪音。
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
revision: str = "b7e2c1a9f4d3"
down_revision: Union[str, Sequence[str], None] = "9258bddde4ea"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
with op.batch_alter_table("comparison_record", schema=None) as batch_op:
batch_op.add_column(sa.Column("best_deeplink", sa.String(length=1024), nullable=True))
def downgrade() -> None:
with op.batch_alter_table("comparison_record", schema=None) as batch_op:
batch_op.drop_column("best_deeplink")
@@ -1,76 +0,0 @@
"""coin reward points phase 1 tables
Revision ID: coin_reward_phase1
Revises: pstat3growth
Create Date: 2026-06-07 15:30:00.000000
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision: str = 'coin_reward_phase1'
down_revision: Union[str, Sequence[str], None] = 'pstat3growth'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
op.create_table(
'signin_boost_record',
sa.Column('id', sa.Integer(), autoincrement=True, nullable=False),
sa.Column('user_id', sa.Integer(), nullable=False),
sa.Column('signin_date', sa.Date(), nullable=False),
sa.Column('coin_awarded', sa.Integer(), nullable=False),
sa.Column('ad_ref_id', sa.String(length=64), 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', 'signin_date', name='uq_signin_boost_user_date'),
)
with op.batch_alter_table('signin_boost_record', schema=None) as batch_op:
batch_op.create_index(batch_op.f('ix_signin_boost_record_user_id'), ['user_id'], unique=False)
op.create_table(
'ad_feed_reward_record',
sa.Column('id', sa.Integer(), autoincrement=True, nullable=False),
sa.Column('client_event_id', sa.String(length=64), nullable=False),
sa.Column('user_id', sa.Integer(), nullable=False),
sa.Column('reward_date', sa.String(length=10), nullable=False),
sa.Column('duration_seconds', sa.Integer(), nullable=False),
sa.Column('unit_count', sa.Integer(), nullable=False),
sa.Column('ecpm_raw', 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('coin', sa.Integer(), nullable=False),
sa.Column('status', sa.String(length=16), 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'),
sa.UniqueConstraint('client_event_id', name='uq_ad_feed_reward_client_event'),
)
with op.batch_alter_table('ad_feed_reward_record', schema=None) as batch_op:
batch_op.create_index(batch_op.f('ix_ad_feed_reward_record_user_id'), ['user_id'], unique=False)
batch_op.create_index(batch_op.f('ix_ad_feed_reward_record_reward_date'), ['reward_date'], unique=False)
batch_op.create_index(batch_op.f('ix_ad_feed_reward_record_created_at'), ['created_at'], unique=False)
def downgrade() -> None:
with op.batch_alter_table('ad_feed_reward_record', schema=None) as batch_op:
batch_op.drop_index(batch_op.f('ix_ad_feed_reward_record_created_at'))
batch_op.drop_index(batch_op.f('ix_ad_feed_reward_record_reward_date'))
batch_op.drop_index(batch_op.f('ix_ad_feed_reward_record_user_id'))
op.drop_table('ad_feed_reward_record')
with op.batch_alter_table('signin_boost_record', schema=None) as batch_op:
batch_op.drop_index(batch_op.f('ix_signin_boost_record_user_id'))
op.drop_table('signin_boost_record')
@@ -1,41 +0,0 @@
"""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')
@@ -1,28 +0,0 @@
"""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')
@@ -1,64 +0,0 @@
"""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')
-77
View File
@@ -1,77 +0,0 @@
"""coupon state tables (领券今日状态:弹窗频控 engagement + 领券记录 claim)
Revision ID: coupon_state_tables
Revises: 11a1d08c6f55
Create Date: 2026-06-08 00:00:00.000000
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
from sqlalchemy.dialects import postgresql
# revision identifiers, used by Alembic.
revision: str = 'coupon_state_tables'
down_revision: Union[str, Sequence[str], None] = '11a1d08c6f55'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
# PG 上为 JSONB,其它(SQLite dev)为 JSON——与模型层 with_variant 对齐
_JSON = sa.JSON().with_variant(postgresql.JSONB(), 'postgresql')
def upgrade() -> None:
# 领券记录(资产层,当前不参与去重判断)
op.create_table(
'coupon_claim_record',
sa.Column('id', sa.Integer(), autoincrement=True, nullable=False),
sa.Column('device_id', sa.String(length=64), nullable=False),
sa.Column('user_id', sa.Integer(), nullable=True),
sa.Column('coupon_id', sa.String(length=64), nullable=False),
sa.Column('claim_date', sa.Date(), nullable=False),
sa.Column('status', sa.String(length=24), nullable=False),
sa.Column('vendor', sa.String(length=48), nullable=True),
sa.Column('coupon_name', sa.String(length=128), nullable=True),
sa.Column('claimed_count', sa.Integer(), nullable=True),
sa.Column('trace_id', sa.String(length=64), nullable=True),
sa.Column('reason', sa.String(length=255), nullable=True),
sa.Column('extra', _JSON, nullable=True),
sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('(CURRENT_TIMESTAMP)'), nullable=False),
sa.Column('updated_at', sa.DateTime(timezone=True), server_default=sa.text('(CURRENT_TIMESTAMP)'), nullable=False),
sa.PrimaryKeyConstraint('id'),
sa.UniqueConstraint('device_id', 'coupon_id', 'claim_date', name='uq_coupon_claim_device_coupon_date'),
)
with op.batch_alter_table('coupon_claim_record', schema=None) as batch_op:
batch_op.create_index('ix_coupon_claim_device_date', ['device_id', 'claim_date'], unique=False)
batch_op.create_index(batch_op.f('ix_coupon_claim_record_user_id'), ['user_id'], unique=False)
batch_op.create_index(batch_op.f('ix_coupon_claim_record_trace_id'), ['trace_id'], unique=False)
# 弹窗频控(今天 engage 过就不再弹)
op.create_table(
'coupon_prompt_engagement',
sa.Column('id', sa.Integer(), autoincrement=True, nullable=False),
sa.Column('device_id', sa.String(length=64), nullable=False),
sa.Column('user_id', sa.Integer(), nullable=True),
sa.Column('engage_date', sa.Date(), nullable=False),
sa.Column('engage_type', sa.String(length=16), nullable=False),
sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('(CURRENT_TIMESTAMP)'), nullable=False),
sa.Column('updated_at', sa.DateTime(timezone=True), server_default=sa.text('(CURRENT_TIMESTAMP)'), nullable=False),
sa.PrimaryKeyConstraint('id'),
sa.UniqueConstraint('device_id', 'engage_date', name='uq_coupon_engage_device_date'),
)
with op.batch_alter_table('coupon_prompt_engagement', schema=None) as batch_op:
batch_op.create_index(batch_op.f('ix_coupon_prompt_engagement_user_id'), ['user_id'], unique=False)
def downgrade() -> None:
with op.batch_alter_table('coupon_prompt_engagement', schema=None) as batch_op:
batch_op.drop_index(batch_op.f('ix_coupon_prompt_engagement_user_id'))
op.drop_table('coupon_prompt_engagement')
with op.batch_alter_table('coupon_claim_record', schema=None) as batch_op:
batch_op.drop_index(batch_op.f('ix_coupon_claim_record_trace_id'))
batch_op.drop_index(batch_op.f('ix_coupon_claim_record_user_id'))
batch_op.drop_index('ix_coupon_claim_device_date')
op.drop_table('coupon_claim_record')
@@ -1,26 +0,0 @@
"""merge app_config and price_report heads
Revision ID: ebb6af5c0b56
Revises: cfg1a2b3c4d5, b7e2c1a9f4d3
Create Date: 2026-06-06 02:29:13.475265
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision: str = 'ebb6af5c0b56'
down_revision: Union[str, Sequence[str], None] = ('cfg1a2b3c4d5', 'b7e2c1a9f4d3')
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
pass
def downgrade() -> None:
pass
@@ -1,54 +0,0 @@
"""user.invite_code + invite_relation table
Revision ID: invite_code_and_relation
Revises: 11a1d08c6f55
Create Date: 2026-06-08 00:00:00.000000
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision: str = 'invite_code_and_relation'
down_revision: Union[str, Sequence[str], None] = '11a1d08c6f55'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
# user 加邀请码列(唯一索引;nullable 允许多个 NULL=未生成的用户)
with op.batch_alter_table('user', schema=None) as batch_op:
batch_op.add_column(sa.Column('invite_code', sa.String(length=16), nullable=True))
batch_op.create_index('ix_user_invite_code', ['invite_code'], unique=True)
# 邀请关系表
op.create_table(
'invite_relation',
sa.Column('id', sa.Integer(), autoincrement=True, nullable=False),
sa.Column('inviter_user_id', sa.Integer(), nullable=False),
sa.Column('invitee_user_id', sa.Integer(), nullable=False),
sa.Column('channel', sa.String(length=16), nullable=False, server_default='clipboard'),
sa.Column('status', sa.String(length=16), nullable=False, server_default='effective'),
sa.Column('inviter_coin', sa.Integer(), nullable=False, server_default='0'),
sa.Column('invitee_coin', sa.Integer(), nullable=False, server_default='0'),
sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
sa.ForeignKeyConstraint(['inviter_user_id'], ['user.id']),
sa.ForeignKeyConstraint(['invitee_user_id'], ['user.id']),
sa.PrimaryKeyConstraint('id'),
)
op.create_index('ix_invite_relation_inviter_user_id', 'invite_relation', ['inviter_user_id'])
op.create_index('ix_invite_relation_invitee_user_id', 'invite_relation', ['invitee_user_id'], unique=True)
op.create_index('ix_invite_relation_created_at', 'invite_relation', ['created_at'])
def downgrade() -> None:
op.drop_index('ix_invite_relation_created_at', table_name='invite_relation')
op.drop_index('ix_invite_relation_invitee_user_id', table_name='invite_relation')
op.drop_index('ix_invite_relation_inviter_user_id', table_name='invite_relation')
op.drop_table('invite_relation')
with op.batch_alter_table('user', schema=None) as batch_op:
batch_op.drop_index('ix_user_invite_code')
batch_op.drop_column('invite_code')
-50
View File
@@ -1,50 +0,0 @@
"""marquee_seed 升级为「生成规则」:用户名可空 + 金额拆 min/max 区间
把种子从「死记录」升级:
- masked_user 改 nullable(留空 → feed 随机合成脱敏用户名)
- saved_amount_cents(单值)拆成 min_cents / max_cents(feed 在区间内随机取值;固定金额则相等)
现有 6 条种子:min_cents = max_cents = 原 saved_amount_cents,masked_user 原样保留。
Revision ID: marquee_seed02
Revises: marquee_seed01
Create Date: 2026-06-07 11:00:00.000000
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
revision: str = 'marquee_seed02'
down_revision: Union[str, Sequence[str], None] = 'marquee_seed01'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
# 1) 先加可空新列,使现有行合法
with op.batch_alter_table('marquee_seed', schema=None) as batch_op:
batch_op.add_column(sa.Column('min_cents', sa.Integer(), nullable=True))
batch_op.add_column(sa.Column('max_cents', sa.Integer(), nullable=True))
# 2) 用旧单值回填区间(固定金额)
op.execute('UPDATE marquee_seed SET min_cents = saved_amount_cents, max_cents = saved_amount_cents')
# 3) 收紧:区间非空、用户名可空、删旧列
with op.batch_alter_table('marquee_seed', schema=None) as batch_op:
batch_op.alter_column('min_cents', existing_type=sa.Integer(), nullable=False)
batch_op.alter_column('max_cents', existing_type=sa.Integer(), nullable=False)
batch_op.alter_column('masked_user', existing_type=sa.String(length=64), nullable=True)
batch_op.drop_column('saved_amount_cents')
def downgrade() -> None:
# 反向:恢复单值列(取下限),用户名回非空
with op.batch_alter_table('marquee_seed', schema=None) as batch_op:
batch_op.add_column(sa.Column('saved_amount_cents', sa.Integer(), nullable=True))
op.execute('UPDATE marquee_seed SET saved_amount_cents = min_cents')
# 自动合成名的种子 masked_user 为 NULL,降级前补占位符以满足非空
op.execute("UPDATE marquee_seed SET masked_user = '用户********000' WHERE masked_user IS NULL")
with op.batch_alter_table('marquee_seed', schema=None) as batch_op:
batch_op.alter_column('saved_amount_cents', existing_type=sa.Integer(), nullable=False)
batch_op.alter_column('masked_user', existing_type=sa.String(length=64), nullable=False)
batch_op.drop_column('max_cents')
batch_op.drop_column('min_cents')
-52
View File
@@ -1,52 +0,0 @@
"""marquee_seed table (首页轮播种子) + 合并 platform_stat / price_observation 两 head
Revision ID: marquee_seed01
Revises: pstat2anchor01, price_observation_table
Create Date: 2026-06-07 10:00:00.000000
兼并当前两个 head 为单 head,同时建 marquee_seed 表并播种客户端原写死的 6 条轮播
(作为初始种子,保证上线即有内容)。
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
revision: str = 'marquee_seed01'
down_revision: Union[str, Sequence[str], None] = ('pstat2anchor01', 'price_observation_table')
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
table = op.create_table(
'marquee_seed',
sa.Column('id', sa.Integer(), autoincrement=True, nullable=False),
sa.Column('masked_user', sa.String(length=64), nullable=False),
sa.Column('saved_amount_cents', sa.Integer(), nullable=False),
sa.Column('enabled', sa.Boolean(), nullable=False, server_default=sa.true()),
sa.Column('sort_order', sa.Integer(), nullable=False, server_default='0'),
sa.Column(
'created_at', sa.DateTime(timezone=True),
server_default=sa.text('(CURRENT_TIMESTAMP)'), nullable=False,
),
sa.PrimaryKeyConstraint('id'),
)
# 初始种子 = 客户端原写死的 6 条(金额转成分)
op.bulk_insert(
table,
[
{'masked_user': '用户********a52', 'saved_amount_cents': 1860, 'enabled': True, 'sort_order': 1},
{'masked_user': '用户********k89', 'saved_amount_cents': 4240, 'enabled': True, 'sort_order': 2},
{'masked_user': '用户********m71', 'saved_amount_cents': 790, 'enabled': True, 'sort_order': 3},
{'masked_user': '用户********p46', 'saved_amount_cents': 15630, 'enabled': True, 'sort_order': 4},
{'masked_user': '用户********r93', 'saved_amount_cents': 2350, 'enabled': True, 'sort_order': 5},
{'masked_user': '用户********c28', 'saved_amount_cents': 8940, 'enabled': True, 'sort_order': 6},
],
)
def downgrade() -> None:
op.drop_table('marquee_seed')
-62
View File
@@ -1,62 +0,0 @@
"""meituan_coupon table(美团 CPS 券本地缓存,供销量/佣金排序从库里捞、本地排序)
Revision ID: meituan_coupon_table
Revises: withdraw_review_ad_watch
Create Date: 2026-06-06 12:00:00.000000
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
from sqlalchemy.dialects import postgresql
# revision identifiers, used by Alembic.
revision: str = 'meituan_coupon_table'
down_revision: Union[str, Sequence[str], None] = 'withdraw_review_ad_watch'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
op.create_table(
'meituan_coupon',
sa.Column('id', sa.Integer(), autoincrement=True, nullable=False),
sa.Column('source', sa.String(length=16), nullable=False),
sa.Column('platform', sa.Integer(), nullable=False),
sa.Column('biz_line', sa.Integer(), nullable=True),
sa.Column('city_id', sa.String(length=32), nullable=False),
sa.Column('product_view_sign', sa.String(length=128), nullable=False),
sa.Column('sku_view_id', sa.String(length=128), nullable=True),
sa.Column('name', sa.String(length=256), nullable=True),
sa.Column('brand_name', sa.String(length=128), nullable=True),
sa.Column('sell_price_cents', sa.Integer(), nullable=True),
sa.Column('original_price_cents', sa.Integer(), nullable=True),
sa.Column('head_url', sa.String(length=512), nullable=True),
sa.Column('sale_volume', sa.String(length=32), nullable=True),
sa.Column('sale_volume_num', sa.Integer(), nullable=True),
sa.Column('commission_percent', sa.Float(), nullable=True),
sa.Column('commission_amount_cents', sa.Integer(), nullable=True),
sa.Column('poi_name', sa.String(length=128), nullable=True),
sa.Column('available_poi_num', sa.Integer(), nullable=True),
sa.Column('delivery_distance_m', sa.Float(), nullable=True),
sa.Column('dedup_key', sa.String(length=64), nullable=False),
sa.Column('raw', sa.JSON().with_variant(postgresql.JSONB(), 'postgresql'), nullable=False),
sa.Column('first_seen', sa.DateTime(timezone=True), server_default=sa.text('(CURRENT_TIMESTAMP)'), nullable=False),
sa.Column('last_seen', sa.DateTime(timezone=True), server_default=sa.text('(CURRENT_TIMESTAMP)'), nullable=False),
sa.Column('updated_at', sa.DateTime(timezone=True), server_default=sa.text('(CURRENT_TIMESTAMP)'), nullable=False),
sa.PrimaryKeyConstraint('id'),
sa.UniqueConstraint('source', 'product_view_sign', name='uq_meituan_coupon_source_sign'),
)
op.create_index('ix_meituan_coupon_source', 'meituan_coupon', ['source'])
op.create_index('ix_meituan_coupon_city_id', 'meituan_coupon', ['city_id'])
op.create_index('ix_meituan_coupon_brand_name', 'meituan_coupon', ['brand_name'])
op.create_index('ix_meituan_coupon_sale_volume_num', 'meituan_coupon', ['sale_volume_num'])
op.create_index('ix_meituan_coupon_commission_percent', 'meituan_coupon', ['commission_percent'])
op.create_index('ix_meituan_coupon_dedup_key', 'meituan_coupon', ['dedup_key'])
op.create_index('ix_meituan_coupon_last_seen', 'meituan_coupon', ['last_seen'])
def downgrade() -> None:
op.drop_table('meituan_coupon')
-28
View File
@@ -1,28 +0,0 @@
"""运营可配表加 ops_ 前缀:platform_stat_display→ops_stat_config / marquee_seed→ops_marquee_seed
把首页门面数据的两张「运营可配」表重命名,与代码侧 ops_ 前缀(模型 OpsStatConfig/OpsMarqueeSeed、
repo ops_stat/ops_marquee、admin ops_stat_config/ops_marquee_seed)对齐。仅改表名,不动列与数据。
Revision ID: opsrename01
Revises: coin_reward_phase1
Create Date: 2026-06-07 16:00:00.000000
"""
from typing import Sequence, Union
from alembic import op
revision: str = 'opsrename01'
down_revision: Union[str, Sequence[str], None] = 'coin_reward_phase1'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
op.rename_table('platform_stat_display', 'ops_stat_config')
op.rename_table('marquee_seed', 'ops_marquee_seed')
def downgrade() -> None:
op.rename_table('ops_stat_config', 'platform_stat_display')
op.rename_table('ops_marquee_seed', 'marquee_seed')
@@ -1,30 +0,0 @@
"""platform_stat_display 增加 random_anchor_minutes(自增长触发时刻对齐偏移)
Revision ID: pstat2anchor01
Revises: pstat1d2e3f4a
Create Date: 2026-06-06 13:00:00.000000
自增长 tick 从「启动时刻 + 间隔」改为「北京时间钟点对齐(anchor + k*interval)」,
新增列存对齐偏移(距 0 点分钟数)。已有行默认 0(= 对齐到自然边界:天→0点、小时→整点)。
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
revision: str = 'pstat2anchor01'
down_revision: Union[str, Sequence[str], None] = 'pstat1d2e3f4a'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
op.add_column(
'platform_stat_display',
sa.Column('random_anchor_minutes', sa.Integer(), nullable=False, server_default='0'),
)
def downgrade() -> None:
op.drop_column('platform_stat_display', 'random_anchor_minutes')
@@ -1,59 +0,0 @@
"""platform_stat_display table (首页三统计展示配置:每指标独立 real/manual/random)
Revision ID: pstat1d2e3f4a
Revises: withdraw_review_ad_watch
Create Date: 2026-06-06 12:00:00.000000
播种 3 行:默认 manual + 客户端原写死门面值(12847 人 / 86532 次 / 37621.4 元),
保证上线前后展示一致,运营再按需切模式。
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision: str = 'pstat1d2e3f4a'
down_revision: Union[str, Sequence[str], None] = 'withdraw_review_ad_watch'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
table = op.create_table(
'platform_stat_display',
sa.Column('metric', sa.String(length=32), nullable=False),
sa.Column('mode', sa.String(length=16), nullable=False),
sa.Column('manual_value', sa.Integer(), nullable=True),
sa.Column('random_mult_min', sa.Integer(), nullable=False),
sa.Column('random_mult_max', sa.Integer(), nullable=False),
sa.Column('random_tick_seconds', sa.Integer(), nullable=False),
sa.Column('random_current', sa.Integer(), nullable=True),
sa.Column('random_last_tick_at', sa.DateTime(timezone=True), nullable=True),
sa.Column('updated_by_admin_id', sa.Integer(), nullable=True),
sa.Column(
'updated_at', sa.DateTime(timezone=True),
server_default=sa.text('(CURRENT_TIMESTAMP)'), nullable=False,
),
sa.PrimaryKeyConstraint('metric'),
)
_common = {
'mode': 'manual',
'random_mult_min': 1000, # 1.000(只增不减下限)
'random_mult_max': 1100, # 1.100
'random_tick_seconds': 86400, # 1 天
}
op.bulk_insert(
table,
[
{'metric': 'help_users', 'manual_value': 12847, **_common},
{'metric': 'total_compares', 'manual_value': 86532, **_common},
{'metric': 'total_saved', 'manual_value': 3762140, **_common}, # 37621.40 元
],
)
def downgrade() -> None:
op.drop_table('platform_stat_display')
@@ -1,39 +0,0 @@
"""platform_stat_display 增强:绝对增量增长 / 真实值偏移 / 只增不减护栏
给首页三统计配置加 5 列:
- random_kind ('mult'/'add') + random_step_min/max:自增长支持「绝对增量」(每周期 +[a,b]),不只是 ×倍率
- real_offset:真实值模式基数偏移(展示=真实+偏移,冷启动既真实又体面)
- allow_decrease:门面数字默认「只增不减」,开此开关才允许展示值下降
Revision ID: pstat3growth
Revises: marquee_seed02
Create Date: 2026-06-07 12:00:00.000000
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
revision: str = 'pstat3growth'
down_revision: Union[str, Sequence[str], None] = 'marquee_seed02'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
with op.batch_alter_table('platform_stat_display', schema=None) as batch_op:
batch_op.add_column(sa.Column('random_kind', sa.String(length=8), nullable=False, server_default='mult'))
batch_op.add_column(sa.Column('random_step_min', sa.Integer(), nullable=False, server_default='0'))
batch_op.add_column(sa.Column('random_step_max', sa.Integer(), nullable=False, server_default='0'))
batch_op.add_column(sa.Column('real_offset', sa.Integer(), nullable=False, server_default='0'))
batch_op.add_column(sa.Column('allow_decrease', sa.Boolean(), nullable=False, server_default=sa.false()))
def downgrade() -> None:
with op.batch_alter_table('platform_stat_display', schema=None) as batch_op:
batch_op.drop_column('allow_decrease')
batch_op.drop_column('real_offset')
batch_op.drop_column('random_step_max')
batch_op.drop_column('random_step_min')
batch_op.drop_column('random_kind')
@@ -1,74 +0,0 @@
"""price_observation table (价格观测:比价沉淀的价格资产层,server 侧无条件落库)
Revision ID: price_observation_table
Revises: wx_transfer_auth
Create Date: 2026-06-07 00:00:00.000000
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision: str = 'price_observation_table'
down_revision: Union[str, Sequence[str], None] = 'wx_transfer_auth'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
op.create_table(
'price_observation',
sa.Column('id', sa.Integer(), autoincrement=True, nullable=False),
sa.Column('observed_at', sa.DateTime(timezone=True), server_default=sa.text('(CURRENT_TIMESTAMP)'), nullable=False),
sa.Column('trace_id', sa.String(length=64), nullable=False),
sa.Column('business_type', sa.String(length=16), nullable=False),
sa.Column('platform', sa.String(length=32), nullable=False),
sa.Column('platform_store_id', sa.String(length=64), nullable=True),
sa.Column('store_name', sa.String(length=128), nullable=True),
sa.Column('city', sa.String(length=64), nullable=True),
sa.Column('geohash', sa.String(length=16), nullable=True),
sa.Column('lng', sa.Float(), nullable=True),
sa.Column('lat', sa.Float(), nullable=True),
sa.Column('is_source', sa.Boolean(), nullable=False),
sa.Column('scope', sa.String(length=16), nullable=False),
sa.Column('price_cents', sa.Integer(), nullable=True),
sa.Column('coupon_saved_cents', sa.Integer(), nullable=True),
sa.Column('coupon_name', sa.String(length=64), nullable=True),
sa.Column('store_closed', sa.String(length=32), nullable=True),
sa.Column('rank', sa.Integer(), nullable=True),
# PG 上为 JSONB,其它(SQLite)为 JSON——与模型层 with_variant 对齐
sa.Column('dishes', sa.JSON().with_variant(sa.dialects.postgresql.JSONB(), 'postgresql'), nullable=True),
sa.Column('attrs', sa.JSON().with_variant(sa.dialects.postgresql.JSONB(), 'postgresql'), nullable=True),
sa.Column('source_device_id', sa.String(length=64), nullable=True),
sa.Column('source_user_id', sa.Integer(), nullable=True),
sa.Column('confidence', sa.Float(), nullable=False),
sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('(CURRENT_TIMESTAMP)'), nullable=False),
sa.PrimaryKeyConstraint('id'),
sa.UniqueConstraint('trace_id', 'platform', 'scope', name='uq_price_obs_trace_platform_scope'),
)
with op.batch_alter_table('price_observation', schema=None) as batch_op:
batch_op.create_index(batch_op.f('ix_price_observation_observed_at'), ['observed_at'], unique=False)
batch_op.create_index(batch_op.f('ix_price_observation_trace_id'), ['trace_id'], unique=False)
batch_op.create_index(batch_op.f('ix_price_observation_business_type'), ['business_type'], unique=False)
batch_op.create_index(batch_op.f('ix_price_observation_platform'), ['platform'], unique=False)
batch_op.create_index(batch_op.f('ix_price_observation_store_name'), ['store_name'], unique=False)
batch_op.create_index(batch_op.f('ix_price_observation_geohash'), ['geohash'], unique=False)
batch_op.create_index(batch_op.f('ix_price_observation_source_device_id'), ['source_device_id'], unique=False)
batch_op.create_index(batch_op.f('ix_price_observation_source_user_id'), ['source_user_id'], unique=False)
def downgrade() -> None:
with op.batch_alter_table('price_observation', schema=None) as batch_op:
batch_op.drop_index(batch_op.f('ix_price_observation_source_user_id'))
batch_op.drop_index(batch_op.f('ix_price_observation_source_device_id'))
batch_op.drop_index(batch_op.f('ix_price_observation_geohash'))
batch_op.drop_index(batch_op.f('ix_price_observation_store_name'))
batch_op.drop_index(batch_op.f('ix_price_observation_platform'))
batch_op.drop_index(batch_op.f('ix_price_observation_business_type'))
batch_op.drop_index(batch_op.f('ix_price_observation_trace_id'))
batch_op.drop_index(batch_op.f('ix_price_observation_observed_at'))
op.drop_table('price_observation')
-49
View File
@@ -1,49 +0,0 @@
"""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')
@@ -1,48 +0,0 @@
"""wechat_transfer_authorization 表(免确认收款授权)
商家转账「用户授权免确认收款模式」:用户授权一次后,后续提现免逐笔确认直接到账。
一个用户一条(user_id 主键)。out_authorization_no 我方生成,authorization_id 微信 active 后返回。
Revision ID: wx_transfer_auth
Revises: withdraw_review_ad_watch
Create Date: 2026-06-06 12:00:00.000000
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision: str = 'wx_transfer_auth'
down_revision: Union[str, Sequence[str], None] = 'withdraw_review_ad_watch'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
op.create_table(
'wechat_transfer_authorization',
sa.Column('user_id', sa.Integer(), nullable=False),
sa.Column('openid', sa.String(length=64), nullable=False),
sa.Column('out_authorization_no', sa.String(length=64), nullable=False),
sa.Column('authorization_id', sa.String(length=64), nullable=True),
sa.Column('state', sa.String(length=16), server_default='pending', nullable=False),
sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('(CURRENT_TIMESTAMP)'), nullable=False),
sa.Column('updated_at', sa.DateTime(timezone=True), server_default=sa.text('(CURRENT_TIMESTAMP)'), nullable=False),
sa.ForeignKeyConstraint(['user_id'], ['user.id'], ),
sa.PrimaryKeyConstraint('user_id'),
)
with op.batch_alter_table('wechat_transfer_authorization', schema=None) as batch_op:
batch_op.create_index(
batch_op.f('ix_wechat_transfer_authorization_out_authorization_no'),
['out_authorization_no'],
unique=True,
)
def downgrade() -> None:
with op.batch_alter_table('wechat_transfer_authorization', schema=None) as batch_op:
batch_op.drop_index(batch_op.f('ix_wechat_transfer_authorization_out_authorization_no'))
op.drop_table('wechat_transfer_authorization')
@@ -1,62 +0,0 @@
"""withdraw_order.user_name 列 + ad_watch_log 表
提现审核功能:WithdrawOrder 加 user_name(审核异步打款时传给微信的实名,达额转账要求)。
看广告时长防刷:新增 ad_watch_log 表(前端 onAdClose 上报观看秒数,按 (user_id, watch_date)
聚合做"每天 50 分钟"硬上限)。
Revision ID: withdraw_review_ad_watch
Revises: ebb6af5c0b56
Create Date: 2026-06-05 12:00:00.000000
注:本迁移原 down_revision=ad60a1b2c3d4(分支创建时的 head);rebase 合 main 后 main 侧已新增
app_config / price_report 等迁移并 merge 出 head=ebb6af5c0b56,故改挂到其上,保持单 head(本迁移
与那些表互不相干,叠加顺序无影响)。
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision: str = 'withdraw_review_ad_watch'
down_revision: Union[str, Sequence[str], None] = 'ebb6af5c0b56'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
# ① 提现单加实名列:审核通过后异步打款时传给微信(达额转账要求),发起提现时存下
with op.batch_alter_table('withdraw_order', schema=None) as batch_op:
batch_op.add_column(sa.Column('user_name', sa.String(length=64), nullable=True))
# ② 看广告观看时长表:前端上报观看秒数,(user_id, watch_date) 聚合做每日 50 分钟防刷主闸
op.create_table(
'ad_watch_log',
sa.Column('id', sa.Integer(), autoincrement=True, nullable=False),
sa.Column('user_id', sa.Integer(), nullable=False),
sa.Column('watch_seconds', sa.Integer(), server_default='0', nullable=False),
sa.Column('watch_date', sa.String(length=10), nullable=False),
sa.Column(
'created_at', sa.DateTime(timezone=True),
server_default=sa.text('(CURRENT_TIMESTAMP)'), nullable=False,
),
sa.ForeignKeyConstraint(['user_id'], ['user.id'], ),
sa.PrimaryKeyConstraint('id'),
)
with op.batch_alter_table('ad_watch_log', schema=None) as batch_op:
batch_op.create_index(batch_op.f('ix_ad_watch_log_user_id'), ['user_id'], unique=False)
batch_op.create_index(batch_op.f('ix_ad_watch_log_watch_date'), ['watch_date'], unique=False)
batch_op.create_index(batch_op.f('ix_ad_watch_log_created_at'), ['created_at'], unique=False)
def downgrade() -> None:
with op.batch_alter_table('ad_watch_log', schema=None) as batch_op:
batch_op.drop_index(batch_op.f('ix_ad_watch_log_created_at'))
batch_op.drop_index(batch_op.f('ix_ad_watch_log_watch_date'))
batch_op.drop_index(batch_op.f('ix_ad_watch_log_user_id'))
op.drop_table('ad_watch_log')
with op.batch_alter_table('withdraw_order', schema=None) as batch_op:
batch_op.drop_column('user_name')
@@ -1,42 +0,0 @@
"""add withdraw concurrency safety indexes
Revision ID: withdraw_safety_indexes
Revises: 0cf18d590b1d
Create Date: 2026-06-08 16:20:00.000000
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision: str = "withdraw_safety_indexes"
down_revision: Union[str, Sequence[str], None] = "0cf18d590b1d"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
op.create_index(
"ux_withdraw_order_user_active",
"withdraw_order",
["user_id"],
unique=True,
sqlite_where=sa.text("status IN ('reviewing', 'pending')"),
postgresql_where=sa.text("status IN ('reviewing', 'pending')"),
)
op.create_index(
"ux_cash_transaction_withdraw_refund_ref",
"cash_transaction",
["ref_id"],
unique=True,
sqlite_where=sa.text("biz_type = 'withdraw_refund' AND ref_id IS NOT NULL"),
postgresql_where=sa.text("biz_type = 'withdraw_refund' AND ref_id IS NOT NULL"),
)
def downgrade() -> None:
op.drop_index("ux_cash_transaction_withdraw_refund_ref", table_name="cash_transaction")
op.drop_index("ux_withdraw_order_user_active", table_name="withdraw_order")
-6
View File
@@ -1,6 +0,0 @@
"""运营 Admin 后台子应用。
独立 FastAPI app(app.admin.main:admin_app),独立进程/端口运行,复用 App 的
models/repositories/integrations + 同一个 DB,但鉴权完全隔离(独立 JWT secret)。
现有 app.main:app 不 import 本包,admin 崩溃不影响 App 主进程。
"""
-38
View File
@@ -1,38 +0,0 @@
"""审计写入门面。
每个 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
@@ -1,84 +0,0 @@
"""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 ""
-86
View File
@@ -1,86 +0,0 @@
"""Admin 后台 FastAPI app(独立进程)。
启动:uvicorn app.admin.main:admin_app --host 127.0.0.1 --port 8771
复用 App 的 DB/models/repositories/integrations;鉴权独立(admin JWT,见 app/admin/security.py)。
现有 app.main:app 不 import 本模块,两进程互不影响。
"""
from __future__ import annotations
import logging
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from app.admin.routers.admins import router as admins_router
from app.admin.routers.audit import router as audit_router
from app.admin.routers.auth import router as auth_router
from app.admin.routers.config import router as config_router
from app.admin.routers.dashboard import router as dashboard_router
from app.admin.routers.ops_stat_config import router as ops_stat_config_router
from app.admin.routers.feedback import router as feedback_router
from app.admin.routers.ops_marquee_seed import router as ops_marquee_seed_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(ops_stat_config_router)
admin_app.include_router(ops_marquee_seed_router)
admin_app.include_router(users_router)
admin_app.include_router(wallet_router)
admin_app.include_router(withdraw_router)
admin_app.include_router(feedback_router)
admin_app.include_router(admins_router)
admin_app.include_router(audit_router)
admin_app.include_router(config_router)
-1
View File
@@ -1 +0,0 @@
"""admin 专用数据访问层(跨用户查询 + admin 账号 + 审计 + 大盘聚合)。"""
-64
View File
@@ -1,64 +0,0 @@
"""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
@@ -1,70 +0,0 @@
"""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
@@ -1,36 +0,0 @@
"""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
-397
View File
@@ -1,397 +0,0 @@
"""admin 跨用户查询(去掉现有 repo 的 user_id 强制过滤)+ 通用游标分页 helper + 用户概览。
现有 app/repositories/ 的 list_* 都强绑单个 user_id(C 端只看自己);admin 要看全量、按条件筛,
所以在这里另起一套。游标约定与现有一致:id 倒序,cursor=上页最后一条 id,返回 (items, next_cursor)。
"""
from __future__ import annotations
from datetime import datetime, timedelta, timezone
from zoneinfo import ZoneInfo
from sqlalchemy import Select, asc, desc, func, or_, select
from sqlalchemy.orm import Session
from app.models.admin import AdminAuditLog
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,
keyword: str | None = None,
date_from: datetime | None = None,
date_to: datetime | None = None,
date_field: str = "created_at",
sort_by: str = "created_at",
sort_order: str = "desc",
quick_filter: str | None = None,
limit: int = 20,
cursor: int | None = None,
) -> tuple[list[WithdrawOrder], int | None]:
stmt = select(WithdrawOrder)
needs_user_join = bool(keyword and keyword.strip()) or quick_filter == "high_risk"
if needs_user_join:
stmt = stmt.outerjoin(User, User.id == WithdrawOrder.user_id)
if user_id is not None:
stmt = stmt.where(WithdrawOrder.user_id == user_id)
if status:
stmt = stmt.where(WithdrawOrder.status == status)
kw = (keyword or "").strip()
if kw:
pattern = f"%{kw}%"
conditions = [
WithdrawOrder.out_bill_no.ilike(pattern),
WithdrawOrder.transfer_bill_no.ilike(pattern),
WithdrawOrder.user_name.ilike(pattern),
WithdrawOrder.wechat_state.ilike(pattern),
WithdrawOrder.fail_reason.ilike(pattern),
User.phone.ilike(pattern),
User.nickname.ilike(pattern),
User.wechat_nickname.ilike(pattern),
]
if kw.isdigit():
conditions.append(WithdrawOrder.user_id == int(kw))
stmt = stmt.where(or_(*conditions))
date_col = WithdrawOrder.updated_at if date_field == "updated_at" else WithdrawOrder.created_at
if date_from is not None:
stmt = stmt.where(date_col >= _as_utc_naive(date_from))
if date_to is not None:
stmt = stmt.where(date_col <= _as_utc_naive(date_to))
now = datetime.now(timezone.utc).replace(tzinfo=None)
today_start = (
datetime.now(ZoneInfo("Asia/Shanghai"))
.replace(hour=0, minute=0, second=0, microsecond=0)
.astimezone(timezone.utc)
.replace(tzinfo=None)
)
if quick_filter == "abnormal":
stmt = stmt.where(
or_(
WithdrawOrder.status.in_(("failed", "rejected")),
WithdrawOrder.fail_reason.is_not(None),
WithdrawOrder.wechat_state.in_(("FAIL", "CANCELLED", "CLOSED")),
)
)
elif quick_filter == "failed":
stmt = stmt.where(WithdrawOrder.status == "failed")
elif quick_filter == "overdue_reviewing":
stmt = stmt.where(
WithdrawOrder.status == "reviewing",
WithdrawOrder.created_at <= now - timedelta(minutes=30),
)
elif quick_filter == "pending_overdue":
stmt = stmt.where(
WithdrawOrder.status == "pending",
WithdrawOrder.updated_at <= now - timedelta(minutes=15),
)
elif quick_filter == "today":
stmt = stmt.where(WithdrawOrder.created_at >= today_start)
elif quick_filter == "high_risk":
stmt = stmt.where(
or_(
WithdrawOrder.user_name.is_(None),
WithdrawOrder.user_name == "",
User.status != "active",
User.created_at >= now - timedelta(hours=24),
WithdrawOrder.status.in_(("failed", "rejected")),
WithdrawOrder.fail_reason.is_not(None),
)
)
sort_cols = {
"id": WithdrawOrder.id,
"created_at": WithdrawOrder.created_at,
"updated_at": WithdrawOrder.updated_at,
"amount_cents": WithdrawOrder.amount_cents,
}
sort_col = sort_cols.get(sort_by, WithdrawOrder.created_at)
order_fn = asc if sort_order == "asc" else desc
id_order = asc(WithdrawOrder.id) if sort_order == "asc" else desc(WithdrawOrder.id)
stmt = stmt.order_by(order_fn(sort_col), id_order)
offset = max(cursor or 0, 0)
rows = list(db.execute(stmt.offset(offset).limit(limit + 1)).scalars().all())
has_more = len(rows) > limit
items = rows[:limit]
next_cursor = offset + limit if has_more else None
return items, next_cursor
def _as_utc_naive(value: datetime) -> datetime:
"""前端传 ISO 时间;DB 当前按 UTC naive 比较最稳(SQLite/本地开发一致)。"""
if value.tzinfo is None:
return value
return value.astimezone(timezone.utc).replace(tzinfo=None)
def list_feedbacks(
db: Session,
*,
status: str | None = None,
user_id: int | None = None,
limit: int = 20,
cursor: int | None = None,
) -> tuple[list[Feedback], int | None]:
stmt = select(Feedback)
if status:
stmt = stmt.where(Feedback.status == status)
if user_id is not None:
stmt = stmt.where(Feedback.user_id == user_id)
return cursor_paginate(db, stmt, Feedback.id, limit=limit, cursor=cursor)
def get_withdraw_by_out_bill_no(db: Session, out_bill_no: str) -> WithdrawOrder | None:
"""按商户单号查提现单(admin 重试打款先拿 user_id 用,M3)。"""
return db.execute(
select(WithdrawOrder).where(WithdrawOrder.out_bill_no == out_bill_no)
).scalar_one_or_none()
def withdraw_summary(db: Session) -> dict:
"""提现审核台顶部统计。金额单位:分。"""
rows = db.execute(
select(
WithdrawOrder.status,
func.count(WithdrawOrder.id),
func.coalesce(func.sum(WithdrawOrder.amount_cents), 0),
).group_by(WithdrawOrder.status)
).all()
by_status = {
status: {"count": int(count), "amount_cents": int(amount_cents)}
for status, count, amount_cents in rows
}
today_start = (
datetime.now(ZoneInfo("Asia/Shanghai"))
.replace(hour=0, minute=0, second=0, microsecond=0)
.astimezone(timezone.utc)
)
def _today_count(status: str) -> int:
return db.execute(
select(func.count(WithdrawOrder.id)).where(
WithdrawOrder.status == status,
WithdrawOrder.updated_at >= today_start,
)
).scalar_one()
today_success_amount = db.execute(
select(func.coalesce(func.sum(WithdrawOrder.amount_cents), 0)).where(
WithdrawOrder.status == "success",
WithdrawOrder.updated_at >= today_start,
)
).scalar_one()
return {
"reviewing_count": by_status.get("reviewing", {}).get("count", 0),
"reviewing_amount_cents": by_status.get("reviewing", {}).get("amount_cents", 0),
"pending_count": by_status.get("pending", {}).get("count", 0),
"failed_count": by_status.get("failed", {}).get("count", 0),
"today_success_count": _today_count("success"),
"today_success_amount_cents": int(today_success_amount),
"today_rejected_count": _today_count("rejected"),
}
def list_withdraw_audit_logs(
db: Session, out_bill_no: str, *, limit: int = 20
) -> list[AdminAuditLog]:
return list(
db.execute(
select(AdminAuditLog)
.where(AdminAuditLog.target_type == "withdraw", AdminAuditLog.target_id == out_bill_no)
.order_by(AdminAuditLog.id.desc())
.limit(limit)
).scalars().all()
)
def withdraw_risk_flags(
order: WithdrawOrder,
user: User | None,
recent_withdraws: list[WithdrawOrder],
cash_balance_cents: int,
) -> tuple[list[str], int]:
flags: list[str] = []
if not order.user_name:
flags.append("缺少提现实名")
if user and user.status != "active":
flags.append(f"账号状态:{user.status}")
if user and user.created_at:
created_at = user.created_at.replace(tzinfo=timezone.utc) if user.created_at.tzinfo is None else user.created_at
if datetime.now(timezone.utc) - created_at < timedelta(hours=24):
flags.append("新注册用户")
failed_or_rejected = sum(1 for item in recent_withdraws if item.status in {"failed", "rejected"})
if failed_or_rejected:
flags.append(f"历史异常提现{failed_or_rejected}")
recent_reviewing = sum(1 for item in recent_withdraws if item.status == "reviewing")
if recent_reviewing >= 3:
flags.append(f"待审核提现偏多:{recent_reviewing}")
if cash_balance_cents < 0:
flags.append("现金余额为负")
score = min(100, len(flags) * 20 + failed_or_rejected * 10)
return flags, score
def withdraw_ledger_check(db: Session) -> dict:
cash_balance_total = int(
db.execute(select(func.coalesce(func.sum(CoinAccount.cash_balance_cents), 0))).scalar_one()
)
cash_txn_total = int(
db.execute(select(func.coalesce(func.sum(CashTransaction.amount_cents), 0))).scalar_one()
)
orders = list(db.execute(select(WithdrawOrder)).scalars().all())
cash_txns = list(
db.execute(
select(CashTransaction).where(
CashTransaction.biz_type.in_(("withdraw", "withdraw_refund"))
)
).scalars().all()
)
withdraw_refs = {txn.ref_id for txn in cash_txns if txn.biz_type == "withdraw"}
refund_counts: dict[str, int] = {}
for txn in cash_txns:
if txn.biz_type == "withdraw_refund" and txn.ref_id:
refund_counts[txn.ref_id] = refund_counts.get(txn.ref_id, 0) + 1
missing_withdraw = 0
missing_refund = 0
refund_on_non_terminal = 0
for order in orders:
if order.out_bill_no not in withdraw_refs:
missing_withdraw += 1
has_refund = refund_counts.get(order.out_bill_no, 0) > 0
if order.status in {"failed", "rejected"} and not has_refund:
missing_refund += 1
if has_refund and order.status not in {"failed", "rejected"}:
refund_on_non_terminal += 1
duplicate_refund = sum(1 for count in refund_counts.values() if count > 1)
diff = cash_balance_total - cash_txn_total
ok = (
diff == 0
and missing_withdraw == 0
and missing_refund == 0
and duplicate_refund == 0
and refund_on_non_terminal == 0
)
return {
"ok": ok,
"cash_balance_total_cents": cash_balance_total,
"cash_transaction_total_cents": cash_txn_total,
"balance_diff_cents": diff,
"missing_withdraw_txn_count": missing_withdraw,
"missing_refund_txn_count": missing_refund,
"duplicate_refund_txn_count": duplicate_refund,
"refund_txn_on_non_terminal_count": refund_on_non_terminal,
}
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
@@ -1,91 +0,0 @@
"""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
@@ -1 +0,0 @@
"""admin 路由(前缀统一 /admin/api)。"""
-69
View File
@@ -1,69 +0,0 @@
"""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
@@ -1,34 +0,0 @@
"""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
@@ -1,48 +0,0 @@
"""Admin 认证:账号密码登录 → admin JWT(独立 secret)。"""
from __future__ import annotations
import logging
from fastapi import APIRouter, Depends, HTTPException
from app.admin.deps import AdminDb, CurrentAdmin
from app.admin.repositories import admin_user as admin_repo
from app.admin.schemas.auth import AdminLoginRequest, AdminLoginResponse, AdminOut
from app.admin.security import create_admin_token
from app.core.ratelimit import rate_limit
from app.core.security import verify_password
logger = logging.getLogger("shagua.admin.auth")
router = APIRouter(prefix="/admin/api/auth", tags=["admin-auth"])
@router.post(
"/login",
response_model=AdminLoginResponse,
summary="管理员登录",
dependencies=[Depends(rate_limit(10, 60, "admin-login"))], # 同 IP 每分钟≤10 次,防爆破
)
def login(req: AdminLoginRequest, db: AdminDb) -> AdminLoginResponse:
admin = admin_repo.get_by_username(db, req.username)
# 用户名不存在 / 密码错统一 401 同文案(防账号枚举)。
# disabled 账号单独 403"账号已禁用":admin 是内部少数已知账号、无枚举价值,
# 明确提示比防枚举更有运维价值(与 App 端 auth.py 对 disabled 用户的 403 一致)。
if admin is None or not verify_password(req.password, admin.password_hash):
raise HTTPException(status_code=401, detail="用户名或密码错误")
if admin.status != "active":
raise HTTPException(status_code=403, detail="账号已禁用")
token, expires_in = create_admin_token(admin_id=admin.id, role=admin.role)
admin_repo.update_last_login(db, admin)
logger.info("admin login ok id=%d username=%s role=%s", admin.id, admin.username, admin.role)
return AdminLoginResponse(
access_token=token,
expires_in=expires_in,
admin=AdminOut.model_validate(admin),
)
@router.get("/me", response_model=AdminOut, summary="当前管理员")
def me(admin: CurrentAdmin) -> AdminOut:
return AdminOut.model_validate(admin)
-81
View File
@@ -1,81 +0,0 @@
"""admin 运营配置:列出所有可配项 + 改某项(带审计 + 改值校验)。
配置项定义见 app.core.config_schema.CONFIG_DEFS;业务读配置 fallback 默认(见 app_config repo)。
权限:operator / finance 都可改(运营调奖励、财务调额度),super 恒可。
"""
from __future__ import annotations
from typing import Annotated, Any
from fastapi import APIRouter, Depends, HTTPException, Request
from app.admin.audit import write_audit
from app.admin.deps import AdminDb, get_client_ip, get_current_admin, require_role
from app.admin.schemas.config import ConfigItemOut, ConfigUpdateRequest
from app.core.config_schema import CONFIG_DEFS
from app.models.admin import AdminUser
from app.repositories import app_config
router = APIRouter(
prefix="/admin/api/config",
tags=["admin-config"],
dependencies=[Depends(get_current_admin)],
)
def _validate(key: str, value: Any) -> None:
"""按配置项 type 校验新值,不合法抛 ValueError(router 转 400)。"""
t = CONFIG_DEFS[key]["type"]
if t == "int":
if not isinstance(value, int) or isinstance(value, bool) or value < 0:
raise ValueError("需为非负整数")
elif t == "int_list":
if not isinstance(value, list) or not value:
raise ValueError("需为非空整数列表")
if not all(isinstance(x, int) and not isinstance(x, bool) and x >= 0 for x in value):
raise ValueError("列表元素需为非负整数")
if key == "signin_rewards" and len(value) != 14:
raise ValueError("签到档位必须正好 14 个(对应 14 天循环)")
elif t == "dict_str_int":
if not isinstance(value, dict) or not all(
isinstance(k, str) and isinstance(v, int) and not isinstance(v, bool)
for k, v in value.items()
):
raise ValueError("需为 {字符串: 整数} 映射")
def _item(db, key: str) -> ConfigItemOut:
for item in app_config.list_all(db):
if item["key"] == key:
return ConfigItemOut(**item)
raise HTTPException(status_code=404, detail="未知配置项")
@router.get("", response_model=list[ConfigItemOut], summary="所有可配项 + 当前值")
def list_config(db: AdminDb) -> list[ConfigItemOut]:
return [ConfigItemOut(**item) for item in app_config.list_all(db)]
@router.patch("/{key}", response_model=ConfigItemOut, summary="改某项配置(带审计)")
def update_config(
key: str,
body: ConfigUpdateRequest,
request: Request,
admin: Annotated[AdminUser, Depends(require_role("operator", "finance"))],
db: AdminDb,
) -> ConfigItemOut:
if key not in CONFIG_DEFS:
raise HTTPException(status_code=404, detail="未知配置项")
try:
_validate(key, body.value)
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e)) from e
before = app_config.get_value(db, key)
app_config.set_value(db, key, body.value, admin_id=admin.id, commit=False)
write_audit(
db, admin, action="config.set", target_type="config", target_id=key,
detail={"before": before, "after": body.value}, ip=get_client_ip(request), commit=False,
)
db.commit()
return _item(db, key)
-19
View File
@@ -1,19 +0,0 @@
"""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
@@ -1,56 +0,0 @@
"""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()
-160
View File
@@ -1,160 +0,0 @@
"""admin 首页轮播种子管理:列出 / 新增 / 改 / 删 / 批量生成 / 预览(带审计)。
种子用于首页轮播「真实+种子混播」的兜底(见 app/repositories/ops_marquee.py)。
权限:operator 可改,super 恒可;预览为只读(任意已登录 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.schemas.ops_marquee_seed import (
OpsMarqueeSeedBulkCreate,
OpsMarqueeSeedCreate,
OpsMarqueeSeedOut,
OpsMarqueeSeedUpdate,
OpsSavingsFeedPreviewOut,
)
from app.models.admin import AdminUser
from app.models.ops_marquee_seed import OpsMarqueeSeed
from app.repositories import ops_marquee
router = APIRouter(
prefix="/admin/api/marquee-seeds",
tags=["admin-marquee-seeds"],
dependencies=[Depends(get_current_admin)],
)
def _out(seed: OpsMarqueeSeed) -> OpsMarqueeSeedOut:
return OpsMarqueeSeedOut(
id=seed.id,
masked_user=seed.masked_user,
min_cents=seed.min_cents,
max_cents=seed.max_cents,
enabled=seed.enabled,
sort_order=seed.sort_order,
created_at=seed.created_at.isoformat() if seed.created_at else None,
)
@router.get("", response_model=list[OpsMarqueeSeedOut], summary="轮播种子列表")
def list_seeds(db: AdminDb) -> list[OpsMarqueeSeedOut]:
return [_out(s) for s in ops_marquee.list_seeds(db)]
@router.get("/preview", response_model=OpsSavingsFeedPreviewOut, summary="预览实际混播 feed")
def preview_feed(
db: AdminDb,
limit: Annotated[int, Query(ge=1, le=30)] = 8,
) -> OpsSavingsFeedPreviewOut:
"""返回客户端实际会看到的 feed(真实记录会插队、种子随机抽取/金额随机/名字合成),供运营对效果。"""
return OpsSavingsFeedPreviewOut(items=ops_marquee.get_feed(db, limit=limit))
@router.post("", response_model=OpsMarqueeSeedOut, summary="新增轮播种子(带审计)")
def create_seed(
body: OpsMarqueeSeedCreate,
request: Request,
admin: Annotated[AdminUser, Depends(require_role("operator"))],
db: AdminDb,
) -> OpsMarqueeSeedOut:
try:
seed = ops_marquee.create_seed(
db,
masked_user=body.masked_user,
min_cents=body.min_cents,
max_cents=body.max_cents,
enabled=body.enabled,
sort_order=body.sort_order,
commit=False,
)
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e)) from e
write_audit(
db, admin, action="ops_marquee_seed.create", target_type="ops_marquee_seed",
target_id=None, detail=body.model_dump(), ip=get_client_ip(request), commit=False,
)
db.commit()
db.refresh(seed)
return _out(seed)
@router.post("/bulk", response_model=list[OpsMarqueeSeedOut], summary="批量生成种子(带审计)")
def bulk_generate(
body: OpsMarqueeSeedBulkCreate,
request: Request,
admin: Annotated[AdminUser, Depends(require_role("operator"))],
db: AdminDb,
) -> list[OpsMarqueeSeedOut]:
try:
seeds = ops_marquee.bulk_generate(
db,
count=body.count,
min_cents=body.min_cents,
max_cents=body.max_cents,
enabled=body.enabled,
commit=False,
)
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e)) from e
write_audit(
db, admin, action="ops_marquee_seed.bulk_generate", target_type="ops_marquee_seed",
target_id=None, detail=body.model_dump(), ip=get_client_ip(request), commit=False,
)
db.commit()
for s in seeds:
db.refresh(s)
return [_out(s) for s in seeds]
@router.patch("/{seed_id}", response_model=OpsMarqueeSeedOut, summary="改轮播种子(带审计)")
def update_seed(
seed_id: int,
body: OpsMarqueeSeedUpdate,
request: Request,
admin: Annotated[AdminUser, Depends(require_role("operator"))],
db: AdminDb,
) -> OpsMarqueeSeedOut:
try:
seed = ops_marquee.update_seed(
db,
seed_id,
masked_user=body.masked_user,
min_cents=body.min_cents,
max_cents=body.max_cents,
enabled=body.enabled,
sort_order=body.sort_order,
commit=False,
)
except ValueError as e:
raise HTTPException(status_code=404 if "不存在" in str(e) else 400, detail=str(e)) from e
write_audit(
db, admin, action="ops_marquee_seed.update", target_type="ops_marquee_seed",
target_id=seed_id, detail=body.model_dump(exclude_none=True),
ip=get_client_ip(request), commit=False,
)
db.commit()
db.refresh(seed)
return _out(seed)
@router.delete("/{seed_id}", summary="删轮播种子(带审计)")
def delete_seed(
seed_id: int,
request: Request,
admin: Annotated[AdminUser, Depends(require_role("operator"))],
db: AdminDb,
) -> dict:
ok = ops_marquee.delete_seed(db, seed_id, commit=False)
if not ok:
raise HTTPException(status_code=404, detail="种子不存在")
write_audit(
db, admin, action="ops_marquee_seed.delete", target_type="ops_marquee_seed",
target_id=seed_id, detail=None, ip=get_client_ip(request), commit=False,
)
db.commit()
return {"deleted": True}
-74
View File
@@ -1,74 +0,0 @@
"""admin 首页三统计展示配置:列出三指标配置(+ 真实值预览)+ 改某指标(带审计 + 校验)。
每指标可独立选 real/manual/random。计算逻辑见 app/repositories/ops_stat.py。
权限:operator 可改(运营调门面数字),super 恒可。
"""
from __future__ import annotations
from typing import Annotated
from fastapi import APIRouter, Depends, HTTPException, Request
from app.admin.audit import write_audit
from app.admin.deps import AdminDb, get_client_ip, get_current_admin, require_role
from app.admin.schemas.ops_stat_config import (
OpsStatItemOut,
OpsStatUpdateRequest,
)
from app.models.admin import AdminUser
from app.repositories import ops_stat
router = APIRouter(
prefix="/admin/api/dashboard-display",
tags=["admin-dashboard-display"],
dependencies=[Depends(get_current_admin)],
)
@router.get("", response_model=list[OpsStatItemOut], summary="首页三统计配置 + 真实值预览")
def list_display(db: AdminDb) -> list[OpsStatItemOut]:
return [OpsStatItemOut(**item) for item in ops_stat.get_config(db)]
@router.patch(
"/{metric}", response_model=OpsStatItemOut, summary="改某指标展示配置(带审计)"
)
def update_display(
metric: str,
body: OpsStatUpdateRequest,
request: Request,
admin: Annotated[AdminUser, Depends(require_role("operator"))],
db: AdminDb,
) -> OpsStatItemOut:
try:
before, after = ops_stat.update_config(
db,
metric,
mode=body.mode,
manual_value=body.manual_value,
random_mult_min=body.random_mult_min,
random_mult_max=body.random_mult_max,
random_tick_seconds=body.random_tick_seconds,
random_anchor_minutes=body.random_anchor_minutes,
random_kind=body.random_kind,
random_step_min=body.random_step_min,
random_step_max=body.random_step_max,
real_offset=body.real_offset,
allow_decrease=body.allow_decrease,
random_initial=body.random_initial,
apply_now=body.apply_now,
admin_id=admin.id,
commit=False,
)
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e)) from e
write_audit(
db, admin, action="ops_stat_config.set", target_type="ops_stat_config",
target_id=metric, detail={"before": before, "after": after},
ip=get_client_ip(request), commit=False,
)
db.commit()
item = next(it for it in ops_stat.get_config(db) if it["metric"] == metric)
return OpsStatItemOut(**item)
-111
View File
@@ -1,111 +0,0 @@
"""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
@@ -1,49 +0,0 @@
"""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,
)
-440
View File
@@ -1,440 +0,0 @@
"""admin 提现:列表(读)+ 单笔重试查单 + 批量对账(写,带审计)。
提现的钱逻辑(查微信/退款/撤单/幂等)全部复用 app.repositories.wallet,admin 只触发 + 记审计。
这些 wallet 函数内部各自 commit(涉及微信调用),审计在其后单独 commit:操作本身幂等,
审计记录"谁触发的 + 结果",事务边界比"改金币"宽松是有意的(不能塞进 wallet 的自有事务)。
"""
from __future__ import annotations
from datetime import datetime
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.admin import AdminAuditLogOut
from app.admin.schemas.wallet import (
CashTxnOut,
ReconcileResult,
WithdrawBulkRejectRequest,
WithdrawBulkRequest,
WithdrawBulkResult,
WithdrawBulkItemResult,
WithdrawDetailOut,
WithdrawLedgerCheckOut,
WithdrawOrderOut,
WithdrawRejectRequest,
WithdrawSummaryOut,
WithdrawUserSnapshot,
WxpayHealthCheckOut,
)
from app.core.config import settings
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,
keyword: Annotated[str | None, Query(max_length=100)] = None,
date_from: Annotated[datetime | None, Query()] = None,
date_to: Annotated[datetime | None, Query()] = None,
date_field: Annotated[str, Query(pattern="^(created_at|updated_at)$")] = "created_at",
sort_by: Annotated[
str, Query(pattern="^(id|created_at|updated_at|amount_cents)$")
] = "created_at",
sort_order: Annotated[str, Query(pattern="^(asc|desc)$")] = "desc",
quick_filter: Annotated[
str | None,
Query(pattern="^(abnormal|failed|overdue_reviewing|pending_overdue|today|high_risk)$"),
] = 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,
keyword=keyword,
date_from=date_from,
date_to=date_to,
date_field=date_field,
sort_by=sort_by,
sort_order=sort_order,
quick_filter=quick_filter,
limit=limit,
cursor=cursor,
)
return CursorPage(
items=[WithdrawOrderOut.model_validate(o) for o in items], next_cursor=next_cursor,
)
@router.get("/summary", response_model=WithdrawSummaryOut, summary="提现审核台统计")
def withdraws_summary(db: AdminDb) -> WithdrawSummaryOut:
return WithdrawSummaryOut(**queries.withdraw_summary(db))
@router.get("/health-check", response_model=WxpayHealthCheckOut, summary="提现配置健康检查")
def withdraw_health_check() -> WxpayHealthCheckOut:
private_path = wxpay._resolve_config_path(settings.WXPAY_MCH_PRIVATE_KEY_PATH) # noqa: SLF001
public_path = wxpay._resolve_config_path(settings.WXPAY_PUBLIC_KEY_PATH) # noqa: SLF001
issues: list[str] = []
private_loadable = False
public_loadable = False
try:
wxpay._load_private_key() # noqa: SLF001
private_loadable = True
except Exception as e: # noqa: BLE001
issues.append(f"商户私钥不可用:{e}")
try:
wxpay._load_public_key() # noqa: SLF001
public_loadable = True
except Exception as e: # noqa: BLE001
issues.append(f"微信支付平台公钥不可用:{e}")
if not settings.wxpay_configured:
issues.append("微信支付基础配置不完整")
if not settings.WXPAY_AUTH_NOTIFY_URL:
issues.append("免确认授权回调地址未配置")
if not settings.WITHDRAW_AUTO_RECONCILE_ENABLED:
issues.append("自动对账未开启")
return WxpayHealthCheckOut(
ok=not issues,
wxpay_configured=settings.wxpay_configured,
wxpay_auth_configured=settings.wxpay_auth_configured,
private_key_path=str(private_path),
private_key_exists=private_path.exists(),
private_key_loadable=private_loadable,
public_key_path=str(public_path),
public_key_exists=public_path.exists(),
public_key_loadable=public_loadable,
auth_notify_url_configured=bool(settings.WXPAY_AUTH_NOTIFY_URL),
auto_reconcile_enabled=settings.WITHDRAW_AUTO_RECONCILE_ENABLED,
auto_reconcile_interval_sec=settings.WITHDRAW_AUTO_RECONCILE_INTERVAL_SEC,
auto_reconcile_older_than_minutes=settings.WITHDRAW_AUTO_RECONCILE_OLDER_THAN_MINUTES,
issues=issues,
)
@router.get("/ledger-check", response_model=WithdrawLedgerCheckOut, summary="提现资金账本校验")
def withdraw_ledger_check(db: AdminDb) -> WithdrawLedgerCheckOut:
return WithdrawLedgerCheckOut(**queries.withdraw_ledger_check(db))
@router.get("/{out_bill_no}", response_model=WithdrawDetailOut, summary="提现单详情")
def withdraw_detail(out_bill_no: str, db: AdminDb) -> WithdrawDetailOut:
order = queries.get_withdraw_by_out_bill_no(db, out_bill_no)
if order is None:
raise HTTPException(status_code=404, detail="提现单不存在")
overview = queries.get_user_overview(db, order.user_id)
user_snapshot = None
if overview is not None:
user = overview["user"]
user_snapshot = WithdrawUserSnapshot(
id=user.id,
phone=user.phone,
nickname=user.nickname,
status=user.status,
wechat_nickname=user.wechat_nickname,
wechat_avatar_url=user.wechat_avatar_url,
created_at=user.created_at,
last_login_at=user.last_login_at,
cash_balance_cents=overview["cash_balance_cents"],
withdraw_total=overview["withdraw_total"],
withdraw_success_cents=overview["withdraw_success_cents"],
)
recent_withdraws, _ = queries.list_all_withdraw_orders(
db, user_id=order.user_id, limit=5, cursor=None,
)
recent_cash_transactions, _ = queries.list_all_cash_transactions(
db, user_id=order.user_id, limit=8, cursor=None,
)
audit_logs = queries.list_withdraw_audit_logs(db, out_bill_no, limit=10)
risk_flags, risk_score = queries.withdraw_risk_flags(
order,
overview["user"] if overview else None,
recent_withdraws,
overview["cash_balance_cents"] if overview else 0,
)
return WithdrawDetailOut(
order=WithdrawOrderOut.model_validate(order),
user=user_snapshot,
risk_flags=risk_flags,
risk_score=risk_score,
recent_withdraws=[WithdrawOrderOut.model_validate(o) for o in recent_withdraws],
recent_cash_transactions=[CashTxnOut.model_validate(t) for t in recent_cash_transactions],
audit_logs=[AdminAuditLogOut.model_validate(log) for log in audit_logs],
)
# 注意:/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)
def _bulk_result(items: list[WithdrawBulkItemResult]) -> WithdrawBulkResult:
success = sum(1 for item in items if item.ok)
return WithdrawBulkResult(
total=len(items),
success=success,
failed=len(items) - success,
items=items,
)
@router.post("/bulk/refresh", response_model=WithdrawBulkResult, summary="批量刷新查单")
def bulk_refresh_withdraws(
body: WithdrawBulkRequest,
request: Request,
admin: Annotated[AdminUser, Depends(require_role("finance"))],
db: AdminDb,
) -> WithdrawBulkResult:
results: list[WithdrawBulkItemResult] = []
ip = get_client_ip(request)
for out_bill_no in body.out_bill_nos:
try:
order = queries.get_withdraw_by_out_bill_no(db, out_bill_no)
if order is None:
raise wallet_repo.WithdrawOrderNotFound
refreshed = wallet_repo.refresh_withdraw_status(
db, order.user_id, out_bill_no, cancel_if_unconfirmed=True,
)
write_audit(
db, admin, action="withdraw.refresh", target_type="withdraw",
target_id=out_bill_no,
detail={
"status": refreshed.status,
"wechat_state": refreshed.wechat_state,
"fail_reason": refreshed.fail_reason,
"bulk": True,
},
ip=ip, commit=True,
)
results.append(
WithdrawBulkItemResult(
out_bill_no=out_bill_no, ok=True, status=refreshed.status
)
)
except wallet_repo.WithdrawOrderNotFound:
db.rollback()
results.append(
WithdrawBulkItemResult(out_bill_no=out_bill_no, ok=False, error="提现单不存在")
)
except wxpay.WxPayNotConfiguredError:
db.rollback()
results.append(
WithdrawBulkItemResult(out_bill_no=out_bill_no, ok=False, error="微信支付未配置")
)
except Exception as e: # noqa: BLE001 - 批量操作单笔失败不打断整批
db.rollback()
results.append(
WithdrawBulkItemResult(out_bill_no=out_bill_no, ok=False, error=f"系统异常: {e}")
)
return _bulk_result(results)
@router.post("/bulk/approve", response_model=WithdrawBulkResult, summary="批量审核通过并打款")
def bulk_approve_withdraws(
body: WithdrawBulkRequest,
request: Request,
admin: Annotated[AdminUser, Depends(require_role("finance"))],
db: AdminDb,
) -> WithdrawBulkResult:
results: list[WithdrawBulkItemResult] = []
ip = get_client_ip(request)
for out_bill_no in body.out_bill_nos:
try:
order = wallet_repo.approve_withdraw(db, out_bill_no)
write_audit(
db, admin, action="withdraw.approve", target_type="withdraw",
target_id=out_bill_no,
detail={
"status": order.status,
"wechat_state": order.wechat_state,
"amount_cents": order.amount_cents,
"bulk": True,
},
ip=ip, commit=True,
)
results.append(
WithdrawBulkItemResult(out_bill_no=out_bill_no, ok=True, status=order.status)
)
except wallet_repo.WithdrawOrderNotFound:
db.rollback()
results.append(
WithdrawBulkItemResult(out_bill_no=out_bill_no, ok=False, error="提现单不存在")
)
except wallet_repo.WithdrawNotReviewable as e:
db.rollback()
results.append(
WithdrawBulkItemResult(out_bill_no=out_bill_no, ok=False, error=str(e))
)
except wxpay.WxPayNotConfiguredError:
db.rollback()
results.append(
WithdrawBulkItemResult(out_bill_no=out_bill_no, ok=False, error="微信支付未配置")
)
except Exception as e: # noqa: BLE001 - 批量操作单笔失败不打断整批
db.rollback()
results.append(
WithdrawBulkItemResult(out_bill_no=out_bill_no, ok=False, error=f"系统异常: {e}")
)
return _bulk_result(results)
@router.post("/bulk/reject", response_model=WithdrawBulkResult, summary="批量审核拒绝并退款")
def bulk_reject_withdraws(
body: WithdrawBulkRejectRequest,
request: Request,
admin: Annotated[AdminUser, Depends(require_role("finance"))],
db: AdminDb,
) -> WithdrawBulkResult:
results: list[WithdrawBulkItemResult] = []
ip = get_client_ip(request)
for out_bill_no in body.out_bill_nos:
try:
order = wallet_repo.reject_withdraw(db, out_bill_no, body.reason)
write_audit(
db, admin, action="withdraw.reject", target_type="withdraw",
target_id=out_bill_no,
detail={
"reason": body.reason,
"amount_cents": order.amount_cents,
"bulk": True,
},
ip=ip, commit=True,
)
results.append(
WithdrawBulkItemResult(out_bill_no=out_bill_no, ok=True, status=order.status)
)
except wallet_repo.WithdrawOrderNotFound:
db.rollback()
results.append(
WithdrawBulkItemResult(out_bill_no=out_bill_no, ok=False, error="提现单不存在")
)
except wallet_repo.WithdrawNotReviewable as e:
db.rollback()
results.append(
WithdrawBulkItemResult(out_bill_no=out_bill_no, ok=False, error=str(e))
)
except Exception as e: # noqa: BLE001 - 批量操作单笔失败不打断整批
db.rollback()
results.append(
WithdrawBulkItemResult(out_bill_no=out_bill_no, ok=False, error=f"系统异常: {e}")
)
return _bulk_result(results)
@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,
"fail_reason": refreshed.fail_reason,
},
ip=get_client_ip(request), commit=True,
)
return WithdrawOrderOut.model_validate(refreshed)
@router.post("/{out_bill_no}/approve", response_model=WithdrawOrderOut, summary="审核通过(发起微信打款)")
def approve_withdraw_order(
out_bill_no: str,
request: Request,
admin: Annotated[AdminUser, Depends(require_role("finance"))],
db: AdminDb,
) -> WithdrawOrderOut:
"""审核通过 → 调微信转账。仅 reviewing 单可通过(非 reviewing 返 409,防重复打款)。
打款的钱逻辑(转账/查单/退款/幂等)复用 wallet_repo.approve_withdraw → execute_withdraw_transfer;
返回的 order 可能是 pending(在途/待用户确认)/success/failed(打款失败已退),admin 前端按 status 展示。
"""
try:
order = wallet_repo.approve_withdraw(db, out_bill_no)
except wallet_repo.WithdrawOrderNotFound as e:
raise HTTPException(status_code=404, detail="提现单不存在") from e
except wallet_repo.WithdrawNotReviewable as e:
raise HTTPException(status_code=409, detail=str(e)) from e
except wxpay.WxPayNotConfiguredError as e:
raise HTTPException(status_code=503, detail="微信支付未配置") from e
write_audit(
db, admin, action="withdraw.approve", target_type="withdraw", target_id=out_bill_no,
detail={
"status": order.status,
"wechat_state": order.wechat_state,
"amount_cents": order.amount_cents,
},
ip=get_client_ip(request), commit=True,
)
return WithdrawOrderOut.model_validate(order)
@router.post("/{out_bill_no}/reject", response_model=WithdrawOrderOut, summary="审核拒绝(退回现金)")
def reject_withdraw_order(
out_bill_no: str,
body: WithdrawRejectRequest,
request: Request,
admin: Annotated[AdminUser, Depends(require_role("finance"))],
db: AdminDb,
) -> WithdrawOrderOut:
"""审核拒绝 → 退回用户现金 + 置 rejected,理由写入 fail_reason(用户可见)。仅 reviewing 单可拒。"""
try:
order = wallet_repo.reject_withdraw(db, out_bill_no, body.reason)
except wallet_repo.WithdrawOrderNotFound as e:
raise HTTPException(status_code=404, detail="提现单不存在") from e
except wallet_repo.WithdrawNotReviewable as e:
raise HTTPException(status_code=409, detail=str(e)) from e
write_audit(
db, admin, action="withdraw.reject", target_type="withdraw", target_id=out_bill_no,
detail={"reason": body.reason, "amount_cents": order.amount_cents},
ip=get_client_ip(request), commit=True,
)
return WithdrawOrderOut.model_validate(order)
-1
View File
@@ -1 +0,0 @@
"""admin 请求/响应 schemas(snake_case,与前端约定一致)。"""
-37
View File
@@ -1,37 +0,0 @@
"""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
@@ -1,29 +0,0 @@
"""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
@@ -1,19 +0,0 @@
"""通用 schema:游标分页响应 + 通用 ok 响应。"""
from __future__ import annotations
from typing import Generic, TypeVar
from pydantic import BaseModel
T = TypeVar("T")
class CursorPage(BaseModel, Generic[T]):
"""游标分页响应:items + 下一页游标(next_cursor=None 表示末页)。"""
items: list[T]
next_cursor: int | None = None
class OkResponse(BaseModel):
ok: bool = True
-22
View File
@@ -1,22 +0,0 @@
"""admin 运营配置 schemas。"""
from __future__ import annotations
from typing import Any
from pydantic import BaseModel
class ConfigItemOut(BaseModel):
key: str
label: str
group: str
type: str # int / int_list / dict_str_int
help: str | None = None
default: Any
value: Any
overridden: bool # 是否被运营覆盖过(false=用默认)
updated_at: str | None = None
class ConfigUpdateRequest(BaseModel):
value: Any
-48
View File
@@ -1,48 +0,0 @@
"""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
@@ -1,18 +0,0 @@
"""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
-53
View File
@@ -1,53 +0,0 @@
"""admin 首页轮播种子 schemas。金额单位:分(前端展示 ÷100 取元)。
种子 = 生成规则:masked_user 可空(空→feed 随机合成名)、金额是 [min_cents, max_cents] 区间。
"""
from __future__ import annotations
from pydantic import BaseModel
class OpsMarqueeSeedOut(BaseModel):
id: int
masked_user: str | None = None
min_cents: int
max_cents: int
enabled: bool
sort_order: int
created_at: str | None = None
class OpsMarqueeSeedCreate(BaseModel):
masked_user: str | None = None # 空 / 不传 → feed 随机合成脱敏名
min_cents: int
max_cents: int
enabled: bool = True
sort_order: int = 0
class OpsMarqueeSeedUpdate(BaseModel):
# masked_user:不传=不改;传空串=清空(改回随机合成);传值=固定该名
masked_user: str | None = None
min_cents: int | None = None
max_cents: int | None = None
enabled: bool | None = None
sort_order: int | None = None
class OpsMarqueeSeedBulkCreate(BaseModel):
"""批量生成 count 条自动合成名种子(冷启动铺量)。"""
count: int
min_cents: int
max_cents: int
enabled: bool = True
class OpsSavingsFeedPreviewItem(BaseModel):
masked_user: str
saved_amount_cents: int
time: str
class OpsSavingsFeedPreviewOut(BaseModel):
"""运营预览:实际混播出来的 feed(真实记录会插队,与客户端一致)。"""
items: list[OpsSavingsFeedPreviewItem]
-48
View File
@@ -1,48 +0,0 @@
"""admin 首页三统计展示配置 schemas。
倍率用千分比整数(1.000→1000);total_saved 的 manual_value / random_current 单位是分。
"""
from __future__ import annotations
from pydantic import BaseModel
class OpsStatItemOut(BaseModel):
metric: str # help_users / total_compares / total_saved
label: str # 帮助用户 / 完成比价 / 累计节省
unit: str # 人 / 次 / 分
mode: str # real / manual / random
manual_value: int | None = None
random_mult_min: int
random_mult_max: int
random_tick_seconds: int
random_anchor_minutes: int
random_kind: str # mult(×倍率) / add(+绝对增量)
random_step_min: int # 绝对增量区间(基础单位;total_saved 为分)
random_step_max: int
real_offset: int # 真实值模式基数偏移(基础单位)
allow_decrease: bool # 是否允许展示值下降(默认 false = 只增不减)
random_current: int | None = None
random_last_tick_at: str | None = None
real_value: int # 当前真实值(纯真实、不含偏移;给运营对比参考)
updated_at: str | None = None
class OpsStatUpdateRequest(BaseModel):
"""改某指标配置(均可选,只改传了的字段)。"""
mode: str | None = None
manual_value: int | None = None
random_mult_min: int | None = None
random_mult_max: int | None = None
random_tick_seconds: int | None = None
random_anchor_minutes: int | None = None
random_kind: str | None = None # mult / add
random_step_min: int | None = None
random_step_max: int | None = None
real_offset: int | None = None
allow_decrease: bool | None = None
# 切 random 时的初始基数;不传则用当前真实值播种
random_initial: int | None = None
# 立即更新:不等更新钟点,保存后马上把展示值刷新一次(real 快照/manual 生效/random 走一档)
apply_now: bool = False
-47
View File
@@ -1,47 +0,0 @@
"""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 不走此接口)"
)
-149
View File
@@ -1,149 +0,0 @@
"""admin 钱包(金币/现金流水 + 提现单)schemas。"""
from __future__ import annotations
from datetime import datetime
from pydantic import BaseModel, ConfigDict, Field
from app.admin.schemas.admin import AdminAuditLogOut
class CoinTxnOut(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: int
user_id: int
amount: int
balance_after: int
biz_type: str
ref_id: str | None = None
remark: str | None = None
created_at: datetime
class CashTxnOut(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: int
user_id: int
amount_cents: int
balance_after_cents: int
biz_type: str
ref_id: str | None = None
remark: str | None = None
created_at: datetime
class WithdrawOrderOut(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: int
user_id: int
out_bill_no: str
amount_cents: int
user_name: str | None = None # 提现实名(审核核对 + 打款用)
status: str
wechat_state: str | None = None
transfer_bill_no: str | None = None
fail_reason: str | None = None
created_at: datetime
updated_at: datetime
class WithdrawSummaryOut(BaseModel):
reviewing_count: int
reviewing_amount_cents: int
pending_count: int
failed_count: int
today_success_count: int
today_success_amount_cents: int
today_rejected_count: int
class WithdrawUserSnapshot(BaseModel):
id: int
phone: str
nickname: str | None = None
status: str
wechat_nickname: str | None = None
wechat_avatar_url: str | None = None
created_at: datetime
last_login_at: datetime
cash_balance_cents: int
withdraw_total: int
withdraw_success_cents: int
class WithdrawDetailOut(BaseModel):
order: WithdrawOrderOut
user: WithdrawUserSnapshot | None = None
risk_flags: list[str]
risk_score: int
recent_withdraws: list[WithdrawOrderOut]
recent_cash_transactions: list[CashTxnOut]
audit_logs: list[AdminAuditLogOut]
class ReconcileResult(BaseModel):
checked: int
resolved: int
class WithdrawBulkRequest(BaseModel):
out_bill_nos: list[str] = Field(
..., min_length=1, max_length=50, description="提现商户单号列表"
)
class WithdrawBulkRejectRequest(WithdrawBulkRequest):
reason: str = Field(
..., min_length=1, max_length=200, description="批量拒绝理由(用户可见)"
)
class WithdrawBulkItemResult(BaseModel):
out_bill_no: str
ok: bool
status: str | None = None
error: str | None = None
class WithdrawBulkResult(BaseModel):
total: int
success: int
failed: int
items: list[WithdrawBulkItemResult]
class WithdrawLedgerCheckOut(BaseModel):
ok: bool
cash_balance_total_cents: int
cash_transaction_total_cents: int
balance_diff_cents: int
missing_withdraw_txn_count: int
missing_refund_txn_count: int
duplicate_refund_txn_count: int
refund_txn_on_non_terminal_count: int
class WxpayHealthCheckOut(BaseModel):
ok: bool
wxpay_configured: bool
wxpay_auth_configured: bool
private_key_path: str
private_key_exists: bool
private_key_loadable: bool
public_key_path: str
public_key_exists: bool
public_key_loadable: bool
auth_notify_url_configured: bool
auto_reconcile_enabled: bool
auto_reconcile_interval_sec: int
auto_reconcile_older_than_minutes: int
issues: list[str]
class WithdrawRejectRequest(BaseModel):
reason: str = Field(
..., min_length=1, max_length=200, description="拒绝理由(写入 fail_reason,用户可见)"
)
-55
View File
@@ -1,55 +0,0 @@
"""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
-1
View File
@@ -1 +0,0 @@
"""内部(server→server)端点。不对客户端开放,靠共享密钥头校验,不走用户 JWT。"""
-63
View File
@@ -1,63 +0,0 @@
"""价格观测内部上报端点(pricebot → app-server)。
pricebot 在比价 done 帧把整批价格事实 POST 到这里落库。**不是给客户端的接口**:
不走用户 JWT,靠 server 间共享密钥头 `X-Internal-Secret` 校验(== settings.INTERNAL_API_SECRET)。
密钥未配置(默认空)时直接 503,避免裸奔的写端点。
与不鉴权的比价透传(compare.py / coupon.py)无关:那两个是客户端→server 的透传壳,
这个是 server→server 的内部写库。
"""
from __future__ import annotations
import hmac
import logging
from typing import Annotated
from fastapi import APIRouter, Header, HTTPException, status
from app.api.deps import DbSession
from app.core.config import settings
from app.repositories import price_observation as repo
from app.schemas.price_observation import (
PriceObservationBatchIn,
PriceObservationBatchOut,
)
logger = logging.getLogger("shagua.internal.price")
router = APIRouter(prefix="/internal", tags=["internal"])
def _check_secret(x_internal_secret: str | None) -> None:
"""共享密钥校验。未配置 → 503(挡住裸奔写端点);不匹配 → 401(常量时间比较)。"""
configured = settings.INTERNAL_API_SECRET
if not configured:
raise HTTPException(
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
detail="internal api not configured",
)
if not x_internal_secret or not hmac.compare_digest(x_internal_secret, configured):
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="invalid internal secret",
)
@router.post(
"/price-observation",
response_model=PriceObservationBatchOut,
summary="价格观测内部上报(pricebot→app-server,落 price_observation)",
)
def report_price_observation(
payload: PriceObservationBatchIn,
db: DbSession,
x_internal_secret: Annotated[str | None, Header()] = None,
) -> PriceObservationBatchOut:
_check_secret(x_internal_secret)
inserted, skipped = repo.insert_batch(db, payload)
logger.info(
"price_observation trace=%s inserted=%d skipped=%d device=%s user=%s",
payload.trace_id, inserted, skipped,
payload.source_device_id, payload.source_user_id,
)
return PriceObservationBatchOut(inserted=inserted, skipped=skipped)
+4 -105
View File
@@ -19,21 +19,8 @@ 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_feed_reward as crud_feed
from app.repositories import ad_reward as crud_ad
from app.repositories import ad_watch as crud_watch
from app.schemas.ad import (
AdRewardStatusOut,
EcpmReportIn,
EcpmReportOut,
FeedRewardIn,
FeedRewardOut,
PangleCallbackOut,
TestGrantOut,
WatchReportIn,
WatchReportOut,
)
from app.schemas.ad import AdRewardStatusOut, PangleCallbackOut, TestGrantOut
logger = logging.getLogger("shagua.ad")
@@ -77,7 +64,7 @@ def pangle_callback(request: Request, db: DbSession) -> PangleCallbackOut:
return PangleCallbackOut(is_verify=False, reason=REASON_BAD_PARAMS)
user_id = int(raw_user_id)
coin = rewards.resolve_ad_reward_coin(db, params.get("reward_amount"))
coin = rewards.resolve_ad_reward_coin(params.get("reward_amount"))
raw = "&".join(f"{k}={v}" for k, v in sorted(params.items()) if k != "sign")
try:
rec = crud_ad.grant_ad_reward(
@@ -97,8 +84,7 @@ 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, round_count, cooldown_until,
watched_sec, watch_limit) = 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,
@@ -106,62 +92,9 @@ def reward_status(user: CurrentUser, db: DbSession) -> AdRewardStatusOut:
coin_per_ad=coin_per,
round_count=round_count,
cooldown_until=cooldown_until,
watched_seconds_today=watched_sec,
watch_seconds_limit=watch_limit,
watch_seconds_remaining=max(0, watch_limit - watched_sec),
)
@router.post(
"/watch-report",
response_model=WatchReportOut,
summary="上报激励视频观看时长(旧客户端兼容字段)",
dependencies=[Depends(rate_limit(120, 60, "ad-watch-report"))],
)
def watch_report(payload: WatchReportIn, user: CurrentUser, db: DbSession) -> WatchReportOut:
"""客户端在激励视频关闭(onAdClose)后上报本次实际观看秒数,服务端累计到当日总时长。
Bearer 鉴权,user_id 取自 JWT(不信 body)。seconds 服务端夹 [0, MAX_SINGLE_WATCH_SECONDS]。
当前产品只保留每日 500 次上限,DAILY_AD_WATCH_SECONDS_LIMIT=0 表示时长闸不启用;该接口
仍保留用于旧客户端兼容和排查观看时长。
"""
total = crud_watch.add_watch_seconds(db, user.id, payload.seconds)
limit = rewards.DAILY_AD_WATCH_SECONDS_LIMIT
logger.info(
"ad watch report user_id=%d +%ds total=%d/%d",
user.id, payload.seconds, total, limit,
)
return WatchReportOut(
watched_seconds_today=total,
watch_seconds_limit=limit,
watch_seconds_remaining=max(0, limit - total),
)
@router.post(
"/ecpm-report",
response_model=EcpmReportOut,
summary="上报本次广告展示的 eCPM(内部收益统计)",
dependencies=[Depends(rate_limit(120, 60, "ad-ecpm-report"))],
)
def ecpm_report(payload: EcpmReportIn, user: CurrentUser, db: DbSession) -> EcpmReportOut:
"""客户端在广告展示后(onAdShow 读 getShowEcpm)上报 eCPM,落库做内部收益统计/对账。
Bearer 鉴权,user_id 取自 JWT(不信 body)。best-effort:落库即 ok,客户端 fire-and-forget,
丢一两条不影响业务(穿山甲后台报表是结算权威)。eCPM 与发奖(S2S)是两条独立流,不逐条关联。
"""
crud_ecpm.create_ecpm_record(
db, user.id,
ad_type=payload.ad_type, ecpm_raw=payload.ecpm,
adn=payload.adn, slot_id=payload.slot_id,
)
logger.info(
"ad ecpm report user_id=%d type=%s ecpm=%s adn=%s slot=%s",
user.id, payload.ad_type, payload.ecpm, payload.adn, payload.slot_id,
)
return EcpmReportOut(ok=True)
@router.post(
"/test-grant",
response_model=TestGrantOut,
@@ -187,8 +120,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, round_count, cooldown_until,
_watched, _watch_limit) = 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"),
@@ -201,36 +133,3 @@ def test_grant(user: CurrentUser, db: DbSession) -> TestGrantOut:
round_count=round_count,
cooldown_until=cooldown_until,
)
@router.post(
"/feed-reward",
response_model=FeedRewardOut,
summary="信息流广告完成后结算金币",
dependencies=[Depends(rate_limit(120, 60, "ad-feed-reward"))],
)
def feed_reward(payload: FeedRewardIn, user: CurrentUser, db: DbSession) -> FeedRewardOut:
"""点位 2:信息流广告每展示满 10 秒累计一份奖励,视频完成后一次性入账。
当前一期由客户端完成回调携带 eCPM / 展示秒数上报;client_event_id 做幂等键,避免重试重复发。
"""
rec = crud_feed.grant_feed_reward(
db,
user.id,
client_event_id=payload.client_event_id,
ecpm=payload.ecpm,
duration_seconds=payload.duration_seconds,
adn=payload.adn,
slot_id=payload.slot_id,
)
logger.info(
"feed ad reward user_id=%d event=%s status=%s units=%d coin=%d",
user.id, rec.client_event_id, rec.status, rec.unit_count, rec.coin,
)
return FeedRewardOut(
granted=(rec.status == "granted"),
status=rec.status,
coin=rec.coin,
unit_count=rec.unit_count,
daily_limit=rewards.get_ad_daily_limit(db),
)
+4 -15
View File
@@ -12,10 +12,9 @@ from __future__ import annotations
import logging
from fastapi import APIRouter, Depends, HTTPException, status
from fastapi import APIRouter, 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
@@ -72,29 +71,19 @@ def jverify_login(req: JverifyLoginRequest, db: DbSession) -> TokenWithUser:
# ===================== 短信登录 =====================
@router.post(
"/sms/send",
response_model=SmsSendResponse,
summary="发送短信验证码",
dependencies=[Depends(rate_limit(10, 60, "sms-send"))], # 同 IP 每分钟≤10 次(防一 IP 刷不同号)
)
@router.post("/sms/send", response_model=SmsSendResponse, summary="发送短信验证码 (mock)")
def sms_send(req: SmsSendRequest) -> SmsSendResponse:
try:
cooldown = send_code(req.phone)
except SmsError as e:
raise HTTPException(status_code=e.status_code, detail=str(e)) from e
raise HTTPException(status_code=429, 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="手机号+验证码登录",
dependencies=[Depends(rate_limit(20, 60, "sms-login"))], # 防撞库爆破(另有单码失败次数上限)
)
@router.post("/sms/login", response_model=TokenWithUser, summary="手机号+验证码登录")
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")
+6 -25
View File
@@ -18,7 +18,6 @@ pricebot 协议文档:
"""
from __future__ import annotations
import json
import logging
from typing import Any
@@ -26,7 +25,6 @@ import httpx
from fastapi import APIRouter, HTTPException, Request, status
from app.core.config import settings
from app.core.pricebot_router import pick_pricebot
logger = logging.getLogger("shagua.compare")
@@ -40,35 +38,25 @@ async def _passthrough(request: Request, upstream_path: str) -> dict[str, Any]:
打日志。比价单帧是大上下文 / 逐帧 LLM,超时用 PRICEBOT_COMPARE_TIMEOUT_SEC(60s,
比领券的 30s 长)。
"""
# 读原始字节,避免"反序列化→再序列化"的双重 JSON(省 ~一半透传 CPU,让 app-server
# 单 worker 也扛得住高并发)。只 json.loads 一次拿 trace_id 做亲和 + 打日志,转发时
# 直接发原始 bytes(content=raw),不重新 dumps。
raw = await request.body()
try:
meta = json.loads(raw)
body = await request.json()
except Exception as e:
raise HTTPException(status_code=400, detail=f"invalid json body: {e}") from e
if not isinstance(meta, dict):
meta = {}
# 按 trace_id 一致性 hash 选 pricebot 实例(同一比价的所有帧落同一进程,内存维护 state)
base = pick_pricebot(meta.get("trace_id"))
url = f"{base.rstrip('/')}{upstream_path}"
url = f"{settings.PRICEBOT_BASE_URL.rstrip('/')}{upstream_path}"
timeout = settings.PRICEBOT_COMPARE_TIMEOUT_SEC
logger.info(
"compare %s device_id=%s trace_id=%s step=%s",
upstream_path,
meta.get("device_id"),
meta.get("trace_id"),
meta.get("step"),
body.get("device_id"),
body.get("trace_id"),
body.get("step"),
)
try:
async with httpx.AsyncClient(timeout=timeout) as client:
resp = await client.post(
url, content=raw, headers={"Content-Type": "application/json"}
)
resp = await client.post(url, json=body)
except httpx.RequestError as e:
logger.error("[pricebot] request failed: %s", e)
raise HTTPException(
@@ -95,13 +83,6 @@ async def intent_recognize(request: Request) -> dict[str, Any]:
return await _passthrough(request, "/api/intent/recognize")
@router.post("/intent/step", summary="外卖比价 Phase 1 多帧意图识别 (透传到 pricebot, 仅淘宝源)")
async def intent_step(request: Request) -> dict[str, Any]:
# 多帧版意图识别(展开+滚动采集→提取): 循环调用直到 done(done 帧顶层带
# result+calibration)。目前仅淘宝源走这条, 其它源走上面单次 /intent/recognize。
return await _passthrough(request, "/api/intent/step")
@router.post("/price/step", summary="外卖比价 Phase 2 步进 (透传到 pricebot)")
async def price_step(request: Request) -> dict[str, Any]:
return await _passthrough(request, "/api/price/step")
-76
View File
@@ -1,76 +0,0 @@
"""比价战绩里程碑 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
)
-97
View File
@@ -1,97 +0,0 @@
"""比价记录 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 (
CompareStatsOut,
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(
"/stats",
response_model=CompareStatsOut,
summary="比价口径战绩(完成比价数 + 累计发现可省)",
)
def stats(user: CurrentUser, db: DbSession) -> CompareStatsOut:
count, saved = crud_compare.get_stats(db, user.id)
return CompareStatsOut(compare_count=count, discovered_saved_cents=saved)
@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 -124
View File
@@ -10,72 +10,19 @@ pricebot 协议文档:
"""
from __future__ import annotations
import json
import logging
from typing import Any
import httpx
from fastapi import APIRouter, HTTPException, Request, status
from fastapi.concurrency import run_in_threadpool
from app.api.deps import DbSession
from app.core.config import settings
from app.core.pricebot_router import pick_pricebot
from app.db.session import SessionLocal
from app.repositories import coupon_state as coupon_repo
from app.schemas.coupon_state import CouponPromptDismissIn, CouponPromptShouldShowOut
logger = logging.getLogger("shagua.coupon")
router = APIRouter(prefix="/api/v1/coupon", tags=["coupon"])
def _to_int(v: object) -> int | None:
"""user_id 协议是字符串(登录态才带),转 int 存库;缺失/非法 → None。"""
if v is None:
return None
try:
return int(v) # type: ignore[arg-type]
except (TypeError, ValueError):
return None
def _extract_coupon_results(resp_json: dict) -> list[dict]:
"""从 pricebot 响应抽本帧券结果,并**按 coupon_id 去重**。
⚠️ done/单券帧同时带 last_coupon_result(最后一张)+ action.params.coupon_results
(全量含最后一张)→ 最后一张出现两次。必须去重:同批里同 coupon_id 两次,
record_claims 在 autoflush=False 下两次 select 都查不到刚 add 的行 → 重复 add →
commit 撞唯一约束 IntegrityError 回滚整批(done/单券记录全丢)。
coupon_results(全量权威)覆盖 last_coupon_result。"""
by_id: dict[str, dict] = {}
last = resp_json.get("last_coupon_result")
if isinstance(last, dict) and last.get("coupon_id"):
by_id[last["coupon_id"]] = last
params = (resp_json.get("action") or {}).get("params") or {}
cr = params.get("coupon_results")
if isinstance(cr, list):
for x in cr:
if isinstance(x, dict) and x.get("coupon_id"):
by_id[x["coupon_id"]] = x
return list(by_id.values())
def _mark_engagement_blocking(
device_id: str, user_id: int | None, engage_type: str
) -> None:
"""独立 session 写 engagement(async 端点经 run_in_threadpool 调,不阻塞事件循环)。"""
with SessionLocal() as db:
coupon_repo.mark_engagement(db, device_id, user_id, engage_type)
def _record_claims_blocking(
device_id: str, user_id: int | None, trace_id: str | None, results: list[dict]
) -> None:
with SessionLocal() as db:
coupon_repo.record_claims(db, device_id, user_id, trace_id, results)
@router.post("/step", summary="领券任务步进 (透传到 pricebot)")
async def coupon_step(
request: Request,
@@ -86,48 +33,24 @@ async def coupon_step(
- 透传: 不做 schema 校验,pricebot 自己校验
- 失败: 网络不可达 / pricebot 5xx → 502 + 友好 message
"""
# 读原始字节,避免"反序列化→再序列化"的双重 JSON(省透传 CPU)。只 loads 一次拿
# trace_id 做亲和 + 打日志,转发时直接发原始 bytes,不重新 dumps。
raw = await request.body()
try:
meta = json.loads(raw)
body = await request.json()
except Exception as e:
raise HTTPException(status_code=400, detail=f"invalid json body: {e}") from e
if not isinstance(meta, dict):
meta = {}
device_id = meta.get("device_id")
user_id = _to_int(meta.get("user_id")) # 登录态才带;判断不靠它,资产留痕用
trace_id = meta.get("trace_id")
# 领券任务首帧(step=0)= 用户已发起领券 → 记一条今日 engagement(claim_started),
# 今天这台设备不再弹引导窗(对齐前台"点一键领取即 markEngaged")。写库失败绝不能
# 连累领券主流程,整段吞掉。
if device_id and meta.get("step") == 0:
try:
await run_in_threadpool(
_mark_engagement_blocking, device_id, user_id, "claim_started"
)
except Exception as e: # noqa: BLE001
logger.warning("coupon engagement write failed: %s", e)
# 按 trace_id 一致性 hash 选 pricebot 实例(同一领券任务的所有帧落同一进程)
base = pick_pricebot(meta.get("trace_id"))
url = f"{base.rstrip('/')}/api/coupon/step"
url = f"{settings.PRICEBOT_BASE_URL.rstrip('/')}/api/coupon/step"
timeout = settings.PRICEBOT_REQUEST_TIMEOUT_SEC
logger.info(
"coupon_step device_id=%s trace_id=%s step=%s",
meta.get("device_id"),
meta.get("trace_id"),
meta.get("step"),
body.get("device_id"),
body.get("trace_id"),
body.get("step"),
)
try:
async with httpx.AsyncClient(timeout=timeout) as client:
resp = await client.post(
url, content=raw, headers={"Content-Type": "application/json"}
)
resp = await client.post(url, json=body)
except httpx.RequestError as e:
logger.error("[pricebot] request failed: %s", e)
raise HTTPException(
@@ -146,44 +69,4 @@ async def coupon_step(
detail=f"pricebot upstream returned {resp.status_code}",
)
resp_json = resp.json()
# 领券结果沉淀(资产):中间帧 last_coupon_result + done 帧 coupon_results 幂等写库。
# 当前只记录、不参与"要不要领"判断(MVP 先不去重)。写库失败不影响返回。
if device_id:
results = _extract_coupon_results(resp_json)
if results:
try:
await run_in_threadpool(
_record_claims_blocking, device_id, user_id, trace_id, results
)
except Exception as e: # noqa: BLE001
logger.warning("coupon claim write failed: %s", e)
return resp_json
@router.post("/prompt/dismiss", summary="用户拒绝/关闭领券引导窗(记今日已 engage)")
def coupon_prompt_dismiss(payload: CouponPromptDismissIn, db: DbSession) -> dict[str, bool]:
"""客户端点关闭引导窗时调用 → 记一条今日 engagement(dismissed),今天不再弹。
server 在透传链路里看不到"用户拒绝"(拒绝不发起领券),故必须客户端通知。
MVP 不鉴权,按 device_id 记。
"""
coupon_repo.mark_engagement(db, payload.device_id, payload.user_id, "dismissed")
return {"ok": True}
@router.get(
"/prompt/should-show",
response_model=CouponPromptShouldShowOut,
summary="切到外卖 App 时是否还应弹领券引导窗",
)
def coupon_prompt_should_show(
device_id: str, db: DbSession
) -> CouponPromptShouldShowOut:
"""今天这台设备已 engage(领或拒)过 → should_show=false。客户端据此决定弹不弹
(前台 SP 缓存做快速路径,这里是权威)。"""
return CouponPromptShouldShowOut(
should_show=not coupon_repo.has_engaged_today(db, device_id)
)
return resp.json()
-64
View File
@@ -1,64 +0,0 @@
"""好友邀请 endpoint。
路由前缀 /api/v1/invite,需 Bearer 鉴权(用户级数据):
GET /me 我的邀请码 + 分享链接 + 已邀人数/已得金币(邀请页展示)
POST /bind 把当前登录用户(被邀请人)绑定到某邀请码,注册即生效,双方各发金币
客户端两种调用 /bind:
- 自动:首启读剪贴板拿到邀请码 → 登录后调本接口(channel=clipboard)
- 手动:用户在邀请页输入好友邀请码(channel=manual)
绑定的真正逻辑(邀请码生成/反查、幂等、自邀屏蔽、发金币)在 repositories/invite.py。
"""
from __future__ import annotations
import logging
from fastapi import APIRouter
from app.api.deps import CurrentUser, DbSession
from app.core.config import settings
from app.repositories import invite as invite_repo
from app.schemas.invite import BindInviteIn, BindInviteOut, InviteInfoOut
logger = logging.getLogger("shagua.invite")
router = APIRouter(prefix="/api/v1/invite", tags=["invite"])
_BIND_MESSAGES = {
"success": "邀请绑定成功,金币已到账",
"already_bound": "你已绑定过邀请人",
"invalid_code": "邀请码无效",
"self_invite": "不能填写自己的邀请码",
"not_eligible": "邀请仅对新注册用户生效",
}
@router.get("/me", response_model=InviteInfoOut, summary="我的邀请码 + 分享链接 + 战绩")
def my_invite(user: CurrentUser, db: DbSession) -> InviteInfoOut:
code = invite_repo.ensure_code(db, user)
invited, coins = invite_repo.get_stats(db, user.id)
sep = "&" if "?" in settings.INVITE_LANDING_URL else "?"
share_url = f"{settings.INVITE_LANDING_URL}{sep}ref={code}"
return InviteInfoOut(
invite_code=code,
share_url=share_url,
invited_count=invited,
coins_earned=coins,
)
@router.post("/bind", response_model=BindInviteOut, summary="绑定邀请人(注册即生效,双方发金币)")
def bind_invite(req: BindInviteIn, user: CurrentUser, db: DbSession) -> BindInviteOut:
result = invite_repo.bind(
db, invitee=user, invite_code=req.invite_code, channel=req.channel,
)
logger.info(
"invite bind invitee=%d code=%s channel=%s -> %s",
user.id, req.invite_code, req.channel, result.status,
)
return BindInviteOut(
status=result.status,
coins_awarded=result.invitee_coin,
message=_BIND_MESSAGES.get(result.status, result.status),
)
+4 -75
View File
@@ -7,14 +7,10 @@ from __future__ import annotations
import logging
from concurrent.futures import ThreadPoolExecutor, as_completed
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy import select
from sqlalchemy.orm import Session
from fastapi import APIRouter, HTTPException
from app.core.config import settings
from app.db.session import get_db
from app.integrations.meituan import MeituanCpsError, get_referral_link, query_coupon
from app.models.meituan_coupon import MeituanCoupon
from app.schemas.meituan import (
CouponCard,
CouponListRequest,
@@ -23,7 +19,6 @@ from app.schemas.meituan import (
FeedResponse,
ReferralLinkRequest,
ReferralLinkResponse,
TopSalesRequest,
)
logger = logging.getLogger("shagua.meituan")
@@ -84,21 +79,13 @@ def _interleave(waimai: list[dict], daodian: list[dict]) -> list[CouponCard]:
return items
def _commission_pct(card: CouponCard) -> float:
"""'1.4%' → 1.4;解析失败按 0(会被智能推荐过滤掉)。"""
try:
return float(card.commission_rate.rstrip("%"))
except (ValueError, AttributeError):
return 0.0
@router.post("/feed", response_model=FeedResponse, summary="混合feed(外卖+到店交叉);tab=rec智能推荐/distance距离最近")
@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
tab = (req.tab or "").strip()
logger.info("[feed] tab=%s page=%s lon=%.6f lat=%.6f", tab or "(default)", req.page, lon, lat)
logger.info("[feed] page=%s lon=%.6f lat=%.6f", req.page, lon, lat)
def _fetch_topic(platform: int, biz_line: int | None, topic: int) -> list[dict]:
try:
@@ -110,28 +97,6 @@ def feed(req: FeedRequest) -> FeedResponse:
except MeituanCpsError:
return []
# 距离最近:拉齐全部榜单轮次,合并去重,后端在【完整池子】上全局按距离由近及远排,一次性返回。
# (距离排序必须在完整池上做、不能逐页排——这正是之前放前端不合理的根因。)
if tab == "distance":
with ThreadPoolExecutor(max_workers=len(_TOPIC_ROUNDS) * 2) as pool:
futs = []
for wm_topic, dd_topic in _TOPIC_ROUNDS:
futs.append(pool.submit(_fetch_topic, 1, None, wm_topic))
futs.append(pool.submit(_fetch_topic, 2, 1, dd_topic))
raws = [f.result() for f in futs]
seen: set[str] = set()
cards: list[CouponCard] = []
for raw_list in raws:
for it in raw_list:
card = CouponCard.from_raw(it)
if card.product_view_sign and card.product_view_sign not in seen:
seen.add(card.product_view_sign)
cards.append(card)
cards.sort(key=lambda c: c.distance_meters if c.distance_meters is not None else float("inf"))
return FeedResponse(items=cards, has_next=False, page=1)
# 智能推荐(rec,默认):沿用逐轮分页的混合 feed,后端过滤掉佣金率 < 3%。
page_idx = req.page - 1
if page_idx >= len(_TOPIC_ROUNDS):
return FeedResponse(items=[], has_next=False, page=req.page)
@@ -142,8 +107,6 @@ def feed(req: FeedRequest) -> FeedResponse:
waimai, daodian = f_wm.result(), f_dd.result()
items = _interleave(waimai, daodian)
if tab == "rec":
items = [c for c in items if _commission_pct(c) >= 3.0]
has_next = page_idx + 1 < len(_TOPIC_ROUNDS)
return FeedResponse(items=items, has_next=has_next, page=req.page)
@@ -167,37 +130,3 @@ def referral_link(req: ReferralLinkRequest) -> ReferralLinkResponse:
link_map = raw.get("referralLinkMap") or {}
link = raw.get("data") or link_map.get("1") or link_map.get("3") or next(iter(link_map.values()), "")
return ReferralLinkResponse(link=link, link_map=link_map)
@router.post("/top-sales", response_model=CouponListResponse,
summary="销量最高(从离线库 meituan_coupon 按销量降序 + 跨源去重,不实时打美团)")
def top_sales(req: TopSalesRequest, db: Session = Depends(get_db)) -> CouponListResponse:
# 只取有销量档的(美团很多券没销量,排不了);按销量降序、同销量再按佣金降序
stmt = select(MeituanCoupon).where(MeituanCoupon.sale_volume_num.isnot(None))
if req.platform is not None:
stmt = stmt.where(MeituanCoupon.platform == req.platform)
stmt = stmt.order_by(
MeituanCoupon.sale_volume_num.desc(),
MeituanCoupon.commission_percent.desc(),
)
rows = db.execute(stmt).scalars().all()
# 跨源去重(dedup_key = 品牌|名|价):按销量降序遍历,每个 dedup_key 只留第一条(=销量最高那条)。
# 用 raw(整条原始返回)重建 CouponCard,字段与实时接口完全一致,前端无需改渲染。
seen: set[str] = set()
cards: list[CouponCard] = []
for row in rows:
if row.dedup_key in seen:
continue
seen.add(row.dedup_key)
try:
card = CouponCard.from_raw(row.raw or {})
except Exception: # noqa: BLE001
continue
if card.product_view_sign:
cards.append(card)
start = (req.page - 1) * req.page_size
page_items = cards[start:start + req.page_size]
has_next = start + req.page_size < len(cards)
return CouponListResponse(items=page_items, has_next=has_next, search_id=None)
-25
View File
@@ -1,25 +0,0 @@
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,
)
-37
View File
@@ -1,37 +0,0 @@
"""首页平台级展示数据 endpoint(全平台门面数字,**不鉴权**——登录前首页也要展示)。
路由前缀 `/api/v1/platform`:
GET /stats 首页三统计(帮助用户 / 完成比价 / 累计节省),按运营后台配的模式算。
展示模式(real/manual/random,每指标独立)与计算逻辑见 app/repositories/ops_stat.py。
"""
from __future__ import annotations
import logging
from fastapi import APIRouter, Query
from app.api.deps import DbSession
from app.repositories import ops_marquee as marquee_crud
from app.repositories import ops_stat as crud
from app.schemas.platform import PlatformStatsOut, SavingsFeedItem, SavingsFeedOut
logger = logging.getLogger("shagua.platform")
router = APIRouter(prefix="/api/v1/platform", tags=["platform"])
@router.get("/stats", response_model=PlatformStatsOut, summary="首页三统计(全平台门面数字)")
def stats(db: DbSession) -> PlatformStatsOut:
v = crud.get_display_values(db)
return PlatformStatsOut(
help_users=v["help_users"],
total_compares=v["total_compares"],
total_saved_cents=v["total_saved"],
)
@router.get("/savings-feed", response_model=SavingsFeedOut, summary="首页轮播 feed(真实+种子混播)")
def savings_feed(db: DbSession, limit: int = Query(8, ge=1, le=30)) -> SavingsFeedOut:
items = marquee_crud.get_feed(db, limit=limit)
return SavingsFeedOut(items=[SavingsFeedItem(**it) for it in items])
-145
View File
@@ -1,145 +0,0 @@
"""上报更低价 endpoint。
路由前缀 `/api/v1/report`,需 Bearer 鉴权(上报绑登录用户)。
POST / 提交上报(multipart:comparison_record_id / reported_platform_id /
reported_price(元) + images 1~4 张)
GET /records 上报记录列表(?status= pending/approved/rejected 可选筛选)
要点:
- 截图复用 [app.core.media] 落盘到 /media/price_report/。
- 原最低价由 comparison_record_id 反查比价记录 best_*(不信任客户端传的快照)。
- 校验(D):上报价必须 < 原最低价,否则 400。
- 提交一律 status=pending;通过发奖励是人工审核后台动作,不在此。
"""
from __future__ import annotations
import logging
from fastapi import APIRouter, File, Form, HTTPException, UploadFile
from app.api.deps import CurrentUser, DbSession
from app.core import media
from app.models.comparison import ComparisonRecord
from app.repositories import report as report_repo
from app.schemas.report import (
ReportRecordCounts,
ReportRecordOut,
ReportRecordsOut,
ReportSubmitOut,
)
logger = logging.getLogger("shagua.report")
router = APIRouter(prefix="/api/v1/report", tags=["report"])
_MAX_IMAGES = 4
_VALID_STATUS = ("pending", "approved", "rejected")
# 上报平台(canonical id → 展示名),与原型 fb-platform-chip data-pf 一致
_PLATFORM_NAMES = {
"meituan-waimai": "美团外卖",
"jd-waimai": "京东外卖",
"taobao-shanguang": "淘宝闪购",
}
def _dish_summary(items: list | None) -> str | None:
"""把比价记录 items[{name,qty,...}] 拼成菜品摘要文案。"""
if not items:
return None
names = [str(it.get("name", "")).strip() for it in items if isinstance(it, dict)]
names = [n for n in names if n]
return "".join(names) if names else None
@router.post("", response_model=ReportSubmitOut, summary="提交上报更低价")
async def submit_report(
user: CurrentUser,
db: DbSession,
comparison_record_id: int = Form(...),
reported_platform_id: str = Form(...),
reported_price: str = Form(..., description="用户填的更低价(元)"),
images: list[UploadFile] = File(default=[]),
) -> ReportSubmitOut:
# 平台
reported_platform_id = reported_platform_id.strip()
platform_name = _PLATFORM_NAMES.get(reported_platform_id)
if platform_name is None:
raise HTTPException(status_code=400, detail="不支持的上报平台")
# 价格(元 → 分)
try:
price_yuan = float(reported_price)
except (TypeError, ValueError):
raise HTTPException(status_code=400, detail="价格格式不正确") from None
if price_yuan <= 0:
raise HTTPException(status_code=400, detail="价格必须大于 0")
reported_price_cents = round(price_yuan * 100)
# 反查比价记录(必须属于本人)→ 取原最低价快照
rec = db.get(ComparisonRecord, comparison_record_id)
if rec is None or rec.user_id != user.id:
raise HTTPException(status_code=404, detail="比价记录不存在")
original_price_cents = rec.best_price_cents
# D:上报价必须低于原最低价
if original_price_cents is not None and reported_price_cents >= original_price_cents:
raise HTTPException(
status_code=400,
detail=f"上报价需低于原最低价 ¥{original_price_cents / 100:.2f}",
)
# 截图(至少 1 张,最多 4 张)
files = [f for f in (images or []) if f is not None and f.filename]
if not files:
raise HTTPException(status_code=400, detail="请至少上传一张截图证明")
if len(files) > _MAX_IMAGES:
raise HTTPException(status_code=400, detail=f"最多上传 {_MAX_IMAGES} 张图片")
urls: list[str] = []
for f in files:
data = await f.read()
try:
urls.append(media.save_report_image(user.id, data))
except media.MediaError as e:
raise HTTPException(status_code=400, detail=str(e)) from e
rep = report_repo.create_report(
db,
user_id=user.id,
comparison_record_id=comparison_record_id,
store_name=rec.store_name,
dish_summary=_dish_summary(rec.items),
original_platform_id=rec.best_platform_id,
original_platform_name=rec.best_platform_name,
original_price_cents=original_price_cents,
reported_platform_id=reported_platform_id,
reported_platform_name=platform_name,
reported_price_cents=reported_price_cents,
images=urls,
)
logger.info("price_report id=%d user_id=%d images=%d", rep.id, user.id, len(urls))
return ReportSubmitOut.model_validate(rep)
@router.get("/records", response_model=ReportRecordsOut, summary="上报记录列表")
def list_report_records(
user: CurrentUser,
db: DbSession,
status: str | None = None,
) -> ReportRecordsOut:
status = (status or "").strip() or None
if status and status not in _VALID_STATUS:
raise HTTPException(status_code=400, detail="无效的状态筛选")
rows = report_repo.list_reports(db, user.id, status)
# counts 始终基于全量(不受 status 筛选影响),供前端 chip 计数
all_rows = rows if status is None else report_repo.list_reports(db, user.id, None)
counts = ReportRecordCounts(
all=len(all_rows),
pending=sum(1 for r in all_rows if r.status == "pending"),
approved=sum(1 for r in all_rows if r.status == "approved"),
rejected=sum(1 for r in all_rows if r.status == "rejected"),
)
return ReportRecordsOut(
records=[ReportRecordOut.model_validate(r) for r in rows],
counts=counts,
)
-1
View File
@@ -42,7 +42,6 @@ def battle(user: CurrentUser, db: DbSession) -> SavingsBattleOut:
week_saved_cents=b.week_saved_cents,
beat_percent=b.beat_percent,
streak_days=b.streak_days,
compare_count=b.compare_count,
)
+2 -32
View File
@@ -1,9 +1,8 @@
"""签到 endpoint。
路由前缀 `/api/v1/signin`:
GET /status 今日签到状态 + 14 天档位
GET /status 今日签到状态 + 7 天档位
POST / 执行今日签到
POST /boost 签到后看广告膨胀金币
"""
from __future__ import annotations
@@ -13,12 +12,7 @@ from fastapi import APIRouter, HTTPException, status
from app.api.deps import CurrentUser, DbSession
from app.repositories import signin as crud_signin
from app.schemas.welfare import (
SigninBoostRequest,
SigninBoostResultOut,
SigninResultOut,
SigninStatusOut,
)
from app.schemas.welfare import SigninResultOut, SigninStatusOut
logger = logging.getLogger("shagua.signin")
@@ -48,27 +42,3 @@ def do_signin(user: CurrentUser, db: DbSession) -> SigninResultOut:
streak=record.streak,
coin_balance=balance,
)
@router.post("/boost", response_model=SigninBoostResultOut, summary="签到后看广告膨胀金币")
def boost_signin(
payload: SigninBoostRequest, user: CurrentUser, db: DbSession
) -> SigninBoostResultOut:
try:
record, balance = crud_signin.boost_today_signin(
db, user.id, ad_ref_id=payload.ad_ref_id
)
except crud_signin.NotSignedTodayError as e:
raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="not signed today") from e
except crud_signin.AlreadyBoostedError as e:
raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="already boosted today") from e
logger.info(
"signin boost ok user_id=%d date=%s coin=%d",
user.id, record.signin_date, record.coin_awarded,
)
return SigninBoostResultOut(
coin_awarded=record.coin_awarded,
coin_balance=balance,
signin_date=record.signin_date.isoformat(),
)
+10 -86
View File
@@ -33,8 +33,6 @@ from app.schemas.welfare import (
ExchangeInfoOut,
ExchangeRequest,
ExchangeResultOut,
TransferAuthResultOut,
TransferAuthStatusOut,
UnbindWechatResultOut,
WithdrawInfoOut,
WithdrawOrderOut,
@@ -157,29 +155,25 @@ def unbind_wechat(user: CurrentUser, db: DbSession) -> UnbindWechatResultOut:
return UnbindWechatResultOut(bound=False)
@router.get("/withdraw-info", response_model=WithdrawInfoOut, summary="提现额度/绑定状态/免确认开关")
@router.get("/withdraw-info", response_model=WithdrawInfoOut, summary="提现额度/绑定状态")
def withdraw_info(user: CurrentUser, db: DbSession) -> WithdrawInfoOut:
u = db.get(User, user.id)
# 顺带同步免确认授权状态(捕获首单确认后已生效的授权 pending→active),让开关展示实时
auth = crud_wallet.sync_transfer_auth(db, user.id)
return WithdrawInfoOut(
min_cents=WITHDRAW_MIN_CENTS,
max_cents=WITHDRAW_MAX_CENTS,
wechat_bound=bool(u and u.wechat_openid),
wechat_nickname=u.wechat_nickname if u else None,
wechat_avatar_url=u.wechat_avatar_url if u else None,
transfer_auth_enabled=bool(auth and auth.state == "active"),
)
@router.post(
"/withdraw",
response_model=WithdrawResultOut,
summary="发起提现(扣款建单,待人工审核;审核通过后才打款)",
dependencies=[Depends(rate_limit(5, 60, "withdraw"))], # IP 级粗限流;用户级未完成单限制在仓库层
summary="发起提现到微信零钱",
dependencies=[Depends(rate_limit(20, 60, "withdraw"))], # #6 同 IP 每分钟≤20 次
)
def withdraw(req: WithdrawRequest, user: CurrentUser, db: DbSession) -> WithdrawResultOut:
# 提现发起本身不调微信(打款在审核通过后),但仍要求微信支付已配置——否则审核通过也打不了款,提前拦
if not settings.wxpay_configured:
raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="wechat pay not configured")
try:
@@ -189,35 +183,18 @@ def withdraw(req: WithdrawRequest, user: CurrentUser, db: DbSession) -> Withdraw
except crud_wallet.InvalidWithdrawAmountError as e:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="提现金额不符合要求",
detail=f"amount_cents must be within [{WITHDRAW_MIN_CENTS}, {WITHDRAW_MAX_CENTS}]",
) from e
except crud_wallet.WechatNotBoundError as e:
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="请先绑定微信") from e
except crud_wallet.WithdrawTooFrequentError as e:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail="已有提现申请正在审核或打款中,请处理完成后再申请",
) from e
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="wechat not bound") from e
except crud_wallet.InsufficientCashError as e:
raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="现金余额不足") from e
# 调试直发(非生产):skip_review=true 时跳过人工审核,立即发起微信转账(等价 admin approve)。
# 双闸保护——客户端仅 debug 包下发此 flag,服务端仅非 prod 才认;任一道闸拦住即恢复正常审核,
# 生产绝不会被客户端 flag 绕过审核。用于本地联调"提现→微信转账/免确认到账"全链路。
if req.skip_review and not settings.is_prod and order.status == "reviewing":
logger.warning(
"withdraw skip_review(非prod调试直发,跳过人工审核立即打款) user_id=%d bill=%s",
user.id, order.out_bill_no,
)
order = crud_wallet.execute_withdraw_transfer(db, order)
raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="insufficient cash balance") from e
except crud_wallet.WithdrawTransferError as e:
# 转账调用失败,余额已退回
raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail=f"wechat transfer failed: {e}") from e
acc = crud_wallet.get_or_create_account(db, user.id)
logger.info(
"withdraw submitted user_id=%d cents=%d bill=%s status=%s",
user.id, req.amount_cents, order.out_bill_no, order.status,
)
# 此刻 status=reviewing,尚未打款 → 无 package_info;App 据 status 提示"已提交,等待审核"。
# 审核通过后客户端轮询 /withdraw/status 拿到 package_info(若需确认)再拉起微信确认页。
logger.info("withdraw user_id=%d cents=%d bill=%s state=%s", user.id, req.amount_cents, order.out_bill_no, order.wechat_state)
return WithdrawResultOut(
out_bill_no=order.out_bill_no,
status=order.status,
@@ -246,11 +223,6 @@ def withdraw_status(
status=order.status,
wechat_state=order.wechat_state,
amount_cents=order.amount_cents,
fail_reason=order.fail_reason,
# 审核通过后 status=pending 且可能带 package_info(WAIT_USER_CONFIRM):供 App 拉起微信确认页
package_info=order.package_info,
mch_id=settings.WXPAY_MCH_ID if order.package_info else None,
app_id=settings.WECHAT_APP_ID if order.package_info else None,
)
@@ -266,51 +238,3 @@ def withdraw_orders(
items=[WithdrawOrderOut.model_validate(it) for it in items],
next_cursor=next_cursor,
)
# ===== 免确认收款授权(用户授权免确认模式)=====
# 开启一次后,后续提现走免确认转账直接到账,不再跳微信确认。绑定 openid 是前提。
@router.post(
"/transfer-auth",
response_model=TransferAuthResultOut,
summary="开启免确认到账(申请授权,返回拉起微信授权页的 package)",
dependencies=[Depends(rate_limit(10, 60, "transfer-auth"))],
)
def open_transfer_auth(user: CurrentUser, db: DbSession) -> TransferAuthResultOut:
if not settings.wxpay_auth_configured:
raise HTTPException(
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
detail="wechat transfer auth not configured",
)
try:
info = crud_wallet.apply_transfer_auth(db, user.id)
except crud_wallet.WechatNotBoundError as e:
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="wechat not bound") from e
except crud_wallet.WithdrawTransferError as e:
raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail=str(e)) from e
logger.info("open transfer-auth user_id=%d already_active=%s", user.id, info["already_active"])
return TransferAuthResultOut(**info)
@router.get(
"/transfer-auth/status",
response_model=TransferAuthStatusOut,
summary="查免确认授权状态(从微信授权页返回后轮询)",
)
def transfer_auth_status(user: CurrentUser, db: DbSession) -> TransferAuthStatusOut:
auth = crud_wallet.sync_transfer_auth(db, user.id)
state = auth.state if auth else "none"
return TransferAuthStatusOut(state=state, enabled=(state == "active"))
@router.post(
"/transfer-auth/close",
response_model=TransferAuthStatusOut,
summary="关闭免确认到账(解除授权)",
)
def close_transfer_auth_endpoint(user: CurrentUser, db: DbSession) -> TransferAuthStatusOut:
crud_wallet.close_transfer_auth(db, user.id)
logger.info("close transfer-auth user_id=%d", user.id)
return TransferAuthStatusOut(state="closed", enabled=False)
-30
View File
@@ -1,30 +0,0 @@
"""微信支付相关回调(一期:免确认收款授权结果通知 stub)。
⚠️ 一期不处理回调内容、不验签:授权状态以主动查询(query_transfer_authorization)为准。
本端点仅向微信回 200 避免重试风暴,**绝不依据回调内容改账**。二期接入时必须先补
V3 平台证书/公钥验签(Wechatpay-Signature) + APIv3 密钥 AEAD 解密,验签通过后方可信任并处理。
授权回调地址通过 settings.WXPAY_AUTH_NOTIFY_URL 配置,需指向本端点的公网地址。
"""
from __future__ import annotations
import logging
from fastapi import APIRouter, Request
logger = logging.getLogger("shagua.wxpay")
router = APIRouter(prefix="/api/v1/wxpay", tags=["wxpay"])
@router.post("/transfer-auth-notify", summary="免确认收款授权结果通知(一期 stub:仅应答,不处理)")
async def transfer_auth_notify(request: Request) -> dict:
# 一期:不验签、不解密、不改账。仅记录 + 应答成功;真实授权状态靠 /transfer-auth/status 查询兜底。
try:
body = await request.json()
logger.info(
"transfer-auth notify id=%s type=%s", body.get("id"), body.get("event_type")
)
except Exception: # noqa: BLE001 — body 解析失败也照常应答 200,避免微信重试
logger.info("transfer-auth notify (unparseable body)")
return {"code": "SUCCESS", "message": "OK"}
-54
View File
@@ -1,54 +0,0 @@
"""看激励视频冷却策略 —— 与发奖记录查询解耦的纯计算。
当前策略:**每 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)
+1 -62
View File
@@ -40,20 +40,6 @@ 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 = ""
@@ -65,14 +51,6 @@ 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 返空(优雅降级),不影响登录/领券等其他业务。
@@ -81,9 +59,6 @@ class Settings(BaseSettings):
MT_CPS_HOST: str = "https://media.meituan.com"
MT_CPS_TIMEOUT_SEC: int = 15
MT_CPS_DEFAULT_SID: str = "sgbjia"
# 美团调用走的代理。本机开发直连美团会 SSL EOF,需填 http://127.0.0.1:7897;
# 线上国内服务器留空(=直连)。见 .env.example 与 integrations/meituan.py。
MT_CPS_PROXY: str = ""
@property
def mt_cps_configured(self) -> bool:
@@ -102,13 +77,6 @@ class Settings(BaseSettings):
WXPAY_PUBLIC_KEY_PATH: str = "./secrets/pub_key.pem" # 微信支付平台公钥
WXPAY_TRANSFER_SCENE_ID: str = "1000" # 转账场景 ID(1000=现金营销)
WXPAY_REQUEST_TIMEOUT_SEC: int = 10
WITHDRAW_AUTO_RECONCILE_ENABLED: bool = False
WITHDRAW_AUTO_RECONCILE_INTERVAL_SEC: int = 300
WITHDRAW_AUTO_RECONCILE_OLDER_THAN_MINUTES: int = 15
# 免确认收款授权(用户授权免确认模式)的授权结果回调地址,必须公网可访问 HTTPS、不带参数。
# 发起授权 / 首单顺带授权时作为 authorization_notify_url 传给微信。一期不处理回调内容
# (授权状态靠 query 查询兜底),但微信要求该字段非空,故启用免确认前必须配置;留空时免确认相关接口返回未配置。
WXPAY_AUTH_NOTIFY_URL: str = ""
@property
def wxpay_configured(self) -> bool:
@@ -120,11 +88,6 @@ class Settings(BaseSettings):
and self.WXPAY_PUBLIC_KEY_ID
)
@property
def wxpay_auth_configured(self) -> bool:
"""免确认收款授权可用 = 微信支付凭证齐全 + 授权回调地址已配。"""
return bool(self.wxpay_configured and self.WXPAY_AUTH_NOTIFY_URL)
# ===== 穿山甲激励视频(服务端发奖回调)=====
# 看完激励视频后穿山甲服务器回调本服务发金币(S2S,客户端被破解也刷不到)。
# PANGLE_REWARD_SECRET 是穿山甲后台配置的"奖励校验密钥",验签用,从后台取到后填 .env。
@@ -141,34 +104,15 @@ class Settings(BaseSettings):
"""回调开关打开且验签密钥已配,才接受发奖回调。"""
return bool(self.PANGLE_CALLBACK_ENABLED and self.PANGLE_REWARD_SECRET)
# ===== Pricebot 上游 (领券/比价业务透传目标) =====
# ===== Pricebot 上游 (领券业务透传目标) =====
# pricebot-backend 默认跑在 8000。/api/v1/coupon/step 会透传到这里的 /api/coupon/step
PRICEBOT_BASE_URL: str = "http://localhost:8000"
# 多实例(单机多进程)时填逗号分隔的实例列表,例如:
# PRICEBOT_INSTANCES=http://127.0.0.1:8001,http://127.0.0.1:8002,http://127.0.0.1:8003
# 透传层按 trace_id 一致性 hash 选实例(见 app/core/pricebot_router.py),保证同一次
# 比价/领券的所有帧落同一 pricebot 进程,从而用进程内内存维护 session/coordinator,
# 无需 Redis。留空 → 退回单实例 [PRICEBOT_BASE_URL],零改动兼容。
PRICEBOT_INSTANCES: str = ""
# 领券一帧最多 wait 6s,加网络往返,timeout 给 30s 比较稳
PRICEBOT_REQUEST_TIMEOUT_SEC: int = 30
# 比价(intent/recognize + price/step)透传超时:意图识别是大上下文 LLM、
# price/step 每帧也是 LLM,可能 >30s;对齐客户端 agent ApiClient 的 60s 读超时。
PRICEBOT_COMPARE_TIMEOUT_SEC: int = 60
@property
def pricebot_instances(self) -> list[str]:
"""pricebot 上游实例列表。空 → 单实例兜底 [PRICEBOT_BASE_URL]。"""
if not self.PRICEBOT_INSTANCES.strip():
return [self.PRICEBOT_BASE_URL]
return [u.strip() for u in self.PRICEBOT_INSTANCES.split(",") if u.strip()]
# ===== 内部(server→server)端点密钥 =====
# pricebot 比价 done 后把价格观测 POST 到 /internal/price-observation 落库,
# 靠这个共享密钥头(X-Internal-Secret)校验,与 pricebot 侧 INTERNAL_API_SECRET 同值。
# 默认空 = 内部写端点关闭(返 503),启用前两边都要配上同一高熵串。
INTERNAL_API_SECRET: str = ""
# ===== 媒体文件(用户头像上传)=====
# 落盘根目录(data/ 已 gitignore,上传不进库);对外经 StaticFiles 挂在 MEDIA_URL_PREFIX。
# 生产可改由 nginx 直接 serve MEDIA_ROOT,绕过应用进程。
@@ -176,11 +120,6 @@ class Settings(BaseSettings):
MEDIA_URL_PREFIX: str = "/media"
AVATAR_MAX_BYTES: int = 5 * 1024 * 1024 # 头像最大 5MB
# ===== 邀请好友 =====
# 分享落地页(二维码 / 分享链接指向这里;扫码 → 落地页 → 引导浏览器下载 APK)。
# MVP 落地页就放 app-server 的 /media 静态目录下(dl.html)。
INVITE_LANDING_URL: str = "https://app-api.shaguabijia.com/media/dl.html"
# ===== CORS =====
CORS_ALLOW_ORIGINS: str = ""
-63
View File
@@ -1,63 +0,0 @@
"""运营可配置项注册表:默认值(= 原 rewards 常量)+ 元信息(label/group/type)。
配置项的 single source of truth:
- 业务读配置时 fallback 到这里的 default(空 DB = 原行为不变)。
- admin 配置页用它展示"有哪些可配项 + 当前值 + 默认值 + 说明"
新增可配项 = 这里加一条 + 业务处改用 app_config.get_value(db, key) 读。
"""
from __future__ import annotations
from typing import Any
from app.core import rewards as r
# type 约定(给前端渲染编辑控件用):int / int_list / dict_str_int
CONFIG_DEFS: dict[str, dict[str, Any]] = {
"signin_rewards": {
"default": list(r.SIGNIN_REWARDS), "label": "签到 14 天金币档位",
"group": "签到", "type": "int_list",
"help": "第 1~14 天每天签到发的金币;断签重置回第 1 天。长度需为 14。",
},
"min_exchange_coin": {
"default": r.MIN_EXCHANGE_COIN, "label": "最低兑换金币",
"group": "钱包", "type": "int", "help": "金币兑现金的单次最低金币(10000 金币=1 元)。",
},
"withdraw_min_cents": {
"default": r.WITHDRAW_MIN_CENTS, "label": "提现最低额(分)",
"group": "钱包", "type": "int", "help": "单次提现最低;微信商家转账最低 10 分(0.1 元)。",
},
"withdraw_max_cents": {
"default": r.WITHDRAW_MAX_CENTS, "label": "提现最高额(分)",
"group": "钱包", "type": "int",
},
"task_rewards": {
"default": dict(r.TASK_REWARDS), "label": "一次性任务奖励",
"group": "任务", "type": "dict_str_int", "help": "task_key → 金币。",
},
"record_milestones": {
"default": list(r.RECORD_MILESTONES), "label": "比价里程碑金币档位",
"group": "里程碑", "type": "int_list",
"help": "累计成功比价第 1~N 档解锁发的金币。",
},
"ad_reward_coin": {
"default": r.AD_REWARD_COIN, "label": "看广告单次金币",
"group": "看广告", "type": "int",
"help": "看完一个激励视频发的金币;须与穿山甲后台代码位'奖励数量'保持一致。",
},
"ad_daily_limit": {
"default": r.DAILY_AD_REWARD_LIMIT, "label": "看广告每日上限(次)",
"group": "看广告", "type": "int", "help": "福利页激励视频每日可发奖次数上限,默认 500。",
},
"ad_max_coin": {
"default": r.MAX_AD_REWARD_COIN, "label": "看广告单次金币上限",
"group": "看广告", "type": "int", "help": "夹紧穿山甲回调里异常的 reward_amount,防刷爆。",
},
"ad_round_count": {
"default": r.VIDEO_ROUND_REQUIRED_COUNT, "label": "每轮看广告次数",
"group": "看广告", "type": "int", "help": "当前为 1,表示每次广告关闭后触发短冷却。",
},
"ad_cooldown_sec": {
"default": r.VIDEO_ROUND_COOLDOWN_SECONDS, "label": "广告关闭后冷却(秒)",
"group": "看广告", "type": "int", "help": "点击退出广告后,下次点击观看前的冷却时间,默认 3 秒。",
},
}
+5 -76
View File
@@ -1,91 +1,20 @@
"""统一日志配置。
业务代码用 `logger = logging.getLogger("shagua.xxx")` 即可, 本模块在 main.py 启动时调一次。
- 控制台(stdout): 人类可读文本, 给 systemd / 本地看。
- 文件 `logs/app-server.log`: 单行 JSON, 供阿里云 SLS/Logtail 采集(JSON 模式零正则);
异常栈作为字段内嵌不换行 → 每条日志一行。
- 环境变量:
- LOG_JSON_CONSOLE=1 控制台也输出 JSON
- LOG_DIR / LOG_FILE 改落盘路径(默认 logs/app-server.log)
⚠️ admin 子进程(app.admin.main, 独立进程)应设不同 LOG_FILE, 避免与主进程争抢同一轮转文件
- LOG_SERVICE_NAME JSON 里的 service 字段(默认 app-server)
业务代码用 `logger = logging.getLogger("shagua.xxx")` 即可,本模块在 main.py 启动时调一次。
"""
from __future__ import annotations
import json
import logging
import os
import sys
from datetime import datetime
from logging.handlers import RotatingFileHandler
from pathlib import Path
class JsonFormatter(logging.Formatter):
"""把 LogRecord 序列化成单行 JSON(SLS/Logtail 友好)。异常栈内嵌为字段, 整条仍是一行。"""
def __init__(self, service: str = "app-server"):
super().__init__()
self.service = service
def format(self, record: logging.LogRecord) -> str:
dt = datetime.fromtimestamp(record.created)
data = {
"time": dt.strftime("%Y-%m-%dT%H:%M:%S.") + f"{int(record.msecs):03d}",
"level": record.levelname,
"service": self.service,
"logger": record.name,
"func": record.funcName,
"line": record.lineno,
"message": record.getMessage(),
}
if record.exc_info:
data["exception"] = self.formatException(record.exc_info)
if record.stack_info:
data["stack"] = self.formatStack(record.stack_info)
return json.dumps(data, ensure_ascii=False, default=str)
_CONFIGURED = False
def setup_logging(debug: bool = False) -> None:
global _CONFIGURED
if _CONFIGURED:
return
level = logging.DEBUG if debug else logging.INFO
service = os.getenv("LOG_SERVICE_NAME", "app-server")
root = logging.getLogger()
root.setLevel(level)
# 清掉 basicConfig/uvicorn 可能预置的 root handler, 避免重复输出
for h in list(root.handlers):
root.removeHandler(h)
# 控制台: 默认文本(systemd/本地看); LOG_JSON_CONSOLE=1 时输出 JSON
console = logging.StreamHandler(sys.stdout)
if os.getenv("LOG_JSON_CONSOLE") == "1":
console.setFormatter(JsonFormatter(service))
else:
console.setFormatter(
logging.Formatter("%(asctime)s %(levelname)s %(name)s: %(message)s")
)
root.addHandler(console)
# 文件: 单行 JSON, 供 Logtail 采集(自动轮转, 单文件 10MB, 保留 5 个)
log_file = os.getenv("LOG_FILE") or str(
Path(os.getenv("LOG_DIR", "logs")) / "app-server.log"
logging.basicConfig(
level=level,
format="%(asctime)s %(levelname)s %(name)s: %(message)s",
stream=sys.stdout,
)
Path(log_file).parent.mkdir(parents=True, exist_ok=True)
file_handler = RotatingFileHandler(
log_file, maxBytes=10 * 1024 * 1024, backupCount=5, encoding="utf-8",
)
file_handler.setFormatter(JsonFormatter(service))
root.addHandler(file_handler)
# 第三方库降噪
logging.getLogger("httpx").setLevel(logging.WARNING)
logging.getLogger("httpcore").setLevel(logging.WARNING)
_CONFIGURED = True
-5
View File
@@ -62,11 +62,6 @@ 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/"
-82
View File
@@ -1,82 +0,0 @@
"""pricebot 上游实例选择:按 trace_id 一致性 hash(ketama 风格)选实例。
单机多进程下,保证同一 trace_id(一次比价/领券的所有帧)落同一 pricebot 进程,
从而用进程内内存维护 session/coordinator,不需要 Redis。
为什么用一致性 hash 而非简单取模(crc32 % N):
加/减实例时,取模会让几乎所有 trace 重新映射(扩缩容全量中断进行中的比价);
一致性 hash(虚拟节点)只重映射约 1/N 的 trace,扩缩容对存量冲击最小。
⚠️ 但内存态下,被重映射的那 1/N trace 仍会丢失 session(状态没外置),
所以扩缩容仍建议挑低峰。
hash 用 md5(纯函数),app-server 多 worker / 多实例算出的环一致,亲和不会因
app-server 自身水平扩展而被破坏。
"""
from __future__ import annotations
import bisect
import hashlib
from typing import Dict, List, Optional
from app.core.config import settings
# 每个真实实例在 hash 环上的虚拟节点数。越多分布越均匀。实例数少时(6~10)需要更多
# 虚拟节点才均匀:实测 150→负载偏差~16%、1000→~6%。环只在启动/实例变更时构建一次,
# 1000×N 个点的构建与 bisect 查找开销都可忽略。
_VNODES_PER_NODE = 1000
def _hash(s: str) -> int:
"""取 md5 前 8 hex(32-bit)做环坐标。确定性、跨进程一致。"""
return int(hashlib.md5(s.encode("utf-8")).hexdigest()[:8], 16)
class _HashRing:
"""ketama 风格一致性 hash 环。不可变,实例列表变化时整体重建(见 _get_ring)。"""
def __init__(self, nodes: List[str]):
self._keys: List[int] = [] # 升序的环坐标
self._key_to_node: Dict[int, str] = {}
for node in nodes:
for v in range(_VNODES_PER_NODE):
h = _hash(f"{node}#{v}")
# 极小概率撞坐标,撞了跳过该虚拟节点(不影响正确性,仅少一个 vnode)
if h not in self._key_to_node:
self._key_to_node[h] = node
self._keys.append(h)
self._keys.sort()
def pick(self, key: str) -> str:
h = _hash(key)
idx = bisect.bisect(self._keys, h)
if idx == len(self._keys):
idx = 0 # 环回绕
return self._key_to_node[self._keys[idx]]
# 按实例列表缓存环,列表变了(改 PRICEBOT_INSTANCES + 重启)才重建。
_ring: Optional[_HashRing] = None
_ring_nodes: tuple = ()
def _get_ring(nodes: List[str]) -> _HashRing:
global _ring, _ring_nodes
key = tuple(nodes)
if _ring is None or key != _ring_nodes:
_ring = _HashRing(nodes)
_ring_nodes = key
return _ring
def pick_pricebot(trace_id: Optional[str]) -> str:
"""按 trace_id 选 pricebot 实例 base url。
- 实例列表来自 settings.pricebot_instances(空则单实例兜底 [PRICEBOT_BASE_URL])
- trace_id 缺失/空 或 单实例 → 直接返回第一个,不进环
"""
nodes = settings.pricebot_instances
if len(nodes) == 1 or not trace_id:
return nodes[0]
# str() 防御:协议保证 trace_id 是 str,但万一传入非 str(如 int)也不至于在
# _hash 的 .encode() 处炸,确定性地选到实例。
return _get_ring(nodes).pick(str(trace_id))
+22 -167
View File
@@ -17,12 +17,9 @@ def cn_today() -> date:
return datetime.now(CN_TZ).date()
# ===== 签到:14 天循环,断签重置回第 1 天 =====
# 第 1→14 天的金币奖励,签到逻辑用 cycle_day(1..14)索引。
SIGNIN_REWARDS: tuple[int, ...] = (
200, 120, 150, 180, 200, 250, 500,
200, 250, 300, 350, 400, 500, 3000,
)
# ===== 签到:7 天循环,断签重置回第 1 天 =====
# 第 1→7 天的金币奖励,签到逻辑用 cycle_day(1..7)索引。
SIGNIN_REWARDS: tuple[int, ...] = (10, 20, 30, 50, 80, 120, 200)
SIGNIN_CYCLE_LEN: int = len(SIGNIN_REWARDS)
@@ -37,9 +34,8 @@ COIN_PER_YUAN: int = 10000
CENTS_PER_YUAN: int = 100
# 1 分对应多少金币 = 100。兑换金币数必须是它的整数倍,否则会出现不足 1 分的零头。
COIN_PER_CENT: int = COIN_PER_YUAN // CENTS_PER_YUAN
# 单次兑换最少 1 分(=COIN_PER_CENT;产品定:可兑 1 分起,与 step 同粒度)。早先为 1 元(=COIN_PER_YUAN),
# 用户 2026-06 改为 1 分:让"换现金"小额即可用(也方便联调验证资产卡飞金币动画)。
MIN_EXCHANGE_COIN: int = COIN_PER_CENT
# 单次兑换最少 1 元(避免大量 1 分级碎兑)
MIN_EXCHANGE_COIN: int = COIN_PER_YUAN
def coins_to_cents(coin_amount: int) -> int:
@@ -57,182 +53,41 @@ WITHDRAW_MAX_CENTS: int = 5_000_000 # 5 万元
TASK_ENABLE_NOTIFICATION = "enable_notification"
# task_key -> 奖励金币
# 打开消息提醒: 1000 金币(=¥0.1, 客户端原型展示口径; 量级与签到/里程碑相称)。
# 注意: 不再 = 兑换下限(下限已降到 MIN_EXCHANGE_COIN=COIN_PER_CENT=100), test_exchange_flow 改走 grant_coins 直接供款。
TASK_REWARDS: dict[str, int] = {
TASK_ENABLE_NOTIFICATION: 1000,
TASK_ENABLE_NOTIFICATION: 10000,
}
# ===== 比价战绩里程碑(累计成功比价 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]
# ===== 邀请好友(注册即生效,邀请人 + 被邀请人各发金币)=====
# 10000 金币 = 1 元,双方各得 1 元。MVP 先用固定常量(不走 app_config)。
INVITE_INVITER_COINS: int = 10000
INVITE_INVITEE_COINS: int = 10000
# "新用户闸":被邀请人必须在注册后此窗口内绑定才发奖(挡存量老用户互相填码薅羊毛)。
# 自动绑(剪贴板)在首次注册登录后几秒内发生;留 72h 给手动填码兜底。
INVITE_NEW_USER_WINDOW_HOURS: int = 72
# ===== 看激励视频 / 信息流广告发金币 =====
# 金币数值体系约定:eCPM 单位按"元/千次展示"处理,单次收入 = eCPM / 1000 元。
AD_ECPM_FACTOR_TABLE: tuple[tuple[float, int, int | None], ...] = (
(0.1, 0, 100),
(0.3, 101, 200),
(0.4, 201, 400),
(0.6, 401, None),
)
AD_LT_FACTOR_TABLE: tuple[tuple[float, int, int | None], ...] = (
(2.0, 1, 1),
(1.5, 2, 2),
(1.3, 3, 3),
(1.1, 4, 10),
(1.0, 11, None),
)
def parse_ecpm_yuan(ecpm: str | int | float | None) -> float:
"""解析 eCPM 原始值。当前产品口径:SDK 返回值按"元/千次展示"处理。"""
if ecpm is None:
return 0.0
try:
value = float(str(ecpm).strip())
except (TypeError, ValueError):
return 0.0
return max(0.0, value)
def ad_ecpm_factor(ecpm_yuan: float) -> float:
"""eCPM 档位因子:0-100=0.1,101-200=0.3,201-400=0.4,>400=0.6。"""
if ecpm_yuan > 400:
return 0.6
if ecpm_yuan > 200:
return 0.4
if ecpm_yuan > 100:
return 0.3
return 0.1
def ad_lt_factor(today_count_after_this: int) -> float:
"""LT 因子。today_count_after_this 是当天累计第 N 条/份广告奖励。"""
count = max(1, today_count_after_this)
for factor, lo, hi in AD_LT_FACTOR_TABLE:
if count >= lo and (hi is None or count <= hi):
return factor
return 1.0
def calculate_ad_reward_coin(ecpm: str | int | float | None, today_count_after_this: int) -> int:
"""按金币数值体系计算单份广告奖励金币。
单次奖励(元)=eCPM/1000 × 因子1(eCPM 档) × 因子2(LT);再按 1 元=10000 金币取整。
"""
ecpm_yuan = parse_ecpm_yuan(ecpm)
yuan = (ecpm_yuan / 1000.0) * ad_ecpm_factor(ecpm_yuan) * ad_lt_factor(today_count_after_this)
return max(0, round(yuan * COIN_PER_YUAN))
# ===== 看激励视频发金币(穿山甲 S2S 服务端回调发奖)=====
# 看完一个激励视频发的金币(666 金币 ≈¥0.0666,汇率 10000 金币=1 元)。
# 看完一个激励视频发的金币(100 金币 ≈¥0.01,汇率 10000 金币=1 元)。
# 作用:① 回调缺/坏 reward_amount 时的回退值;② 客户端进度接口展示的"单次预告金币";
# ③ test-grant 本地联调的发奖额。
# 真实发放以穿山甲回调带回的 reward_amount 为准(见 resolve_ad_reward_coin),后台应把
# 代码位"奖励数量"配成与本值一致(=666),保证"广告内展示 / 进度预告 / 实际到账"三者一致。
AD_REWARD_COIN: int = 666
# 代码位"奖励数量"配成与本值一致(=100),保证"广告内展示 / 进度预告 / 实际到账"三者一致。
AD_REWARD_COIN: int = 100
# 单次发奖金币上限:夹紧穿山甲回调里异常的 reward_amount(如后台多打一个 0),防刷爆余额。
MAX_AD_REWARD_COIN: int = 1000
# 每用户每日发奖次数上限。产品口径:一天最多看 500 次广告
DAILY_AD_REWARD_LIMIT: int = 500
# ===== 看激励视频每日总时长上限(已停用)=====
# 旧版本用 50 分钟时长闸;当前产品只保留每日 500 次上限。保留字段返回给旧客户端,
# 但值为 0,客户端按 limit<=0 视为未启用
DAILY_AD_WATCH_SECONDS_LIMIT: int = 0
# 单次上报观看时长的合理上限(夹紧异常值):激励视频一般 15~60 秒,超 120 秒按 120 记,防刷大值。
MAX_SINGLE_WATCH_SECONDS: int = 120
# 每次广告关闭后 3 秒内不允许再次点击观看。沿用 round_count/cooldown_until 字段表达:
# round_size=1 表示每次成功发奖后都会派生一个 3 秒 cooldown_until。
VIDEO_ROUND_REQUIRED_COUNT: int = 1
VIDEO_ROUND_COOLDOWN_SECONDS: int = 3
# 每用户每日发奖次数上限,防刷 + 控成本
# ⚠️ 测试阶段值=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(db, reward_amount: str | int | None) -> int: # noqa: ANN001
def resolve_ad_reward_coin(reward_amount: str | int | None) -> int:
"""把穿山甲回调的 reward_amount(后台代码位配的"奖励数量")解析成本次发放金币。
缺失 / 非数字 / ≤0 → 回退配置单次金币;超过配置单次上限 → 夹紧(均从 app_config 读)
缺失 / 非数字 / ≤0 → 回退 AD_REWARD_COIN;超过 MAX_AD_REWARD_COIN → 夹紧
注:reward_amount 不参与穿山甲 sign(只签 trans_id),但伪造回调需先伪造合法 sign
(要 SecurityKey),故能过验签的 reward_amount 即视为可信,这里只防后台配错。
"""
default_coin = get_ad_reward_coin(db)
try:
coin = int(reward_amount) # type: ignore[arg-type]
except (TypeError, ValueError):
return default_coin
return AD_REWARD_COIN
if coin <= 0:
return default_coin
return min(coin, get_ad_max_coin(db))
# ===== 配置读取(运营后台可覆盖;app_config 表空时返回上面的常量默认 = 现状不变)=====
# 业务处用这些 getter(传 db)取值,而非直接用常量,这样 admin 改 app_config 即生效。
# 函数内延迟 import,避免 rewards ← app_config ← config_schema ← rewards 的模块级循环。
def _cfg(db, key: str): # noqa: ANN001
from app.repositories import app_config
return app_config.get_value(db, key)
def get_signin_rewards(db) -> list[int]: # noqa: ANN001
return list(_cfg(db, "signin_rewards"))
def get_min_exchange_coin(db) -> int: # noqa: ANN001
return int(_cfg(db, "min_exchange_coin"))
def get_withdraw_min_cents(db) -> int: # noqa: ANN001
return int(_cfg(db, "withdraw_min_cents"))
def get_withdraw_max_cents(db) -> int: # noqa: ANN001
return int(_cfg(db, "withdraw_max_cents"))
def get_task_rewards(db) -> dict[str, int]: # noqa: ANN001
return dict(_cfg(db, "task_rewards"))
def get_record_milestones(db) -> list[int]: # noqa: ANN001
return list(_cfg(db, "record_milestones"))
def get_ad_reward_coin(db) -> int: # noqa: ANN001
return int(_cfg(db, "ad_reward_coin"))
def get_ad_daily_limit(db) -> int: # noqa: ANN001
return int(_cfg(db, "ad_daily_limit"))
def get_ad_max_coin(db) -> int: # noqa: ANN001
return int(_cfg(db, "ad_max_coin"))
def get_ad_round_count(db) -> int: # noqa: ANN001
return int(_cfg(db, "ad_round_count"))
def get_ad_cooldown_sec(db) -> int: # noqa: ANN001
return int(_cfg(db, "ad_cooldown_sec"))
return AD_REWARD_COIN
return min(coin, MAX_AD_REWARD_COIN)
-22
View File
@@ -14,7 +14,6 @@ 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
@@ -85,24 +84,3 @@ 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
-118
View File
@@ -1,118 +0,0 @@
"""提现 pending 单自动对账后台任务。"""
from __future__ import annotations
import asyncio
import contextlib
import logging
import os
import time
from collections.abc import Iterator
from pathlib import Path
from sqlalchemy.exc import SQLAlchemyError
from app.core.config import settings
from app.db.session import SessionLocal
from app.integrations.wxpay import WxPayNotConfiguredError
from app.repositories import wallet as wallet_repo
logger = logging.getLogger("shagua.withdraw_reconcile")
_LOCK_PATH = Path(__file__).resolve().parents[2] / "data" / "withdraw_reconcile.lock"
def _touch_lock() -> None:
with contextlib.suppress(FileNotFoundError):
os.utime(_LOCK_PATH, None)
@contextlib.contextmanager
def _single_instance_lock(stale_after_sec: int) -> Iterator[bool]:
"""同机多进程保护:同一时间只允许一个自动查单 worker 运行。"""
_LOCK_PATH.parent.mkdir(parents=True, exist_ok=True)
fd: int | None = None
try:
try:
fd = os.open(str(_LOCK_PATH), os.O_CREAT | os.O_EXCL | os.O_WRONLY)
except FileExistsError:
try:
age = time.time() - _LOCK_PATH.stat().st_mtime
except FileNotFoundError:
age = stale_after_sec + 1
if age > stale_after_sec:
with contextlib.suppress(FileNotFoundError):
_LOCK_PATH.unlink()
try:
fd = os.open(str(_LOCK_PATH), os.O_CREAT | os.O_EXCL | os.O_WRONLY)
except FileExistsError:
fd = None
if fd is None:
yield False
return
os.write(fd, f"pid={os.getpid()} started_at={int(time.time())}\n".encode("ascii"))
yield True
finally:
if fd is not None:
os.close(fd)
with contextlib.suppress(FileNotFoundError):
_LOCK_PATH.unlink()
def _reconcile_once(older_than_minutes: int) -> dict:
with SessionLocal() as db:
return wallet_repo.reconcile_pending_withdraws(db, older_than_minutes=older_than_minutes)
async def _run_loop() -> None:
interval = max(30, int(settings.WITHDRAW_AUTO_RECONCILE_INTERVAL_SEC))
older_than = max(1, int(settings.WITHDRAW_AUTO_RECONCILE_OLDER_THAN_MINUTES))
lock_stale_after = max(interval * 3, 600)
with _single_instance_lock(lock_stale_after) as lock_acquired:
if not lock_acquired:
logger.warning("withdraw auto reconcile skipped: another worker owns lock")
return
await _run_locked_loop(interval, older_than)
async def _run_locked_loop(interval: int, older_than: int) -> None:
logger.info(
"withdraw auto reconcile started interval=%ss older_than=%sm",
interval,
older_than,
)
try:
while True:
try:
_touch_lock()
result = await asyncio.to_thread(_reconcile_once, older_than)
if result["checked"] or result["resolved"]:
logger.info("withdraw auto reconcile result=%s", result)
except WxPayNotConfiguredError:
logger.warning("withdraw auto reconcile skipped: wxpay not configured")
except SQLAlchemyError:
logger.exception("withdraw auto reconcile db error")
except Exception: # noqa: BLE001 - 后台任务不能因单次异常退出
logger.exception("withdraw auto reconcile unexpected error")
await asyncio.sleep(interval)
except asyncio.CancelledError:
logger.info("withdraw auto reconcile stopped")
raise
def start_withdraw_reconcile_worker() -> asyncio.Task | None:
if not settings.WITHDRAW_AUTO_RECONCILE_ENABLED:
logger.info("withdraw auto reconcile disabled")
return None
if not settings.wxpay_configured:
logger.warning("withdraw auto reconcile enabled but wxpay not configured")
return asyncio.create_task(_run_loop(), name="withdraw-auto-reconcile")
async def stop_withdraw_reconcile_worker(task: asyncio.Task | None) -> None:
if task is None:
return
task.cancel()
with contextlib.suppress(asyncio.CancelledError):
await task
+1 -1
View File
@@ -4,7 +4,7 @@
"""
from __future__ import annotations
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import DeclarativeBase, MappedAsDataclass
class Base(DeclarativeBase):
+1 -5
View File
@@ -62,12 +62,8 @@ def _call(path: str, body_obj: dict[str, Any]) -> dict[str, Any]:
}
url = f"{settings.MT_CPS_HOST}{path}"
# 美团调用走 MT_CPS_PROXY(本机开发直连会 SSL EOF,必须走代理;线上留空=直连)。
# trust_env=False:不读进程环境的 HTTP_PROXY,只认配置,避免被错误/失效代理误导。
proxy = settings.MT_CPS_PROXY or None
try:
with httpx.Client(proxy=proxy, trust_env=False, timeout=settings.MT_CPS_TIMEOUT_SEC) as client:
resp = client.post(url, content=body, headers=headers)
resp = httpx.post(url, content=body, headers=headers, timeout=settings.MT_CPS_TIMEOUT_SEC)
except httpx.HTTPError as e:
logger.exception("[MT] http error calling %s", url)
raise MeituanCpsError(f"meituan http error: {e}") from e
+31 -174
View File
@@ -1,211 +1,68 @@
"""短信验证码服务。
两种运行模式由 `SMS_MOCK` 切换:
- **mock**(开发/测试,默认):不真发短信,验证码打到日志;校验**放行任意 N 位数字**
(测试/开发便利)。真实校验逻辑(比对存码 / 一次性 / 防爆破)由 real 分支 + 单测覆盖。
- **real**(生产 `SMS_MOCK=false`):本服务生成 N 位验证码 → 调极光短信 REST
`/v1/messages` 发送(自定义验证码模式,极光只负责发,code 由本服务生成/保管/
校验)→ 鉴权复用极光一键登录的 `JG_APP_KEY`/`JG_MASTER_SECRET`(同一极光应用)。
当前实现:**mock 模式**。
- send_code(): 不真发短信,仅 log。记录 phone → 发送时间(防 60s 内重复发)
- verify_code(): 任意 6 位数字均通过(配合前台 demo 行为)
验证码存储:**进程内存**(单 worker uvicorn 够用)。重启丢失(用户重发即可)。多
worker / 多机时内存不共享 → 冷却、每日上限、校验都会失效,届时迁移到 DB/Redis。
见 docs/待办与技术债.md。
后续接真供应商(阿里云 / 腾讯云)时:
- send_code() 改为调供应商 API,把生成的 6 位码存到 cache(redis / sqlite)
- verify_code() 比对 cache 里的码,且验过即作废
防刷三层(短信花钱 + `/sms/send` 在登录前无法 JWT 鉴权):
1. 单号 `SMS_SEND_INTERVAL_SEC` 冷却(本文件)
2. 单号每日 `SMS_DAILY_LIMIT_PER_PHONE` 条上限(本文件)
3. 单 IP 频控(api 层 rate_limit 依赖)+ 极光控制台 IP 白名单/防轰炸(运维侧)
另:单码校验失败 `SMS_MAX_VERIFY_ATTEMPTS` 次即作废(防爆破),验过即作废(一次性)。
进程内存方案占坑期够用;切真实供应商时一并切 redis。
"""
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):
"""业务异常。`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)
"""业务异常:发送过频 / 验证码不对。api 层 catch 后翻成 4xx。"""
def send_code(phone: str) -> int:
"""发送验证码。
"""发送(或 mock 发送)验证码。
Returns: 距下次可发的秒数(= SMS_SEND_INTERVAL_SEC)
Raises: SmsError(过频 429 / 当日超限 429 / 供应商失败 503 / 手机号无效 400)
Returns: 距下次可发的秒数(=0 表示刚刚已发,需等 SMS_SEND_INTERVAL_SEC)
Raises: SmsError 如果上次发送在 SMS_SEND_INTERVAL_SEC 内
"""
now = time.time()
# --- lock 内:防刷检查 + 预占(防并发重复发烧钱)---
with _lock:
_gc(now) # 顺手清过期内存(超阈值才扫)
elapsed = now - _last_sent.get(phone, 0.0)
last = _last_sent.get(phone, 0.0)
elapsed = now - last
if elapsed < settings.SMS_SEND_INTERVAL_SEC:
remain = int(settings.SMS_SEND_INTERVAL_SEC - elapsed)
raise SmsError(f"发送过于频繁,请 {remain}s 后再试")
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()
# 预占:先记冷却/计数/存码,释放锁后再发网络(发失败保留冷却+计数,见下)
raise SmsError(f"send too frequent, retry in {remain}s")
_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
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")
return settings.SMS_SEND_INTERVAL_SEC
def verify_code(phone: str, code: str) -> bool:
"""校验验证码。
- **mock 模式**:放行任意 N 位数字(测试/开发便利,不真校验)。
- **real 模式**:比对本服务存的码,匹配即作废(一次性);失败累计到上限也作废(防爆破)。
"""
"""校验验证码。mock 模式下任意 6 位数字均通过。"""
if settings.SMS_MOCK:
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) # 验过即作废
if len(code) == 6 and code.isdigit():
logger.info("[SMS-MOCK] verify ok for %s****", phone[:3])
return True
rec.attempts += 1
logger.info("[SMS-MOCK] verify fail (need 6 digits) for %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)
# TODO: 比对供应商发出的真实验证码
logger.warning("[SMS] real provider not implemented yet, phone=%s****", phone[:3])
return False
+40 -208
View File
@@ -14,9 +14,7 @@ import base64
import json
import time
import uuid
from pathlib import Path
import certifi
import httpx
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import padding
@@ -25,12 +23,7 @@ from cryptography.hazmat.primitives.asymmetric.rsa import RSAPrivateKey, RSAPubl
from app.core.config import settings
_API_HOST = "https://api.mch.weixin.qq.com"
_PROJECT_ROOT = Path(__file__).resolve().parents[2]
_TRANSFER_PATH = "/v3/fund-app/mch-transfer/transfer-bills"
# 免确认收款授权(用户授权免确认模式)
_AUTH_PATH = "/v3/fund-app/mch-transfer/user-confirm-authorization"
_PRE_TRANSFER_AUTH_PATH = "/v3/fund-app/mch-transfer/transfer-bills/pre-transfer-with-authorization"
_TRANSFER_WITH_AUTH_PATH = "/v3/fund-app/mch-transfer/transfer-bills/transfer"
# 懒加载缓存
_private_key: RSAPrivateKey | None = None
@@ -41,30 +34,15 @@ class WxPayNotConfiguredError(Exception):
"""微信支付凭证 / 证书缺失,无法调用。"""
def _http_client() -> httpx.Client:
"""微信相关 HTTP 客户端。
显式使用当前 Python 环境的 certifi 证书,避免被外部 SSL_CERT_FILE 指到不存在文件。
代理仍允许从环境变量读取,方便本地开发。
"""
return httpx.Client(verify=certifi.where())
def _resolve_config_path(value: str) -> Path:
path = Path(value)
return path if path.is_absolute() else _PROJECT_ROOT / path
def _load_private_key() -> RSAPrivateKey:
global _private_key
if _private_key is None:
key_path = _resolve_config_path(settings.WXPAY_MCH_PRIVATE_KEY_PATH)
try:
with key_path.open("rb") as f:
with open(settings.WXPAY_MCH_PRIVATE_KEY_PATH, "rb") as f:
_private_key = serialization.load_pem_private_key(f.read(), password=None)
except FileNotFoundError as e:
raise WxPayNotConfiguredError(
f"商户私钥不存在: {key_path}"
f"商户私钥不存在: {settings.WXPAY_MCH_PRIVATE_KEY_PATH}"
) from e
return _private_key
@@ -72,13 +50,12 @@ def _load_private_key() -> RSAPrivateKey:
def _load_public_key() -> RSAPublicKey:
global _public_key
if _public_key is None:
key_path = _resolve_config_path(settings.WXPAY_PUBLIC_KEY_PATH)
try:
with key_path.open("rb") as f:
with open(settings.WXPAY_PUBLIC_KEY_PATH, "rb") as f:
_public_key = serialization.load_pem_public_key(f.read())
except FileNotFoundError as e:
raise WxPayNotConfiguredError(
f"微信支付平台公钥不存在: {key_path}"
f"微信支付平台公钥不存在: {settings.WXPAY_PUBLIC_KEY_PATH}"
) from e
return _public_key
@@ -145,7 +122,7 @@ def create_transfer(
"Content-Type": "application/json",
"Wechatpay-Serial": settings.WXPAY_PUBLIC_KEY_ID,
}
with _http_client() as client:
with httpx.Client() as client:
resp = client.post(
f"{_API_HOST}{_TRANSFER_PATH}",
content=body_str,
@@ -162,7 +139,7 @@ def query_transfer(out_bill_no: str) -> dict:
"Authorization": _build_authorization("GET", path, ""),
"Accept": "application/json",
}
with _http_client() as client:
with httpx.Client() as client:
resp = client.get(
f"{_API_HOST}{path}", headers=headers, timeout=settings.WXPAY_REQUEST_TIMEOUT_SEC
)
@@ -178,29 +155,46 @@ def cancel_transfer(out_bill_no: str) -> dict:
"Accept": "application/json",
"Content-Type": "application/json",
}
with _http_client() as client:
with httpx.Client() as client:
resp = client.post(
f"{_API_HOST}{path}", headers=headers, timeout=settings.WXPAY_REQUEST_TIMEOUT_SEC
)
return {"status_code": resp.status_code, "data": resp.json()}
def 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。
注意:微信隐私新政下,部分 app 的 sns/userinfo 可能返回脱敏值(昵称"微信用户"/灰头像);
nickname/avatar_url 可能为空,调用方需兜底。"""
with _http_client() as client:
r1 = client.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,
)
r1 = 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,
)
d1 = r1.json()
if "openid" not in d1 or "access_token" not in d1:
raise ValueError(f"微信授权失败: {d1.get('errmsg', d1)}")
@@ -211,12 +205,11 @@ def code_to_userinfo(code: str) -> dict:
raw: dict = {}
# userinfo 拉取失败不应让绑定失败(openid 已拿到),吞掉异常只是没昵称头像
try:
with _http_client() as client:
r2 = client.get(
"https://api.weixin.qq.com/sns/userinfo",
params={"access_token": d1["access_token"], "openid": openid, "lang": "zh_CN"},
timeout=settings.WXPAY_REQUEST_TIMEOUT_SEC,
)
r2 = httpx.get(
"https://api.weixin.qq.com/sns/userinfo",
params={"access_token": d1["access_token"], "openid": openid, "lang": "zh_CN"},
timeout=settings.WXPAY_REQUEST_TIMEOUT_SEC,
)
raw = r2.json()
nickname = raw.get("nickname") or None
avatar_url = raw.get("headimgurl") or None
@@ -224,164 +217,3 @@ def code_to_userinfo(code: str) -> dict:
pass
return {"openid": openid, "nickname": nickname, "avatar_url": avatar_url, "raw": raw}
# ===== 免确认收款授权(用户授权免确认模式)=====
# 用户授权一次后,后续转账走 transfer_with_authorization 免逐笔确认直接到账。
# 复用上面同一套 V3 签名(_build_authorization)、敏感字段加密(encrypt_sensitive)、懒加载密钥。
def _transfer_report_infos() -> list[dict]:
"""现金营销(scene 1000)转账场景报备信息,与 create_transfer 同口径。"""
return [
{"info_type": "活动名称", "info_content": "比价返现"},
{"info_type": "奖励说明", "info_content": "现金提现到微信零钱"},
]
def apply_transfer_authorization(
out_authorization_no: str,
openid: str,
user_display_name: str,
notify_url: str,
*,
scene_info: dict | None = None,
) -> dict:
"""发起免确认收款授权(方式二:不转账,仅申请授权)。返回 {status_code, data}。
成功(200 + state=WAIT_USER_CONFIRM)时 data 带 package_info,供 App 拉起微信授权页。"""
body: dict = {
"out_authorization_no": out_authorization_no,
"appid": settings.WECHAT_APP_ID,
"openid": openid,
"transfer_scene_id": settings.WXPAY_TRANSFER_SCENE_ID,
"user_display_name": user_display_name,
"authorization_notify_url": notify_url,
}
if scene_info:
body["scene_info"] = scene_info
body_str = json.dumps(body, ensure_ascii=False)
headers = {
"Authorization": _build_authorization("POST", _AUTH_PATH, body_str),
"Accept": "application/json",
"Content-Type": "application/json",
}
with _http_client() as client:
resp = client.post(
f"{_API_HOST}{_AUTH_PATH}",
content=body_str,
headers=headers,
timeout=settings.WXPAY_REQUEST_TIMEOUT_SEC,
)
return {"status_code": resp.status_code, "data": resp.json()}
def query_transfer_authorization(out_authorization_no: str) -> dict:
"""按商户授权单号查授权结果。返回 {status_code, data}。
data.state: WAIT_USER_CONFIRM / TAKING_EFFECT(已生效) / CLOSED;TAKING_EFFECT 时带 authorization_id。"""
path = f"{_AUTH_PATH}/out-authorization-no/{out_authorization_no}"
headers = {
"Authorization": _build_authorization("GET", path, ""),
"Accept": "application/json",
}
with _http_client() as client:
resp = client.get(
f"{_API_HOST}{path}", headers=headers, timeout=settings.WXPAY_REQUEST_TIMEOUT_SEC
)
return {"status_code": resp.status_code, "data": resp.json()}
def close_transfer_authorization(out_authorization_no: str) -> dict:
"""解除免确认收款授权。返回 {status_code, data}(成功 data.state=CLOSED)。"""
path = f"{_AUTH_PATH}/out-authorization-no/{out_authorization_no}/close"
headers = {
"Authorization": _build_authorization("POST", path, ""),
"Accept": "application/json",
"Content-Type": "application/json",
}
with _http_client() as client:
resp = client.post(
f"{_API_HOST}{path}", headers=headers, timeout=settings.WXPAY_REQUEST_TIMEOUT_SEC
)
return {"status_code": resp.status_code, "data": resp.json()}
def pre_transfer_with_authorization(
openid: str,
amount_fen: int,
out_bill_no: str,
out_authorization_no: str,
user_display_name: str,
notify_url: str,
user_name: str | None = None,
) -> dict:
"""方式一:发起转账并同时申请免确认收款授权(用户在确认这笔收款时一并完成授权)。
返回 {status_code, data};成功 state=WAIT_USER_CONFIRM 时带 package_info(拉确认+授权页)。"""
body: dict = {
"appid": settings.WECHAT_APP_ID,
"out_bill_no": out_bill_no,
"transfer_scene_id": settings.WXPAY_TRANSFER_SCENE_ID,
"openid": openid,
"transfer_amount": amount_fen,
"transfer_remark": "金币提现",
"transfer_scene_report_infos": _transfer_report_infos(),
"authorization_info": {
"user_display_name": user_display_name,
"out_authorization_no": out_authorization_no,
"authorization_notify_url": notify_url,
},
}
if user_name and amount_fen >= 30:
body["user_name"] = encrypt_sensitive(user_name)
body_str = json.dumps(body, ensure_ascii=False)
headers = {
"Authorization": _build_authorization("POST", _PRE_TRANSFER_AUTH_PATH, body_str),
"Accept": "application/json",
"Content-Type": "application/json",
"Wechatpay-Serial": settings.WXPAY_PUBLIC_KEY_ID,
}
with _http_client() as client:
resp = client.post(
f"{_API_HOST}{_PRE_TRANSFER_AUTH_PATH}",
content=body_str,
headers=headers,
timeout=settings.WXPAY_REQUEST_TIMEOUT_SEC,
)
return {"status_code": resp.status_code, "data": resp.json()}
def transfer_with_authorization(
authorization_id: str,
amount_fen: int,
out_bill_no: str,
user_name: str | None = None,
) -> dict:
"""方式二 / 二次起:用户已授权后免确认转账(无需用户逐笔确认,直接到账)。返回 {status_code, data}。
成功 state ∈ ACCEPTED/PROCESSING/TRANSFERING/SUCCESS(无 WAIT_USER_CONFIRM、无 package_info)。"""
body: dict = {
"appid": settings.WECHAT_APP_ID,
"out_bill_no": out_bill_no,
"transfer_scene_id": settings.WXPAY_TRANSFER_SCENE_ID,
"transfer_amount": amount_fen,
"transfer_remark": "金币提现",
"transfer_scene_report_infos": _transfer_report_infos(),
"authorization_id": authorization_id,
}
if user_name and amount_fen >= 30:
body["user_name"] = encrypt_sensitive(user_name)
body_str = json.dumps(body, ensure_ascii=False)
headers = {
"Authorization": _build_authorization("POST", _TRANSFER_WITH_AUTH_PATH, body_str),
"Accept": "application/json",
"Content-Type": "application/json",
"Wechatpay-Serial": settings.WXPAY_PUBLIC_KEY_ID,
}
with _http_client() as client:
resp = client.post(
f"{_API_HOST}{_TRANSFER_WITH_AUTH_PATH}",
content=body_str,
headers=headers,
timeout=settings.WXPAY_REQUEST_TIMEOUT_SEC,
)
return {"status_code": resp.status_code, "data": resp.json()}
+2 -27
View File
@@ -17,28 +17,16 @@ 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.internal.price import router as internal_price_router
from app.api.v1.feedback import router as feedback_router
from app.api.v1.invite import router as invite_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.platform import router as platform_router
from app.api.v1.report import router as report_router
from app.api.v1.savings import router as savings_router
from app.api.v1.signin import router as signin_router
from app.api.v1.tasks import router as tasks_router
from app.api.v1.user import router as user_router
from app.api.v1.wallet import router as wallet_router
from app.api.v1.wxpay import router as wxpay_router
from app.core.config import settings
from app.core.logging import setup_logging
from app.core.withdraw_reconcile_worker import (
start_withdraw_reconcile_worker,
stop_withdraw_reconcile_worker,
)
setup_logging(debug=settings.APP_DEBUG)
logger = logging.getLogger("shagua.main")
@@ -54,12 +42,8 @@ async def lifespan(_: FastAPI) -> AsyncIterator[None]:
settings.APP_DEBUG,
settings.DATABASE_URL.split("://", 1)[0],
)
reconcile_task = start_withdraw_reconcile_worker()
try:
yield
finally:
await stop_withdraw_reconcile_worker(reconcile_task)
logger.info("shutting down")
yield
logger.info("shutting down")
app = FastAPI(
@@ -88,23 +72,14 @@ def health() -> dict[str, str]:
app.include_router(auth_router)
app.include_router(user_router)
app.include_router(feedback_router)
app.include_router(invite_router)
app.include_router(coupon_router)
app.include_router(compare_router)
app.include_router(compare_record_router)
app.include_router(compare_milestone_router)
app.include_router(meituan_router)
app.include_router(wallet_router)
app.include_router(wxpay_router)
app.include_router(signin_router)
app.include_router(tasks_router)
app.include_router(savings_router)
app.include_router(ad_router)
app.include_router(order_router)
app.include_router(report_router)
# 内部(server→server)端点:pricebot 上报价格观测,靠共享密钥头校验,不对客户端开放。
app.include_router(internal_price_router)
app.include_router(platform_router)
# 用户上传文件(头像)静态服务。生产可改由 nginx 直接 serve MEDIA_ROOT。
_media_root = Path(settings.MEDIA_ROOT)
+1 -18
View File
@@ -1,25 +1,8 @@
"""所有 ORM model 必须在这里 import 一次,Alembic / metadata 才能扫到。"""
from app.models.ad_ecpm import AdEcpmRecord # noqa: F401
from app.models.ad_feed_reward import AdFeedRewardRecord # noqa: F401
from app.models.ad_reward import AdRewardRecord # noqa: F401
from app.models.ad_watch_log import AdWatchLog # noqa: F401
from app.models.admin import AdminAuditLog, AdminUser # noqa: F401
from app.models.app_config import AppConfig # noqa: F401
from app.models.comparison import ComparisonRecord # noqa: F401
from app.models.comparison_milestone import ComparisonMilestoneClaim # noqa: F401
from app.models.coupon_state import ( # noqa: F401
CouponClaimRecord,
CouponPromptEngagement,
)
from app.models.feedback import Feedback # noqa: F401
from app.models.invite import InviteRelation # noqa: F401
from app.models.meituan_coupon import MeituanCoupon # noqa: F401
from app.models.ops_marquee_seed import OpsMarqueeSeed # noqa: F401
from app.models.ops_stat_config import OpsStatConfig # noqa: F401
from app.models.price_observation import PriceObservation # noqa: F401
from app.models.price_report import PriceReport # noqa: F401
from app.models.savings import SavingsRecord # noqa: F401
from app.models.signin import SigninBoostRecord, SigninRecord # noqa: F401
from app.models.signin import SigninRecord # noqa: F401
from app.models.task import UserTask # noqa: F401
from app.models.user import User # noqa: F401
from app.models.wallet import ( # noqa: F401
-49
View File
@@ -1,49 +0,0 @@
"""广告展示 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}>"
)
-43
View File
@@ -1,43 +0,0 @@
"""信息流广告奖励记录。
点位 2:比价等待 / 领券信息流广告。每展示满 10 秒累计一份奖励,视频完成后一次性入账。
"""
from __future__ import annotations
from datetime import datetime
from sqlalchemy import DateTime, ForeignKey, Integer, String, UniqueConstraint, func
from sqlalchemy.orm import Mapped, mapped_column
from app.db.base import Base
class AdFeedRewardRecord(Base):
__tablename__ = "ad_feed_reward_record"
__table_args__ = (
UniqueConstraint("client_event_id", name="uq_ad_feed_reward_client_event"),
)
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
client_event_id: Mapped[str] = mapped_column(String(64), nullable=False)
user_id: Mapped[int] = mapped_column(
Integer, ForeignKey("user.id"), index=True, nullable=False
)
reward_date: Mapped[str] = mapped_column(String(10), index=True, nullable=False)
duration_seconds: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
unit_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
ecpm_raw: Mapped[str] = mapped_column(String(32), nullable=False)
adn: Mapped[str | None] = mapped_column(String(32), nullable=True)
slot_id: Mapped[str | None] = mapped_column(String(64), nullable=True)
coin: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
status: Mapped[str] = mapped_column(String(16), nullable=False, default="granted")
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"<AdFeedRewardRecord user_id={self.user_id} event={self.client_event_id} "
f"{self.status} coin={self.coin}>"
)
-40
View File
@@ -1,40 +0,0 @@
"""看激励视频观看时长记录(旧版每日总时长字段)。
每条 = 客户端 onAdClose 上报的一次激励视频实际观看秒数。按 (user_id, watch_date) 聚合
SUM(watch_seconds) 得当日总观看时长。当前产品只保留每日 500 次上限,
core.rewards.DAILY_AD_WATCH_SECONDS_LIMIT=0 表示不启用时长闸;本表保留用于旧客户端兼容和排查。
这是看广告的第三条数据流,与另两条并列、互不关联:
- ad_reward(AdRewardRecord):穿山甲 S2S 回调发金币(后端,有 trans_id);
- ad_ecpm(AdEcpmRecord):客户端上报展示收益(前端,有 ecpm);
- ad_watch_log(本表):客户端上报观看时长(前端,有 watch_seconds)。
前端上报的时长不可信,不能作为唯一发奖依据;当前发奖闸口以 ad_reward 的每日次数上限为准。
"""
from __future__ import annotations
from datetime import datetime
from sqlalchemy import DateTime, ForeignKey, Integer, String, func
from sqlalchemy.orm import Mapped, mapped_column
from app.db.base import Base
class AdWatchLog(Base):
__tablename__ = "ad_watch_log"
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
user_id: Mapped[int] = mapped_column(
Integer, ForeignKey("user.id"), index=True, nullable=False
)
# 本次激励视频实际观看秒数(前端 onAdShow→onAdClose);服务端已夹 [0, MAX_SINGLE_WATCH_SECONDS]
watch_seconds: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
# 北京时间日期串 'YYYY-MM-DD',按它等值做"当日总时长"聚合(不在 SQL 里跨时区比较)
watch_date: Mapped[str] = mapped_column(String(10), index=True, nullable=False)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), server_default=func.now(), index=True, nullable=False
)
def __repr__(self) -> str: # pragma: no cover
return f"<AdWatchLog user_id={self.user_id} sec={self.watch_seconds} {self.watch_date}>"
-70
View File
@@ -1,70 +0,0 @@
"""Admin 后台:管理员账号 + 操作审计日志。
与 App 用户(user 表)完全隔离:admin 走独立 JWT secret、独立鉴权链(见 app/admin/)。
- admin_user:账号密码(bcrypt)登录,带角色(super_admin / finance / operator)。
- admin_audit_log:每个写操作落一条,记前后值,不可删,用于追溯"谁在何时改了谁的钱/状态"
admin_username / target_id 冗余存字符串,即使关联对象被删/改名也能追溯。
"""
from __future__ import annotations
from datetime import datetime
from sqlalchemy import JSON, DateTime, ForeignKey, Integer, String, func
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.orm import Mapped, mapped_column
from app.db.base import Base
# PG 用 JSONB,SQLite(本地/测试)退化为通用 JSON(同 comparison_record.raw_payload)。
_JSON = JSON().with_variant(JSONB(), "postgresql")
class AdminUser(Base):
__tablename__ = "admin_user"
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
username: Mapped[str] = mapped_column(String(64), unique=True, index=True, nullable=False)
password_hash: Mapped[str] = mapped_column(String(255), nullable=False)
# super_admin(全权+管账号)/ finance(钱:提现+金币)/ operator(用户+反馈+大盘)
role: Mapped[str] = mapped_column(String(20), nullable=False, default="operator")
# active / disabled
status: Mapped[str] = mapped_column(String(20), nullable=False, default="active")
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), server_default=func.now(), nullable=False
)
last_login_at: Mapped[datetime | None] = mapped_column(
DateTime(timezone=True), nullable=True
)
def __repr__(self) -> str: # pragma: no cover
return f"<AdminUser id={self.id} username={self.username} role={self.role}>"
class AdminAuditLog(Base):
__tablename__ = "admin_audit_log"
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
admin_id: Mapped[int] = mapped_column(
Integer, ForeignKey("admin_user.id"), index=True, nullable=False
)
# 冗余存操作者用户名:admin 改名/禁用后仍可追溯是谁干的
admin_username: Mapped[str] = mapped_column(String(64), nullable=False)
# 操作类型,如 user.coins.grant / user.status.set / withdraw.refresh / feedback.handle
action: Mapped[str] = mapped_column(String(64), index=True, nullable=False)
# 被操作对象类型 + id(id 用字符串以兼容 out_bill_no 等非整型主键)
target_type: Mapped[str] = mapped_column(String(32), nullable=False)
target_id: Mapped[str | None] = mapped_column(String(64), nullable=True)
# 上下文 + 前后值,如 {"amount": 1000, "reason": "...", "before": {...}, "after": {...}}
detail: Mapped[dict | None] = mapped_column(_JSON, nullable=True)
ip: Mapped[str | None] = mapped_column(String(64), nullable=True)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), server_default=func.now(), index=True, nullable=False
)
def __repr__(self) -> str: # pragma: no cover
return (
f"<AdminAuditLog id={self.id} admin={self.admin_username} "
f"action={self.action} target={self.target_type}:{self.target_id}>"
)
-32
View File
@@ -1,32 +0,0 @@
"""运营可配置项(把 rewards.py 等硬编码规则挪到 DB,运营后台可改)。
key 是配置标识(见 app.core.config_schema.CONFIG_DEFS),value 用 JSON 存任意值(int/list/dict)。
表里没有的 key,业务读配置时 fallback 到 CONFIG_DEFS 默认值(= 原 rewards 常量),
所以**空表 = 现有行为完全不变**。admin 改某项 → 写一行 → 业务下次读到新值(跨进程一致)。
"""
from __future__ import annotations
from datetime import datetime
from typing import Any
from sqlalchemy import JSON, DateTime, Integer, String, func
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.orm import Mapped, mapped_column
from app.db.base import Base
_JSON = JSON().with_variant(JSONB(), "postgresql")
class AppConfig(Base):
__tablename__ = "app_config"
key: Mapped[str] = mapped_column(String(64), primary_key=True)
value: Mapped[Any] = mapped_column(_JSON, nullable=False)
updated_by_admin_id: Mapped[int | None] = mapped_column(Integer, nullable=True)
updated_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), server_default=func.now(), onupdate=func.now(), nullable=False
)
def __repr__(self) -> str: # pragma: no cover
return f"<AppConfig key={self.key}>"

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