Compare commits

..

5 Commits

Author SHA1 Message Date
unknown c72e052528 修复:提现详情兼容旧数据库字段 2026-07-24 11:59:48 +08:00
guke f7a7a49281 fix(withdraw): 免确认授权已开启判定加 authorization_id 非空,与打款一致 (#168)
免确认授权已开启判定加 authorization_id 非空,与打款一致

---------

Co-authored-by: guke <guke@autohome.com.cn>
Reviewed-on: #168
2026-07-24 11:17:09 +08:00
linkeyu 71aef455f4 修复:签到膨胀金币归入看视频分类 (#169)
## 修改内容

- 将历史 `signin_boost` 金币从“常规任务金币”分类移出。
- 将 `signin_boost` 与 `reward_video/ad_reward` 一起计入“看视频金币”。
- 保留独立的 `signin_boost_coin_total` 历史审计字段。
- 增加不重不漏回归测试,确认分类调整前后本期发放总额保持不变。

## 本地验证

- `tests/test_admin_read.py`:16 项通过。
- Ruff 与 `git diff --check`:通过。
- 全量后端测试:505 项通过;8 项为 `main` 现有无关失败。

---------

Co-authored-by: unknown <798648091@qq.com>
Reviewed-on: #169
Co-authored-by: linkeyu <linkeyu@wonderable.ai>
Co-committed-by: linkeyu <linkeyu@wonderable.ai>
2026-07-24 11:16:38 +08:00
linkeyu 66527f6cdc 修复:领券单券成功率按任务独立统计 (#166)
## 修改内容

- 新增按 `trace_id + coupon_id` 幂等的逐次单券事件表。
- 保留每日资产记录,同时独立保存每次领券事件,避免同设备同日重跑串场。
- 后台逐场成功率和单券明细改为读取逐次事件。
- 增加历史数据回填迁移、本地 mock 脚本和回归测试。

## 验证结果

- 领券相关测试:22 项通过。
- 全新 SQLite 数据库执行 `alembic upgrade head`:通过。
- 全量测试:508 项通过;另外 8 项为现有无关失败。

---------

Co-authored-by: unknown <798648091@qq.com>
Reviewed-on: #166
Co-authored-by: linkeyu <linkeyu@wonderable.ai>
Co-committed-by: linkeyu <linkeyu@wonderable.ai>
2026-07-24 11:15:09 +08:00
linkeyu b4a2a8c31d 功能:限制每位用户每天最多发起 100 次比价 (#165)
## 变更内容

- 登录用户按北京时间自然日计算比价发起次数,每人每天最多 100 次。
- 第 101 次起返回 HTTP 429,并提示“今日已比价超过100次,请明天再试”。
- 同一 trace_id 的网络重试按幂等处理,不会重复计数。
- 使用用户行锁串行化同一账号的并发请求,避免并发突破上限。
- 复用现有 comparison_record 的 running 记录,无需新增数据库迁移。

## 本地验证

- 比价额度专项测试 11 项通过。
- 本次修改涉及文件的 Ruff 检查通过。
- 已覆盖未登录、重复 trace_id、跨自然日、第 100 次放行及第 101 次拒绝。

---------

Co-authored-by: CodexSandboxOffline <798648091@qq.com>
Reviewed-on: #165
Co-authored-by: linkeyu <linkeyu@wonderable.ai>
Co-committed-by: linkeyu <linkeyu@wonderable.ai>
2026-07-24 11:14:31 +08:00
25 changed files with 1284 additions and 57 deletions
+93
View File
@@ -0,0 +1,93 @@
"""add per-session coupon claim event table
Revision ID: coupon_claim_event
Revises: 8e04cc13a211
Create Date: 2026-07-23
"""
from collections.abc import Sequence
import sqlalchemy as sa
from sqlalchemy.dialects import postgresql
from alembic import op
revision: str = "coupon_claim_event"
down_revision: str | Sequence[str] | None = "8e04cc13a211"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
_JSON = sa.JSON().with_variant(postgresql.JSONB(), "postgresql")
def upgrade() -> None:
op.create_table(
"coupon_claim_event",
sa.Column("id", sa.Integer(), autoincrement=True, nullable=False),
sa.Column("trace_id", sa.String(length=64), 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("app_env", sa.String(length=16), nullable=True),
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("reason", sa.String(length=255), nullable=True),
sa.Column("extra", _JSON, nullable=True),
sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
sa.PrimaryKeyConstraint("id"),
sa.UniqueConstraint(
"trace_id", "coupon_id",
name="uq_coupon_claim_event_trace_coupon",
),
)
op.create_index(
"ix_coupon_claim_event_date_env",
"coupon_claim_event",
["claim_date", "app_env"],
unique=False,
)
op.create_index(
op.f("ix_coupon_claim_event_app_env"),
"coupon_claim_event",
["app_env"],
unique=False,
)
op.create_index(
op.f("ix_coupon_claim_event_trace_id"),
"coupon_claim_event",
["trace_id"],
unique=False,
)
op.create_index(
op.f("ix_coupon_claim_event_user_id"),
"coupon_claim_event",
["user_id"],
unique=False,
)
# 旧表只能回填当前仍保留的 trace;历史上已被每日去重覆盖的关联无法恢复。
op.execute(
"""
INSERT INTO coupon_claim_event (
trace_id, device_id, user_id, coupon_id, claim_date, status, app_env,
vendor, coupon_name, claimed_count, reason, extra, created_at, updated_at
)
SELECT
trace_id, device_id, user_id, coupon_id, claim_date, status, app_env,
vendor, coupon_name, claimed_count, reason, extra, created_at, updated_at
FROM coupon_claim_record
WHERE trace_id IS NOT NULL
"""
)
def downgrade() -> None:
op.drop_index(op.f("ix_coupon_claim_event_user_id"), table_name="coupon_claim_event")
op.drop_index(op.f("ix_coupon_claim_event_trace_id"), table_name="coupon_claim_event")
op.drop_index(op.f("ix_coupon_claim_event_app_env"), table_name="coupon_claim_event")
op.drop_index("ix_coupon_claim_event_date_env", table_name="coupon_claim_event")
op.drop_table("coupon_claim_event")
+12 -12
View File
@@ -16,7 +16,7 @@ from sqlalchemy import case, func, or_, select
from sqlalchemy.orm import Session
from app.core import rewards
from app.models.coupon_state import CouponClaimRecord, CouponSession
from app.models.coupon_state import CouponClaimEvent, CouponClaimRecord, CouponSession
from app.models.user import User
from app.repositories import ad_ecpm as crud_ecpm
from app.repositories.coupon_state import DEFAULT_PLATFORMS, coupon_id_to_platform
@@ -193,18 +193,18 @@ def _point_scores_by_trace(db: Session, trace_ids: list[str]) -> dict[str, dict[
"""聚合查询批量返回逐场点位分数,不加载逐券明细。"""
if not trace_ids:
return {}
succeeded = func.sum(case((CouponClaimRecord.status.in_(_SLOT_OK), 1), else_=0))
succeeded = func.sum(case((CouponClaimEvent.status.in_(_SLOT_OK), 1), else_=0))
rows = db.execute(
select(
CouponClaimRecord.trace_id,
CouponClaimEvent.trace_id,
succeeded.label("succeeded"),
func.count().label("tried"),
)
.where(
CouponClaimRecord.trace_id.in_(trace_ids),
CouponClaimRecord.status.in_(_SLOT_TRIED),
CouponClaimEvent.trace_id.in_(trace_ids),
CouponClaimEvent.status.in_(_SLOT_TRIED),
)
.group_by(CouponClaimRecord.trace_id)
.group_by(CouponClaimEvent.trace_id)
).all()
return {
trace_id: {"succeeded": int(success_count or 0), "tried": int(tried or 0)}
@@ -217,13 +217,13 @@ def coupon_point_details(db: Session, *, trace_id: str) -> list[dict]:
"""按单个 trace 查询逐券结果;仅在后台用户点击分数时调用。"""
rows = db.execute(
select(
CouponClaimRecord.coupon_id,
CouponClaimRecord.coupon_name,
CouponClaimRecord.status,
CouponClaimRecord.reason,
CouponClaimEvent.coupon_id,
CouponClaimEvent.coupon_name,
CouponClaimEvent.status,
CouponClaimEvent.reason,
)
.where(CouponClaimRecord.trace_id == trace_id)
.order_by(CouponClaimRecord.id)
.where(CouponClaimEvent.trace_id == trace_id)
.order_by(CouponClaimEvent.id)
).all()
return [
{
+29 -12
View File
@@ -1171,24 +1171,30 @@ def user_reward_stats(
acc = db.get(CoinAccount, user_id) # 现金余额:当前快照,不随窗口
cash_balance = acc.cash_balance_cents if acc else 0
rv = list(db.execute(
select(AdRewardRecord).where(
# 只投影本统计实际使用的列。避免滚动发布或旧本地库尚未补齐无关新列时,
# SQLAlchemy 因 select(ORM) 自动展开整表字段而让提现详情整体 500。
rv = db.execute(
select(AdRewardRecord.ecpm_raw, AdRewardRecord.coin).where(
AdRewardRecord.user_id == user_id,
AdRewardRecord.reward_scene == "reward_video",
AdRewardRecord.status == "granted",
*_window_conds(AdRewardRecord.created_at, date_from, date_to),
)
).scalars())
).all()
rv_ecpms = [rewards.parse_ecpm_fen(r.ecpm_raw) for r in rv if r.ecpm_raw]
rv_coins = sum(r.coin for r in rv)
feed = list(db.execute(
select(AdFeedRewardRecord).where(
feed = db.execute(
select(
AdFeedRewardRecord.unit_count,
AdFeedRewardRecord.ecpm_raw,
AdFeedRewardRecord.coin,
).where(
AdFeedRewardRecord.user_id == user_id,
AdFeedRewardRecord.status == "granted",
*_window_conds(AdFeedRewardRecord.created_at, date_from, date_to),
)
).scalars())
).all()
feed_ecpms = [rewards.parse_ecpm_fen(f.ecpm_raw) for f in feed if f.ecpm_raw]
feed_coins = sum(f.coin for f in feed)
@@ -1246,8 +1252,14 @@ def user_coin_records(
signin_from = date_from + timedelta(hours=8) if date_from is not None else None
signin_to = date_to + timedelta(hours=8) if date_to is not None else None
# 三类来源都只取页面需要的列,避免无关 ORM 新列造成旧库查询失败。
for rec in db.execute(
select(AdRewardRecord)
select(
AdRewardRecord.reward_scene,
AdRewardRecord.created_at,
AdRewardRecord.ecpm_raw,
AdRewardRecord.coin,
)
.where(
AdRewardRecord.user_id == user_id,
AdRewardRecord.status == "granted",
@@ -1255,7 +1267,7 @@ def user_coin_records(
)
.order_by(AdRewardRecord.created_at.desc())
.limit(fetch)
).scalars():
).all():
is_video = rec.reward_scene == "reward_video"
rows.append({
"source": rec.reward_scene,
@@ -1266,7 +1278,12 @@ def user_coin_records(
})
for rec in db.execute(
select(AdFeedRewardRecord)
select(
AdFeedRewardRecord.feed_scene,
AdFeedRewardRecord.created_at,
AdFeedRewardRecord.ecpm_raw,
AdFeedRewardRecord.coin,
)
.where(
AdFeedRewardRecord.user_id == user_id,
AdFeedRewardRecord.status == "granted",
@@ -1274,7 +1291,7 @@ def user_coin_records(
)
.order_by(AdFeedRewardRecord.created_at.desc())
.limit(fetch)
).scalars():
).all():
rows.append({
"source": "feed",
"source_label": _FEED_SCENE_LABEL.get(rec.feed_scene, "信息流广告"),
@@ -1284,7 +1301,7 @@ def user_coin_records(
})
for rec in db.execute(
select(CoinTransaction)
select(CoinTransaction.created_at, CoinTransaction.amount)
.where(
CoinTransaction.user_id == user_id,
CoinTransaction.biz_type == "signin",
@@ -1292,7 +1309,7 @@ def user_coin_records(
)
.order_by(CoinTransaction.created_at.desc())
.limit(fetch)
).scalars():
).all():
rows.append({
"source": "signin",
"source_label": "签到",
+3 -4
View File
@@ -30,18 +30,17 @@ from app.models.user import User
from app.models.wallet import CoinTransaction, WithdrawOrder
_BEIJING = timezone(timedelta(hours=8))
REWARD_VIDEO_BIZ_TYPES = ("reward_video", "ad_reward")
REWARD_VIDEO_BIZ_TYPES = ("reward_video", "ad_reward", "signin_boost")
# 领券/比价奖励金币的真实来源是信息流广告发奖(ad_feed_reward_record,按 feed_scene 分场景);
# coin_transaction 里只有扁平的 feed_ad_reward、biz_type 不分 coupon/comparison,故这俩桶历史从未
# 被写入,仅留作未来兜底,实际金额在下方按 feed_scene 汇总 ad_feed_reward_record 得出。reward_video/
# ad_reward 是激励视频,单独成桶、不再混进领券奖励(历史误并会把激励视频金币双计进领券)
# 被写入,仅留作未来兜底,实际金额在下方按 feed_scene 汇总 ad_feed_reward_record 得出。
# reward_video/ad_reward 及历史 signin_boost 均归看视频桶,不再混进领券奖励或常规任务
COUPON_REWARD_BIZ_TYPES = ("coupon", "coupon_reward")
COMPARISON_REWARD_BIZ_TYPES = ("comparison", "compare_reward", "comparison_reward")
# 常规任务必须按明确来源相加;不能从全部正向流水反减排除项,否则新增广告/运营
# biz_type 时会在排除清单更新前自动混入该桶。task_ 前缀在查询处单独覆盖现有及未来任务。
REGULAR_TASK_EXACT_BIZ_TYPES = (
"signin",
"signin_boost",
"price_report_reward",
"feedback_reward",
)
+2 -2
View File
@@ -79,10 +79,10 @@ class CouponDataRow(BaseModel):
started_at: datetime = Field(..., description="发起时刻(明细「时间」列)")
claimed_count: int | None = None
point_success_count: int | None = Field(
None, description="本次成功券点位数(success+already_claimed);无逐券埋点为空"
None, description="本次成功券数(success+already_claimed);无逐券事件为空"
)
point_total_count: int | None = Field(
None, description="本次尝试券点位数(success+already_claimed+failed,不含 skipped);无逐券埋点为空"
None, description="本次尝试券数(success+already_claimed+failed,不含 skipped);无逐券事件为空"
)
trace_url: str | None = Field(None, description="pricebot 公网 trace 链接(仅 completed 有);admin 渲染可点链接,无则显示可复制 trace_id")
ad_revenue_yuan: float = Field(
+37
View File
@@ -20,6 +20,8 @@ from app.db.session import SessionLocal
from app.models.comparison import ComparisonRecord
from app.repositories import comparison as crud_compare
from app.schemas.compare_record import (
CompareStartReserveIn,
CompareStartReserveOut,
CompareStatsOut,
ComparisonRecordCreatedOut,
ComparisonRecordDetailOut,
@@ -35,6 +37,41 @@ logger = logging.getLogger("shagua.compare_record")
router = APIRouter(prefix="/api/v1/compare", tags=["compare-record"])
@router.post(
"/start",
response_model=CompareStartReserveOut,
summary="预占一次当日比价发起次数(每人每天最多100次)",
)
def reserve_compare_start(
payload: CompareStartReserveIn,
user: CurrentUser,
db: DbSession,
) -> CompareStartReserveOut:
try:
_, used = crud_compare.reserve_daily_start(
db,
user_id=user.id,
trace_id=payload.trace_id,
business_type=payload.business_type,
device_id=payload.device_id,
)
except crud_compare.DailyCompareStartLimitExceeded:
raise HTTPException(
status_code=status.HTTP_429_TOO_MANY_REQUESTS,
detail="今日已比价超过100次,请明天再试",
) from None
except crud_compare.ComparisonTraceOwnershipError:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail="比价任务标识冲突,请重新发起",
) from None
return CompareStartReserveOut(
limit=crud_compare.DAILY_COMPARE_START_LIMIT,
used=used,
remaining=max(crud_compare.DAILY_COMPARE_START_LIMIT - used, 0),
)
@router.post(
"/record",
response_model=ComparisonRecordCreatedOut,
+2 -2
View File
@@ -81,7 +81,7 @@ def _record_claims_blocking(
device_id: str, user_id: int | None, trace_id: str | None, results: list[dict]
) -> None:
with SessionLocal() as db:
# 取本次 session 环境,给 coupon_claim_record 打 app_env 标(每券成功率表按它过滤;设计 §13)
# 取本次 session 环境,给每日资产和逐次事件同时打环境标
app_env = coupon_repo.session_app_env(db, trace_id)
coupon_repo.record_claims(db, device_id, user_id, trace_id, results, app_env=app_env)
# 顺带把本帧「成功平台」并入 coupon_session.platform_success(admin 领券数据 ②整单/③点位成功率;
@@ -176,7 +176,7 @@ async def coupon_step(
resp_json = resp.json()
# 领券结果沉淀(资产):中间帧 last_coupon_result + done 帧 coupon_results 幂等写库。
# 领券结果沉淀:每日资产 + 逐次事件;中间帧和 done 全量帧均幂等写库。
# 当前只记录、不参与"要不要领"判断(MVP 先不去重)。写库失败不影响返回。
if device_id:
results = _extract_coupon_results(resp_json)
+3 -2
View File
@@ -192,7 +192,7 @@ def withdraw_info(
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"),
transfer_auth_enabled=bool(auth and auth.state == "active" and auth.authorization_id),
tiers=[WithdrawTierOut(**t) for t in crud_wallet.withdraw_tier_states(db, user.id, source)],
)
@@ -335,7 +335,8 @@ def open_transfer_auth(user: CurrentUser, db: DbSession) -> TransferAuthResultOu
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"))
enabled = bool(auth and auth.state == "active" and auth.authorization_id)
return TransferAuthStatusOut(state=state, enabled=enabled)
@router.post(
+51 -2
View File
@@ -22,13 +22,13 @@ from __future__ import annotations
import json
import logging
import os
import shutil
import sys
from contextvars import ContextVar
from datetime import datetime
from logging.handlers import RotatingFileHandler
from pathlib import Path
# 请求级 trace_id:入口(如 compare.py 透传壳)set 之后, 本请求上下文(含 run_in_threadpool
# 拷贝出去的线程)内所有日志自动带上。默认空串 = 非请求上下文(启动/后台 worker)。
trace_id_ctx: ContextVar[str] = ContextVar("trace_id", default="")
@@ -91,6 +91,55 @@ class TextFormatter(logging.Formatter):
return f"{base} trace={tid}" if tid else base
class SafeRotatingFileHandler(RotatingFileHandler):
"""Windows 下不会被外部句柄卡死的 RotatingFileHandler。
stdlib 轮转靠 rename 活动文件(app-server.log .1);Windows 只要有别的句柄(IDE 索引
app.admin.main 第二进程残留 --reload worker杀软扫描)开着它, rename WinError 32,
轮转永久卡死文件停在 maxBytes之后每条日志被丢这里 Windows 改用 copytruncate:把活动
文件拷进备份再通过自己的句柄原地清空, 从不 rename 活动文件, 故外部句柄开着也能转
POSIX(生产 Linux)rename 打开中的文件本就合法, 保留 stdlib 的原子轮转不变
代价:copytruncate 拷贝清空极窄窗口内并发写可能丢几行(仅跨进程;同进程 emit
handler 锁串行, 无此问题)对本地开发日志可接受
"""
def doRollover(self) -> None:
if os.name != "nt":
super().doRollover()
return
if self.stream is None:
self.stream = self._open()
else:
self.stream.flush()
try:
self._copytruncate_backups()
except OSError:
# 备份腾挪是尽力而为:任一备份被占用也绝不能挡住下面的清空, 否则活动文件继续涨、
# 轮转又卡死——那就白改了。
pass
# 通过自己独占的句柄原地清空:不涉及 rename, 外部只读句柄不受影响。
self.stream.seek(0)
self.stream.truncate()
self.stream.flush()
def _copytruncate_backups(self) -> None:
"""把 .N-1→.N 逐级腾挪, 再把活动文件拷到 .1(不动活动文件本身)。"""
if self.backupCount <= 0:
return
for i in range(self.backupCount - 1, 0, -1):
sfn = self.rotation_filename(f"{self.baseFilename}.{i}")
dfn = self.rotation_filename(f"{self.baseFilename}.{i + 1}")
if os.path.exists(sfn):
if os.path.exists(dfn):
os.remove(dfn)
os.replace(sfn, dfn)
dfn = self.rotation_filename(f"{self.baseFilename}.1")
if os.path.exists(dfn):
os.remove(dfn)
shutil.copyfile(self.baseFilename, dfn)
_CONFIGURED = False
@@ -126,7 +175,7 @@ def setup_logging(debug: bool = False) -> None:
Path(os.getenv("LOG_DIR", "logs")) / "app-server.log"
)
Path(log_file).parent.mkdir(parents=True, exist_ok=True)
file_handler = RotatingFileHandler(
file_handler = SafeRotatingFileHandler(
log_file, maxBytes=10 * 1024 * 1024, backupCount=5, encoding="utf-8",
)
file_handler.setFormatter(JsonFormatter(service))
+1
View File
@@ -21,6 +21,7 @@ from app.models.cps_wx_user import CpsWxUser # noqa: F401
from app.models.comparison_milestone import ComparisonMilestoneClaim # noqa: F401
from app.models.device import DeviceLiveness # noqa: F401
from app.models.coupon_state import ( # noqa: F401
CouponClaimEvent,
CouponClaimRecord,
CouponDailyCompletion,
CouponPromptEngagement,
+34
View File
@@ -96,6 +96,40 @@ class CouponClaimRecord(Base):
)
class CouponClaimEvent(Base):
"""一次领券任务中的单券结果,按 ``(trace_id, coupon_id)`` 幂等。"""
__tablename__ = "coupon_claim_event"
__table_args__ = (
UniqueConstraint(
"trace_id", "coupon_id",
name="uq_coupon_claim_event_trace_coupon",
),
Index("ix_coupon_claim_event_date_env", "claim_date", "app_env"),
)
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
trace_id: Mapped[str] = mapped_column(String(64), nullable=False, index=True)
device_id: Mapped[str] = mapped_column(String(64), nullable=False)
user_id: Mapped[int | None] = mapped_column(Integer, index=True, nullable=True)
coupon_id: Mapped[str] = mapped_column(String(64), nullable=False)
claim_date: Mapped[date] = mapped_column(Date, nullable=False)
status: Mapped[str] = mapped_column(String(24), nullable=False)
app_env: Mapped[str | None] = mapped_column(String(16), index=True, nullable=True)
vendor: Mapped[str | None] = mapped_column(String(48), nullable=True)
coupon_name: Mapped[str | None] = mapped_column(String(128), nullable=True)
claimed_count: Mapped[int | None] = mapped_column(Integer, nullable=True)
reason: Mapped[str | None] = mapped_column(String(255), nullable=True)
extra: Mapped[dict | None] = mapped_column(_JSON, nullable=True)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), server_default=func.now(), nullable=False
)
updated_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), server_default=func.now(), onupdate=func.now(),
nullable=False,
)
class CouponDailyCompletion(Base):
"""按 (device, 自然日) 记"今天是否已跑完整轮领券(到 done 帧)"——首页置灰源。
+84 -1
View File
@@ -5,7 +5,7 @@
"""
from __future__ import annotations
from datetime import datetime
from datetime import datetime, timedelta
from sqlalchemy import func, or_, select
from sqlalchemy.orm import Session, defer
@@ -14,8 +14,19 @@ from app.core.rewards import CN_TZ
from app.models.ad_feed_reward import AdFeedRewardRecord
from app.models.comparison import ComparisonRecord
from app.models.savings import SavingsRecord
from app.models.user import User
from app.schemas.compare_record import ComparisonRecordIn
DAILY_COMPARE_START_LIMIT = 100
class DailyCompareStartLimitExceeded(Exception):
"""The authenticated user has consumed today's comparison-start quota."""
class ComparisonTraceOwnershipError(Exception):
"""A trace id already belongs to a different authenticated user."""
def _yuan_to_cents(yuan: float | None) -> int | None:
"""元(float)→ 分(int)。None 透传。"""
@@ -243,6 +254,78 @@ def _get_by_trace(db: Session, trace_id: str) -> ComparisonRecord | None:
).scalar_one_or_none()
def reserve_daily_start(
db: Session,
*,
user_id: int,
trace_id: str,
business_type: str = "food",
device_id: str | None = None,
now: datetime | None = None,
) -> tuple[ComparisonRecord, int]:
"""Atomically reserve one of a user's 100 Beijing-day comparison starts.
``trace_id`` makes client retries idempotent. Locking the user row serializes
concurrent starts for one account, so parallel requests cannot both consume
the final available slot. The reservation is the existing ``running``
comparison row; later result reporting updates that same row.
"""
db.execute(select(User.id).where(User.id == user_id).with_for_update()).scalar_one()
existing = _get_by_trace(db, trace_id)
if existing is not None:
if existing.user_id not in (None, user_id):
raise ComparisonTraceOwnershipError
if existing.user_id is None:
existing.user_id = user_id
if existing.device_id is None and device_id:
existing.device_id = device_id
db.commit()
db.refresh(existing)
existing_at = existing.created_at
if existing_at.tzinfo is not None:
existing_at = existing_at.astimezone(CN_TZ).replace(tzinfo=None)
day_start = existing_at.replace(hour=0, minute=0, second=0, microsecond=0)
day_end = day_start + timedelta(days=1)
used = db.scalar(
select(func.count(ComparisonRecord.id)).where(
ComparisonRecord.user_id == user_id,
ComparisonRecord.created_at >= day_start,
ComparisonRecord.created_at < day_end,
)
) or 0
return existing, int(used)
current = now or datetime.now(CN_TZ)
if current.tzinfo is not None:
current = current.astimezone(CN_TZ).replace(tzinfo=None)
day_start = current.replace(hour=0, minute=0, second=0, microsecond=0)
day_end = day_start + timedelta(days=1)
used = db.scalar(
select(func.count(ComparisonRecord.id)).where(
ComparisonRecord.user_id == user_id,
ComparisonRecord.created_at >= day_start,
ComparisonRecord.created_at < day_end,
)
) or 0
if used >= DAILY_COMPARE_START_LIMIT:
raise DailyCompareStartLimitExceeded
rec = ComparisonRecord(
trace_id=trace_id,
user_id=user_id,
business_type=business_type or "food",
device_id=device_id,
status="running",
created_at=current,
)
db.add(rec)
db.commit()
db.refresh(rec)
return rec, int(used) + 1
def harvest_running(
db: Session,
*,
+39 -2
View File
@@ -14,6 +14,7 @@ from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session
from app.models.coupon_state import (
CouponClaimEvent,
CouponClaimRecord,
CouponDailyCompletion,
CouponPromptEngagement,
@@ -164,11 +165,12 @@ def record_claims(
results: list[dict],
app_env: str | None = None,
) -> int:
"""一批券领取结果幂等写入,返回写入(新增 + 更新)条数
"""一批券领取结果同时写入每日资产表和逐次事件表
results 单项取自 pricebot last_coupon_result / done.coupon_results,识别字段:
coupon_id(必需)/ status(必需)/ name / vendor / reason /(display_count)
(device, coupon_id, 今天) 唯一:重复上报同张券走更新(status 以最后一次为准)
- CouponClaimRecord (device, coupon_id, 今天) 幂等供每日资产口径使用
- CouponClaimEvent (trace_id, coupon_id) 幂等 admin 逐场统计使用
"""
today = today_cn()
written = 0
@@ -208,6 +210,41 @@ def record_claims(
claimed_count=count, trace_id=trace_id, reason=r.get("reason"),
extra=r,
))
if trace_id:
event = db.execute(
select(CouponClaimEvent).where(
CouponClaimEvent.trace_id == trace_id,
CouponClaimEvent.coupon_id == coupon_id,
)
).scalar_one_or_none()
if event is not None:
event.device_id = device_id
event.status = status
event.reason = r.get("reason")
event.vendor = r.get("vendor")
event.coupon_name = r.get("name")
event.extra = r
if user_id is not None:
event.user_id = user_id
if count is not None:
event.claimed_count = count
if app_env is not None:
event.app_env = app_env
else:
db.add(CouponClaimEvent(
trace_id=trace_id,
device_id=device_id,
user_id=user_id,
coupon_id=coupon_id,
claim_date=today,
status=status,
app_env=app_env,
vendor=r.get("vendor"),
coupon_name=r.get("name"),
claimed_count=count,
reason=r.get("reason"),
extra=r,
))
written += 1
if written == 0:
return 0
+14 -1
View File
@@ -13,7 +13,6 @@ from datetime import datetime
from pydantic import BaseModel, ConfigDict, Field, field_validator
# ===== 上报请求 =====
class ComparisonItemIn(BaseModel):
@@ -198,6 +197,20 @@ class ComparisonRecordCreatedOut(BaseModel):
id: int = Field(..., description="写入(或已存在)的记录 id")
class CompareStartReserveIn(BaseModel):
"""Reserve one authenticated comparison start before the agent begins."""
trace_id: str = Field(..., min_length=1, max_length=64)
business_type: str = Field(default="food", min_length=1, max_length=16)
device_id: str | None = Field(default=None, max_length=64)
class CompareStartReserveOut(BaseModel):
limit: int
used: int
remaining: int
class CompareStatsOut(BaseModel):
"""「我的」页省钱战绩卡(比价口径)聚合。"""
+197
View File
@@ -0,0 +1,197 @@
"""生成 admin「领券记录」本地联调数据。
用法:
python scripts/seed_coupon_session_mock.py
脚本只清理 ``mock-coupon-repeat-*`` 前缀的数据并重新生成打开后台领券记录
日期选今天分别切换 prod/dev可验证同一设备同一天多次领券仍各自显示正确分数
"""
from __future__ import annotations
import sys
from datetime import UTC, datetime
from pathlib import Path
from zoneinfo import ZoneInfo
from sqlalchemy import delete, select
ROOT = Path(__file__).resolve().parents[1]
if str(ROOT) not in sys.path:
sys.path.insert(0, str(ROOT))
from app.db.session import SessionLocal, engine # noqa: E402
from app.models.coupon_state import ( # noqa: E402
CouponClaimEvent,
CouponClaimRecord,
CouponSession,
)
from app.models.user import User # noqa: E402
from app.repositories.coupon_state import record_claims, today_cn # noqa: E402
PREFIX = "mock-coupon-repeat-"
CN_TZ = ZoneInfo("Asia/Shanghai")
DEVICE_REPEAT = f"{PREFIX}device"
DEVICE_CONTROL = f"{PREFIX}control-device"
PHONE = "19900009001"
USERNAME = "80000009001"
PLATFORM_ELAPSED = {
"meituan-waimai": 46_800,
"taobao-shanguang": 19_500,
"jd-waimai": 24_000,
}
FIRST_RESULTS = [
{"coupon_id": "mt_mock_1", "name": "美团模拟券1", "vendor": "meituan", "status": "success"},
{"coupon_id": "mt_mock_2", "name": "美团模拟券2", "vendor": "meituan", "status": "success"},
{"coupon_id": "mt_mock_3", "name": "美团模拟券3", "vendor": "meituan", "status": "success"},
{"coupon_id": "mt_mock_4", "name": "美团模拟券4", "vendor": "meituan", "status": "failed", "reason": "模拟失败"},
{"coupon_id": "tb_mock_1", "name": "淘宝模拟券1", "vendor": "taobao", "status": "success"},
{"coupon_id": "tb_mock_2", "name": "淘宝模拟券2", "vendor": "taobao", "status": "success"},
{"coupon_id": "tb_mock_3", "name": "淘宝模拟券3", "vendor": "taobao", "status": "success"},
{"coupon_id": "jd_mock_1", "name": "京东模拟券", "vendor": "jingdong", "status": "success"},
]
SECOND_RESULTS = [
{"coupon_id": "mt_mock_1", "name": "美团模拟券1", "vendor": "meituan", "status": "skipped", "reason": "模拟跳过,不计分母"},
{"coupon_id": "mt_mock_2", "name": "美团模拟券2", "vendor": "meituan", "status": "already_claimed"},
{"coupon_id": "mt_mock_3", "name": "美团模拟券3", "vendor": "meituan", "status": "success"},
{"coupon_id": "mt_mock_4", "name": "美团模拟券4", "vendor": "meituan", "status": "success"},
{"coupon_id": "tb_mock_1", "name": "淘宝模拟券1", "vendor": "taobao", "status": "success"},
{"coupon_id": "tb_mock_2", "name": "淘宝模拟券2", "vendor": "taobao", "status": "success"},
{"coupon_id": "tb_mock_3", "name": "淘宝模拟券3", "vendor": "taobao", "status": "already_claimed"},
{"coupon_id": "jd_mock_1", "name": "京东模拟券", "vendor": "jingdong", "status": "success"},
]
def _started_at(hour: int, minute: int) -> datetime:
local = datetime.combine(today_cn(), datetime.min.time()).replace(
hour=hour, minute=minute, tzinfo=CN_TZ
)
return local.astimezone(UTC)
def _session(
*,
trace_id: str,
device_id: str,
user_id: int,
app_env: str,
hour: int,
minute: int,
status: str = "completed",
elapsed_ms: int | None = 91_900,
platform_elapsed: dict[str, int] | None = None,
) -> CouponSession:
started_at = _started_at(hour, minute)
return CouponSession(
trace_id=trace_id,
device_id=device_id,
user_id=user_id,
status=status,
app_env=app_env,
platforms=[],
origin_package=None,
device_model="Mock Phone",
rom="MockOS 1",
started_at=started_at,
started_date=today_cn(),
finished_at=started_at if status != "started" else None,
elapsed_ms=elapsed_ms,
platform_elapsed=platform_elapsed,
platform_success=(
["meituan-waimai", "taobao-shanguang", "jd-waimai"]
if status == "completed" else None
),
claimed_count=7 if status == "completed" else 0,
)
def main() -> None:
# 本地旧库 Alembic 版本链可能未同步;仅为联调补建新事件表,正式环境仍走 migration。
CouponClaimEvent.__table__.create(bind=engine, checkfirst=True)
with SessionLocal() as db:
db.execute(delete(CouponClaimEvent).where(
CouponClaimEvent.trace_id.startswith(PREFIX)
))
db.execute(delete(CouponClaimRecord).where(
CouponClaimRecord.device_id.startswith(PREFIX)
))
db.execute(delete(CouponSession).where(
CouponSession.trace_id.startswith(PREFIX)
))
user = db.execute(select(User).where(User.phone == PHONE)).scalar_one_or_none()
if user is None:
user = User(
phone=PHONE,
username=USERNAME,
register_channel="sms",
nickname="领券重复测试",
)
db.add(user)
db.flush()
first_trace = f"{PREFIX}dev-first"
second_trace = f"{PREFIX}prod-second"
abandoned_trace = f"{PREFIX}prod-abandoned"
control_trace = f"{PREFIX}prod-control"
db.add_all([
_session(
trace_id=first_trace,
device_id=DEVICE_REPEAT,
user_id=user.id,
app_env="dev",
hour=10,
minute=0,
platform_elapsed=PLATFORM_ELAPSED,
),
_session(
trace_id=second_trace,
device_id=DEVICE_REPEAT,
user_id=user.id,
app_env="prod",
hour=15,
minute=0,
platform_elapsed=PLATFORM_ELAPSED,
),
_session(
trace_id=abandoned_trace,
device_id=DEVICE_REPEAT,
user_id=user.id,
app_env="prod",
hour=16,
minute=0,
status="abandoned",
elapsed_ms=21_500,
platform_elapsed={"meituan-waimai": 20_500},
),
_session(
trace_id=control_trace,
device_id=DEVICE_CONTROL,
user_id=user.id,
app_env="prod",
hour=17,
minute=0,
platform_elapsed=PLATFORM_ELAPSED,
),
])
db.commit()
record_claims(
db, DEVICE_REPEAT, user.id, first_trace, FIRST_RESULTS, app_env="dev"
)
record_claims(
db, DEVICE_REPEAT, user.id, second_trace, SECOND_RESULTS, app_env="prod"
)
record_claims(
db, DEVICE_CONTROL, user.id, control_trace, FIRST_RESULTS, app_env="prod"
)
print(f"已生成 {today_cn()} 的领券 mock 数据。")
print("筛选用户 19900009001。")
print("prod 应有:7/7100.0%)、-、7/887.5%)三条。")
print("dev 应有:7/887.5%)一条。")
if __name__ == "__main__":
main()
+277
View File
@@ -0,0 +1,277 @@
"""一次性 mock:造几条不同状态的用户反馈,供运营后台「反馈工单」页联调验收。
覆盖:
- 三个状态 tab(待审核 pending / 已采纳 adopted / 未采纳 rejected),重点铺待审核;
- 两种反馈类型 source(普通反馈 profile /我的页入口比价反馈 comparison / 比价结果页入口),
比价反馈带问题场景scene(找错商品/优惠不对/比价太慢);
- 提交端环境快照(app_version / device_model / rom_name / android_version)新端反馈才有,
另留 1~2 条历史反馈(env NULLcontact 有值)旧数据展示;
- 截图: data/media/feedback/ 生成真实可加载的纯色 PNG(手写字节,无需 Pillow),
让审核抽屉的图能真加载出来( seed_mock_price_reports 同法)
- 已采纳条带 reward_coins + admin_reply + review_note;未采纳条带 reject_reason + admin_reply
幂等:每次运行先按固定 mock 手机号清掉旧 mock 用户/反馈 + mock 截图再重建仅清理用 --clean-only
.venv\\Scripts\\python.exe scripts\\seed_mock_feedback.py
.venv\\Scripts\\python.exe scripts\\seed_mock_feedback.py --clean-only
看图:前端反馈工单(http://localhost:3001 反馈)图经 NEXT_PUBLIC_MEDIA_BASE
(本地 = http://localhost:8770) App 后端 /media 加载改过 .env.local 后需重启 next dev,
App 后端(:8770)要在跑
"""
from __future__ import annotations
import argparse
import struct
import sys
import zlib
from datetime import UTC, datetime, timedelta
from pathlib import Path
from sqlalchemy import delete, select
from app.core.config import settings
from app.db.session import SessionLocal
from app.models.feedback import Feedback
from app.models.user import User
from app.repositories.user import _gen_unique_username
# Windows GBK 控制台下也能正常打印中文/¥(避免 UnicodeEncodeError)
if hasattr(sys.stdout, "reconfigure"):
sys.stdout.reconfigure(encoding="utf-8")
# 固定 mock 手机号:脚本只动这些号,便于幂等重建 / 清理。
# 刻意与 seed_mock_withdraws / seed_mock_price_reports 的号段错开,互不干扰。
MOCK_PHONES = [
"13255550001",
"13255550002",
"13255550003",
"13255550004",
"13255550005",
]
_FEEDBACK_DIR = Path(settings.MEDIA_ROOT) / "feedback"
_MOCK_IMG_GLOB = "mock_fb_*.png" # 本脚本生成的图前缀,清理时按此删
def _naive_utc_now() -> datetime:
"""与 func.now() 在 SQLite 的口径一致:naive UTC。反馈 created_at 走 server_default=func.now(),
这里显式造数据也用 naive UTC,和真实反馈行同源,前端展示口径一致"""
return datetime.now(UTC).replace(tzinfo=None)
def _solid_png(width: int, height: int, rgb: tuple[int, int, int]) -> bytes:
"""生成一张纯色 PNG(truecolor RGB)的字节,无需 Pillow。颜色块即可肉眼判断「图加载出来了」。"""
def _chunk(typ: bytes, data: bytes) -> bytes:
body = typ + data
return struct.pack(">I", len(data)) + body + struct.pack(">I", zlib.crc32(body) & 0xFFFFFFFF)
ihdr = struct.pack(">IIBBBBB", width, height, 8, 2, 0, 0, 0) # 8bit/通道, color type 2 = RGB
row = b"\x00" + bytes(rgb) * width # 每行前缀 filter byte 0
idat = zlib.compress(row * height, 9)
return b"\x89PNG\r\n\x1a\n" + _chunk(b"IHDR", ihdr) + _chunk(b"IDAT", idat) + _chunk(b"IEND", b"")
def _write_mock_image(name: str, rgb: tuple[int, int, int]) -> str:
"""写一张 mock 截图到 media/feedback/,返回其相对 URL(/media/feedback/<name>)。"""
_FEEDBACK_DIR.mkdir(parents=True, exist_ok=True)
(_FEEDBACK_DIR / name).write_bytes(_solid_png(320, 320, rgb))
return f"{settings.MEDIA_URL_PREFIX}/feedback/{name}"
def clean(db) -> int:
uids = list(db.execute(select(User.id).where(User.phone.in_(MOCK_PHONES))).scalars())
if uids:
db.execute(delete(Feedback).where(Feedback.user_id.in_(uids)))
db.execute(delete(User).where(User.id.in_(uids)))
db.commit()
# 删 mock 截图文件
if _FEEDBACK_DIR.exists():
for f in _FEEDBACK_DIR.glob(_MOCK_IMG_GLOB):
f.unlink(missing_ok=True)
return len(uids)
def seed(db) -> list[Feedback]:
now = _naive_utc_now()
def ago(**kw) -> datetime:
return now - timedelta(**kw)
# 1) 建 5 个 mock 用户(U5 昵称留空测 "-" 展示)
users_spec = [
("13255550001", "反馈小达人"),
("13255550002", "比价挑刺王"),
("13255550003", "热心用户阿明"),
("13255550004", "老用户张姐"),
("13255550005", None),
]
users: dict[str, User] = {}
for phone, nickname in users_spec:
u = User(
phone=phone,
username=_gen_unique_username(db), # 列 NOT NULL + unique,仿真实注册生成
nickname=nickname,
register_channel="sms",
status="active",
created_at=ago(days=15),
last_login_at=ago(hours=1),
)
db.add(u)
users[phone] = u
db.flush() # 拿 user.id
# 2) 生成 mock 截图(不同颜色块,便于肉眼区分「都加载出来了」)
palette = [
(24, 144, 255), # 蓝
(82, 196, 26), # 绿
(250, 173, 20), # 橙
(245, 34, 45), # 红
]
imgs = [_write_mock_image(f"mock_fb_{i}.png", palette[i]) for i in range(len(palette))]
# 3) 造反馈记录
def fb(
phone: str,
content: str,
*,
source: str = "profile",
scene: str | None = None,
images: list[str] | None = None,
contact: str = "",
status: str = "pending",
reject_reason: str | None = None,
reward_coins: int | None = None,
review_note: str | None = None,
admin_reply: str | None = None,
app_version: str | None = None,
device_model: str | None = None,
rom_name: str | None = None,
android_version: str | None = None,
created: datetime,
) -> Feedback:
return Feedback(
user_id=users[phone].id,
content=content,
contact=contact, # 列 NOT NULL:新端存空串,历史数据有值
source=source,
scene=scene,
images=images,
status=status,
reject_reason=reject_reason,
reward_coins=reward_coins,
review_note=review_note,
admin_reply=admin_reply,
app_version=app_version,
device_model=device_model,
rom_name=rom_name,
android_version=android_version,
reviewed_at=(created + timedelta(hours=2)) if status != "pending" else None,
created_at=created,
)
feedbacks = [
# ===== 待审核 pending(默认 tab,重点铺量)=====
# 普通反馈 · 新端(带环境快照)· 无图
fb("13255550001", "签到金币到账有时候会延迟一两分钟,能不能做成实时到账?",
source="profile",
app_version="2.3.1", device_model="PJA110", rom_name="ColorOS", android_version="14",
created=ago(minutes=6)),
# 比价反馈 · scene=优惠不对 · 新端 · 2 图
fb("13255550002", "这家店京东外卖的到手价比你们算出来的最低价还低,截图为证,麻烦核实。",
source="comparison", scene="优惠不对", images=[imgs[0], imgs[1]],
app_version="2.3.1", device_model="M2012K11AC", rom_name="MIUI", android_version="13",
created=ago(minutes=22)),
# 比价反馈 · scene=找错商品 · 新端 · 无图
fb("13255550002", "比价结果里的商品跟我搜的不是同一个规格,数量对不上。",
source="comparison", scene="找错商品",
app_version="2.3.0", device_model="V2309A", rom_name="OriginOS", android_version="14",
created=ago(hours=1)),
# 普通反馈 · 新端 · 1 图(表扬 + 小问题)
fb("13255550003", "提现秒到账,好评!顺手反馈个小 bug:金币记录页偶尔白屏,要退出去重进。",
source="profile", images=[imgs[2]],
app_version="2.3.1", device_model="23078RKD5C", rom_name="MIUI", android_version="14",
created=ago(hours=3)),
# 比价反馈 · scene=比价太慢 · 历史数据(env 全 NULL、contact 有值)
fb("13255550004", "比价转圈太久了,经常要等十几秒才出结果,体验不太好。",
source="comparison", scene="比价太慢", contact="微信 zhangjie_66",
created=ago(days=1, hours=2)),
# 普通反馈 · 历史数据(env 全 NULL、contact 有值)· 无昵称用户
fb("13255550005", "希望能增加支付宝提现,微信零钱用不太习惯。",
source="profile", contact="QQ 100200300",
created=ago(days=1, hours=8)),
# ===== 已采纳 adopted(发金币 + 回复)=====
fb("13255550003", "建议在比价结果页加个「一键复制口令」,分享给家人更方便。",
source="profile", images=[imgs[3]],
status="adopted", reward_coins=2000,
review_note="有效产品建议,已排期到 2.4.0", admin_reply="感谢反馈!该功能已在规划中,金币奖励已发放~",
app_version="2.2.8", device_model="PJA110", rom_name="ColorOS", android_version="13",
created=ago(days=2)),
# ===== 未采纳 rejected(带原因 + 回复)=====
fb("13255550002", "你们算的价格不准,我看到的更便宜。",
source="comparison", scene="价格不准",
status="rejected", reject_reason="截图价格为限时活动价且已过期,不满足「长期可复现更低价」条件,暂不采纳。",
admin_reply="感谢参与,本次未通过,欢迎继续上报有效更低价~",
app_version="2.3.0", device_model="M2012K11AC", rom_name="MIUI", android_version="13",
created=ago(days=3)),
]
db.add_all(feedbacks)
db.commit()
for f in feedbacks:
db.refresh(f)
return feedbacks
def main() -> None:
parser = argparse.ArgumentParser(description="造用户反馈 mock 数据(运营后台反馈工单页联调用)")
parser.add_argument("--clean-only", action="store_true", help="只清理 mock 数据,不重建")
args = parser.parse_args()
db = SessionLocal()
try:
removed = clean(db)
if removed:
print(f"🧹 已清理旧 mock:{removed} 个用户及其反馈 + mock 截图")
if args.clean_only:
print("✅ 仅清理,已完成。")
return
feedbacks = seed(db)
status_label = {"pending": "待审核", "adopted": "已采纳", "rejected": "未采纳"}
source_label = {"profile": "普通反馈", "comparison": "比价反馈"}
by_status: dict[str, list[Feedback]] = {}
for f in feedbacks:
by_status.setdefault(f.status, []).append(f)
print(f"\n✅ 已生成 {len(feedbacks)} 条反馈(截图落 {_FEEDBACK_DIR}),分布:")
for st in ("pending", "adopted", "rejected"):
lst = by_status.get(st, [])
print(f" {status_label[st]:<4} {len(lst)}")
print("\n 明细(#id | 状态 | 类型/场景 | 图 | 内容):")
uid2phone = dict(
db.execute(select(User.id, User.phone).where(User.phone.in_(MOCK_PHONES))).all()
)
for f in feedbacks:
src = source_label.get(f.source, f.source)
scene = f"·{f.scene}" if f.scene else ""
nimg = len(f.images or [])
snippet = f.content[:20] + ("" if len(f.content) > 20 else "")
print(
f" #{f.id} [{status_label.get(f.status, f.status)}] "
f"{src}{scene} {nimg}{uid2phone.get(f.user_id, '?')} {snippet}"
)
print(
"\n👉 打开 http://localhost:3001 → 反馈 查看(默认「待审核」tab)。"
"\n 图加载不出来时排查:① 是否重启过 next dev(读 .env.local 的 NEXT_PUBLIC_MEDIA_BASE)"
" ② App 后端(:8770)是否在跑(它托管 /media)。"
)
finally:
db.close()
if __name__ == "__main__":
main()
+3 -1
View File
@@ -30,6 +30,7 @@ from app.core.config import settings
from app.db.session import SessionLocal
from app.models.price_report import PriceReport
from app.models.user import User
from app.repositories.user import _gen_unique_username
if hasattr(sys.stdout, "reconfigure"):
sys.stdout.reconfigure(encoding="utf-8") # Windows 控制台输出中文/¥
@@ -84,10 +85,11 @@ def seed(db) -> list[PriceReport]:
# 1) 建 3 个 mock 用户
users: dict[str, User] = {}
for i, (phone, nickname) in enumerate(
zip(MOCK_PHONES, ["省钱小王", "比价老李", "薅羊毛阿珍"])
zip(MOCK_PHONES, ["省钱小王", "比价老李", "薅羊毛阿珍"], strict=True)
):
u = User(
phone=phone,
username=_gen_unique_username(db), # 列 NOT NULL + unique,仿真实注册生成
nickname=nickname,
register_channel="sms",
status="active",
+4 -2
View File
@@ -21,7 +21,7 @@ import argparse
import sys
import uuid
from collections import defaultdict
from datetime import datetime, timedelta, timezone
from datetime import UTC, datetime, timedelta
from sqlalchemy import delete, select
@@ -31,6 +31,7 @@ from app.models.ad_feed_reward import AdFeedRewardRecord
from app.models.ad_reward import AdRewardRecord
from app.models.user import User
from app.models.wallet import CashTransaction, CoinAccount, CoinTransaction, WithdrawOrder
from app.repositories.user import _gen_unique_username
# Windows GBK 控制台下也能正常打印中文(避免 UnicodeEncodeError)
if hasattr(sys.stdout, "reconfigure"):
@@ -51,7 +52,7 @@ REMARK = {"exchange_in": "金币兑入", "withdraw": "提现扣款", "withdraw_r
def _naive_utc_now() -> datetime:
"""与 func.now() 在 SQLite 的口径一致:naive UTC。前端按 UTC 解析再转北京时间。"""
return datetime.now(timezone.utc).replace(tzinfo=None)
return datetime.now(UTC).replace(tzinfo=None)
def _mock_transfer_no() -> str:
@@ -96,6 +97,7 @@ def seed(db) -> list[WithdrawOrder]:
for phone, nickname in users_spec:
u = User(
phone=phone,
username=_gen_unique_username(db), # 列 NOT NULL + unique,仿真实注册生成
nickname=nickname,
register_channel="sms",
status="active",
+80 -2
View File
@@ -5,10 +5,11 @@ from datetime import datetime
import pytest
from fastapi.testclient import TestClient
from sqlalchemy import event
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.db.session import SessionLocal, engine
from app.models.comparison import ComparisonRecord
from app.models.feedback import Feedback
from app.models.wallet import CashTransaction, WithdrawOrder
@@ -132,6 +133,35 @@ def test_user_list_and_detail(admin_client: TestClient, admin_token: str) -> Non
assert admin_client.get("/admin/api/users/999999", headers=_auth(admin_token)).status_code == 404
def test_user_reward_detail_does_not_select_unrelated_new_ad_columns(
admin_client: TestClient, admin_token: str
) -> None:
"""旧库缺少无关新列时,提现详情的统计和金币记录仍应可读。"""
uid = _seed_user_with_data("13800000022")
def reject_full_ad_reward_projection(
_conn, _cursor, statement: str, _parameters, _context, _executemany
) -> None:
if "ad_reward_record.boost_round_id" in statement:
raise AssertionError("提现详情不应查询未使用的 boost_round_id")
event.listen(engine, "before_cursor_execute", reject_full_ad_reward_projection)
try:
stats = admin_client.get(
f"/admin/api/users/{uid}/reward-stats", headers=_auth(admin_token)
)
records = admin_client.get(
f"/admin/api/users/{uid}/coin-records",
params={"limit": 10, "cursor": 0},
headers=_auth(admin_token),
)
finally:
event.remove(engine, "before_cursor_execute", reject_full_ad_reward_projection)
assert stats.status_code == 200, stats.text
assert records.status_code == 200, records.text
def test_user_filter_by_status(admin_client: TestClient, admin_token: str) -> None:
_seed_user_with_data("13800000003")
r = admin_client.get("/admin/api/users", params={"status": "active"}, headers=_auth(admin_token))
@@ -468,13 +498,13 @@ def test_period_regular_task_coin_uses_explicit_allowlist(
d = "2021-06-18"
included = {
"signin": 100,
"signin_boost": 200,
"task_enable_notification": 300,
"task_other": 400,
"price_report_reward": 500,
"feedback_reward": 600,
}
excluded = {
"signin_boost": 200,
"feed_ad_reward_coupon": 700,
"feed_ad_reward_comparison": 800,
"feed_ad_reward": 900,
@@ -514,3 +544,51 @@ def test_period_regular_task_coin_uses_explicit_allowlist(
coins = response.json()["period"]["coins"]
assert coins["regular_task_coin_total"] == sum(included.values())
assert coins["task_coin_total"] == 700
assert coins["reward_video_coin_total"] == 1200
def test_period_signin_boost_moves_to_reward_video_without_double_count(
admin_client: TestClient, admin_token: str
) -> None:
"""历史签到膨胀归看视频桶,不再进常规任务;本期发放总额不变且不重复计算。"""
from datetime import datetime
from app.models.wallet import CoinTransaction
d = "2021-06-19"
rows = [
("signin_boost", 200),
("reward_video", 100),
("signin", 50),
]
db = SessionLocal()
try:
uid = user_repo.upsert_user_for_login(
db, phone="13800008805", register_channel="sms"
).id
balance = 0
for index, (biz_type, amount) in enumerate(rows, start=1):
balance += amount
db.add(CoinTransaction(
user_id=uid,
amount=amount,
balance_after=balance,
biz_type=biz_type,
ref_id=f"signin-boost-route-{index}",
created_at=datetime(2021, 6, 19, 12, 0, index),
))
db.commit()
finally:
db.close()
response = admin_client.get(
"/admin/api/stats/overview",
params={"date_from": d, "date_to": d},
headers=_auth(admin_token),
)
assert response.status_code == 200, response.text
coins = response.json()["period"]["coins"]
assert coins["granted_total"] == 350
assert coins["reward_video_coin_total"] == 300
assert coins["regular_task_coin_total"] == 50
assert coins["signin_boost_coin_total"] == 200
+120
View File
@@ -0,0 +1,120 @@
from __future__ import annotations
import time
from datetime import datetime, timedelta
from sqlalchemy import func, select
from app.core.rewards import CN_TZ
from app.core.security import decode_token
from app.db.session import SessionLocal
from app.models.comparison import ComparisonRecord
def _login(client) -> tuple[str, int]:
phone = f"137{int(time.time() * 1000) % 100000000:08d}"
sent = client.post("/api/v1/auth/sms/send", json={"phone": phone})
assert sent.status_code == 200, sent.text
logged_in = client.post(
"/api/v1/auth/sms/login",
json={"phone": phone, "code": "123456"},
)
assert logged_in.status_code == 200, logged_in.text
token = logged_in.json()["access_token"]
return token, int(decode_token(token, expected_type="access")["sub"])
def _headers(token: str) -> dict[str, str]:
return {"Authorization": f"Bearer {token}"}
def test_compare_start_requires_login(client) -> None:
response = client.post(
"/api/v1/compare/start",
json={"trace_id": "quota-no-auth", "business_type": "food"},
)
assert response.status_code == 401
def test_compare_start_is_idempotent_by_trace_id(client) -> None:
token, user_id = _login(client)
payload = {
"trace_id": f"quota-idempotent-{user_id}",
"business_type": "ecom",
"device_id": "quota-device",
}
first = client.post("/api/v1/compare/start", json=payload, headers=_headers(token))
retry = client.post("/api/v1/compare/start", json=payload, headers=_headers(token))
assert first.status_code == 200, first.text
assert first.json() == {"limit": 100, "used": 1, "remaining": 99}
assert retry.status_code == 200, retry.text
assert retry.json() == first.json()
with SessionLocal() as db:
count = db.scalar(
select(func.count(ComparisonRecord.id)).where(
ComparisonRecord.trace_id == payload["trace_id"]
)
)
record = db.execute(
select(ComparisonRecord).where(
ComparisonRecord.trace_id == payload["trace_id"]
)
).scalar_one()
assert count == 1
assert record.user_id == user_id
assert record.status == "running"
assert record.business_type == "ecom"
assert record.device_id == "quota-device"
def test_compare_start_rejects_101st_beijing_day_attempt(client) -> None:
token, user_id = _login(client)
now = datetime.now(CN_TZ).replace(tzinfo=None)
with SessionLocal() as db:
db.add_all(
[
ComparisonRecord(
user_id=user_id,
trace_id=f"quota-full-{user_id}-{index}",
status="failed",
created_at=now,
)
for index in range(99)
]
)
db.add(
ComparisonRecord(
user_id=user_id,
trace_id=f"quota-yesterday-{user_id}",
status="success",
created_at=now - timedelta(days=1),
)
)
db.commit()
final_allowed_trace = f"quota-final-allowed-{user_id}"
allowed = client.post(
"/api/v1/compare/start",
json={"trace_id": final_allowed_trace, "business_type": "food"},
headers=_headers(token),
)
assert allowed.status_code == 200, allowed.text
assert allowed.json() == {"limit": 100, "used": 100, "remaining": 0}
rejected_trace = f"quota-rejected-{user_id}"
response = client.post(
"/api/v1/compare/start",
json={"trace_id": rejected_trace, "business_type": "food"},
headers=_headers(token),
)
assert response.status_code == 429
assert response.json()["detail"] == "今日已比价超过100次,请明天再试"
with SessionLocal() as db:
assert db.scalar(
select(func.count(ComparisonRecord.id)).where(
ComparisonRecord.trace_id == rejected_trace
)
) == 0
+54
View File
@@ -0,0 +1,54 @@
"""逐次单券事件不能被同设备同日的每日去重记录串场。"""
from sqlalchemy import delete, select
from app.admin.repositories.coupon_data import _point_scores_by_trace, coupon_point_details
from app.db.session import SessionLocal
from app.models.coupon_state import CouponClaimEvent, CouponClaimRecord
from app.repositories.coupon_state import record_claims
def test_same_device_same_day_keeps_scores_for_each_trace() -> None:
db = SessionLocal()
device = "event-repeat-device"
first_trace = "event-repeat-first"
second_trace = "event-repeat-second"
first_results = [
{"coupon_id": "mt-repeat", "name": "美团测试券", "status": "success"},
{"coupon_id": "tb-repeat", "name": "淘宝测试券", "status": "failed"},
]
second_results = [
{"coupon_id": "mt-repeat", "name": "美团测试券", "status": "already_claimed"},
{"coupon_id": "tb-repeat", "name": "淘宝测试券", "status": "success"},
]
try:
record_claims(
db, device, None, first_trace, first_results, app_env="dev"
)
record_claims(
db, device, None, second_trace, second_results, app_env="prod"
)
assets = db.execute(
select(CouponClaimRecord).where(CouponClaimRecord.device_id == device)
).scalars().all()
assert len(assets) == 2
assert {row.trace_id for row in assets} == {first_trace}
events = db.execute(
select(CouponClaimEvent).where(CouponClaimEvent.device_id == device)
).scalars().all()
assert len(events) == 4
assert {row.trace_id for row in events} == {first_trace, second_trace}
scores = _point_scores_by_trace(db, [first_trace, second_trace])
assert scores[first_trace] == {"succeeded": 1, "tried": 2}
assert scores[second_trace] == {"succeeded": 2, "tried": 2}
assert [row["status"] for row in coupon_point_details(
db, trace_id=second_trace
)] == ["already_claimed", "success"]
finally:
db.execute(delete(CouponClaimEvent).where(CouponClaimEvent.device_id == device))
db.execute(delete(CouponClaimRecord).where(CouponClaimRecord.device_id == device))
db.commit()
db.close()
+10 -10
View File
@@ -13,7 +13,7 @@ from app.admin.repositories.coupon_data import (
)
from app.admin.security import create_admin_token
from app.db.session import SessionLocal
from app.models.coupon_state import CouponClaimRecord, CouponSession
from app.models.coupon_state import CouponClaimEvent, CouponSession
def test_point_scores_by_trace() -> None:
@@ -22,14 +22,14 @@ def test_point_scores_by_trace() -> None:
trace = "point-score-trace"
try:
db.add_all([
CouponClaimRecord(
CouponClaimEvent(
trace_id=trace,
device_id="score-device",
coupon_id=f"mt-score-{status}",
claim_date=date(2020, 1, 2),
status=status,
coupon_name=f"测试点位-{status}",
reason="测试失败" if status == "failed" else None,
trace_id=trace,
)
for status in ("success", "already_claimed", "failed", "skipped")
])
@@ -53,12 +53,12 @@ def test_skipped_detail_does_not_create_a_score() -> None:
db = SessionLocal()
trace = "point-score-skipped"
try:
db.add(CouponClaimRecord(
db.add(CouponClaimEvent(
trace_id=trace,
device_id="score-device-skipped",
coupon_id="mt-score-skipped-only",
claim_date=date(2020, 1, 2),
status="skipped",
trace_id=trace,
))
db.flush()
@@ -87,12 +87,12 @@ def test_coupon_data_report_returns_scores_without_embedding_details() -> None:
started_date=report_date,
))
db.add_all([
CouponClaimRecord(
CouponClaimEvent(
trace_id=trace,
device_id="score-report-device",
coupon_id=f"mt-report-{status}",
claim_date=report_date,
status=status,
trace_id=trace,
)
for status in ("success", "failed")
])
@@ -128,14 +128,14 @@ def test_coupon_point_details_endpoint() -> None:
role="super_admin",
)
token, _expires_at = create_admin_token(admin_id=admin.id, role=admin.role)
db.add(CouponClaimRecord(
db.add(CouponClaimEvent(
trace_id=trace,
device_id="point-details-endpoint-device",
coupon_id="mt-point-details-endpoint",
coupon_name="接口测试券",
claim_date=date(2020, 1, 5),
status="failed",
reason="接口测试失败",
trace_id=trace,
))
db.commit()
@@ -156,6 +156,6 @@ def test_coupon_point_details_endpoint() -> None:
}
finally:
db.rollback()
db.execute(delete(CouponClaimRecord).where(CouponClaimRecord.trace_id == trace))
db.execute(delete(CouponClaimEvent).where(CouponClaimEvent.trace_id == trace))
db.commit()
db.close()
+2 -1
View File
@@ -7,7 +7,7 @@ from sqlalchemy import delete, select
from app.admin.repositories.coupon_data import coupon_slot_report
from app.db.session import SessionLocal
from app.models.coupon_state import CouponClaimRecord, CouponSession
from app.models.coupon_state import CouponClaimEvent, CouponClaimRecord, CouponSession
from app.repositories.coupon_state import record_claims, session_app_env
@@ -49,6 +49,7 @@ def test_record_claims_stamps_app_env() -> None:
assert row.app_env == "prod"
assert row.status == "already_claimed"
finally:
db.execute(delete(CouponClaimEvent).where(CouponClaimEvent.trace_id == "t-stamp"))
db.execute(delete(CouponClaimRecord).where(CouponClaimRecord.device_id == dev))
db.commit()
db.close()
+73
View File
@@ -0,0 +1,73 @@
"""SafeRotatingFileHandler:Windows 轮转不被外部句柄卡死(WinError 32)。
stdlib RotatingFileHandler rename 活动文件轮转;Windows 上只要有第二个句柄(admin
第二进程残留 --reload workerIDE 索引杀软)开着它, rename WinError 32轮转永久
卡死Safe 版在 Windows 改走 copytruncate(拷贝就地清空, 从不 rename 活动文件)
这些用例用 monkeypatch os.name 强制成 "nt", 使 copytruncate 分支在任何 OS CI 上都被
覆盖;在真实 Windows 上则天然命中
"""
from __future__ import annotations
import logging
from pathlib import Path
from app.core.logging import SafeRotatingFileHandler
def _emit(handler: logging.Handler, msg: str) -> None:
handler.emit(logging.LogRecord("t", logging.INFO, __file__, 0, msg, (), None))
def test_rollover_survives_second_open_handle(tmp_path: Path, monkeypatch) -> None:
"""第二个句柄开着活动文件时轮转:不抛异常, 且确实转了(生成 .1、活动文件就地清空)。"""
monkeypatch.setattr("app.core.logging.os.name", "nt")
log_file = tmp_path / "app-server.log"
# maxBytes 放大, 避免 emit 期间自动轮转干扰;本用例手动触发 doRollover。
handler = SafeRotatingFileHandler(
str(log_file), maxBytes=10**9, backupCount=3, encoding="utf-8",
)
handler.setFormatter(logging.Formatter("%(message)s"))
try:
for i in range(20):
_emit(handler, f"line-{i:03d}")
handler.flush()
before = log_file.stat().st_size
assert before > 0
# 正是 Windows 上 rename 失败的条件:另一个句柄开着活动文件。
with open(log_file, "a", encoding="utf-8"):
handler.doRollover() # 不应抛 PermissionError / WinError 32
backup = tmp_path / "app-server.log.1"
assert backup.exists()
assert backup.stat().st_size == before # 轮转前内容完整进了备份
assert log_file.stat().st_size == 0 # 活动文件就地清空(不是 rename)
# 句柄没被 rename 破坏, 仍能继续写。
_emit(handler, "after-rollover")
handler.flush()
assert log_file.stat().st_size > 0
finally:
handler.close()
def test_backups_shift_and_capped(tmp_path: Path, monkeypatch) -> None:
"""多次轮转:.1/.2 逐级腾挪, 超过 backupCount 的丢弃(不出现 .3)。"""
monkeypatch.setattr("app.core.logging.os.name", "nt")
log_file = tmp_path / "app-server.log"
handler = SafeRotatingFileHandler(
str(log_file), maxBytes=10**9, backupCount=2, encoding="utf-8",
)
handler.setFormatter(logging.Formatter("%(message)s"))
try:
for _ in range(4):
_emit(handler, "x" * 50)
handler.doRollover()
assert (tmp_path / "app-server.log.1").exists()
assert (tmp_path / "app-server.log.2").exists()
assert not (tmp_path / "app-server.log.3").exists()
finally:
handler.close()
+60 -1
View File
@@ -9,7 +9,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.models.wallet import CoinAccount, WithdrawOrder, WechatTransferAuthorization
from app.repositories import wallet as crud_wallet
@@ -426,3 +426,62 @@ def test_withdraw_reject_refunds(client, monkeypatch) -> None:
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"] == "测试拒绝"
# ===== §fix 免确认授权 enabled 判定收严:active+authorization_id 非空才算已授权 =====
def _seed_transfer_auth(phone: str, state: str, authorization_id: str | None) -> None:
"""直接在库里写/改该用户的免确认授权记录(绕过微信,用于判定测试)。"""
db = SessionLocal()
try:
user = db.execute(select(User).where(User.phone == phone)).scalar_one()
auth = db.get(WechatTransferAuthorization, user.id)
if auth is None:
auth = WechatTransferAuthorization(
user_id=user.id, openid=user.wechat_openid or "openid_test_abc",
out_authorization_no=f"oan_{user.id}",
)
db.add(auth)
auth.state = state
auth.authorization_id = authorization_id
db.commit()
finally:
db.close()
def test_withdraw_info_auth_enabled_requires_authorization_id(client, monkeypatch) -> None:
_patch_userinfo(monkeypatch)
token = _login(client, "13800002051")
# 触发建号 + 绑定微信(withdraw-info 读 openid)
client.get("/api/v1/wallet/withdraw-info", headers=_auth(token))
client.post("/api/v1/wallet/bind-wechat", json={"code": "c1"}, headers=_auth(token))
# active 但无 authorization_id → 视为未授权
_seed_transfer_auth("13800002051", "active", None)
r = client.get("/api/v1/wallet/withdraw-info", headers=_auth(token))
assert r.status_code == 200, r.text
assert r.json()["transfer_auth_enabled"] is False
# active 且有 authorization_id → 已授权
_seed_transfer_auth("13800002051", "active", "wx_auth_123")
r = client.get("/api/v1/wallet/withdraw-info", headers=_auth(token))
assert r.status_code == 200, r.text
assert r.json()["transfer_auth_enabled"] is True
def test_transfer_auth_status_requires_authorization_id(client, monkeypatch) -> None:
_patch_userinfo(monkeypatch)
token = _login(client, "13800002052")
client.get("/api/v1/wallet/withdraw-info", headers=_auth(token))
client.post("/api/v1/wallet/bind-wechat", json={"code": "c2"}, headers=_auth(token))
_seed_transfer_auth("13800002052", "active", None)
r = client.get("/api/v1/wallet/transfer-auth/status", headers=_auth(token))
assert r.status_code == 200, r.text
assert r.json()["state"] == "active"
assert r.json()["enabled"] is False
_seed_transfer_auth("13800002052", "active", "wx_auth_456")
r = client.get("/api/v1/wallet/transfer-auth/status", headers=_auth(token))
assert r.status_code == 200, r.text
assert r.json()["enabled"] is True