Merge branch 'main' into feat/admin-api-db-docs

This commit is contained in:
OuYingJun1024
2026-06-06 10:33:33 +08:00
39 changed files with 1560 additions and 113 deletions
@@ -0,0 +1,65 @@
"""新增 price_report 上报更低价表
Revision ID: 9258bddde4ea
Revises: ad60a1b2c3d4
Create Date: 2026-06-05 10:15:51.508598
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision: str = '9258bddde4ea'
down_revision: Union[str, Sequence[str], None] = 'ad60a1b2c3d4'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
# ### commands auto generated by Alembic - please adjust! ###
op.create_table('price_report',
sa.Column('id', sa.Integer(), autoincrement=True, nullable=False),
sa.Column('user_id', sa.Integer(), nullable=False),
sa.Column('comparison_record_id', sa.Integer(), nullable=True),
sa.Column('store_name', sa.String(length=128), nullable=True),
sa.Column('dish_summary', sa.String(length=256), nullable=True),
sa.Column('original_platform_id', sa.String(length=32), nullable=True),
sa.Column('original_platform_name', sa.String(length=32), nullable=True),
sa.Column('original_price_cents', sa.Integer(), nullable=True),
sa.Column('reported_platform_id', sa.String(length=32), nullable=False),
sa.Column('reported_platform_name', sa.String(length=32), nullable=False),
sa.Column('reported_price_cents', sa.Integer(), nullable=False),
sa.Column('images', sa.JSON(), nullable=False),
sa.Column('status', sa.String(length=16), nullable=False),
sa.Column('reject_reason', sa.String(length=256), nullable=True),
sa.Column('reward_coins', sa.Integer(), nullable=True),
sa.Column('reviewed_at', sa.DateTime(timezone=True), nullable=True),
sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('(CURRENT_TIMESTAMP)'), nullable=False),
sa.ForeignKeyConstraint(['comparison_record_id'], ['comparison_record.id'], ),
sa.ForeignKeyConstraint(['user_id'], ['user.id'], ),
sa.PrimaryKeyConstraint('id')
)
with op.batch_alter_table('price_report', schema=None) as batch_op:
batch_op.create_index(batch_op.f('ix_price_report_comparison_record_id'), ['comparison_record_id'], unique=False)
batch_op.create_index(batch_op.f('ix_price_report_created_at'), ['created_at'], unique=False)
batch_op.create_index(batch_op.f('ix_price_report_status'), ['status'], unique=False)
batch_op.create_index(batch_op.f('ix_price_report_user_id'), ['user_id'], unique=False)
# 注:autogenerate 另检测到「删除 order_record 表」——那是已降级为审计的历史表
# (model 已移除、库表保留),与本次无关,手动剔除,避免误删他人审计数据。
# ### end Alembic commands ###
def downgrade() -> None:
# ### commands auto generated by Alembic - please adjust! ###
# (对称:upgrade 未删 order_record,downgrade 也不重建它)
with op.batch_alter_table('price_report', schema=None) as batch_op:
batch_op.drop_index(batch_op.f('ix_price_report_user_id'))
batch_op.drop_index(batch_op.f('ix_price_report_status'))
batch_op.drop_index(batch_op.f('ix_price_report_created_at'))
batch_op.drop_index(batch_op.f('ix_price_report_comparison_record_id'))
op.drop_table('price_report')
# ### end Alembic commands ###
+33
View File
@@ -0,0 +1,33 @@
"""app_config table (运营可配置项:rewards 常量等后台化)
Revision ID: cfg1a2b3c4d5
Revises: ad60a1b2c3d4
Create Date: 2026-06-04 12:00:00.000000
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision: str = 'cfg1a2b3c4d5'
down_revision: Union[str, Sequence[str], None] = 'ad60a1b2c3d4'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
op.create_table(
'app_config',
sa.Column('key', sa.String(length=64), nullable=False),
sa.Column('value', sa.JSON().with_variant(sa.dialects.postgresql.JSONB(), 'postgresql'), nullable=False),
sa.Column('updated_by_admin_id', sa.Integer(), nullable=True),
sa.Column('updated_at', sa.DateTime(timezone=True), server_default=sa.text('(CURRENT_TIMESTAMP)'), nullable=False),
sa.PrimaryKeyConstraint('key'),
)
def downgrade() -> None:
op.drop_table('app_config')
@@ -0,0 +1,32 @@
"""comparison_record 加 best_deeplink(再次比价直达商家深链)
Revision ID: b7e2c1a9f4d3
Revises: 9258bddde4ea
Create Date: 2026-06-05
「再次比价」要直达上次最低价平台的商家/商品页。深链(link)在比价时客户端已从剪贴板采到
(collectedLinks[best_index]),本列把它落库;再次比价时写剪贴板 + launch 该平台 App 直达。
旧记录无此列值(None) → 前端降级为打开对应 App 首页。
手写迁移(只加一列):autogenerate 会误报"删 order_record"(降级为审计的历史表,model 已移除、
库表保留),手写规避该噪音。
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
revision: str = "b7e2c1a9f4d3"
down_revision: Union[str, Sequence[str], None] = "9258bddde4ea"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
with op.batch_alter_table("comparison_record", schema=None) as batch_op:
batch_op.add_column(sa.Column("best_deeplink", sa.String(length=1024), nullable=True))
def downgrade() -> None:
with op.batch_alter_table("comparison_record", schema=None) as batch_op:
batch_op.drop_column("best_deeplink")
@@ -0,0 +1,26 @@
"""merge app_config and price_report heads
Revision ID: ebb6af5c0b56
Revises: cfg1a2b3c4d5, b7e2c1a9f4d3
Create Date: 2026-06-06 02:29:13.475265
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision: str = 'ebb6af5c0b56'
down_revision: Union[str, Sequence[str], None] = ('cfg1a2b3c4d5', 'b7e2c1a9f4d3')
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
pass
def downgrade() -> None:
pass
@@ -0,0 +1,62 @@
"""withdraw_order.user_name 列 + ad_watch_log 表
提现审核功能:WithdrawOrder 加 user_name(审核异步打款时传给微信的实名,达额转账要求)。
看广告时长防刷:新增 ad_watch_log 表(前端 onAdClose 上报观看秒数,按 (user_id, watch_date)
聚合做"每天 50 分钟"硬上限)。
Revision ID: withdraw_review_ad_watch
Revises: ebb6af5c0b56
Create Date: 2026-06-05 12:00:00.000000
注:本迁移原 down_revision=ad60a1b2c3d4(分支创建时的 head);rebase 合 main 后 main 侧已新增
app_config / price_report 等迁移并 merge 出 head=ebb6af5c0b56,故改挂到其上,保持单 head(本迁移
与那些表互不相干,叠加顺序无影响)。
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision: str = 'withdraw_review_ad_watch'
down_revision: Union[str, Sequence[str], None] = 'ebb6af5c0b56'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
# ① 提现单加实名列:审核通过后异步打款时传给微信(达额转账要求),发起提现时存下
with op.batch_alter_table('withdraw_order', schema=None) as batch_op:
batch_op.add_column(sa.Column('user_name', sa.String(length=64), nullable=True))
# ② 看广告观看时长表:前端上报观看秒数,(user_id, watch_date) 聚合做每日 50 分钟防刷主闸
op.create_table(
'ad_watch_log',
sa.Column('id', sa.Integer(), autoincrement=True, nullable=False),
sa.Column('user_id', sa.Integer(), nullable=False),
sa.Column('watch_seconds', sa.Integer(), server_default='0', nullable=False),
sa.Column('watch_date', sa.String(length=10), nullable=False),
sa.Column(
'created_at', sa.DateTime(timezone=True),
server_default=sa.text('(CURRENT_TIMESTAMP)'), nullable=False,
),
sa.ForeignKeyConstraint(['user_id'], ['user.id'], ),
sa.PrimaryKeyConstraint('id'),
)
with op.batch_alter_table('ad_watch_log', schema=None) as batch_op:
batch_op.create_index(batch_op.f('ix_ad_watch_log_user_id'), ['user_id'], unique=False)
batch_op.create_index(batch_op.f('ix_ad_watch_log_watch_date'), ['watch_date'], unique=False)
batch_op.create_index(batch_op.f('ix_ad_watch_log_created_at'), ['created_at'], unique=False)
def downgrade() -> None:
with op.batch_alter_table('ad_watch_log', schema=None) as batch_op:
batch_op.drop_index(batch_op.f('ix_ad_watch_log_created_at'))
batch_op.drop_index(batch_op.f('ix_ad_watch_log_watch_date'))
batch_op.drop_index(batch_op.f('ix_ad_watch_log_user_id'))
op.drop_table('ad_watch_log')
with op.batch_alter_table('withdraw_order', schema=None) as batch_op:
batch_op.drop_column('user_name')
+2
View File
@@ -16,6 +16,7 @@ from fastapi.middleware.cors import CORSMiddleware
from app.admin.routers.admins import router as admins_router
from app.admin.routers.audit import router as audit_router
from app.admin.routers.auth import router as auth_router
from app.admin.routers.config import router as config_router
from app.admin.routers.dashboard import router as dashboard_router
from app.admin.routers.feedback import router as feedback_router
from app.admin.routers.users import router as users_router
@@ -78,3 +79,4 @@ 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)
+81
View File
@@ -0,0 +1,81 @@
"""admin 运营配置:列出所有可配项 + 改某项(带审计 + 改值校验)。
配置项定义见 app.core.config_schema.CONFIG_DEFS;业务读配置 fallback 默认(见 app_config repo)。
权限:operator / finance 都可改(运营调奖励、财务调额度),super 恒可。
"""
from __future__ import annotations
from typing import Annotated, Any
from fastapi import APIRouter, Depends, HTTPException, Request
from app.admin.audit import write_audit
from app.admin.deps import AdminDb, get_client_ip, get_current_admin, require_role
from app.admin.schemas.config import ConfigItemOut, ConfigUpdateRequest
from app.core.config_schema import CONFIG_DEFS
from app.models.admin import AdminUser
from app.repositories import app_config
router = APIRouter(
prefix="/admin/api/config",
tags=["admin-config"],
dependencies=[Depends(get_current_admin)],
)
def _validate(key: str, value: Any) -> None:
"""按配置项 type 校验新值,不合法抛 ValueError(router 转 400)。"""
t = CONFIG_DEFS[key]["type"]
if t == "int":
if not isinstance(value, int) or isinstance(value, bool) or value < 0:
raise ValueError("需为非负整数")
elif t == "int_list":
if not isinstance(value, list) or not value:
raise ValueError("需为非空整数列表")
if not all(isinstance(x, int) and not isinstance(x, bool) and x >= 0 for x in value):
raise ValueError("列表元素需为非负整数")
if key == "signin_rewards" and len(value) != 7:
raise ValueError("签到档位必须正好 7 个(对应 7 天循环)")
elif t == "dict_str_int":
if not isinstance(value, dict) or not all(
isinstance(k, str) and isinstance(v, int) and not isinstance(v, bool)
for k, v in value.items()
):
raise ValueError("需为 {字符串: 整数} 映射")
def _item(db, key: str) -> ConfigItemOut:
for item in app_config.list_all(db):
if item["key"] == key:
return ConfigItemOut(**item)
raise HTTPException(status_code=404, detail="未知配置项")
@router.get("", response_model=list[ConfigItemOut], summary="所有可配项 + 当前值")
def list_config(db: AdminDb) -> list[ConfigItemOut]:
return [ConfigItemOut(**item) for item in app_config.list_all(db)]
@router.patch("/{key}", response_model=ConfigItemOut, summary="改某项配置(带审计)")
def update_config(
key: str,
body: ConfigUpdateRequest,
request: Request,
admin: Annotated[AdminUser, Depends(require_role("operator", "finance"))],
db: AdminDb,
) -> ConfigItemOut:
if key not in CONFIG_DEFS:
raise HTTPException(status_code=404, detail="未知配置项")
try:
_validate(key, body.value)
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e)) from e
before = app_config.get_value(db, key)
app_config.set_value(db, key, body.value, admin_id=admin.id, commit=False)
write_audit(
db, admin, action="config.set", target_type="config", target_id=key,
detail={"before": before, "after": body.value}, ip=get_client_ip(request), commit=False,
)
db.commit()
return _item(db, key)
+56 -1
View File
@@ -14,7 +14,7 @@ from app.admin.audit import write_audit
from app.admin.deps import AdminDb, get_client_ip, get_current_admin, require_role
from app.admin.repositories import queries
from app.admin.schemas.common import CursorPage
from app.admin.schemas.wallet import ReconcileResult, WithdrawOrderOut
from app.admin.schemas.wallet import ReconcileResult, WithdrawOrderOut, WithdrawRejectRequest
from app.integrations import wxpay
from app.models.admin import AdminUser
from app.repositories import wallet as wallet_repo
@@ -83,3 +83,58 @@ def refresh_withdraw(
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)
+22
View File
@@ -0,0 +1,22 @@
"""admin 运营配置 schemas。"""
from __future__ import annotations
from typing import Any
from pydantic import BaseModel
class ConfigItemOut(BaseModel):
key: str
label: str
group: str
type: str # int / int_list / dict_str_int
help: str | None = None
default: Any
value: Any
overridden: bool # 是否被运营覆盖过(false=用默认)
updated_at: str | None = None
class ConfigUpdateRequest(BaseModel):
value: Any
+8 -1
View File
@@ -3,7 +3,7 @@ from __future__ import annotations
from datetime import datetime
from pydantic import BaseModel, ConfigDict
from pydantic import BaseModel, ConfigDict, Field
class CoinTxnOut(BaseModel):
@@ -39,6 +39,7 @@ class WithdrawOrderOut(BaseModel):
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
@@ -50,3 +51,9 @@ class WithdrawOrderOut(BaseModel):
class ReconcileResult(BaseModel):
checked: int
resolved: int
class WithdrawRejectRequest(BaseModel):
reason: str = Field(
..., min_length=1, max_length=200, description="拒绝理由(写入 fail_reason,用户可见)"
)
+37 -3
View File
@@ -21,12 +21,15 @@ from app.integrations import pangle
from app.core.ratelimit import rate_limit
from app.repositories import ad_ecpm as crud_ecpm
from app.repositories import ad_reward as crud_ad
from app.repositories import ad_watch as crud_watch
from app.schemas.ad import (
AdRewardStatusOut,
EcpmReportIn,
EcpmReportOut,
PangleCallbackOut,
TestGrantOut,
WatchReportIn,
WatchReportOut,
)
logger = logging.getLogger("shagua.ad")
@@ -71,7 +74,7 @@ def pangle_callback(request: Request, db: DbSession) -> PangleCallbackOut:
return PangleCallbackOut(is_verify=False, reason=REASON_BAD_PARAMS)
user_id = int(raw_user_id)
coin = rewards.resolve_ad_reward_coin(params.get("reward_amount"))
coin = rewards.resolve_ad_reward_coin(db, params.get("reward_amount"))
raw = "&".join(f"{k}={v}" for k, v in sorted(params.items()) if k != "sign")
try:
rec = crud_ad.grant_ad_reward(
@@ -91,7 +94,8 @@ 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 = crud_ad.today_status(db, user.id)
(used, limit, coin_per, round_count, cooldown_until,
watched_sec, watch_limit) = crud_ad.today_status(db, user.id)
return AdRewardStatusOut(
used_today=used,
daily_limit=limit,
@@ -99,6 +103,35 @@ 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="上报激励视频观看时长(每日 50 分钟防刷计时)",
dependencies=[Depends(rate_limit(120, 60, "ad-watch-report"))],
)
def watch_report(payload: WatchReportIn, user: CurrentUser, db: DbSession) -> WatchReportOut:
"""客户端在激励视频关闭(onAdClose)后上报本次实际观看秒数,服务端累计到当日总时长。
Bearer 鉴权,user_id 取自 JWT(不信 body)。seconds 服务端夹 [0, MAX_SINGLE_WATCH_SECONDS]。
当日累计达 DAILY_AD_WATCH_SECONDS_LIMIT(50 分钟)后:① 本接口 / reward-status 返回 remaining=0,
客户端不再展示广告;② 后端发奖(S2S 回调)也会因时长闸记 capped 不发金币(双闸,前端绕不过)。
"""
total = crud_watch.add_watch_seconds(db, user.id, payload.seconds)
limit = rewards.DAILY_AD_WATCH_SECONDS_LIMIT
logger.info(
"ad watch report user_id=%d +%ds total=%d/%d",
user.id, payload.seconds, total, limit,
)
return WatchReportOut(
watched_seconds_today=total,
watch_seconds_limit=limit,
watch_seconds_remaining=max(0, limit - total),
)
@@ -151,7 +184,8 @@ def test_grant(user: CurrentUser, db: DbSession) -> TestGrantOut:
except crud_ad.UnknownUserError as e:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="user not found") from e
used, limit, coin_per, round_count, cooldown_until = crud_ad.today_status(db, user.id)
(used, limit, coin_per, round_count, cooldown_until,
_watched, _watch_limit) = crud_ad.today_status(db, user.id)
logger.info("ad TEST grant user_id=%d status=%s coin=%d", user.id, rec.status, rec.coin)
return TestGrantOut(
granted=(rec.status == "granted"),
+145
View File
@@ -0,0 +1,145 @@
"""上报更低价 endpoint。
路由前缀 `/api/v1/report`,需 Bearer 鉴权(上报绑登录用户)。
POST / 提交上报(multipart:comparison_record_id / reported_platform_id /
reported_price(元) + images 1~4 张)
GET /records 上报记录列表(?status= pending/approved/rejected 可选筛选)
要点:
- 截图复用 [app.core.media] 落盘到 /media/price_report/。
- 原最低价由 comparison_record_id 反查比价记录 best_*(不信任客户端传的快照)。
- 校验(D):上报价必须 < 原最低价,否则 400。
- 提交一律 status=pending;通过发奖励是人工审核后台动作,不在此。
"""
from __future__ import annotations
import logging
from fastapi import APIRouter, File, Form, HTTPException, UploadFile
from app.api.deps import CurrentUser, DbSession
from app.core import media
from app.models.comparison import ComparisonRecord
from app.repositories import report as report_repo
from app.schemas.report import (
ReportRecordCounts,
ReportRecordOut,
ReportRecordsOut,
ReportSubmitOut,
)
logger = logging.getLogger("shagua.report")
router = APIRouter(prefix="/api/v1/report", tags=["report"])
_MAX_IMAGES = 4
_VALID_STATUS = ("pending", "approved", "rejected")
# 上报平台(canonical id → 展示名),与原型 fb-platform-chip data-pf 一致
_PLATFORM_NAMES = {
"meituan-waimai": "美团外卖",
"jd-waimai": "京东外卖",
"taobao-shanguang": "淘宝闪购",
}
def _dish_summary(items: list | None) -> str | None:
"""把比价记录 items[{name,qty,...}] 拼成菜品摘要文案。"""
if not items:
return None
names = [str(it.get("name", "")).strip() for it in items if isinstance(it, dict)]
names = [n for n in names if n]
return "".join(names) if names else None
@router.post("", response_model=ReportSubmitOut, summary="提交上报更低价")
async def submit_report(
user: CurrentUser,
db: DbSession,
comparison_record_id: int = Form(...),
reported_platform_id: str = Form(...),
reported_price: str = Form(..., description="用户填的更低价(元)"),
images: list[UploadFile] = File(default=[]),
) -> ReportSubmitOut:
# 平台
reported_platform_id = reported_platform_id.strip()
platform_name = _PLATFORM_NAMES.get(reported_platform_id)
if platform_name is None:
raise HTTPException(status_code=400, detail="不支持的上报平台")
# 价格(元 → 分)
try:
price_yuan = float(reported_price)
except (TypeError, ValueError):
raise HTTPException(status_code=400, detail="价格格式不正确") from None
if price_yuan <= 0:
raise HTTPException(status_code=400, detail="价格必须大于 0")
reported_price_cents = round(price_yuan * 100)
# 反查比价记录(必须属于本人)→ 取原最低价快照
rec = db.get(ComparisonRecord, comparison_record_id)
if rec is None or rec.user_id != user.id:
raise HTTPException(status_code=404, detail="比价记录不存在")
original_price_cents = rec.best_price_cents
# D:上报价必须低于原最低价
if original_price_cents is not None and reported_price_cents >= original_price_cents:
raise HTTPException(
status_code=400,
detail=f"上报价需低于原最低价 ¥{original_price_cents / 100:.2f}",
)
# 截图(至少 1 张,最多 4 张)
files = [f for f in (images or []) if f is not None and f.filename]
if not files:
raise HTTPException(status_code=400, detail="请至少上传一张截图证明")
if len(files) > _MAX_IMAGES:
raise HTTPException(status_code=400, detail=f"最多上传 {_MAX_IMAGES} 张图片")
urls: list[str] = []
for f in files:
data = await f.read()
try:
urls.append(media.save_report_image(user.id, data))
except media.MediaError as e:
raise HTTPException(status_code=400, detail=str(e)) from e
rep = report_repo.create_report(
db,
user_id=user.id,
comparison_record_id=comparison_record_id,
store_name=rec.store_name,
dish_summary=_dish_summary(rec.items),
original_platform_id=rec.best_platform_id,
original_platform_name=rec.best_platform_name,
original_price_cents=original_price_cents,
reported_platform_id=reported_platform_id,
reported_platform_name=platform_name,
reported_price_cents=reported_price_cents,
images=urls,
)
logger.info("price_report id=%d user_id=%d images=%d", rep.id, user.id, len(urls))
return ReportSubmitOut.model_validate(rep)
@router.get("/records", response_model=ReportRecordsOut, summary="上报记录列表")
def list_report_records(
user: CurrentUser,
db: DbSession,
status: str | None = None,
) -> ReportRecordsOut:
status = (status or "").strip() or None
if status and status not in _VALID_STATUS:
raise HTTPException(status_code=400, detail="无效的状态筛选")
rows = report_repo.list_reports(db, user.id, status)
# counts 始终基于全量(不受 status 筛选影响),供前端 chip 计数
all_rows = rows if status is None else report_repo.list_reports(db, user.id, None)
counts = ReportRecordCounts(
all=len(all_rows),
pending=sum(1 for r in all_rows if r.status == "pending"),
approved=sum(1 for r in all_rows if r.status == "approved"),
rejected=sum(1 for r in all_rows if r.status == "rejected"),
)
return ReportRecordsOut(
records=[ReportRecordOut.model_validate(r) for r in rows],
counts=counts,
)
+13 -5
View File
@@ -170,10 +170,11 @@ def withdraw_info(user: CurrentUser, db: DbSession) -> WithdrawInfoOut:
@router.post(
"/withdraw",
response_model=WithdrawResultOut,
summary="发起提现到微信零钱",
summary="发起提现(扣款建单,待人工审核;审核通过后才打款)",
dependencies=[Depends(rate_limit(20, 60, "withdraw"))], # #6 同 IP 每分钟≤20 次
)
def withdraw(req: WithdrawRequest, user: CurrentUser, db: DbSession) -> WithdrawResultOut:
# 提现发起本身不调微信(打款在审核通过后),但仍要求微信支付已配置——否则审核通过也打不了款,提前拦
if not settings.wxpay_configured:
raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="wechat pay not configured")
try:
@@ -189,12 +190,14 @@ def withdraw(req: WithdrawRequest, user: CurrentUser, db: DbSession) -> Withdraw
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="wechat not bound") from e
except crud_wallet.InsufficientCashError as e:
raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="insufficient cash balance") from e
except crud_wallet.WithdrawTransferError as e:
# 转账调用失败,余额已退回
raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail=f"wechat transfer failed: {e}") from e
acc = crud_wallet.get_or_create_account(db, user.id)
logger.info("withdraw user_id=%d cents=%d bill=%s state=%s", user.id, req.amount_cents, order.out_bill_no, order.wechat_state)
logger.info(
"withdraw submitted user_id=%d cents=%d bill=%s status=%s(待审核)",
user.id, req.amount_cents, order.out_bill_no, order.status,
)
# 此刻 status=reviewing,尚未打款 → 无 package_info;App 据 status 提示"已提交,等待审核"。
# 审核通过后客户端轮询 /withdraw/status 拿到 package_info(若需确认)再拉起微信确认页。
return WithdrawResultOut(
out_bill_no=order.out_bill_no,
status=order.status,
@@ -223,6 +226,11 @@ def withdraw_status(
status=order.status,
wechat_state=order.wechat_state,
amount_cents=order.amount_cents,
fail_reason=order.fail_reason,
# 审核通过后 status=pending 且可能带 package_info(WAIT_USER_CONFIRM):供 App 拉起微信确认页
package_info=order.package_info,
mch_id=settings.WXPAY_MCH_ID if order.package_info else None,
app_id=settings.WECHAT_APP_ID if order.package_info else None,
)
+63
View File
@@ -0,0 +1,63 @@
"""运营可配置项注册表:默认值(= 原 rewards 常量)+ 元信息(label/group/type)。
配置项的 single source of truth:
- 业务读配置时 fallback 到这里的 default( DB = 原行为不变)
- admin 配置页用它展示"有哪些可配项 + 当前值 + 默认值 + 说明"
新增可配项 = 这里加一条 + 业务处改用 app_config.get_value(db, key)
"""
from __future__ import annotations
from typing import Any
from app.core import rewards as r
# type 约定(给前端渲染编辑控件用):int / int_list / dict_str_int
CONFIG_DEFS: dict[str, dict[str, Any]] = {
"signin_rewards": {
"default": list(r.SIGNIN_REWARDS), "label": "签到 7 天金币档位",
"group": "签到", "type": "int_list",
"help": "第 1~7 天每天签到发的金币;断签重置回第 1 天。长度需为 7。",
},
"min_exchange_coin": {
"default": r.MIN_EXCHANGE_COIN, "label": "最低兑换金币",
"group": "钱包", "type": "int", "help": "金币兑现金的单次最低金币(10000 金币=1 元)。",
},
"withdraw_min_cents": {
"default": r.WITHDRAW_MIN_CENTS, "label": "提现最低额(分)",
"group": "钱包", "type": "int", "help": "单次提现最低;微信商家转账最低 10 分(0.1 元)。",
},
"withdraw_max_cents": {
"default": r.WITHDRAW_MAX_CENTS, "label": "提现最高额(分)",
"group": "钱包", "type": "int",
},
"task_rewards": {
"default": dict(r.TASK_REWARDS), "label": "一次性任务奖励",
"group": "任务", "type": "dict_str_int", "help": "task_key → 金币。",
},
"record_milestones": {
"default": list(r.RECORD_MILESTONES), "label": "比价里程碑金币档位",
"group": "里程碑", "type": "int_list",
"help": "累计成功比价第 1~N 档解锁发的金币。",
},
"ad_reward_coin": {
"default": r.AD_REWARD_COIN, "label": "看广告单次金币",
"group": "看广告", "type": "int",
"help": "看完一个激励视频发的金币;须与穿山甲后台代码位'奖励数量'保持一致。",
},
"ad_daily_limit": {
"default": r.DAILY_AD_REWARD_LIMIT, "label": "看广告每日上限(次)",
"group": "看广告", "type": "int",
},
"ad_max_coin": {
"default": r.MAX_AD_REWARD_COIN, "label": "看广告单次金币上限",
"group": "看广告", "type": "int", "help": "夹紧穿山甲回调里异常的 reward_amount,防刷爆。",
},
"ad_round_count": {
"default": r.VIDEO_ROUND_REQUIRED_COUNT, "label": "每轮看广告次数",
"group": "看广告", "type": "int", "help": "看完一轮触发冷却。",
},
"ad_cooldown_sec": {
"default": r.VIDEO_ROUND_COOLDOWN_SECONDS, "label": "每轮冷却(秒)",
"group": "看广告", "type": "int",
},
}
+5
View File
@@ -62,6 +62,11 @@ def save_feedback_image(user_id: int, data: bytes) -> str:
return _save_image("feedback", user_id, data)
def save_report_image(user_id: int, data: bytes) -> str:
"""保存上报更低价截图,返回相对 URL(`/media/price_report/<file>`)。"""
return _save_image("price_report", user_id, data)
def delete_avatar(url: str | None) -> None:
"""删除本服务托管的旧头像文件;外部 URL(如微信头像)或空值不处理。"""
prefix = f"{settings.MEDIA_URL_PREFIX}/avatars/"
+73 -8
View File
@@ -83,9 +83,20 @@ def record_milestone_reward(milestone: int) -> int:
AD_REWARD_COIN: int = 666
# 单次发奖金币上限:夹紧穿山甲回调里异常的 reward_amount(如后台多打一个 0),防刷爆余额。
MAX_AD_REWARD_COIN: int = 1000
# 每用户每日发奖次数上限,防刷 + 控成本
# ⚠️ 测试阶段值=20:正式上线前按产品策略 + 广告 eCPM 实测收益重新核定,改这里即可。
DAILY_AD_REWARD_LIMIT: int = 20
# 每用户每日发奖次数上限——现作为"看广告时长上限"的**次数兜底**(防前端少报观看时长绕过时长闸)
# 主闸是 DAILY_AD_WATCH_SECONDS_LIMIT(50 分钟);次数兜底要 ≥ 时长闸内的"合理最大次数",
# 否则会比主闸先掐、误伤看大量短广告的正常用户:50 分钟里若广告各 ~15s,理论可看 ~200 次,
# 故取 200(≈时长闸的次数等价),正常用户先撞 50 分钟时长闸,仅前端时长全报 0 等异常时本闸才兜住。
# 不要为"更严"而下调到 120 之类(会在 ~30 分钟就掐短广告用户);要更严应在客户端如实上报时长。
DAILY_AD_REWARD_LIMIT: int = 200
# ===== 看激励视频每日总时长上限(防刷主闸)=====
# 用户每天最多看 50 分钟激励视频,超了不发奖 / 不给看。时长由前端 onAdClose 上报真实观看
# 秒数累计(POST /api/v1/ad/watch-report → ad_watch_log 表 SUM)。前端不可信,故配合上面
# DAILY_AD_REWARD_LIMIT 次数兜底一起防刷。要调上限直接改这里。
DAILY_AD_WATCH_SECONDS_LIMIT: int = 50 * 60 # 3000 秒 = 50 分钟
# 单次上报观看时长的合理上限(夹紧异常值):激励视频一般 15~60 秒,超 120 秒按 120 记,防刷大值。
MAX_SINGLE_WATCH_SECONDS: int = 120
# 一轮看 N 次激励视频,看完一轮强制 10 分钟冷却(reward-status 派生 cooldown_until 给客户端,
# 客户端任务行 CTA 在冷却期间显示"MM:SS"且不可点)。冷却是 UX 约束,后端发奖只看 daily_limit,
# 冷却期间穿山甲回调仍照常发奖,不另设拒发分支。
@@ -93,17 +104,71 @@ VIDEO_ROUND_REQUIRED_COUNT: int = 3
VIDEO_ROUND_COOLDOWN_SECONDS: int = 600
def resolve_ad_reward_coin(reward_amount: str | int | None) -> int:
def resolve_ad_reward_coin(db, reward_amount: str | int | None) -> int: # noqa: ANN001
"""把穿山甲回调的 reward_amount(后台代码位配的"奖励数量")解析成本次发放金币。
缺失 / 非数字 / 0 回退 AD_REWARD_COIN;超过 MAX_AD_REWARD_COIN 夹紧
缺失 / 非数字 / 0 回退配置单次金币;超过配置单次上限 夹紧(均从 app_config )
:reward_amount 不参与穿山甲 sign(只签 trans_id),但伪造回调需先伪造合法 sign
( SecurityKey),故能过验签的 reward_amount 即视为可信,这里只防后台配错
"""
default_coin = get_ad_reward_coin(db)
try:
coin = int(reward_amount) # type: ignore[arg-type]
except (TypeError, ValueError):
return AD_REWARD_COIN
return default_coin
if coin <= 0:
return AD_REWARD_COIN
return min(coin, MAX_AD_REWARD_COIN)
return default_coin
return min(coin, get_ad_max_coin(db))
# ===== 配置读取(运营后台可覆盖;app_config 表空时返回上面的常量默认 = 现状不变)=====
# 业务处用这些 getter(传 db)取值,而非直接用常量,这样 admin 改 app_config 即生效。
# 函数内延迟 import,避免 rewards ← app_config ← config_schema ← rewards 的模块级循环。
def _cfg(db, key: str): # noqa: ANN001
from app.repositories import app_config
return app_config.get_value(db, key)
def get_signin_rewards(db) -> list[int]: # noqa: ANN001
return list(_cfg(db, "signin_rewards"))
def get_min_exchange_coin(db) -> int: # noqa: ANN001
return int(_cfg(db, "min_exchange_coin"))
def get_withdraw_min_cents(db) -> int: # noqa: ANN001
return int(_cfg(db, "withdraw_min_cents"))
def get_withdraw_max_cents(db) -> int: # noqa: ANN001
return int(_cfg(db, "withdraw_max_cents"))
def get_task_rewards(db) -> dict[str, int]: # noqa: ANN001
return dict(_cfg(db, "task_rewards"))
def get_record_milestones(db) -> list[int]: # noqa: ANN001
return list(_cfg(db, "record_milestones"))
def get_ad_reward_coin(db) -> int: # noqa: ANN001
return int(_cfg(db, "ad_reward_coin"))
def get_ad_daily_limit(db) -> int: # noqa: ANN001
return int(_cfg(db, "ad_daily_limit"))
def get_ad_max_coin(db) -> int: # noqa: ANN001
return int(_cfg(db, "ad_max_coin"))
def get_ad_round_count(db) -> int: # noqa: ANN001
return int(_cfg(db, "ad_round_count"))
def get_ad_cooldown_sec(db) -> int: # noqa: ANN001
return int(_cfg(db, "ad_cooldown_sec"))
+2
View File
@@ -23,6 +23,7 @@ from app.api.v1.coupon import router as coupon_router
from app.api.v1.feedback import router as feedback_router
from app.api.v1.meituan import router as meituan_router
from app.api.v1.order import router as order_router
from app.api.v1.report import router as report_router
from app.api.v1.savings import router as savings_router
from app.api.v1.signin import router as signin_router
from app.api.v1.tasks import router as tasks_router
@@ -86,6 +87,7 @@ app.include_router(tasks_router)
app.include_router(savings_router)
app.include_router(ad_router)
app.include_router(order_router)
app.include_router(report_router)
# 用户上传文件(头像)静态服务。生产可改由 nginx 直接 serve MEDIA_ROOT。
_media_root = Path(settings.MEDIA_ROOT)
+3
View File
@@ -1,10 +1,13 @@
"""所有 ORM model 必须在这里 import 一次,Alembic / metadata 才能扫到。"""
from app.models.ad_ecpm import AdEcpmRecord # noqa: F401
from app.models.ad_reward import AdRewardRecord # noqa: F401
from app.models.ad_watch_log import AdWatchLog # noqa: F401
from app.models.admin import AdminAuditLog, AdminUser # noqa: F401
from app.models.app_config import AppConfig # noqa: F401
from app.models.comparison import ComparisonRecord # noqa: F401
from app.models.comparison_milestone import ComparisonMilestoneClaim # noqa: F401
from app.models.feedback import Feedback # noqa: F401
from app.models.price_report import PriceReport # noqa: F401
from app.models.savings import SavingsRecord # noqa: F401
from app.models.signin import SigninRecord # noqa: F401
from app.models.task import UserTask # noqa: F401
+41
View File
@@ -0,0 +1,41 @@
"""看激励视频观看时长记录(每日总时长防刷)。
每条 = 客户端 onAdClose 上报的一次激励视频实际观看秒数 (user_id, watch_date) 聚合
SUM(watch_seconds) 得当日总观看时长,用于"每天最多看 50 分钟激励视频"硬上限(超了不再发奖 /
不给看, core.rewards.DAILY_AD_WATCH_SECONDS_LIMIT)
这是看广告的第三条数据流,与另两条并列互不关联:
- ad_reward(AdRewardRecord):穿山甲 S2S 回调发金币(后端, trans_id);
- ad_ecpm(AdEcpmRecord):客户端上报展示收益(前端, ecpm);
- ad_watch_log(本表):客户端上报观看时长(前端, watch_seconds)
前端上报的时长不可信 时长是防刷"主闸",配合 ad_reward 的每日次数上限(DAILY_AD_REWARD_LIMIT)
"次数兜底",前端少报时长也刷不过次数闸
"""
from __future__ import annotations
from datetime import datetime
from sqlalchemy import DateTime, ForeignKey, Integer, String, func
from sqlalchemy.orm import Mapped, mapped_column
from app.db.base import Base
class AdWatchLog(Base):
__tablename__ = "ad_watch_log"
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
user_id: Mapped[int] = mapped_column(
Integer, ForeignKey("user.id"), index=True, nullable=False
)
# 本次激励视频实际观看秒数(前端 onAdShow→onAdClose);服务端已夹 [0, MAX_SINGLE_WATCH_SECONDS]
watch_seconds: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
# 北京时间日期串 'YYYY-MM-DD',按它等值做"当日总时长"聚合(不在 SQL 里跨时区比较)
watch_date: Mapped[str] = mapped_column(String(10), index=True, nullable=False)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), server_default=func.now(), index=True, nullable=False
)
def __repr__(self) -> str: # pragma: no cover
return f"<AdWatchLog user_id={self.user_id} sec={self.watch_seconds} {self.watch_date}>"
+32
View File
@@ -0,0 +1,32 @@
"""运营可配置项(把 rewards.py 等硬编码规则挪到 DB,运营后台可改)。
key 是配置标识( app.core.config_schema.CONFIG_DEFS),value JSON 存任意值(int/list/dict)
表里没有的 key,业务读配置时 fallback CONFIG_DEFS 默认值(= rewards 常量),
所以**空表 = 现有行为完全不变**admin 改某项 写一行 业务下次读到新值(跨进程一致)
"""
from __future__ import annotations
from datetime import datetime
from typing import Any
from sqlalchemy import JSON, DateTime, Integer, String, func
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.orm import Mapped, mapped_column
from app.db.base import Base
_JSON = JSON().with_variant(JSONB(), "postgresql")
class AppConfig(Base):
__tablename__ = "app_config"
key: Mapped[str] = mapped_column(String(64), primary_key=True)
value: Mapped[Any] = mapped_column(_JSON, nullable=False)
updated_by_admin_id: Mapped[int | None] = mapped_column(Integer, nullable=True)
updated_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), server_default=func.now(), onupdate=func.now(), nullable=False
)
def __repr__(self) -> str: # pragma: no cover
return f"<AppConfig key={self.key}>"
+3
View File
@@ -64,6 +64,9 @@ class ComparisonRecord(Base):
best_platform_id: Mapped[str | None] = mapped_column(String(32), nullable=True)
best_platform_name: Mapped[str | None] = mapped_column(String(32), nullable=True)
best_price_cents: Mapped[int | None] = mapped_column(Integer, nullable=True)
# 最优平台的商家/商品深链(客户端比价时从剪贴板采到 = collectedLinks[best_index]);
# 「再次比价」写剪贴板 + launch 该平台 App 直达。仅 2026-06 起新比价有,旧记录为 None。
best_deeplink: Mapped[str | None] = mapped_column(String(1024), nullable=True)
# 源价 - 最优价(可为 0 / 负:源平台本来就最便宜时没省到)
saved_amount_cents: Mapped[int | None] = mapped_column(Integer, nullable=True)
# 源平台就是最便宜的一家(= 这次没省到钱)
+67
View File
@@ -0,0 +1,67 @@
"""上报更低价记录表(price_report)。
用户在比价记录里选一条记录,上报某平台有比我们算出的最低价更低的价格,带截图证明,
人工审核通过奖励金币 feedback(自由文本反馈)不同:本表结构化关联具体比价记录
原最低价 vs 上报价审核状态奖励
- 原最低价快照(original_*)由后端按 comparison_record_id 反查比价记录的 best_* 冗余填入,
避免被关联记录后续删/改后对不上
- 价格统一存(cents), comparison_record / savings_record 一致
- status: pending(审核中)/ approved(已通过)/ rejected(未通过),对齐前端筛选与状态徽标
- images: 截图相对 URL 列表(/media/price_report/...,复用 app.core.media)
- 奖励发放(通过 钱包 +reward_coins)是人工审核后台动作,提交时一律 pending,本表只留字段
"""
from __future__ import annotations
from datetime import datetime
from sqlalchemy import JSON, DateTime, ForeignKey, Integer, String, func
from sqlalchemy.orm import Mapped, mapped_column
from app.db.base import Base
class PriceReport(Base):
__tablename__ = "price_report"
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
user_id: Mapped[int] = mapped_column(
Integer, ForeignKey("user.id"), index=True, nullable=False
)
# 选中的那条比价记录(可空:记录被删后仍保留上报历史)
comparison_record_id: Mapped[int | None] = mapped_column(
Integer, ForeignKey("comparison_record.id"), index=True, nullable=True
)
# ===== 选中比价记录的快照(后端反查 comparison_record 填入,冗余存) =====
store_name: Mapped[str | None] = mapped_column(String(128), nullable=True)
dish_summary: Mapped[str | None] = mapped_column(String(256), nullable=True)
original_platform_id: Mapped[str | None] = mapped_column(String(32), nullable=True)
original_platform_name: Mapped[str | None] = mapped_column(String(32), nullable=True)
original_price_cents: Mapped[int | None] = mapped_column(Integer, nullable=True)
# ===== 用户上报的更低价 =====
reported_platform_id: Mapped[str] = mapped_column(String(32), nullable=False)
reported_platform_name: Mapped[str] = mapped_column(String(32), nullable=False)
reported_price_cents: Mapped[int] = mapped_column(Integer, nullable=False)
# 截图证明 URL 列表(相对路径,如 ["/media/price_report/u1_ab.jpg"])
images: Mapped[list[str]] = mapped_column(JSON, nullable=False, default=list)
# ===== 审核(人工后台) =====
status: Mapped[str] = mapped_column(
String(16), nullable=False, default="pending", index=True
)
reject_reason: Mapped[str | None] = mapped_column(String(256), nullable=True)
reward_coins: Mapped[int | None] = mapped_column(Integer, nullable=True)
reviewed_at: Mapped[datetime | None] = mapped_column(
DateTime(timezone=True), nullable=True
)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), server_default=func.now(), index=True, nullable=False
)
def __repr__(self) -> str: # pragma: no cover
return (
f"<PriceReport id={self.id} user_id={self.user_id} status={self.status}>"
)
+11 -6
View File
@@ -65,11 +65,14 @@ class CoinTransaction(Base):
class WithdrawOrder(Base):
"""提现单(现金 → 微信零钱)。状态机:pending → success / failed。
"""提现单(现金 → 微信零钱)。状态机:
reviewing (管理员通过) pending success / failed
reviewing (管理员拒绝) rejected(已退款)
扣现金 + cash_transaction(withdraw) 与建本单在同一事务;失败/取消时退回现金
并写 cash_transaction(withdraw_refund)wechat_state 存微信侧原始状态(WAIT_USER_CONFIRM /
SUCCESS / FAIL / CANCELLED ...),status 是我们归一化后的三态
扣现金 + cash_transaction(withdraw) 与建本单在同一事务,初始 reviewing(待人工审核,
**不打款**);管理员审核通过才调微信转账(execute_withdraw_transfer) pending,拒绝则退回
现金 + rejected微信侧失败/取消时退回现金并写 cash_transaction(withdraw_refund) failed
wechat_state 存微信侧原始状态(WAIT_USER_CONFIRM / SUCCESS / FAIL / CANCELLED ...)
"""
__tablename__ = "withdraw_order"
@@ -81,8 +84,10 @@ class WithdrawOrder(Base):
# 商户单号(我们生成,微信查单的 out_bill_no),唯一
out_bill_no: Mapped[str] = mapped_column(String(64), unique=True, index=True, nullable=False)
amount_cents: Mapped[int] = mapped_column(Integer, nullable=False)
# 归一化状态:pending / success / failed
status: Mapped[str] = mapped_column(String(16), nullable=False, default="pending")
# 提现实名(微信达额转账要求):审核后异步打款时要用,发起提现时存下,可空
user_name: Mapped[str | None] = mapped_column(String(64), nullable=True)
# 归一化状态:reviewing(待审核) / pending(打款在途) / success / failed(打款失败已退) / rejected(审核拒绝已退)
status: Mapped[str] = mapped_column(String(16), nullable=False, default="reviewing")
# 微信侧原始状态
wechat_state: Mapped[str | None] = mapped_column(String(32), nullable=True)
# 微信转账单号
+32 -18
View File
@@ -6,7 +6,7 @@
3. 当日发奖次数 上限 记一条 status='capped' 但不发金币
发金币复用 `wallet.grant_coins`(biz_type='ad_reward', ref_id=trans_id),与发奖记录同事务,
保证"记一笔 + 加金币"原子化
保证"记一笔 + 加金币"原子化单次金币 / 每日上限 / 每轮冷却 都从 app_config (运营后台可改)
"""
from __future__ import annotations
@@ -16,15 +16,13 @@ from sqlalchemy import func, select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session
from app.core import rewards
from app.core.ad_cooldown import compute_cooldown
from app.core.rewards import (
AD_REWARD_COIN,
DAILY_AD_REWARD_LIMIT,
cn_today,
)
from app.repositories import wallet as crud_wallet
from app.core.rewards import DAILY_AD_WATCH_SECONDS_LIMIT, cn_today
from app.models.ad_reward import AdRewardRecord
from app.models.user import User
from app.repositories import wallet as crud_wallet
from app.repositories.ad_watch import watched_seconds_today
class UnknownUserError(Exception):
@@ -54,14 +52,14 @@ def grant_ad_reward(
user_id: int,
trans_id: str,
*,
coin: int = AD_REWARD_COIN,
coin: int | None = None,
reward_name: str | None = None,
raw: str | None = None,
) -> AdRewardRecord:
"""看广告发奖(幂等 + 每日限额)。返回发奖记录(status=granted/capped)。
coin 为本次发放金币(由调用方按穿山甲回调 reward_amount 解析,
rewards.resolve_ad_reward_coin);默认 AD_REWARD_COIN test-grant / 缺省场景
rewards.resolve_ad_reward_coin);None 读配置 get_ad_reward_coin(test-grant / 缺省场景)
user_id 不存在抛 UnknownUserError(不建账户,防伪造 user_id 刷出脏数据)
"""
# #2 幂等:同 trans_id 已处理过 → 原样返回,不重复发
@@ -74,14 +72,21 @@ def grant_ad_reward(
today = cn_today().isoformat()
# #3 每日上限:超了记一条 capped(不发金币),让审计能看到"今天到顶了"
if _granted_today(db, user_id, today) >= DAILY_AD_REWARD_LIMIT:
# #3 每日上限:观看时长(主闸,50 分钟) 或 发奖次数(兜底) 任一到顶 → 记 capped 不发金币,
# 让审计能看到"今天到顶了"。时长由前端 watch-report 累计(防刷主闸),次数防前端少报时长
# 绕过;次数上限走 app_config(运营后台可改,默认 rewards.DAILY_AD_REWARD_LIMIT)。
over_time = watched_seconds_today(db, user_id, today=today) >= DAILY_AD_WATCH_SECONDS_LIMIT
over_count = _granted_today(db, user_id, today) >= rewards.get_ad_daily_limit(db)
if over_time or over_count:
rec = AdRewardRecord(
trans_id=trans_id, user_id=user_id, coin=0, status="capped",
reward_date=today, reward_name=reward_name, raw=raw,
)
return _commit_record(db, rec, trans_id)
if coin is None:
coin = rewards.get_ad_reward_coin(db)
# 发金币 + 记一笔,同事务
crud_wallet.grant_coins(
db, user_id, coin,
@@ -126,20 +131,29 @@ def _granted_times_today_desc(db: Session, user_id: int, reward_date: str) -> li
def today_status(
db: Session, user_id: int
) -> tuple[int, int, int, int, datetime | None]:
) -> tuple[int, int, int, int, datetime | None, int, int]:
"""客户端查"今日看广告发奖"进度。
返回 (今日已发次数, 每日上限, 单次金币, 本轮已看次数, 本轮冷却结束时间(UTC))
本函数只**取数据**(今日 granted created_at 倒序),"本轮已看几次 + 冷却到几点"
**策略判断**委托给 [app.core.ad_cooldown.compute_cooldown](纯函数)换冷却策略只动那个模块
返回 (今日已发次数, 每日次数上限, 单次金币, 本轮已看次数, 本轮冷却结束时间(UTC),
今日已观看总秒数, 每日观看总时长上限())
次数上限/单次金币/每轮次数/冷却秒 均从 app_config (运营后台可改);次数维度的"本轮已看
几次 + 冷却到几点"委托 [app.core.ad_cooldown.compute_cooldown](纯函数);时长维度(50 分钟
主闸) ad_watch_log 当日累计前端据此展示"今日已看 X/50 分钟"+ 满则不给看
"""
today = cn_today().isoformat()
granted_desc = _granted_times_today_desc(db, user_id, today)
state = compute_cooldown(granted_desc, datetime.now(timezone.utc))
state = compute_cooldown(
granted_desc,
datetime.now(timezone.utc),
round_size=rewards.get_ad_round_count(db),
cooldown_seconds=rewards.get_ad_cooldown_sec(db),
)
return (
len(granted_desc),
DAILY_AD_REWARD_LIMIT,
AD_REWARD_COIN,
rewards.get_ad_daily_limit(db),
rewards.get_ad_reward_coin(db),
state.round_count,
state.cooldown_until,
watched_seconds_today(db, user_id, today=today),
DAILY_AD_WATCH_SECONDS_LIMIT,
)
+36
View File
@@ -0,0 +1,36 @@
"""看激励视频观看时长 CRUD(每日总时长防刷)。
前端 onAdClose 上报本次观看秒数 add_watch_seconds 落库 + 返回当日累计;
watched_seconds_today "时长闸"(发奖 / 展示前)判断当天是否已达 50 分钟上限
鉴权接口已确保 user 存在(JWT),故不做 UnknownUser 校验
"""
from __future__ import annotations
from sqlalchemy import func, select
from sqlalchemy.orm import Session
from app.core.rewards import MAX_SINGLE_WATCH_SECONDS, cn_today
from app.models.ad_watch_log import AdWatchLog
def watched_seconds_today(db: Session, user_id: int, *, today: str | None = None) -> int:
"""该用户今日(北京时间)累计观看秒数。coalesce 防无记录时 SUM 返 None。"""
d = today or cn_today().isoformat()
return db.execute(
select(func.coalesce(func.sum(AdWatchLog.watch_seconds), 0)).where(
AdWatchLog.user_id == user_id,
AdWatchLog.watch_date == d,
)
).scalar_one()
def add_watch_seconds(db: Session, user_id: int, seconds: int) -> int:
"""记一次观看时长,返回当日累计秒数。
seconds [0, MAX_SINGLE_WATCH_SECONDS] 防前端报异常大值(刷时长无意义,但夹紧更稳)
"""
clamped = max(0, min(int(seconds), MAX_SINGLE_WATCH_SECONDS))
today = cn_today().isoformat()
db.add(AdWatchLog(user_id=user_id, watch_seconds=clamped, watch_date=today))
db.commit()
return watched_seconds_today(db, user_id, today=today)
+65
View File
@@ -0,0 +1,65 @@
"""运营配置读写。
get_value:业务读配置,DB 没有则 fallback CONFIG_DEFS 默认(= rewards 常量,空表行为不变)
不缓存:配置读频率低(每次福利操作读一次,key 主键查极快),admin 改了立即生效跨进程一致
"""
from __future__ import annotations
from typing import Any
from sqlalchemy import select
from sqlalchemy.orm import Session
from app.core.config_schema import CONFIG_DEFS
from app.models.app_config import AppConfig
def get_value(db: Session, key: str) -> Any:
"""读配置值。DB 有用 DB,没有 fallback 到 CONFIG_DEFS 默认。未知 key 抛 KeyError。"""
row = db.get(AppConfig, key)
if row is not None:
return row.value
return CONFIG_DEFS[key]["default"]
def set_value(
db: Session, key: str, value: Any, *, admin_id: int, commit: bool = True
) -> AppConfig:
"""写配置(admin 用)。未知 key 抛 KeyError。commit=False 让调用方与审计同事务。"""
if key not in CONFIG_DEFS:
raise KeyError(f"unknown config key: {key}")
row = db.get(AppConfig, key)
if row is None:
row = AppConfig(key=key, value=value, updated_by_admin_id=admin_id)
db.add(row)
else:
row.value = value
row.updated_by_admin_id = admin_id
if commit:
db.commit()
db.refresh(row)
else:
db.flush()
return row
def list_all(db: Session) -> list[dict]:
"""所有可配项:default + 当前值(DB 有则用 DB) + 元信息。给 admin 配置页渲染。"""
overrides = {r.key: r for r in db.execute(select(AppConfig)).scalars().all()}
out: list[dict] = []
for key, defn in CONFIG_DEFS.items():
row = overrides.get(key)
out.append(
{
"key": key,
"label": defn["label"],
"group": defn["group"],
"type": defn["type"],
"help": defn.get("help"),
"default": defn["default"],
"value": row.value if row is not None else defn["default"],
"overridden": row is not None,
"updated_at": row.updated_at.isoformat() if row is not None else None,
}
)
return out
+28 -2
View File
@@ -9,6 +9,7 @@ from sqlalchemy import func, select
from sqlalchemy.orm import Session
from app.models.comparison import ComparisonRecord
from app.models.savings import SavingsRecord
from app.schemas.compare_record import ComparisonRecordIn
@@ -79,6 +80,7 @@ def upsert_record(
source_platform_name=payload.source_platform_name,
source_package=payload.source_package,
information=payload.information,
best_deeplink=payload.best_deeplink,
total_dish_count=payload.total_dish_count,
skipped_dish_count=payload.skipped_dish_count,
items=[it.model_dump(exclude_none=True) for it in payload.items],
@@ -109,6 +111,23 @@ def upsert_record(
return rec
def _ordered_shop_names(db: Session, user_id: int) -> set[str]:
"""该用户「真实下单」(source='compare')覆盖到的店名集合,用来给比价记录打「已下单」。
只认 compare(归因命中后真实上报),demo 演示数据不算下单上报不带 trace_id,
只能按店名对齐两边店名同源(都来自比价意图识别阶段的门店名 query),精确相等即视为同店
语义=店级:同一家店比价过多次,这些记录会一并标已下单
"""
rows = db.execute(
select(SavingsRecord.shop_name).where(
SavingsRecord.user_id == user_id,
SavingsRecord.source == "compare",
SavingsRecord.shop_name.is_not(None),
)
).scalars().all()
return {s for s in rows if s}
def list_records(
db: Session,
user_id: int,
@@ -116,14 +135,21 @@ def list_records(
limit: int = 20,
cursor: int | None = None,
) -> tuple[list[ComparisonRecord], int | None]:
"""比价记录分页(按 id 倒序,游标式)。"""
"""比价记录分页(按创建时间倒序、id 兜底,游标式)。附「已下单」店级标记(瞬态,不写库)。"""
stmt = select(ComparisonRecord).where(ComparisonRecord.user_id == user_id)
if cursor is not None:
stmt = stmt.where(ComparisonRecord.id < cursor)
stmt = stmt.order_by(ComparisonRecord.id.desc()).limit(limit)
stmt = stmt.order_by(ComparisonRecord.created_at.desc(), ComparisonRecord.id.desc()).limit(limit)
items = list(db.execute(stmt).scalars().all())
next_cursor = items[-1].id if len(items) == limit else None
# 「已下单」标记:本页记录的 store_name 若落在该用户真实下单的店名集合里即 True。
# ordered 非 ORM 列,仅挂实例上供 ComparisonRecordOut(from_attributes) 读出,不持久化。
ordered_shops = _ordered_shop_names(db, user_id)
for it in items:
it.ordered = bool(it.store_name and it.store_name in ordered_shops)
return items, next_cursor
+7 -8
View File
@@ -2,6 +2,7 @@
进度 = comparison_record status='success' 的条数(crud_compare.count_success)
N 档在"成功次数 >= N"时解锁;每档领一次, comparison_milestone_claim 去重(标记已领)
档位金额来自 rewards.get_record_milestones(运营后台可改)
当前产品定暂不真发金币(后续整体删除该功能):claim 只写领取记录,不调 grant_coins
不写 coin_transaction,coin_awarded 恒为 0,余额不变仿一次性任务 (task.py)
"""
@@ -12,17 +13,14 @@ from dataclasses import dataclass
from sqlalchemy import select
from sqlalchemy.orm import Session
from app.core.rewards import (
RECORD_MILESTONE_COUNT,
RECORD_MILESTONES,
)
from app.core import rewards
from app.models.comparison_milestone import ComparisonMilestoneClaim
from app.repositories import comparison as crud_compare
from app.repositories import wallet as crud_wallet
class UnknownMilestoneError(Exception):
"""档位序号越界(不在 1..RECORD_MILESTONE_COUNT)。"""
"""档位序号越界(不在 1..档位数)。"""
class MilestoneLockedError(Exception):
@@ -60,10 +58,11 @@ def get_status(db: Session, user_id: int) -> MilestoneStatus:
"""各档领取状态:已领=claimed,已解锁未领=active(可领),未解锁=locked。"""
success_count = crud_compare.count_success(db, user_id)
claimed = _claimed_set(db, user_id)
milestones_def = rewards.get_record_milestones(db)
milestones: list[MilestoneState] = []
claimable = 0
for m in range(1, RECORD_MILESTONE_COUNT + 1):
for m in range(1, len(milestones_def) + 1):
if m in claimed:
state = "claimed"
elif success_count >= m:
@@ -72,7 +71,7 @@ def get_status(db: Session, user_id: int) -> MilestoneStatus:
else:
state = "locked"
milestones.append(
MilestoneState(milestone=m, coin=RECORD_MILESTONES[m - 1], state=state)
MilestoneState(milestone=m, coin=milestones_def[m - 1], state=state)
)
return MilestoneStatus(
@@ -87,7 +86,7 @@ def claim(db: Session, user_id: int, milestone: int) -> tuple[int, int]:
越界抛 UnknownMilestoneError;未解锁抛 MilestoneLockedError;重复领抛 AlreadyClaimedError
"""
if milestone < 1 or milestone > RECORD_MILESTONE_COUNT:
if milestone < 1 or milestone > len(rewards.get_record_milestones(db)):
raise UnknownMilestoneError
existing = db.execute(
+53
View File
@@ -0,0 +1,53 @@
"""price_report 表读写。"""
from __future__ import annotations
from sqlalchemy import select
from sqlalchemy.orm import Session
from app.models.price_report import PriceReport
def create_report(
db: Session,
*,
user_id: int,
comparison_record_id: int | None,
store_name: str | None,
dish_summary: str | None,
original_platform_id: str | None,
original_platform_name: str | None,
original_price_cents: int | None,
reported_platform_id: str,
reported_platform_name: str,
reported_price_cents: int,
images: list[str],
) -> PriceReport:
rep = PriceReport(
user_id=user_id,
comparison_record_id=comparison_record_id,
store_name=store_name,
dish_summary=dish_summary,
original_platform_id=original_platform_id,
original_platform_name=original_platform_name,
original_price_cents=original_price_cents,
reported_platform_id=reported_platform_id,
reported_platform_name=reported_platform_name,
reported_price_cents=reported_price_cents,
images=images,
status="pending",
)
db.add(rep)
db.commit()
db.refresh(rep)
return rep
def list_reports(
db: Session, user_id: int, status: str | None = None
) -> list[PriceReport]:
"""该用户的上报记录,按时间倒序;status 非空时按状态筛选。"""
stmt = select(PriceReport).where(PriceReport.user_id == user_id)
if status:
stmt = stmt.where(PriceReport.status == status)
stmt = stmt.order_by(PriceReport.created_at.desc(), PriceReport.id.desc())
return list(db.execute(stmt).scalars().all())
+8 -6
View File
@@ -3,7 +3,7 @@
7 天循环规则:
- 昨天签过 今天 cycle_day = 昨天 % 7 + 1,streak += 1(连续)
- 否则(首签 / 断签) cycle_day = 1,streak = 1(重置)
今天的 cycle_day 决定发多少金币( app.core.rewards.SIGNIN_REWARDS)
今天的 cycle_day 决定发多少金币(档位来自 rewards.get_signin_rewards,运营后台可改)
"""
from __future__ import annotations
@@ -13,9 +13,10 @@ from datetime import timedelta
from sqlalchemy import select
from sqlalchemy.orm import Session
from app.core.rewards import SIGNIN_CYCLE_LEN, SIGNIN_REWARDS, cn_today, signin_reward
from app.repositories import wallet as crud_wallet
from app.core import rewards
from app.core.rewards import SIGNIN_CYCLE_LEN, cn_today
from app.models.signin import SigninRecord
from app.repositories import wallet as crud_wallet
class AlreadySignedError(Exception):
@@ -58,6 +59,7 @@ def _next_cycle_day(last: SigninRecord | None, today) -> int:
def get_status(db: Session, user_id: int) -> SigninStatus:
today = cn_today()
rewards_list = rewards.get_signin_rewards(db) # 配置 or 默认(长度=SIGNIN_CYCLE_LEN)
last = _latest_record(db, user_id)
today_signed = last is not None and last.signin_date == today
@@ -83,13 +85,13 @@ def get_status(db: Session, user_id: int) -> SigninStatus:
status = "today"
else:
status = "locked"
steps.append(SigninStep(day=day, coin=SIGNIN_REWARDS[day - 1], status=status))
steps.append(SigninStep(day=day, coin=rewards_list[day - 1], status=status))
return SigninStatus(
today_signed=today_signed,
consecutive_days=consecutive_days,
today_cycle_day=today_cycle_day,
today_coin=signin_reward(today_cycle_day),
today_coin=rewards_list[today_cycle_day - 1],
can_claim=not today_signed,
steps=steps,
)
@@ -109,7 +111,7 @@ def do_signin(db: Session, user_id: int) -> tuple[SigninRecord, int]:
cycle_day = 1
streak = 1
coin = signin_reward(cycle_day)
coin = rewards.get_signin_rewards(db)[cycle_day - 1]
record = SigninRecord(
user_id=user_id,
signin_date=today,
+8 -5
View File
@@ -1,6 +1,7 @@
"""一次性任务 CRUD:列表状态 + 领奖。
领奖原子化: user_task(唯一约束防重复)+ 发金币,同一事务 commit
任务列表 + 奖励来自 rewards.get_task_rewards(运营后台可改)
"""
from __future__ import annotations
@@ -9,9 +10,9 @@ from dataclasses import dataclass
from sqlalchemy import select
from sqlalchemy.orm import Session
from app.core.rewards import TASK_REWARDS
from app.repositories import wallet as crud_wallet
from app.core import rewards
from app.models.task import UserTask
from app.repositories import wallet as crud_wallet
class UnknownTaskError(Exception):
@@ -31,11 +32,12 @@ class TaskState:
def list_tasks(db: Session, user_id: int) -> list[TaskState]:
"""返回所有已知一次性任务及其领取状态。"""
task_rewards = rewards.get_task_rewards(db)
stmt = select(UserTask.task_key).where(UserTask.user_id == user_id)
claimed_keys = set(db.execute(stmt).scalars().all())
return [
TaskState(task_key=key, coin=coin, claimed=key in claimed_keys)
for key, coin in TASK_REWARDS.items()
for key, coin in task_rewards.items()
]
@@ -44,7 +46,8 @@ def claim_task(db: Session, user_id: int, task_key: str) -> tuple[int, int]:
未知任务抛 UnknownTaskError;重复领取抛 AlreadyClaimedError
"""
if task_key not in TASK_REWARDS:
task_rewards = rewards.get_task_rewards(db)
if task_key not in task_rewards:
raise UnknownTaskError
existing = db.execute(
@@ -55,7 +58,7 @@ def claim_task(db: Session, user_id: int, task_key: str) -> tuple[int, int]:
if existing is not None:
raise AlreadyClaimedError
coin = TASK_REWARDS[task_key]
coin = task_rewards[task_key]
db.add(
UserTask(
user_id=user_id,
+96 -31
View File
@@ -13,13 +13,8 @@ from datetime import datetime, timedelta, timezone
from sqlalchemy import select, update
from sqlalchemy.orm import Session
from app.core.rewards import (
COIN_PER_CENT,
MIN_EXCHANGE_COIN,
WITHDRAW_MAX_CENTS,
WITHDRAW_MIN_CENTS,
coins_to_cents,
)
from app.core import rewards
from app.core.rewards import COIN_PER_CENT, coins_to_cents
from app.integrations import wxpay
from app.models.user import User
from app.models.wallet import CashTransaction, CoinAccount, CoinTransaction, WithdrawOrder
@@ -62,6 +57,10 @@ class WithdrawOrderNotFound(Exception):
"""提现单不存在。"""
class WithdrawNotReviewable(Exception):
"""提现单当前状态不可审核(非 reviewing,可能已被处理过)。"""
def get_or_create_account(db: Session, user_id: int, *, commit: bool = True) -> CoinAccount:
"""取用户金币账户,不存在则建一个空账户。"""
acc = db.get(CoinAccount, user_id)
@@ -142,7 +141,7 @@ def exchange_coins_to_cash(
校验:金额 >= MIN_EXCHANGE_COIN 且为整分倍数(InvalidExchangeAmountError),
余额充足(InsufficientCoinError)扣金币 + 加现金,两条流水同事务 commit
"""
if coin_amount < MIN_EXCHANGE_COIN or coin_amount % COIN_PER_CENT != 0:
if coin_amount < rewards.get_min_exchange_coin(db) or coin_amount % COIN_PER_CENT != 0:
raise InvalidExchangeAmountError
acc = get_or_create_account(db, user_id, commit=False)
@@ -259,10 +258,18 @@ def _add_cash(db: Session, user_id: int, amount_cents: int) -> int:
return bal
def _refund_withdraw(db: Session, order: WithdrawOrder, reason: str) -> None:
"""转账失败/取消:退回现金 + 写 withdraw_refund 流水 + 单子置 failed。幂等:已 failed 不重复退。"""
if order.status == "failed":
return # 防重复退款(并发/对账与查单同时触发)
def _refund_withdraw(
db: Session, order: WithdrawOrder, reason: str, *, final_status: str = "failed"
) -> None:
"""退回现金 + 写 withdraw_refund 流水 + 单子置终态。幂等:已是退款终态不重复退。
final_status:
- "failed" (默认):微信转账失败/取消 自动退款
- "rejected":管理员审核拒绝 退款( reject_withdraw)
两种都已退款, in 判定防重复退(并发/对账/拒绝重复点)
"""
if order.status in ("failed", "rejected"):
return # 防重复退款(并发/对账与查单/重复拒绝同时触发)
bal = _add_cash(db, order.user_id, order.amount_cents)
db.add(
CashTransaction(
@@ -271,10 +278,11 @@ def _refund_withdraw(db: Session, order: WithdrawOrder, reason: str) -> None:
balance_after_cents=bal,
biz_type="withdraw_refund",
ref_id=order.out_bill_no,
remark="提现未成功,金额已退回", # 用户可见;技术原因记在 order.fail_reason
# 用户可见文案区分"未成功(自动退)"vs"审核未通过";技术原因记在 order.fail_reason
remark="提现审核未通过,金额已退回" if final_status == "rejected" else "提现未成功,金额已退回",
)
)
order.status = "failed"
order.status = final_status
order.fail_reason = reason[:256]
db.commit()
@@ -332,22 +340,26 @@ def create_withdraw(
user_name: str | None = None,
out_bill_no: str | None = None,
) -> WithdrawOrder:
"""发起提现(原子扣款 + 幂等 + 模糊失败查单)
"""发起提现:原子扣款 + 建单 reviewing(待人工审核),**不打款**
#1 用原子条件 UPDATE 扣款,杜绝并发超额。
#2 out_bill_no 由客户端提供作幂等键:同号重试不会重复转账,直接返回该单当前状态。
#3 微信调用结果不明时先查单再决定,绝不盲目退款。
返回提现(可能含 package_info App 拉起确认页)
打款推迟到管理员在 admin 后台审核通过(approve_withdraw execute_withdraw_transfer)才发起;
拒绝则退回现金(reject_withdraw)
#1 用原子条件 UPDATE 扣款,杜绝并发超额——且"待审核期间钱已扣减",防止用户拿同一笔余额
重复发起多笔提现(审核拒绝再退回)
#2 out_bill_no 客户端幂等键:同号重试返回该单现状(reviewing 等审核),不重复扣款建单。
实名 user_name 在此存下(WithdrawOrder.user_name),供异步审核打款时传给微信(达额需实名)
"""
if amount_cents < WITHDRAW_MIN_CENTS or amount_cents > WITHDRAW_MAX_CENTS:
if amount_cents < rewards.get_withdraw_min_cents(db) or amount_cents > rewards.get_withdraw_max_cents(db):
raise InvalidWithdrawAmountError
# 提现即要求已绑微信:否则审核通过也打不了款,提前拦更友好
user = db.get(User, user_id)
openid = user.wechat_openid if user else None
if not openid:
raise WechatNotBoundError
# #2 幂等:客户端传了合法 out_bill_no 且该单已存在 → 返回现状(pending 的顺手查一下),不重复发起
# #2 幂等:该单已存在 → 直接返回现状(reviewing 等审核 / 已处理态原样返回),不重复扣款。
# 不再像旧版对 pending 查微信:本函数不再涉及微信,查单交给 /withdraw/status 与对账任务。
if out_bill_no and _OUT_BILL_NO_RE.match(out_bill_no):
existing = db.execute(
select(WithdrawOrder).where(
@@ -355,8 +367,6 @@ def create_withdraw(
)
).scalar_one_or_none()
if existing is not None:
if existing.status == "pending":
return refresh_withdraw_status(db, user_id, out_bill_no)
return existing
else:
out_bill_no = uuid.uuid4().hex
@@ -379,32 +389,53 @@ def create_withdraw(
balance_after_cents=bal,
biz_type="withdraw",
ref_id=out_bill_no,
remark="提现到微信零钱",
remark="提现到微信零钱(待审核)",
)
)
order = WithdrawOrder(
user_id=user_id, out_bill_no=out_bill_no, amount_cents=amount_cents, status="pending"
user_id=user_id,
out_bill_no=out_bill_no,
amount_cents=amount_cents,
user_name=user_name,
status="reviewing",
)
db.add(order)
db.commit()
db.refresh(order)
return order # 待管理员审核;**不在此处打款**
def execute_withdraw_transfer(db: Session, order: WithdrawOrder) -> WithdrawOrder:
"""审核通过后真正发起微信转账(复用原"转账 + 模糊失败查单"逻辑,绝不盲目退款)。
流程:reviewing pending 调微信转账 SUCCESS=success / 失败/取消=退款 failed /
结果不明=先查单再决定**不抛异常**(admin 调用方按 order.status 判断结果)
用户在审核期间解绑微信 退款 failed(打不了款)
可能返回 package_info(WAIT_USER_CONFIRM 场景:需用户在 App 端确认页确认后才真正到账)
"""
user = db.get(User, order.user_id)
openid = user.wechat_openid if user else None
if not openid:
_refund_withdraw(db, order, reason="用户已解绑微信,无法打款")
db.refresh(order)
return order
order.status = "pending" # 进入打款在途(转账调用前崩溃留 pending,交对账兜底)
db.commit()
# 2) 调微信转账;结果不明先查单再决定(#3)
try:
result = wxpay.create_transfer(openid, amount_cents, out_bill_no, user_name=user_name)
result = wxpay.create_transfer(
openid, order.amount_cents, order.out_bill_no, user_name=order.user_name
)
except Exception as e: # noqa: BLE001 — 超时/网络异常≠失败,查单确认
_settle_after_ambiguous(db, order, reason=f"转账调用异常: {e}")
db.refresh(order)
if order.status == "failed":
raise WithdrawTransferError(str(e)) from e
return order
if result["status_code"] != 200:
msg = str(result["data"].get("message") or result["data"])
_settle_after_ambiguous(db, order, reason=msg)
db.refresh(order)
if order.status == "failed":
raise WithdrawTransferError(msg)
return order
data = result["data"]
@@ -418,6 +449,40 @@ def create_withdraw(
return order
def _get_withdraw_or_raise(db: Session, out_bill_no: str) -> WithdrawOrder:
"""按 out_bill_no 取提现单(admin 跨用户,不限 user_id)。不存在抛 WithdrawOrderNotFound。"""
order = db.execute(
select(WithdrawOrder).where(WithdrawOrder.out_bill_no == out_bill_no)
).scalar_one_or_none()
if order is None:
raise WithdrawOrderNotFound
return order
def approve_withdraw(db: Session, out_bill_no: str) -> WithdrawOrder:
"""管理员审核通过:校验 reviewing → 发起微信转账。返回最终 order(pending/success/failed)。
幂等防误操作: reviewing(已通过/已拒绝/已打款) WithdrawNotReviewable,不会重复打款
"""
order = _get_withdraw_or_raise(db, out_bill_no)
if order.status != "reviewing":
raise WithdrawNotReviewable(f"提现单状态为 {order.status},非待审核,不能通过")
return execute_withdraw_transfer(db, order)
def reject_withdraw(db: Session, out_bill_no: str, reason: str) -> WithdrawOrder:
"""管理员审核拒绝:校验 reviewing → 退回现金 + 置 rejected。
reviewing WithdrawNotReviewable(防对已打款单误退导致重复退款)
"""
order = _get_withdraw_or_raise(db, out_bill_no)
if order.status != "reviewing":
raise WithdrawNotReviewable(f"提现单状态为 {order.status},非待审核,不能拒绝")
_refund_withdraw(db, order, reason=reason or "审核未通过", final_status="rejected")
db.refresh(order)
return order
def refresh_withdraw_status(
db: Session, user_id: int, out_bill_no: str, *, cancel_if_unconfirmed: bool = False
) -> WithdrawOrder:
+21
View File
@@ -37,6 +37,10 @@ class AdRewardStatusOut(BaseModel):
description="本轮看完后 10 分钟冷却结束时间(UTC ISO),null 表示不在冷却。"
"冷却是 UX 约束,后端发奖不受影响,客户端 CTA 据此显示 MM:SS 倒计时且不可点",
)
# ===== 看广告每日总时长(50 分钟防刷主闸):客户端据此展示"今日已看 X/50 分钟"、满则不给看 =====
watched_seconds_today: int = Field(0, description="今日已观看激励视频总秒数(前端累计上报)")
watch_seconds_limit: int = Field(0, description="每日观看总时长上限(秒,默认 3000=50 分钟)")
watch_seconds_remaining: int = Field(0, description="今日剩余可观看秒数(<=0 时客户端不应再展示广告)")
class EcpmReportIn(BaseModel):
@@ -57,6 +61,23 @@ class EcpmReportOut(BaseModel):
ok: bool = True
class WatchReportIn(BaseModel):
"""客户端上报一次激励视频的实际观看时长(onAdClose 时,onAdShow→onAdClose 墙钟秒数)。
user_id JWT (Bearer),不在 bodyseconds 服务端会夹 [0, MAX_SINGLE_WATCH_SECONDS]
"""
seconds: int = Field(..., ge=0, description="本次观看秒数")
class WatchReportOut(BaseModel):
"""上报后返回当日累计 + 剩余,客户端即时更新"今日还能看多久""""
watched_seconds_today: int = Field(..., description="今日累计观看秒数")
watch_seconds_limit: int = Field(..., description="每日上限(秒)")
watch_seconds_remaining: int = Field(..., description="今日剩余可观看秒数")
class TestGrantOut(BaseModel):
"""[仅本地联调]模拟发奖结果。带上今日进度,客户端可直接据此刷新展示。"""
+6
View File
@@ -65,6 +65,8 @@ class ComparisonRecordIn(BaseModel):
information: str | None = Field(None, description="done 帧文案,留存备查")
# 不传则服务端按 comparison_results 派生(有非源有效价=success,否则 failed)
status: str | None = Field(None, description="success / failed,可不传由服务端派生")
# 最优平台商家/商品深链(客户端从 collectedLinks[best_index] 取);「再次比价」直达用
best_deeplink: str | None = Field(None, description="最优平台深链,再次比价直达")
# ===== 读取出参 =====
@@ -84,6 +86,7 @@ class ComparisonRecordOut(BaseModel):
best_platform_id: str | None = None
best_platform_name: str | None = None
best_price_cents: int | None = None
best_deeplink: str | None = None
saved_amount_cents: int | None = None
is_source_best: bool | None = None
store_name: str | None = None
@@ -94,6 +97,9 @@ class ComparisonRecordOut(BaseModel):
items: list = []
comparison_results: list = []
skipped_dish_names: list = []
# 「已下单」(店级):该店名在该用户真实下单(source='compare')里出现过即 True。
# 由 list_records 动态算出挂在 ORM 实例上(非 DB 列),from_attributes 读出;缺省 False。
ordered: bool = False
created_at: datetime
+57
View File
@@ -0,0 +1,57 @@
"""上报更低价 响应 schema。
请求是 multipart 表单(comparison_record_id / reported_platform_id / reported_price / images),
在路由里直接校验,不单独建请求 schema( feedback)价格对外统一(cents)
"""
from __future__ import annotations
from datetime import datetime
from pydantic import BaseModel, ConfigDict, Field
class ReportSubmitOut(BaseModel):
"""提交上报后的回执。"""
model_config = ConfigDict(from_attributes=True)
id: int
status: str
created_at: datetime
class ReportRecordOut(BaseModel):
"""一条上报记录(「上报记录」列表用)。价格单位:分。"""
model_config = ConfigDict(from_attributes=True)
id: int
store_name: str | None = None
dish_summary: str | None = None
# 原最低价(上报前系统给出的最低,反查比价记录 best_*)
original_platform_id: str | None = None
original_platform_name: str | None = None
original_price_cents: int | None = None
# 用户上报的更低价
reported_platform_id: str
reported_platform_name: str
reported_price_cents: int
images: list[str] = Field(default_factory=list)
status: str # pending / approved / rejected
reject_reason: str | None = None
reward_coins: int | None = None
created_at: datetime
class ReportRecordCounts(BaseModel):
"""四态计数,供前端筛选 chip 显示(全部/审核中/已通过/未通过)。"""
all: int = 0
pending: int = 0
approved: int = 0
rejected: int = 0
class ReportRecordsOut(BaseModel):
records: list[ReportRecordOut]
counts: ReportRecordCounts
+8 -3
View File
@@ -106,7 +106,7 @@ class WithdrawRequest(BaseModel):
class WithdrawResultOut(BaseModel):
out_bill_no: str = Field(..., description="商户提现单号(查单用)")
status: str = Field(..., description="pending / success / failed")
status: str = Field(..., description="reviewing(待审核) / pending / success / failed / rejected")
wechat_state: str | None = Field(None, description="微信侧原始状态")
amount_cents: int
cash_balance_cents: int = Field(..., description="提现后现金余额(分)")
@@ -118,9 +118,14 @@ class WithdrawResultOut(BaseModel):
class WithdrawStatusOut(BaseModel):
out_bill_no: str
status: str = Field(..., description="pending / success / failed")
status: str = Field(..., description="reviewing(待审核) / pending / success / failed / rejected")
wechat_state: str | None = None
amount_cents: int
fail_reason: str | None = Field(None, description="failed/rejected 时的原因(用户可读)")
# 审核通过进入打款、且微信要求用户确认(WAIT_USER_CONFIRM)时,带回这三项供 App 拉起微信确认页
package_info: str | None = None
mch_id: str | None = None
app_id: str | None = None
class WithdrawOrderOut(BaseModel):
@@ -129,7 +134,7 @@ class WithdrawOrderOut(BaseModel):
id: int
out_bill_no: str
amount_cents: int
status: str
status: str = Field(..., description="reviewing(待审核) / pending / success / failed / rejected")
wechat_state: str | None = None
fail_reason: str | None = None
created_at: datetime
+44 -6
View File
@@ -120,27 +120,65 @@ def test_callback_bad_sign(client) -> None:
assert _coin_balance(client, token) == 0
def test_daily_cap(client) -> None:
"""达到每日上限后继续回调 → 受理但不发(capped),余额封顶。"""
def test_daily_count_cap(client, monkeypatch) -> None:
"""达到每日发奖**次数兜底上限**后继续回调 → 受理但不发(capped),余额封顶。
次数上限现为时长闸的兜底(默认 200),这里 monkeypatch 调小到 3 加速(不真跑 200 );
本用例不上报观看时长 时长主闸不触发,纯验次数兜底
"""
# 次数兜底现走 app_config getter(rebase 合 main 的运营可配后);patch getter 调小到 3
monkeypatch.setattr("app.core.rewards.get_ad_daily_limit", lambda db: 3)
phone = "13800003004"
token = _login(client, phone)
uid = _user_id(phone)
for i in range(DAILY_AD_REWARD_LIMIT):
for i in range(3):
r = _callback(client, _signed(uid, f"trans_cap_{i}"))
assert r.status_code == 200, r.text
# 第 limit+1 次:capped,仍 is_valid不让穿山甲重试),但不加币
# 第 4 次:capped,仍 is_verify(不让穿山甲重试),但不加币
r = _callback(client, _signed(uid, "trans_cap_over"))
assert r.status_code == 200
assert r.json() == {"is_verify": True, "reason": 0}
assert _coin_balance(client, token) == DAILY_AD_REWARD_LIMIT * AD_REWARD_COIN
assert _coin_balance(client, token) == 3 * AD_REWARD_COIN
st = client.get("/api/v1/ad/reward-status", headers=_auth(token)).json()
assert st["used_today"] == DAILY_AD_REWARD_LIMIT
assert st["used_today"] == 3
assert st["remaining"] == 0
def test_daily_watch_time_cap(client, monkeypatch) -> None:
"""看广告累计观看时长达每日上限(主闸,这里调小到 100s)后,再回调发奖 → capped 不发金币。
时长由前端 watch-report 上报累计;到顶后 watch-report/reward-status remaining=0(客户端不再展示),
后端发奖也因时长闸记 capped(双闸,前端绕不过)次数兜底不动它(此时只看了 1 ,远没到次数上限)
"""
# 时长上限两处引用都要 patch:api 用 rewards.X、grant 闸/today_status 用 ad_reward 内绑定的 X
monkeypatch.setattr("app.core.rewards.DAILY_AD_WATCH_SECONDS_LIMIT", 100)
monkeypatch.setattr("app.repositories.ad_reward.DAILY_AD_WATCH_SECONDS_LIMIT", 100)
phone = "13800003201"
token = _login(client, phone)
uid = _user_id(phone)
# 上报一次 100s 观看 → 当日累计达上限,remaining=0
r = client.post("/api/v1/ad/watch-report", json={"seconds": 100}, headers=_auth(token))
assert r.status_code == 200, r.text
assert r.json()["watched_seconds_today"] == 100
assert r.json()["watch_seconds_remaining"] == 0
# 此时回调发奖 → 时长闸命中 → capped(is_verify 仍 true,不让重试),不加金币
before = _coin_balance(client, token)
r = _callback(client, _signed(uid, "trans_time_cap"))
assert r.status_code == 200
assert r.json() == {"is_verify": True, "reason": 0}
assert _coin_balance(client, token) == before # 未加币
st = client.get("/api/v1/ad/reward-status", headers=_auth(token)).json()
assert st["watched_seconds_today"] == 100
assert st["watch_seconds_limit"] == 100
assert st["watch_seconds_remaining"] == 0
def test_callback_unknown_user(client) -> None:
"""验签过但 user_id 不存在 → 不发奖(is_verify false + reason),不崩。"""
params = _signed(999999, "trans_ghost")
+130
View File
@@ -0,0 +1,130 @@
"""Admin M5 配置后台化测试:列出/改配置 + 改配**真生效** + 校验 + 审计。
autouse 清理每个用例后清空 app_config,避免改配污染其他文件的福利测试
"""
from __future__ import annotations
from collections.abc import Iterator
import pytest
from fastapi.testclient import TestClient
from sqlalchemy import delete, select
from app.admin.main import admin_app
from app.admin.repositories import admin_user as admin_repo
from app.db.session import SessionLocal
from app.models.admin import AdminAuditLog
from app.models.app_config import AppConfig
from app.repositories import ad_reward, signin
from app.repositories import user as user_repo
@pytest.fixture()
def admin_client() -> TestClient:
return TestClient(admin_app)
@pytest.fixture()
def token() -> str:
db = SessionLocal()
try:
if admin_repo.get_by_username(db, "cfg_admin") is None:
admin_repo.create_admin(
db, username="cfg_admin", password="cfgpass12", role="super_admin"
)
finally:
db.close()
c = TestClient(admin_app)
return c.post(
"/admin/api/auth/login", json={"username": "cfg_admin", "password": "cfgpass12"}
).json()["access_token"]
@pytest.fixture(autouse=True)
def _clean_config() -> Iterator[None]:
"""每个用例后清空 app_config,避免改配污染其他文件的福利测试(它们假设默认值)。"""
yield
db = SessionLocal()
try:
db.execute(delete(AppConfig))
db.commit()
finally:
db.close()
def _auth(t: str) -> dict:
return {"Authorization": f"Bearer {t}"}
def _seed_user(phone: str) -> int:
db = SessionLocal()
try:
return user_repo.upsert_user_for_login(db, phone=phone, register_channel="sms").id
finally:
db.close()
def test_list_config(admin_client: TestClient, token: str) -> None:
r = admin_client.get("/admin/api/config", headers=_auth(token))
assert r.status_code == 200, r.text
items = {i["key"]: i for i in r.json()}
assert "signin_rewards" in items and "ad_daily_limit" in items
assert items["signin_rewards"]["value"] == [10, 20, 30, 50, 80, 120, 200]
assert items["signin_rewards"]["overridden"] is False
def test_update_signin_takes_effect(admin_client: TestClient, token: str) -> None:
r = admin_client.patch(
"/admin/api/config/signin_rewards",
json={"value": [100, 200, 300, 400, 500, 600, 700]},
headers=_auth(token),
)
assert r.status_code == 200, r.text
assert r.json()["value"][0] == 100 and r.json()["overridden"] is True
# 业务真的用上新值(配置后台化的核心验证)
uid = _seed_user("13912340001")
db = SessionLocal()
try:
st = signin.get_status(db, uid)
assert st.steps[0].coin == 100
assert st.today_coin == 100 # 首签=第 1 天=100(新值)
logs = db.execute(
select(AdminAuditLog).where(
AdminAuditLog.action == "config.set",
AdminAuditLog.target_id == "signin_rewards",
)
).scalars().all()
assert len(logs) == 1 and logs[0].detail["after"][0] == 100
finally:
db.close()
def test_update_ad_limit_takes_effect(admin_client: TestClient, token: str) -> None:
r = admin_client.patch(
"/admin/api/config/ad_daily_limit", json={"value": 5}, headers=_auth(token)
)
assert r.status_code == 200, r.text
uid = _seed_user("13912340002")
db = SessionLocal()
try:
# today_status 现返 7 元组(末两位为观看时长闸:已看秒数 / 上限),取次数上限位
_used, limit, _coin, _rc, _cd, _ws, _wl = ad_reward.today_status(db, uid)
assert limit == 5 # 新配置生效
finally:
db.close()
def test_config_validation(admin_client: TestClient, token: str) -> None:
# 签到档位长度≠7
assert admin_client.patch(
"/admin/api/config/signin_rewards", json={"value": [1, 2, 3]}, headers=_auth(token)
).status_code == 400
# int 负数
assert admin_client.patch(
"/admin/api/config/ad_daily_limit", json={"value": -5}, headers=_auth(token)
).status_code == 400
# 未知 key
assert admin_client.patch(
"/admin/api/config/nope", json={"value": 1}, headers=_auth(token)
).status_code == 404
+79 -10
View File
@@ -10,6 +10,7 @@ from sqlalchemy import select
from app.db.session import SessionLocal
from app.models.user import User
from app.models.wallet import CoinAccount, WithdrawOrder
from app.repositories import wallet as crud_wallet
def _login(client, phone: str) -> str:
@@ -43,6 +44,24 @@ def _patch_userinfo(monkeypatch, openid="openid_test_abc", nickname="测试昵
)
def _approve(bill: str) -> None:
"""模拟管理员审核通过(走 repo,等价 admin 端点 approve → execute_withdraw_transfer 发起转账)。"""
db = SessionLocal()
try:
crud_wallet.approve_withdraw(db, bill)
finally:
db.close()
def _reject(bill: str, reason: str = "不符合提现规则") -> None:
"""模拟管理员审核拒绝(走 repo,等价 admin 端点 reject → 退款 + rejected)。"""
db = SessionLocal()
try:
crud_wallet.reject_withdraw(db, bill, reason)
finally:
db.close()
def test_bind_wechat(client, monkeypatch) -> None:
_patch_userinfo(monkeypatch)
token = _login(client, "13800002001")
@@ -64,7 +83,8 @@ def test_bind_wechat(client, monkeypatch) -> None:
assert j["wechat_nickname"] == "测试昵称"
def test_withdraw_success_flow(client, monkeypatch) -> None:
def test_withdraw_submit_then_approve_success_flow(client, monkeypatch) -> None:
"""提现先进 reviewing(扣款不打款) → 管理员审核通过才发起转账 → 查单 SUCCESS → success。"""
monkeypatch.setattr("app.integrations.wxpay.code_to_userinfo", lambda code: {"openid": "openid_ok", "nickname": None, "avatar_url": None, "raw": {}})
monkeypatch.setattr(
"app.integrations.wxpay.create_transfer",
@@ -77,13 +97,12 @@ def test_withdraw_success_flow(client, monkeypatch) -> None:
_seed_cash(client, token, "13800002002", 100)
client.post("/api/v1/wallet/bind-wechat", json={"code": "c"}, headers=_auth(token))
# 发起提现 50 分
# 发起提现 50 分 → 进入待审核(reviewing),已扣款但未打款,无 package_info
r = client.post("/api/v1/wallet/withdraw", json={"amount_cents": 50}, headers=_auth(token))
assert r.status_code == 200, r.text
body = r.json()
assert body["status"] == "pending"
assert body["package_info"] == "pkg123"
assert body["mch_id"] and body["app_id"]
assert body["status"] == "reviewing"
assert body["package_info"] is None
assert body["cash_balance_cents"] == 50 # 已扣
bill = body["out_bill_no"]
@@ -92,7 +111,10 @@ def test_withdraw_success_flow(client, monkeypatch) -> None:
txns = r.json()["items"]
assert any(t["biz_type"] == "withdraw" and t["amount_cents"] == -50 for t in txns)
# 查单 → 微信返回 SUCCESS → 归一化 success
# 管理员审核通过 → 发起微信转账(mock WAIT_USER_CONFIRM,单进 pending)
_approve(bill)
# 用户确认到账后查单 → SUCCESS → 归一化 success
monkeypatch.setattr(
"app.integrations.wxpay.query_transfer",
lambda out_bill_no: {"status_code": 200, "data": {"state": "SUCCESS"}},
@@ -123,8 +145,12 @@ def test_withdraw_abandoned_confirm_cancels_and_refunds(client, monkeypatch) ->
r = client.post("/api/v1/wallet/withdraw", json={"amount_cents": 50}, headers=_auth(token))
bill = r.json()["out_bill_no"]
assert r.json()["status"] == "reviewing"
assert r.json()["cash_balance_cents"] == 50 # 已扣
# 管理员审核通过 → 发起转账(mock WAIT_USER_CONFIRM,单进 pending)
_approve(bill)
# 回到 app 查单:微信仍是 WAIT_USER_CONFIRM(没确认) + 撤单成功
monkeypatch.setattr(
"app.integrations.wxpay.query_transfer",
@@ -166,7 +192,8 @@ def test_withdraw_not_bound(client) -> None:
assert r.status_code == 400, r.text
def test_withdraw_transfer_fail_refunds(client, monkeypatch) -> None:
def test_withdraw_approve_transfer_fail_refunds(client, monkeypatch) -> None:
"""审核通过发起转账失败(非200) + 查单 NOT_FOUND(未创建) → 自动退款 + 单 failed。"""
monkeypatch.setattr("app.integrations.wxpay.code_to_userinfo", lambda code: {"openid": "openid_fail", "nickname": None, "avatar_url": None, "raw": {}})
monkeypatch.setattr(
"app.integrations.wxpay.create_transfer",
@@ -184,8 +211,14 @@ def test_withdraw_transfer_fail_refunds(client, monkeypatch) -> None:
_seed_cash(client, token, "13800002005", 100)
client.post("/api/v1/wallet/bind-wechat", json={"code": "c"}, headers=_auth(token))
# 发起提现 → reviewing(已扣款),此刻不打款不会失败
r = client.post("/api/v1/wallet/withdraw", json={"amount_cents": 50}, headers=_auth(token))
assert r.status_code == 502, r.text
assert r.status_code == 200, r.text
assert r.json()["status"] == "reviewing"
bill = r.json()["out_bill_no"]
# 管理员审核通过 → 转账失败 → 自动退款 + 单 failed
_approve(bill)
# 余额已退回
r = client.get("/api/v1/wallet/account", headers=_auth(token))
@@ -222,7 +255,9 @@ def test_withdraw_idempotent_same_bill_no(client, monkeypatch) -> None:
r2 = client.post("/api/v1/wallet/withdraw", json={"amount_cents": 50, "out_bill_no": bill}, headers=_auth(token))
assert r2.status_code == 200, r2.text
assert calls["n"] == 1 # 只真发起一次转账
# 提现阶段不打款 → create_transfer 一次都不会被调;两次都返回同一张待审核单
assert calls["n"] == 0
assert r1.json()["status"] == "reviewing" and r2.json()["status"] == "reviewing"
assert r1.json()["out_bill_no"] == bill
assert r2.json()["out_bill_no"] == bill
# 余额只扣一次(100-50=50)
@@ -249,7 +284,13 @@ def test_withdraw_ambiguous_timeout_then_success_no_refund(client, monkeypatch)
r = client.post("/api/v1/wallet/withdraw", json={"amount_cents": 50}, headers=_auth(token))
assert r.status_code == 200, r.text
assert r.json()["status"] == "success"
assert r.json()["status"] == "reviewing"
bill = r.json()["out_bill_no"]
# 审核通过 → 转账调用超时(异常) → 查单确认 SUCCESS → 不退款,单 success
_approve(bill)
r = client.get("/api/v1/wallet/withdraw-orders", headers=_auth(token))
assert r.json()["items"][0]["status"] == "success"
# 没退款:余额仍是扣后的 50,且无 withdraw_refund
r = client.get("/api/v1/wallet/account", headers=_auth(token))
assert r.json()["cash_balance_cents"] == 50
@@ -269,3 +310,31 @@ def test_bind_rejects_openid_already_bound(client, monkeypatch) -> None:
token_b = _login(client, "13800002010")
r = client.post("/api/v1/wallet/bind-wechat", json={"code": "cb"}, headers=_auth(token_b))
assert r.status_code == 409, r.text
def test_withdraw_reject_refunds(client, monkeypatch) -> None:
"""管理员审核拒绝 → 退回现金 + 单 rejected + 理由写入 fail_reason(用户可见)。"""
monkeypatch.setattr("app.integrations.wxpay.code_to_userinfo", lambda code: {"openid": "openid_reject", "nickname": None, "avatar_url": None, "raw": {}})
token = _login(client, "13800002011")
_seed_cash(client, token, "13800002011", 100)
client.post("/api/v1/wallet/bind-wechat", json={"code": "c"}, headers=_auth(token))
# 发起提现 → reviewing(已扣款,余额 50)
r = client.post("/api/v1/wallet/withdraw", json={"amount_cents": 50}, headers=_auth(token))
assert r.json()["status"] == "reviewing"
assert r.json()["cash_balance_cents"] == 50
bill = r.json()["out_bill_no"]
# 管理员拒绝 → 退款 + rejected(不调微信,无需 mock 转账)
_reject(bill, "测试拒绝")
# 余额退回 100
r = client.get("/api/v1/wallet/account", headers=_auth(token))
assert r.json()["cash_balance_cents"] == 100
# 有退款流水
r = client.get("/api/v1/wallet/cash-transactions", headers=_auth(token))
assert any(t["biz_type"] == "withdraw_refund" and t["amount_cents"] == 50 for t in r.json()["items"])
# 单 rejected + 理由透出
r = client.get("/api/v1/wallet/withdraw/status", params={"out_bill_no": bill}, headers=_auth(token))
assert r.json()["status"] == "rejected"
assert r.json()["fail_reason"] == "测试拒绝"