Compare commits

...

11 Commits

Author SHA1 Message Date
guke ab034d3c15 Merge branch 'main' into codex/comparison-device-display 2026-07-27 16:38:50 +08:00
linkeyu 58d609e5d2 修复提现审核汇总按来源和状态统计 (#185)
## 问题

邀请提现页的汇总接口未按提现来源过滤,导致页签数字、顶部“待审核/待审核金额”与邀请提现细则不一致;同时接口缺少已到账、已拒绝和全部的历史总数。

## 改动

- 汇总接口支持并校验 `source=coin_cash|invite_cash`
- 页签数量及顶部待审核数量、金额统一按提现来源过滤
- 增加 `success_count`、`rejected_count` 和 `total_count`
- 保留今日到账/拒绝指标,并同步按来源过滤
- 增加逐来源、逐状态校验“汇总数 = 列表 total”的回归测试
- 增加“顶部待审核金额 = 当前来源待审核细则金额之和”的回归测试

## 验证

- 提现审核专项测试通过
- 后台读取接口测试:19 项通过
- Ruff(排除主干既有规则告警)通过
- 线上数据库只读复核:其他提现与邀请提现的待审核数量、金额已按来源分别核算

---------

Co-authored-by: unknown <798648091@qq.com>
Reviewed-on: #185
Co-authored-by: linkeyu <linkeyu@wonderable.ai>
Co-committed-by: linkeyu <linkeyu@wonderable.ai>
2026-07-27 16:26:02 +08:00
guke 8e01ae0bf9 修复:ComparisonResultIn 补 skipped_dish_count(逐平台缺菜数落库) (#186)
pricebot 把逐平台缺菜数冗余进 comparison_results 每行,但上报入参 schema
ComparisonResultIn 未声明该字段,model_dump() 会在 POST 路径静默丢弃,
记录页三平台网格拿不到逐格「缺少 X 个菜品」。显式声明补齐 + 回归测试。

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: guke <guke@autohome.com.cn>
Reviewed-on: #186
2026-07-27 16:13:54 +08:00
linkeyu 7e17df4130 完善提现审核与用户风险管理 (#175)
## 改动内容
- 拆分邀请提现与其他提现的筛选、统计和详情数据口径
- 增加人工高风险标识、备注、权限和数据库迁移
- 完善微信失败原因查单、退款兜底与审核操作
- 移除人工处理超时订单接口和提现账本校验功能
- 补充提现审核、邀请详情、风险操作与微信失败原因回归测试

## 验证
- 相关后端测试:82 passed
- Ruff F/I:通过
- 新数据库迁移 upgrade/downgrade/upgrade:通过

## 说明
- 本地 seed mock 脚本未提交

---------

Co-authored-by: guke <guke@wonderable.ai>
Co-authored-by: unknown <798648091@qq.com>
Reviewed-on: #175
Co-authored-by: linkeyu <linkeyu@wonderable.ai>
Co-committed-by: linkeyu <linkeyu@wonderable.ai>
2026-07-27 16:03:31 +08:00
linkeyu 46f68b88e3 修复:统一广告收益分类 eCPM 口径 (#178)
## 本次改动

- Draw 信息流 eCPM 合并 `draw + 历史 feed`
- 看视频 eCPM 合并 `reward_video + withdrawal_video`
- 直接按每次真实展示的 SDK eCPM 加权,不再由发奖状态修正后的收益反推
- 增加经营分类聚合接口字段和专项测试

## 验证

`tests/test_admin_ad_revenue_scope.py`:3 passed。

---------

Co-authored-by: unknown <798648091@qq.com>
Reviewed-on: #178
Co-authored-by: linkeyu <linkeyu@wonderable.ai>
Co-committed-by: linkeyu <linkeyu@wonderable.ai>
2026-07-27 16:02:40 +08:00
linkeyu 1226bc8365 修复比价记录平均 TOKEN 成本采集与回填 (#182)
## 问题原因

- 新版客户端通过服务端 harvest 落库,但完成路径没有触发 LLM 调用明细与 TOKEN 成本回填。
- 内部共享密钥不一致或 PriceBot 实例切换后,拉取失败只留下空值,后续没有自动补偿。

## 本次改动

- harvest 完成后立即异步回填 LLM 调用、TOKEN 数及成本快照。
- 抽取统一、幂等的成本回填服务,并增加定时补偿 worker。
- 增加 PriceBot 内部鉴权预检、错误日志和多实例查找兜底。
- 仅回填当前价格配置生效后的终态记录,避免用现价误算更早历史数据。
- 补充环境配置、部署说明和单元测试。

## 验证

- 相关测试:36 passed。
- 静态检查通过,diff check 通过。
- 本地页面显示 ¥0.0139,与数据库精确均值 0.013916 的四舍五入结果一致。

## 上线注意

部署时需确保 app-server 与 PriceBot 的 INTERNAL_API_SECRET 完全一致并重启两个服务;worker 启动后会自动补齐符合条件的历史空值。

---------

Co-authored-by: guke <guke@wonderable.ai>
Co-authored-by: unknown <798648091@qq.com>
Reviewed-on: #182
Co-authored-by: linkeyu <linkeyu@wonderable.ai>
Co-committed-by: linkeyu <linkeyu@wonderable.ai>
2026-07-27 15:51:19 +08:00
unknown 973c440a8a 修复:补全比价记录设备信息展示
为比价记录列表与详情补充可读机型名和 ROM 版本字段,并增加接口回归测试。
2026-07-27 14:36:56 +08:00
linkeyu 22a1105000 修复用户反馈机型与系统版本补全 (#181)
## 改动说明
- 反馈列表按同一用户、同一设备编码和反馈提交时间,补全可读机型、厂商及 ROM 大版本
- 补充常见线上机型编码映射
- 更新本地 mock 数据并增加时间边界回归测试

## 验证
- 反馈相关测试:3 项通过
- Ruff 检查通过

---------

Co-authored-by: unknown <798648091@qq.com>
Reviewed-on: #181
Co-authored-by: linkeyu <linkeyu@wonderable.ai>
Co-committed-by: linkeyu <linkeyu@wonderable.ai>
2026-07-27 13:33:55 +08:00
guke b2a528eba1 fix(comparison): 派生 best 排除缺菜店, 避免虚低价当"最低价" (#176)
pricebot comparison_results[].rank 是纯价格排序(含缺菜店), server 派生 best
若照单全收, 会把缺菜(漏菜)店的虚低总价当 best_price → 记录页戴"最低"红框 +
算出虚假省额。

- _derive / _derive_from_results 派生 best 时按 platform_results[pid].
  skipped_dish_count 排除缺菜店; 源平台永远全菜, 全目标缺菜时回落到源
  (is_source_best、saved=0), 不虚报省额。
- platform_results 内层结构宽松(老客户端透传可伪造), _is_short 用
  isinstance 兜底, 值非 dict 时按"不缺菜"处理, 不打 500。
- harvest_done 传入 done_params.platform_results; 不传→纯 rank/price 老行为不变。
- 新增纯函数测试: 排除缺菜 / 全缺菜回落源 / 不传保持老行为 / 内层非 dict 不崩,
  覆盖 _derive 与 _derive_from_results 两条路径。

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

---------

Co-authored-by: guke <guke@autohome.com.cn>
Reviewed-on: #176
2026-07-27 10:35:15 +08:00
linkeyu cdd49c6421 修复:补齐收益明细广告网络来源 (#179)
## 本次改动

- 激励视频发奖记录按用户与 `ad_session_id` 回填展示侧 ADN
- 信息流明细保留每条发奖自身上报的 ADN 与底层代码位
- 历史记录仅在 `用户 + trace_id + eCPM` 候选网络唯一时安全回填
- 多网络候选保持空值,避免错误归因

## 验证

专项测试 4 passed,覆盖不同 ADN 明细及唯一/歧义回填场景。

---------

Co-authored-by: guke <guke@wonderable.ai>
Co-authored-by: unknown <798648091@qq.com>
Reviewed-on: #179
Co-authored-by: linkeyu <linkeyu@wonderable.ai>
Co-committed-by: linkeyu <linkeyu@wonderable.ai>
2026-07-27 10:24:32 +08:00
linkeyu 775a503d6f 修复广告收益用户详情加载失败 (#180)
## 问题原因
- 奖励统计接口用完整 User ORM 判断用户存在,滚动发布或表结构未同步时会因无关字段导致 500
- 金币记录合并广告与签到数据后直接排序,PostgreSQL 中 aware/naive datetime 混排会抛异常

## 修复内容
- 新增只投影 user.id 的用户存在性检查,保持不存在用户返回 404
- 金币记录排序前统一转换为 aware UTC 排序键
- 增加旧表结构投影与混合时区回归测试

## 验证结果
- tests/test_admin_read.py:18 项全部通过
- 线上只读数据库回归:近期 7 个活跃用户统计与金币明细全部正常返回;用户 #33 返回 1514 条记录
- 语法/未定义引用检查通过
- 全量测试 532 通过、9 失败;失败均在未修改的 origin/main 基线上复现,和本 PR 无关

---------

Co-authored-by: unknown <798648091@qq.com>
Reviewed-on: #180
Co-authored-by: linkeyu <linkeyu@wonderable.ai>
Co-committed-by: linkeyu <linkeyu@wonderable.ai>
2026-07-27 10:15:54 +08:00
42 changed files with 2323 additions and 471 deletions
+5
View File
@@ -139,6 +139,11 @@ PRICEBOT_COMPARE_TIMEOUT_SEC=60
# 必须与 pricebot 侧的 INTERNAL_API_SECRET **同值**;留空 = 内部写端点关闭(返 503)。
# 启用前两边都填同一高熵串:python -c "import secrets; print(secrets.token_urlsafe(48))"
INTERNAL_API_SECRET=
# 新版比价 harvest 完成后即时回填;以下 worker 再补偿短暂故障期间漏掉的记录。
LLM_COST_BACKFILL_ENABLED=true
LLM_COST_BACKFILL_INTERVAL_SEC=300
LLM_COST_BACKFILL_BATCH_SIZE=100
LLM_COST_BACKFILL_LOOKBACK_DAYS=30
# ===== CORS =====
# 逗号分隔,生产留空(只让 app 调,不开放 web)。本地开发可加 http://localhost:5173 之类
@@ -0,0 +1,76 @@
"""add manual high-risk flag and note to user
Revision ID: user_manual_risk_fields
Revises: risk_monitor_generic
"""
from __future__ import annotations
from collections.abc import Sequence
import sqlalchemy as sa
from sqlalchemy.dialects import postgresql
from alembic import op
revision: str = "user_manual_risk_fields"
down_revision: str | None = "risk_monitor_generic"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
_JSON = sa.JSON().with_variant(postgresql.JSONB(), "postgresql")
_INVITE_WITHDRAW_PAGE = "invite-withdraws"
def _finance_role() -> sa.TableClause:
return sa.table(
"admin_role",
sa.column("name", sa.String),
sa.column("pages", _JSON),
)
def upgrade() -> None:
with op.batch_alter_table("user") as batch_op:
batch_op.add_column(
sa.Column(
"is_high_risk",
sa.Boolean(),
server_default=sa.false(),
nullable=False,
)
)
batch_op.add_column(sa.Column("high_risk_note", sa.Text(), nullable=True))
batch_op.create_index("ix_user_is_high_risk", ["is_high_risk"], unique=False)
role = _finance_role()
conn = op.get_bind()
row = conn.execute(
sa.select(role.c.pages).where(role.c.name == "finance")
).scalar_one_or_none()
if row is not None and _INVITE_WITHDRAW_PAGE not in (row or []):
conn.execute(
role.update()
.where(role.c.name == "finance")
.values(pages=[*(row or []), _INVITE_WITHDRAW_PAGE])
)
def downgrade() -> None:
role = _finance_role()
conn = op.get_bind()
row = conn.execute(
sa.select(role.c.pages).where(role.c.name == "finance")
).scalar_one_or_none()
if row is not None:
conn.execute(
role.update()
.where(role.c.name == "finance")
.values(
pages=[
page for page in (row or []) if page != _INVITE_WITHDRAW_PAGE
]
)
)
with op.batch_alter_table("user") as batch_op:
batch_op.drop_index("ix_user_is_high_risk")
batch_op.drop_column("high_risk_note")
batch_op.drop_column("is_high_risk")
+3 -2
View File
@@ -23,7 +23,8 @@ PERMISSION_CATALOG: list[dict] = [
{"key": "cps", "label": "CPS收益"},
]},
{"group": "奖励审核", "pages": [
{"key": "withdraws", "label": "提现审核"},
{"key": "invite-withdraws", "label": "邀请提现审核"},
{"key": "withdraws", "label": "其他提现审核"},
{"key": "price-reports", "label": "低价审核"},
{"key": "feedbacks", "label": "用户反馈"},
]},
@@ -59,7 +60,7 @@ BUILTIN_ROLES: list[dict] = [
"cps", "risk-monitor", "device-liveness", "price-reports", "feedbacks", "huawei-review",
]},
{"name": "finance", "label": "财务", "pages": [
"dashboard", "ad-revenue-report", "cps", "withdraws",
"dashboard", "ad-revenue-report", "cps", "invite-withdraws", "withdraws",
]},
{"name": "tech", "label": "技术", "pages": [
"dashboard", "risk-monitor", "device-liveness", "analytics-health", "config", "ad-revenue", "huawei-review",
+101 -3
View File
@@ -12,10 +12,11 @@
"""
from __future__ import annotations
from sqlalchemy import func, select
from sqlalchemy import func, or_, select
from sqlalchemy.orm import Session
from app.core import rewards
from app.models.ad_ecpm import AdEcpmRecord
from app.models.ad_feed_reward import AdFeedRewardRecord
from app.models.ad_reward import AdRewardRecord
from app.repositories.ad_feed_reward import FEED_REWARD_UNIT_SECONDS
@@ -55,10 +56,21 @@ def _reward_video_rows(
if user_id is not None:
stmt = stmt.where(AdRewardRecord.user_id == user_id)
records = list(db.execute(stmt).scalars())
# S2S 发奖回调不携带实际填充 ADN;用相同用户和 ad_session_id 的展示记录回填。
session_ids = {record.ad_session_id for record in records if record.ad_session_id}
impression_by_session = {
(record.user_id, record.ad_session_id): record
for record in db.execute(
select(AdEcpmRecord).where(AdEcpmRecord.ad_session_id.in_(session_ids))
).scalars()
} if session_ids else {}
# 用本日之前的累计份数做起点,当日 granted 在其上继续递增 → 与 _granted_cumulative+1 对齐
granted_n: dict[int, int] = _prior_granted_counts(db, date=date, user_id=user_id)
rows: list[dict] = []
for rec in db.execute(stmt).scalars():
for rec in records:
impression = impression_by_session.get((rec.user_id, rec.ad_session_id))
if rec.status == "granted":
nth = granted_n.get(rec.user_id, 0) + 1
granted_n[rec.user_id] = nth
@@ -68,6 +80,8 @@ def _reward_video_rows(
"record_id": rec.id,
"user_id": rec.user_id,
"ad_session_id": rec.ad_session_id,
"adn": impression.adn if impression is not None else None,
"slot_id": impression.slot_id if impression is not None else None,
"app_env": rec.app_env,
"our_code_id": rec.our_code_id,
"created_at": rec.created_at,
@@ -90,6 +104,8 @@ def _reward_video_rows(
"record_id": rec.id,
"user_id": rec.user_id,
"ad_session_id": rec.ad_session_id,
"adn": impression.adn if impression is not None else None,
"slot_id": impression.slot_id if impression is not None else None,
"app_env": rec.app_env,
"our_code_id": rec.our_code_id,
"created_at": rec.created_at,
@@ -149,6 +165,78 @@ def _feed_scene_matches(rec: AdFeedRewardRecord, scene: str | None) -> bool:
return True
def _nonblank(value: str | None) -> str | None:
value = value.strip() if value else None
return value or None
def _unique_ad_source(records: list[AdEcpmRecord]) -> tuple[str | None, str | None]:
"""仅在候选展示记录指向唯一 ADN 时回填来源,避免错误归因。"""
adns = {_nonblank(record.adn) for record in records}
adns.discard(None)
if len(adns) != 1:
return None, None
slots = {_nonblank(record.slot_id) for record in records}
slots.discard(None)
return next(iter(adns)), next(iter(slots)) if len(slots) == 1 else None
def _feed_source_fallbacks(
db: Session, records: list[AdFeedRewardRecord]
) -> tuple[
dict[tuple[int, str], tuple[str | None, str | None]],
dict[tuple[int, str, str], tuple[str | None, str | None]],
]:
"""为旧信息流发奖记录构建安全来源索引。"""
session_ids = {record.ad_session_id for record in records if record.ad_session_id}
trace_ids = {record.trace_id for record in records if record.trace_id}
if not session_ids and not trace_ids:
return {}, {}
filters = []
if session_ids:
filters.append(AdEcpmRecord.ad_session_id.in_(session_ids))
if trace_ids:
filters.append(AdEcpmRecord.trace_id.in_(trace_ids))
impressions = list(db.execute(select(AdEcpmRecord).where(or_(*filters))).scalars())
by_session: dict[tuple[int, str], list[AdEcpmRecord]] = {}
by_trace_ecpm: dict[tuple[int, str, str], list[AdEcpmRecord]] = {}
for impression in impressions:
if impression.ad_session_id:
by_session.setdefault((impression.user_id, impression.ad_session_id), []).append(impression)
if impression.trace_id:
by_trace_ecpm.setdefault(
(impression.user_id, impression.trace_id, impression.ecpm_raw), []
).append(impression)
return (
{key: _unique_ad_source(value) for key, value in by_session.items()},
{key: _unique_ad_source(value) for key, value in by_trace_ecpm.items()},
)
def _feed_source(
record: AdFeedRewardRecord,
*,
by_session: dict[tuple[int, str], tuple[str | None, str | None]],
by_trace_ecpm: dict[tuple[int, str, str], tuple[str | None, str | None]],
) -> tuple[str | None, str | None]:
"""返回本条发奖广告的来源;无唯一证据时保留原始空值。"""
adn, slot_id = _nonblank(record.adn), _nonblank(record.slot_id)
if adn and slot_id:
return adn, slot_id
candidate = by_session.get((record.user_id, record.ad_session_id or ""))
if candidate is None and record.trace_id:
candidate = by_trace_ecpm.get((record.user_id, record.trace_id, record.ecpm_raw))
if candidate is None:
return adn, slot_id
candidate_adn, candidate_slot_id = candidate
return adn or candidate_adn, slot_id or candidate_slot_id
def _feed_rows(
db: Session, *, date: str, user_id: int | None, scene: str | None = None
) -> list[dict]:
@@ -167,11 +255,17 @@ def _feed_rows(
if user_id is not None:
stmt = stmt.where(AdFeedRewardRecord.user_id == user_id)
records = list(db.execute(stmt).scalars())
by_session, by_trace_ecpm = _feed_source_fallbacks(db, records)
# 本日之前的累计**条数**做起点,与发奖侧 granted_unit_total(COUNT granted)对齐
granted_count: dict[int, int] = _feed_prior_granted_count(db, date=date, user_id=user_id)
rows: list[dict] = []
for rec in db.execute(stmt).scalars():
for rec in records:
keep = _feed_scene_matches(rec, scene) # 累计照常推进,这里只决定是否展示本行
adn, slot_id = _feed_source(
rec, by_session=by_session, by_trace_ecpm=by_trace_ecpm
)
if rec.status == "granted":
# 一条广告 = 1 份(与 grant_feed_reward 同口径:看满一份即发该条满额,不按 unit_count 累加)。
# nth = 账号累计第几**条**(含本日之前),与发奖侧 granted_unit_total+1 对齐;累计照常推进
@@ -188,6 +282,8 @@ def _feed_rows(
"record_id": rec.id,
"user_id": rec.user_id,
"ad_session_id": rec.ad_session_id,
"adn": adn,
"slot_id": slot_id,
"trace_id": rec.trace_id,
"app_env": rec.app_env,
"our_code_id": rec.our_code_id,
@@ -214,6 +310,8 @@ def _feed_rows(
"record_id": rec.id,
"user_id": rec.user_id,
"ad_session_id": rec.ad_session_id,
"adn": adn,
"slot_id": slot_id,
"trace_id": rec.trace_id,
"app_env": rec.app_env,
"our_code_id": rec.our_code_id,
+40 -3
View File
@@ -96,7 +96,11 @@ _REWARD_DETAIL_KEYS = (
def _reward_detail(row: dict) -> dict:
"""从 audit 行抽出发奖复算明细(给前端展开行渲染因子1/因子2/份数/LT/应发实发)。"""
return {k: row[k] for k in _REWARD_DETAIL_KEYS}
detail = {key: row[key] for key in _REWARD_DETAIL_KEYS}
# 聚合父行可能包含多个 ADN,来源必须保留在每一条发奖明细上。
detail["adn"] = row.get("adn")
detail["slot_id"] = row.get("slot_id")
return detail
def ad_revenue_report(
@@ -244,8 +248,8 @@ def ad_revenue_report(
"impressions": 0,
"ecpm": row["ecpm"],
"revenue_yuan": 0.0,
"adn": None,
"slot_id": None,
"adn": row.get("adn"),
"slot_id": row.get("slot_id"),
"has_reward": True,
"status": row["status"],
"expected_coin": int(row["expected_coin"]),
@@ -427,6 +431,38 @@ def ad_revenue_report(
for k, v in type_map.items()
}
# 经营看板的两类 eCPM 必须按真实展示的 SDK eCPM 加权,不能用发奖状态修正后的
# 收益反推。Draw 包含新 draw 与历史 feed;看视频包含福利与提现视频。
category_map: dict[str, dict] = {}
for event in events:
category = (
"draw" if event["ad_type"] in {"draw", "feed"}
else "video" if event["ad_type"] in {"reward_video", "withdrawal_video"}
else None
)
if category is None:
continue
stat = category_map.setdefault(category, {
"impressions": 0,
"revenue_yuan": 0.0,
"ecpm_fen_sum": 0.0,
})
impressions = int(event["impressions"])
stat["impressions"] += impressions
stat["revenue_yuan"] += event["revenue_yuan"]
stat["ecpm_fen_sum"] += rewards.parse_ecpm_fen(event["ecpm"]) * impressions
category_stats = {
key: {
"impressions": value["impressions"],
"revenue_yuan": round(value["revenue_yuan"], 6),
"ecpm_yuan": round(
value["ecpm_fen_sum"] / value["impressions"] / 100.0,
6,
) if value["impressions"] else 0.0,
}
for key, value in category_map.items()
}
# 分场景小计(按 feed_scene:展示条数 + 预估收益),同 type_stats 基于全量 events——
# 供数据大盘「领券广告 / 比价广告」卡用。此前大盘是在分页 items 里按 feed_scene 现算,
# 2026-07-02 起信息流逐条展示行(唯一带收益 + 场景的行)不再进主表 items,现算恒为 0;
@@ -481,6 +517,7 @@ def ad_revenue_report(
"daily": daily,
"hourly": hourly,
"type_stats": type_stats,
"category_stats": category_stats,
"scene_stats": scene_stats,
"dau": dau,
"items": main_rows[offset:offset + limit],
+19
View File
@@ -44,6 +44,25 @@ def set_user_debug_trace(
return user
def set_user_risk(
db: Session,
user: User,
*,
is_high_risk: bool,
note: str | None,
commit: bool = True,
) -> User:
"""设置人工风险结论和备注,支持与审计日志共用同一事务。"""
user.is_high_risk = is_high_risk
user.high_risk_note = note.strip() if is_high_risk and note else None
if commit:
db.commit()
db.refresh(user)
else:
db.flush()
return user
def update_feedback_status(
db: Session, feedback: Feedback, *, status: str, commit: bool = True
) -> Feedback:
+270 -169
View File
@@ -9,7 +9,7 @@ from datetime import date, datetime, time, timedelta, timezone
from decimal import ROUND_HALF_UP, Decimal
from zoneinfo import ZoneInfo
from sqlalchemy import Select, asc, case, desc, func, or_, select
from sqlalchemy import Select, and_, asc, case, desc, func, or_, select
from sqlalchemy.orm import Session
from app.core import rewards
@@ -22,14 +22,15 @@ from app.models.comparison import ComparisonRecord
from app.models.coupon_state import CouponPromptEngagement
from app.models.device import DeviceLiveness
from app.models.feedback import Feedback
from app.models.invite import InviteRelation
from app.models.onboarding import OnboardingCompletion
from app.models.price_report import PriceReport
from app.models.savings import SavingsRecord
from app.models.user import User
from app.models.wallet import (
CashTransaction,
CoinAccount,
CoinTransaction,
InviteCashTransaction,
WithdrawOrder,
)
from app.repositories import activity, ad_ecpm
@@ -45,6 +46,78 @@ _FEED_SCENE_LABEL = {
}
_DEVICE_MARKETING_NAMES = {
"23078RKD5C": "Redmi K60 至尊版",
"M2012K11AC": "Redmi K40",
"PJA110": "一加 Ace 2 Pro",
"PPG-AN00": "荣耀 GT Pro",
"V2166BA": "vivo Y77e",
"V2309A": "vivo X100",
}
def _device_marketing_name(model: str | None) -> str | None:
"""把线上已知 Build.MODEL 编码转成用户可识别的商品名。"""
if not model:
return None
return _DEVICE_MARKETING_NAMES.get(model.strip().upper())
def _attach_comparison_device_details(items: list[ComparisonRecord]) -> None:
"""给比价记录补充可读机型名,同时保留原始设备编码。"""
for item in items:
item.device_model_name = _device_marketing_name(item.device_model)
def _attach_feedback_device_details(db: Session, feedbacks: list[Feedback]) -> None:
"""按同一用户、同一设备编码及提交时间补齐厂商和 ROM 大版本。"""
candidates = [
item for item in feedbacks if item.device_model and item.device_model.strip()
]
for item in candidates:
item.device_model_name = _device_marketing_name(item.device_model)
item.device_manufacturer = None
item.rom_version = None
if not candidates:
return
ranked = (
select(
Feedback.id.label("feedback_id"),
ComparisonRecord.device_manufacturer.label("device_manufacturer"),
ComparisonRecord.rom_version.label("rom_version"),
func.row_number()
.over(
partition_by=Feedback.id,
order_by=(
ComparisonRecord.created_at.desc(),
ComparisonRecord.id.desc(),
),
)
.label("row_num"),
)
.join(
ComparisonRecord,
and_(
ComparisonRecord.user_id == Feedback.user_id,
ComparisonRecord.device_model == Feedback.device_model,
ComparisonRecord.created_at <= Feedback.created_at,
),
)
.where(Feedback.id.in_([item.id for item in candidates]))
.subquery()
)
details = {
row.feedback_id: row
for row in db.execute(select(ranked).where(ranked.c.row_num == 1)).all()
}
for item in candidates:
detail = details.get(item.id)
item.device_manufacturer = detail.device_manufacturer if detail else None
item.rom_version = detail.rom_version if detail else None
def cursor_paginate(
db: Session, stmt: Select, id_col, *, limit: int, cursor: int | None
) -> tuple[list, int | None]:
@@ -277,6 +350,7 @@ def list_comparison_records(
limit=limit, cursor=cursor,
)
_attach_user_info(db, items)
_attach_comparison_device_details(items)
# 「本次比价看广告的预估收益」:按本页 trace_id 一次性聚合(同 _attach_user_info 逐页范式)。
# ad_revenue_yuan 非 ORM 列,仅瞬态挂实例上供 AdminComparisonListItem(from_attributes)读出。
rev = ad_ecpm.revenue_yuan_by_trace(db, [it.trace_id for it in items])
@@ -423,6 +497,7 @@ def get_comparison_record(db: Session, record_id: int) -> ComparisonRecord | Non
rec = db.get(ComparisonRecord, record_id)
if rec is not None:
_attach_user_info(db, [rec])
_attach_comparison_device_details([rec])
return rec
@@ -710,16 +785,7 @@ def list_all_withdraw_orders(
elif quick_filter == "today":
stmt = stmt.where(WithdrawOrder.created_at >= today_start)
elif quick_filter == "high_risk":
stmt = stmt.where(
or_(
WithdrawOrder.user_name.is_(None),
WithdrawOrder.user_name == "",
User.status != "active",
User.created_at >= now - timedelta(hours=24),
WithdrawOrder.status.in_(("failed", "rejected")),
WithdrawOrder.fail_reason.is_not(None),
)
)
stmt = stmt.where(User.is_high_risk.is_(True))
sort_cols = {
"id": WithdrawOrder.id,
@@ -745,7 +811,9 @@ def _as_utc(value: datetime) -> datetime:
return value.astimezone(timezone.utc)
def withdraw_list_enrichment(db: Session, user_ids: list[int]) -> dict[int, dict]:
def withdraw_list_enrichment(
db: Session, user_ids: list[int], *, source: str | None = None
) -> dict[int, dict]:
"""批量富化提现单列表:按本页 user_id 取 手机号/昵称 + 各自累计成功提现金额(分)。
两条聚合查询搞定(避免逐行 N+1)。累计口径与 get_user_overview 的 withdraw_success_cents
@@ -755,25 +823,32 @@ def withdraw_list_enrichment(db: Session, user_ids: list[int]) -> dict[int, dict
uniq = list(set(user_ids))
if not uniq:
return {}
success_rows = db.execute(
select(
WithdrawOrder.user_id,
func.coalesce(func.sum(WithdrawOrder.amount_cents), 0),
)
.where(WithdrawOrder.user_id.in_(uniq), WithdrawOrder.status == "success")
.group_by(WithdrawOrder.user_id)
).all()
success_stmt = select(
WithdrawOrder.user_id,
func.coalesce(func.sum(WithdrawOrder.amount_cents), 0),
).where(WithdrawOrder.user_id.in_(uniq), WithdrawOrder.status == "success")
if source:
success_stmt = success_stmt.where(WithdrawOrder.source == source)
success_rows = db.execute(success_stmt.group_by(WithdrawOrder.user_id)).all()
success_map = {uid: int(total) for uid, total in success_rows}
users = db.execute(
select(User.id, User.phone, User.nickname).where(User.id.in_(uniq))
select(
User.id,
User.phone,
User.nickname,
User.is_high_risk,
User.high_risk_note,
).where(User.id.in_(uniq))
).all()
return {
uid: {
"phone": phone,
"nickname": nickname,
"is_high_risk": is_high_risk,
"high_risk_note": high_risk_note,
"cumulative_success_cents": success_map.get(uid, 0),
}
for uid, phone, nickname in users
for uid, phone, nickname, is_high_risk, high_risk_note in users
}
@@ -817,6 +892,7 @@ def list_feedbacks(
db, stmt, (order_fn(sort_col), id_order), limit=limit, cursor=cursor
)
_attach_user_info(db, items) # 列表展示完整手机号(点手机号查该用户全部反馈)
_attach_feedback_device_details(db, items)
return items, next_cursor, total
@@ -879,15 +955,16 @@ def get_withdraw_by_out_bill_no(db: Session, out_bill_no: str) -> WithdrawOrder
).scalar_one_or_none()
def withdraw_summary(db: Session) -> dict:
"""提现审核台顶部统计。金额单位:分。"""
rows = db.execute(
select(
WithdrawOrder.status,
func.count(WithdrawOrder.id),
func.coalesce(func.sum(WithdrawOrder.amount_cents), 0),
).group_by(WithdrawOrder.status)
).all()
def withdraw_summary(db: Session, *, source: str | None = None) -> dict:
"""提现审核台顶部统计。金额单位:分;口径与列表的 source 筛选一致"""
summary_stmt = select(
WithdrawOrder.status,
func.count(WithdrawOrder.id),
func.coalesce(func.sum(WithdrawOrder.amount_cents), 0),
)
if source:
summary_stmt = summary_stmt.where(WithdrawOrder.source == source)
rows = db.execute(summary_stmt.group_by(WithdrawOrder.status)).all()
by_status = {
status: {"count": int(count), "amount_cents": int(amount_cents)}
for status, count, amount_cents in rows
@@ -900,31 +977,154 @@ def withdraw_summary(db: Session) -> dict:
)
def _today_count(status: str) -> int:
return db.execute(
select(func.count(WithdrawOrder.id)).where(
WithdrawOrder.status == status,
WithdrawOrder.updated_at >= today_start,
)
).scalar_one()
today_success_amount = db.execute(
select(func.coalesce(func.sum(WithdrawOrder.amount_cents), 0)).where(
WithdrawOrder.status == "success",
stmt = select(func.count(WithdrawOrder.id)).where(
WithdrawOrder.status == status,
WithdrawOrder.updated_at >= today_start,
)
).scalar_one()
if source:
stmt = stmt.where(WithdrawOrder.source == source)
return db.execute(stmt).scalar_one()
today_success_amount_stmt = select(
func.coalesce(func.sum(WithdrawOrder.amount_cents), 0)
).where(
WithdrawOrder.status == "success",
WithdrawOrder.updated_at >= today_start,
)
if source:
today_success_amount_stmt = today_success_amount_stmt.where(
WithdrawOrder.source == source
)
today_success_amount = db.execute(today_success_amount_stmt).scalar_one()
def _status_count(status: str) -> int:
return by_status.get(status, {}).get("count", 0)
return {
"reviewing_count": by_status.get("reviewing", {}).get("count", 0),
"reviewing_count": _status_count("reviewing"),
"reviewing_amount_cents": by_status.get("reviewing", {}).get("amount_cents", 0),
"pending_count": by_status.get("pending", {}).get("count", 0),
"failed_count": by_status.get("failed", {}).get("count", 0),
"pending_count": _status_count("pending"),
"success_count": _status_count("success"),
"rejected_count": _status_count("rejected"),
"failed_count": _status_count("failed"),
"total_count": sum(item["count"] for item in by_status.values()),
"today_success_count": _today_count("success"),
"today_success_amount_cents": int(today_success_amount),
"today_rejected_count": _today_count("rejected"),
}
def invite_overview(
db: Session,
inviter_user_id: int,
*,
date_from: datetime | None = None,
date_to: datetime | None = None,
) -> dict:
"""邀请提现详情:邀请关系统计及受邀用户首次比价/首单信息。
首次记录按自增 id 取最早一条,批量查询避免按受邀用户逐行查询。
邀请成功口径为完成首次比价并已发邀请奖励(compare_reward_granted)。
"""
relation_stmt = (
select(InviteRelation, User)
.join(User, User.id == InviteRelation.invitee_user_id)
.where(InviteRelation.inviter_user_id == inviter_user_id)
.order_by(InviteRelation.created_at.asc(), InviteRelation.id.asc())
)
if date_from is not None:
relation_stmt = relation_stmt.where(User.created_at >= _as_utc_naive(date_from))
if date_to is not None:
relation_stmt = relation_stmt.where(User.created_at <= _as_utc_naive(date_to))
relation_rows = db.execute(relation_stmt).all()
invitee_ids = [relation.invitee_user_id for relation, _ in relation_rows]
if not invitee_ids:
return {"invite_total": 0, "invite_success_total": 0, "items": []}
first_compare_ids = list(
db.execute(
select(func.min(ComparisonRecord.id))
.where(
ComparisonRecord.user_id.in_(invitee_ids),
ComparisonRecord.status == "success",
)
.group_by(ComparisonRecord.user_id)
).scalars()
)
comparisons = (
db.execute(
select(
ComparisonRecord.user_id,
ComparisonRecord.store_name,
ComparisonRecord.product_names,
).where(ComparisonRecord.id.in_(first_compare_ids))
).all()
if first_compare_ids
else []
)
comparison_map = {
user_id: (store_name, product_names)
for user_id, store_name, product_names in comparisons
}
first_order_ids = list(
db.execute(
select(func.min(SavingsRecord.id))
.where(
SavingsRecord.user_id.in_(invitee_ids),
SavingsRecord.source == "compare",
)
.group_by(SavingsRecord.user_id)
).scalars()
)
orders = (
db.execute(
select(
SavingsRecord.user_id,
SavingsRecord.shop_name,
SavingsRecord.title,
SavingsRecord.dishes,
SavingsRecord.order_amount_cents,
).where(SavingsRecord.id.in_(first_order_ids))
).all()
if first_order_ids
else []
)
order_map = {
user_id: (
shop_name,
title or "".join(dishes or []),
order_amount_cents,
)
for user_id, shop_name, title, dishes, order_amount_cents in orders
}
items = []
for relation, user in relation_rows:
compare_store, compare_products = comparison_map.get(user.id, (None, None))
order_store, order_products, order_amount = order_map.get(
user.id, (None, None, None)
)
items.append(
{
"user_id": user.id,
"phone": user.phone,
"registered_at": user.created_at,
"invite_success": bool(relation.compare_reward_granted),
"first_compare_store": compare_store,
"first_compare_products": compare_products,
"first_order_store": order_store,
"first_order_products": order_products,
"first_order_amount_cents": order_amount,
}
)
return {
"invite_total": len(items),
"invite_success_total": sum(1 for item in items if item["invite_success"]),
"items": items,
}
def list_withdraw_audit_logs(
db: Session, out_bill_no: str, *, limit: int = 20
) -> list[AdminAuditLog]:
@@ -945,8 +1145,6 @@ def withdraw_risk_flags(
cash_balance_cents: int,
) -> tuple[list[str], int]:
flags: list[str] = []
if not order.user_name:
flags.append("缺少提现实名")
if user and user.status != "active":
flags.append(f"账号状态:{user.status}")
if user and user.created_at:
@@ -969,125 +1167,6 @@ def withdraw_risk_flags(
return flags, score
def _check_withdraw_ledger_side(
orders: list[WithdrawOrder], txns: list, *, withdraw_biz: str, refund_biz: str
) -> dict:
"""对某一本账(普通现金 / 邀请奖励金)做提现单 ↔ 流水的交叉校验。
orders 已按 source 过滤到本账;txns 是本账流水表里 withdraw_biz/refund_biz 两类流水。
规则:每单发起应有一条扣款流水(ref_id=out_bill_no);失败/拒绝单应有且仅一条退款流水;
非退款终态不应出现退款流水。四个计数全为 0 即本账自洽。
"""
withdraw_refs = {txn.ref_id for txn in txns if txn.biz_type == withdraw_biz}
refund_counts: dict[str, int] = {}
for txn in txns:
if txn.biz_type == refund_biz and txn.ref_id:
refund_counts[txn.ref_id] = refund_counts.get(txn.ref_id, 0) + 1
missing_withdraw = 0
missing_refund = 0
refund_on_non_terminal = 0
for order in orders:
if order.out_bill_no not in withdraw_refs:
missing_withdraw += 1
has_refund = refund_counts.get(order.out_bill_no, 0) > 0
if order.status in {"failed", "rejected"} and not has_refund:
missing_refund += 1
if has_refund and order.status not in {"failed", "rejected"}:
refund_on_non_terminal += 1
return {
"missing_withdraw": missing_withdraw,
"missing_refund": missing_refund,
"duplicate_refund": sum(1 for count in refund_counts.values() if count > 1),
"refund_on_non_terminal": refund_on_non_terminal,
}
def withdraw_ledger_check(db: Session) -> dict:
"""现金账本校验:两本物理隔离的账各自对账(产品红线:coin_cash / invite_cash 不串)。
普通现金:CoinAccount.cash_balance_cents ↔ cash_transaction(withdraw/withdraw_refund);
邀请奖励金:CoinAccount.invite_cash_balance_cents ↔ invite_cash_transaction
(invite_withdraw/invite_withdraw_refund)。
提现单按 source 分流到对应账核对——邀请提现的流水写在 invite_cash_transaction 表,
绝不能拿去和普通现金流水比(否则每笔邀请提现单都会被误报「缺扣款/缺退款流水」)。
分流口径与 create_withdraw 一致:仅 source==invite_cash 走邀请账,其余(含历史空值)归普通现金。
"""
orders = list(db.execute(select(WithdrawOrder)).scalars().all())
coin_orders = [o for o in orders if o.source != "invite_cash"]
invite_orders = [o for o in orders if o.source == "invite_cash"]
# —— 普通现金账(coin_cash) ——
cash_balance_total = int(
db.execute(select(func.coalesce(func.sum(CoinAccount.cash_balance_cents), 0))).scalar_one()
)
cash_txn_total = int(
db.execute(select(func.coalesce(func.sum(CashTransaction.amount_cents), 0))).scalar_one()
)
cash_txns = list(
db.execute(
select(CashTransaction).where(
CashTransaction.biz_type.in_(("withdraw", "withdraw_refund"))
)
).scalars().all()
)
coin = _check_withdraw_ledger_side(
coin_orders, cash_txns, withdraw_biz="withdraw", refund_biz="withdraw_refund"
)
cash_diff = cash_balance_total - cash_txn_total
# —— 邀请奖励金账(invite_cash,独立账户 + 独立流水表) ——
invite_balance_total = int(
db.execute(
select(func.coalesce(func.sum(CoinAccount.invite_cash_balance_cents), 0))
).scalar_one()
)
invite_txn_total = int(
db.execute(
select(func.coalesce(func.sum(InviteCashTransaction.amount_cents), 0))
).scalar_one()
)
invite_txns = list(
db.execute(
select(InviteCashTransaction).where(
InviteCashTransaction.biz_type.in_(("invite_withdraw", "invite_withdraw_refund"))
)
).scalars().all()
)
invite = _check_withdraw_ledger_side(
invite_orders, invite_txns,
withdraw_biz="invite_withdraw", refund_biz="invite_withdraw_refund",
)
invite_diff = invite_balance_total - invite_txn_total
ok = (
cash_diff == 0
and invite_diff == 0
and all(v == 0 for v in coin.values())
and all(v == 0 for v in invite.values())
)
return {
"ok": ok,
# 普通现金账(coin_cash:金币兑换的现金)
"cash_balance_total_cents": cash_balance_total,
"cash_transaction_total_cents": cash_txn_total,
"balance_diff_cents": cash_diff,
"missing_withdraw_txn_count": coin["missing_withdraw"],
"missing_refund_txn_count": coin["missing_refund"],
"duplicate_refund_txn_count": coin["duplicate_refund"],
"refund_txn_on_non_terminal_count": coin["refund_on_non_terminal"],
# 邀请奖励金账(invite_cash:与普通现金物理隔离,各自对账)
"invite_cash_balance_total_cents": invite_balance_total,
"invite_cash_transaction_total_cents": invite_txn_total,
"invite_balance_diff_cents": invite_diff,
"invite_missing_withdraw_txn_count": invite["missing_withdraw"],
"invite_missing_refund_txn_count": invite["missing_refund"],
"invite_duplicate_refund_txn_count": invite["duplicate_refund"],
"invite_refund_txn_on_non_terminal_count": invite["refund_on_non_terminal"],
}
def get_user_overview(db: Session, user_id: int) -> dict | None:
"""用户 360 概览:基础资料 + 钱包余额 + 各项 count。历史明细走各自分页接口(带 user_id 过滤)。"""
user = db.get(User, user_id)
@@ -1147,6 +1226,7 @@ def user_reward_stats(
*,
date_from: datetime | None = None,
date_to: datetime | None = None,
withdraw_source: str | None = None,
) -> dict:
"""提现详情「用户统计区」10 项。窗口作用于除「现金余额」外的所有项(余额是当前快照)。
@@ -1154,22 +1234,35 @@ def user_reward_stats(
平均 eCPM 用原始分值(分/千次)按记录取算术平均;各「提现」= 该来源累计金币折现。
传统任务 = 窗口内正向金币中,排除广告(reward_video/feed_ad_reward)与人工调整后的折现。
"""
withdraw_source_conds = (
[WithdrawOrder.source == withdraw_source] if withdraw_source else []
)
wd_success = db.execute(
select(func.coalesce(func.sum(WithdrawOrder.amount_cents), 0)).where(
WithdrawOrder.user_id == user_id,
WithdrawOrder.status == "success",
*withdraw_source_conds,
*_window_conds(WithdrawOrder.created_at, date_from, date_to),
)
).scalar_one()
wd_total = db.execute(
select(func.count(WithdrawOrder.id)).where(
WithdrawOrder.user_id == user_id,
*withdraw_source_conds,
*_window_conds(WithdrawOrder.created_at, date_from, date_to),
)
).scalar_one()
acc = db.get(CoinAccount, user_id) # 现金余额:当前快照,不随窗口
cash_balance = acc.cash_balance_cents if acc else 0
cash_balance = (
(
acc.invite_cash_balance_cents
if withdraw_source == "invite_cash"
else acc.cash_balance_cents
)
if acc
else 0
)
# 只投影本统计实际使用的列。避免滚动发布或旧本地库尚未补齐无关新列时,
# SQLAlchemy 因 select(ORM) 自动展开整表字段而让提现详情整体 500。
@@ -1228,6 +1321,11 @@ def _cn_wall_to_utc(dt: datetime) -> datetime:
return dt.replace(tzinfo=rewards.CN_TZ).astimezone(timezone.utc).replace(tzinfo=None)
def _coin_record_sort_key(row: dict) -> datetime:
"""金币明细跨数据源排序键:兼容 SQLite naive 与 PostgreSQL aware 时间。"""
return _as_utc(row["created_at"])
def user_coin_records(
db: Session,
user_id: int,
@@ -1319,7 +1417,10 @@ def user_coin_records(
"coin": rec.amount,
})
rows.sort(key=lambda r: r["created_at"], reverse=True)
# SQLite 常返回 naive datetimePostgreSQL timestamptz 返回 aware datetime
# 统一成 aware UTC 排序,避免线上合并广告记录与签到记录时抛
# “can't compare offset-naive and offset-aware datetimes”。
rows.sort(key=_coin_record_sort_key, reverse=True)
has_more = len(rows) > offset + limit
# 总数 = 三源在窗口内 granted 计数之和(供前端页码分页渲染页码/共 N 条)
+56 -2
View File
@@ -16,6 +16,7 @@ from app.admin.schemas.user import (
GrantCashRequest,
GrantCoinsRequest,
SetDebugTraceRequest,
SetUserRiskRequest,
SetUserStatusRequest,
UserCoinRecord,
UserRewardStats,
@@ -84,12 +85,21 @@ def get_user_reward_stats(
db: AdminDb,
date_from: Annotated[datetime | None, Query()] = None,
date_to: Annotated[datetime | None, Query()] = None,
withdraw_source: Annotated[
str | None, Query(pattern="^(coin_cash|invite_cash)$")
] = None,
) -> UserRewardStats:
"""提现详情抽屉「用户统计区」。date_from/date_to 都不传 = 注册至今(全量)。"""
if user_repo.get_user_by_id(db, user_id) is None:
if not user_repo.user_exists(db, user_id):
raise HTTPException(status_code=404, detail="用户不存在")
return UserRewardStats(
**queries.user_reward_stats(db, user_id, date_from=date_from, date_to=date_to)
**queries.user_reward_stats(
db,
user_id,
date_from=date_from,
date_to=date_to,
withdraw_source=withdraw_source,
)
)
@@ -138,6 +148,50 @@ def set_user_status(
return OkResponse()
@router.post("/{user_id}/risk", response_model=OkResponse, summary="设置人工高风险标记与备注")
def set_user_risk(
user_id: int,
body: SetUserRiskRequest,
request: Request,
admin: Annotated[
AdminUser, Depends(require_role("operator", "finance"))
],
db: AdminDb,
) -> OkResponse:
user = user_repo.get_user_by_id(db, user_id)
if user is None:
raise HTTPException(status_code=404, detail="用户不存在")
before = {
"is_high_risk": user.is_high_risk,
"high_risk_note": user.high_risk_note,
}
mutations.set_user_risk(
db,
user,
is_high_risk=body.is_high_risk,
note=body.note,
commit=False,
)
write_audit(
db,
admin,
action="user.risk.set",
target_type="user",
target_id=user_id,
detail={
"before": before,
"after": {
"is_high_risk": user.is_high_risk,
"high_risk_note": user.high_risk_note,
},
},
ip=get_client_ip(request),
commit=False,
)
db.commit()
return OkResponse()
@router.post("/{user_id}/debug-trace", response_model=OkResponse, summary="开关调试链接权限")
def set_user_debug_trace(
user_id: int,
+47 -34
View File
@@ -1,4 +1,4 @@
"""admin 提现:列表(读)+ 单笔重试查单 + 批量对账(写,带审计)。
"""admin 提现:列表(读)+ 单笔/批量查单与审核操作(写,带审计)。
提现的钱逻辑(查微信/退款/撤单/幂等)全部复用 app.repositories.wallet,admin 只触发 + 记审计。
这些 wallet 函数内部各自 commit(涉及微信调用),审计在其后单独 commit:操作本身幂等,
@@ -14,17 +14,16 @@ from fastapi import APIRouter, Depends, HTTPException, Query, Request
from app.admin.audit import write_audit
from app.admin.deps import AdminDb, get_client_ip, get_current_admin, require_role
from app.admin.repositories import queries
from app.admin.schemas.common import CursorPage
from app.admin.schemas.admin import AdminAuditLogOut
from app.admin.schemas.common import CursorPage
from app.admin.schemas.wallet import (
CashTxnOut,
ReconcileResult,
InviteOverviewOut,
WithdrawBulkItemResult,
WithdrawBulkRejectRequest,
WithdrawBulkRequest,
WithdrawBulkResult,
WithdrawBulkItemResult,
WithdrawDetailOut,
WithdrawLedgerCheckOut,
WithdrawListItemOut,
WithdrawOrderOut,
WithdrawRejectRequest,
@@ -83,8 +82,16 @@ def list_withdraws(
cursor=cursor,
)
# 联表带出手机号/昵称 + 累计成功提现(本页 user_id 批量富化,2 条聚合查询,无 N+1)。
enrichment = queries.withdraw_list_enrichment(db, [o.user_id for o in items])
_empty = {"phone": None, "nickname": None, "cumulative_success_cents": 0}
enrichment = queries.withdraw_list_enrichment(
db, [o.user_id for o in items], source=source
)
_empty = {
"phone": None,
"nickname": None,
"is_high_risk": False,
"high_risk_note": None,
"cumulative_success_cents": 0,
}
out_items = [
WithdrawListItemOut.model_validate(o).model_copy(
update=enrichment.get(o.user_id, _empty)
@@ -95,8 +102,13 @@ def list_withdraws(
@router.get("/summary", response_model=WithdrawSummaryOut, summary="提现审核台统计")
def withdraws_summary(db: AdminDb) -> WithdrawSummaryOut:
return WithdrawSummaryOut(**queries.withdraw_summary(db))
def withdraws_summary(
db: AdminDb,
source: Annotated[
str | None, Query(pattern="^(coin_cash|invite_cash)$")
] = None,
) -> WithdrawSummaryOut:
return WithdrawSummaryOut(**queries.withdraw_summary(db, source=source))
@router.get(
@@ -153,13 +165,13 @@ def withdraw_health_check(db: AdminDb) -> WxpayHealthCheckOut:
)
@router.get("/ledger-check", response_model=WithdrawLedgerCheckOut, summary="提现资金账本校验")
def withdraw_ledger_check(db: AdminDb) -> WithdrawLedgerCheckOut:
return WithdrawLedgerCheckOut(**queries.withdraw_ledger_check(db))
@router.get("/{out_bill_no}", response_model=WithdrawDetailOut, summary="提现单详情")
def withdraw_detail(out_bill_no: str, db: AdminDb) -> WithdrawDetailOut:
def withdraw_detail(
out_bill_no: str,
db: AdminDb,
date_from: Annotated[datetime | None, Query()] = None,
date_to: Annotated[datetime | None, Query()] = None,
) -> WithdrawDetailOut:
order = queries.get_withdraw_by_out_bill_no(db, out_bill_no)
if order is None:
raise HTTPException(status_code=404, detail="提现单不存在")
@@ -172,6 +184,8 @@ def withdraw_detail(out_bill_no: str, db: AdminDb) -> WithdrawDetailOut:
id=user.id,
phone=user.phone,
nickname=user.nickname,
is_high_risk=user.is_high_risk,
high_risk_note=user.high_risk_note,
status=user.status,
wechat_nickname=user.wechat_nickname,
wechat_avatar_url=user.wechat_avatar_url,
@@ -195,37 +209,36 @@ def withdraw_detail(out_bill_no: str, db: AdminDb) -> WithdrawDetailOut:
recent_withdraws,
overview["cash_balance_cents"] if overview else 0,
)
detail_enrichment = queries.withdraw_list_enrichment(
db, [order.user_id], source=order.source
).get(order.user_id, {})
return WithdrawDetailOut(
order=WithdrawOrderOut.model_validate(order),
user=user_snapshot,
cumulative_success_cents=int(
detail_enrichment.get("cumulative_success_cents", 0)
),
risk_flags=risk_flags,
risk_score=risk_score,
recent_withdraws=[WithdrawOrderOut.model_validate(o) for o in recent_withdraws],
recent_cash_transactions=[CashTxnOut.model_validate(t) for t in recent_cash_transactions],
audit_logs=[AdminAuditLogOut.model_validate(log) for log in audit_logs],
invite_overview=(
InviteOverviewOut(
**queries.invite_overview(
db,
order.user_id,
date_from=date_from,
date_to=date_to,
)
)
if order.source == "invite_cash"
else None
),
)
# 注意:/reconcile 必须在 /{out_bill_no}/refresh 之前声明(静态路径优先于路径参数)
@router.post("/reconcile", response_model=ReconcileResult, summary="批量对账(扫超时 pending 单)")
def reconcile(
request: Request,
admin: Annotated[AdminUser, Depends(require_role("finance"))],
db: AdminDb,
older_than_minutes: Annotated[int, Query(ge=0)] = 15,
) -> ReconcileResult:
try:
result = wallet_repo.reconcile_pending_withdraws(db, older_than_minutes=older_than_minutes)
except wxpay.WxPayNotConfiguredError as e:
raise HTTPException(status_code=503, detail="微信支付未配置") from e
write_audit(
db, admin, action="withdraw.reconcile", target_type="withdraw", target_id=None,
detail=result, ip=get_client_ip(request), commit=True,
)
return ReconcileResult(**result)
def _bulk_result(items: list[WithdrawBulkItemResult]) -> WithdrawBulkResult:
success = sum(1 for item in items if item.ok)
return WithdrawBulkResult(
+7
View File
@@ -40,6 +40,8 @@ class AdRevenueRecord(BaseModel):
expected_coin: int = Field(..., description="按公式复算应发金币")
actual_coin: int = Field(..., description="实际入账金币")
matched: bool = Field(..., description="复算与实发是否一致")
adn: str | None = Field(None, description="本条发奖对应的实际填充 ADN 子渠道")
slot_id: str | None = Field(None, description="本条发奖对应的底层 mediation rit")
class AdRevenueDaily(BaseModel):
@@ -73,6 +75,7 @@ class AdRevenueTypeStat(BaseModel):
impressions: int = Field(..., description="该类型展示条数合计")
revenue_yuan: float = Field(..., description="该类型预估收益合计(元)")
ecpm_yuan: float | None = Field(None, description="按真实展示次数加权的 SDK eCPM(元/千次)")
class AdRevenueRow(BaseModel):
@@ -142,6 +145,10 @@ class AdRevenueReportOut(BaseModel):
default_factory=dict,
description="按广告类型(ad_type)小计 {ad_type: {impressions, revenue_yuan}};前端取 draw / reward_video 做分类大盘",
)
category_stats: dict[str, AdRevenueTypeStat] = Field(
default_factory=dict,
description="按经营分类小计:draw=draw+历史 feedvideo=reward_video+withdrawal_video",
)
scene_stats: dict[str, AdRevenueTypeStat] = Field(
default_factory=dict,
description="按信息流场景(feed_scene)小计 {comparison/coupon/welfare: {impressions, revenue_yuan}};"
+2 -1
View File
@@ -39,8 +39,10 @@ class AdminComparisonListItem(BaseModel):
# 本次比价 LLM 总成本(元,按当时价冻结);旧记录/未回填为 None → 前端「成本」列回退估算。见 services/llm_cost.py。
llm_cost_yuan: float | None = None
device_model: str | None = None
device_model_name: str | None = None
rom_vendor: str | None = None
rom_name: str | None = None
rom_version: int | None = None
android_version: str | None = None
app_version: str | None = None
ad_revenue_yuan: float = 0.0 # 本次比价看的信息流广告预估收益(元),queries 瞬态挂 ORM 实例上
@@ -84,7 +86,6 @@ class AdminComparisonDetail(AdminComparisonListItem):
skipped_dish_names: list = []
# 全量环境
device_manufacturer: str | None = None
rom_version: int | None = None
android_sdk: int | None = None
app_version_code: int | None = None
source_app_version: str | None = None
+3
View File
@@ -32,7 +32,10 @@ class FeedbackOut(BaseModel):
# 提交端环境快照(feedback 表列):提交版本号 / 机型OS版本;改版前的历史反馈为 None
app_version: str | None = None
device_model: str | None = None
device_model_name: str | None = None
device_manufacturer: str | None = None
rom_name: str | None = None
rom_version: int | None = None
android_version: str | None = None
# 联表瞬态字段(queries._attach_user_info 挂):列表展示完整手机号,点手机号查该用户全部反馈
phone: str | None = None
+18
View File
@@ -16,6 +16,8 @@ class AdminUserListItem(BaseModel):
register_channel: str
status: str
debug_trace_enabled: bool = False
is_high_risk: bool = False
high_risk_note: str | None = None
wechat_openid: str | None = None
wechat_nickname: str | None = None
created_at: datetime
@@ -118,3 +120,19 @@ class SetUserStatusRequest(BaseModel):
class SetDebugTraceRequest(BaseModel):
enabled: bool = Field(..., description="是否给该用户开「复制调试链接」权限")
class SetUserRiskRequest(BaseModel):
is_high_risk: bool = Field(..., description="是否标记为高风险用户")
note: str | None = Field(
None,
max_length=500,
description="高风险备注;标记高风险时必填,解除后清空",
)
@field_validator("note")
@classmethod
def validate_note(cls, value: str | None, info):
if info.data.get("is_high_risk") and not (value or "").strip():
raise ValueError("标记高风险时必须填写原因")
return value.strip() if value and value.strip() else None
+27 -25
View File
@@ -60,6 +60,8 @@ class WithdrawListItemOut(WithdrawOrderOut):
phone: str | None = None
nickname: str | None = None
is_high_risk: bool = False
high_risk_note: str | None = None
cumulative_success_cents: int = 0 # 累计成功提现 = SUM(amount_cents) WHERE status='success'
@@ -67,7 +69,10 @@ class WithdrawSummaryOut(BaseModel):
reviewing_count: int
reviewing_amount_cents: int
pending_count: int
success_count: int
rejected_count: int
failed_count: int
total_count: int
today_success_count: int
today_success_amount_cents: int
today_rejected_count: int
@@ -77,6 +82,8 @@ class WithdrawUserSnapshot(BaseModel):
id: int
phone: str
nickname: str | None = None
is_high_risk: bool = False
high_risk_note: str | None = None
status: str
wechat_nickname: str | None = None
wechat_avatar_url: str | None = None
@@ -87,19 +94,34 @@ class WithdrawUserSnapshot(BaseModel):
withdraw_success_cents: int
class InviteeDetailOut(BaseModel):
user_id: int
phone: str
registered_at: datetime
invite_success: bool
first_compare_store: str | None = None
first_compare_products: str | None = None
first_order_store: str | None = None
first_order_products: str | None = None
first_order_amount_cents: int | None = None
class InviteOverviewOut(BaseModel):
invite_total: int
invite_success_total: int
items: list[InviteeDetailOut]
class WithdrawDetailOut(BaseModel):
order: WithdrawOrderOut
user: WithdrawUserSnapshot | None = None
cumulative_success_cents: int = 0
risk_flags: list[str]
risk_score: int
recent_withdraws: list[WithdrawOrderOut]
recent_cash_transactions: list[CashTxnOut]
audit_logs: list[AdminAuditLogOut]
class ReconcileResult(BaseModel):
checked: int
resolved: int
invite_overview: InviteOverviewOut | None = None
class WithdrawBulkRequest(BaseModel):
@@ -128,26 +150,6 @@ class WithdrawBulkResult(BaseModel):
items: list[WithdrawBulkItemResult]
class WithdrawLedgerCheckOut(BaseModel):
ok: bool
# 普通现金账(coin_cash:金币兑换的现金)
cash_balance_total_cents: int
cash_transaction_total_cents: int
balance_diff_cents: int
missing_withdraw_txn_count: int
missing_refund_txn_count: int
duplicate_refund_txn_count: int
refund_txn_on_non_terminal_count: int
# 邀请奖励金账(invite_cash:与普通现金物理隔离,各自对账)。默认 0 向后兼容。
invite_cash_balance_total_cents: int = 0
invite_cash_transaction_total_cents: int = 0
invite_balance_diff_cents: int = 0
invite_missing_withdraw_txn_count: int = 0
invite_missing_refund_txn_count: int = 0
invite_duplicate_refund_txn_count: int = 0
invite_refund_txn_on_non_terminal_count: int = 0
class WxpayHealthCheckOut(BaseModel):
ok: bool
wxpay_configured: bool
+14 -4
View File
@@ -25,7 +25,7 @@ import uuid
from typing import Any
import httpx
from fastapi import APIRouter, HTTPException, Request, status
from fastapi import APIRouter, BackgroundTasks, HTTPException, Request, status
from fastapi.concurrency import run_in_threadpool
from app.api.deps import DbSession, OptionalUser
@@ -36,6 +36,7 @@ from app.core.pricebot_router import pick_pricebot
from app.db.session import SessionLocal
from app.repositories import comparison as crud_compare
from app.repositories import risk as risk_repo
from app.services.comparison_llm_backfill import backfill_comparison_llm_cost
logger = logging.getLogger("shagua.compare")
@@ -80,7 +81,7 @@ def _harvest_running_blocking(
def _harvest_done_blocking(
trace_id: str, user_id: int | None, done_params: dict, business_type: str,
device_id: str | None, device_info: dict | None, trace_url: str | None,
) -> None:
) -> int:
with SessionLocal() as db:
rec, newly_success = crud_compare.harvest_done(
db, trace_id=trace_id, user_id=user_id, done_params=done_params,
@@ -102,6 +103,7 @@ def _harvest_done_blocking(
# 不在此处发邀请奖:#113 已把发奖口径从「比价」移到「实际下单」(order.py),harvest
# 只记录比价、不发奖。否则比价先于下单 + try_reward 幂等闸会让奖落在「比价」这步,
# 架空 #113 的「下单才发奖」防刷意图(newly_success 仅留作日志观测)。
return rec.id
def _harvest_abort_blocking(
@@ -243,7 +245,12 @@ async def intent_precoupon_step(
@router.post("/price/step", summary="外卖比价 Phase 2 步进 (透传 + done 落库)")
async def price_step(request: Request, user: OptionalUser, db: DbSession) -> dict[str, Any]:
async def price_step(
request: Request,
background_tasks: BackgroundTasks,
user: OptionalUser,
db: DbSession,
) -> dict[str, Any]:
_ensure_compare_allowed(user, db)
resp, trace_id, meta = await _forward(request, "/api/price/step", user)
# 最终 done 帧(command=done 且 continue=false)→ harvest 更新成终态。
@@ -252,12 +259,15 @@ async def price_step(request: Request, user: OptionalUser, db: DbSession) -> dic
if action.get("command") == "done" and not resp.get("continue", True):
done_params = action.get("params") or {}
try:
await run_in_threadpool(
record_id = await run_in_threadpool(
_harvest_done_blocking, trace_id, (user.id if user else None),
done_params, "food",
meta.get("device_id"), meta.get("device_info"),
resp.get("trace_url") or done_params.get("trace_url"),
)
background_tasks.add_task(
backfill_comparison_llm_cost, record_id, trace_id
)
except Exception as e: # noqa: BLE001
logger.warning("harvest_done failed trace=%s: %s", trace_id, e)
return resp
+2 -30
View File
@@ -16,8 +16,6 @@ import logging
from fastapi import APIRouter, BackgroundTasks, HTTPException, Query, status
from app.api.deps import CurrentUser, DbSession
from app.db.session import SessionLocal
from app.models.comparison import ComparisonRecord
from app.repositories import comparison as crud_compare
from app.repositories import risk as risk_repo
from app.schemas.compare_record import (
@@ -30,8 +28,7 @@ from app.schemas.compare_record import (
ComparisonRecordOut,
ComparisonRecordPage,
)
from app.services.llm_cost import compute_llm_cost, get_llm_prices
from app.services.pricebot_llm_calls import fetch_llm_calls
from app.services.comparison_llm_backfill import backfill_comparison_llm_cost
logger = logging.getLogger("shagua.compare_record")
@@ -121,32 +118,7 @@ def report_record(
def _backfill_llm_calls(record_id: int, trace_id: str) -> None:
"""后台回填本次比价的 LLM 调用明细 + 派生 llm_call_count/retry_count。
独立 DB session(请求 session 此时已关);拉取/写库失败只 log,绝不影响已落库的上报"""
calls = fetch_llm_calls(trace_id)
if not calls:
return
db = SessionLocal()
try:
rec = db.get(ComparisonRecord, record_id)
if rec is None:
return
rec.llm_calls = calls
rec.llm_call_count = len(calls)
rec.retry_count = sum(1 for c in calls if c.get("error"))
# token 累加(usage 已被 pricebot llm_client 归一为 prompt/completion_tokens;
# error 的调用 usage 可能为 None,or {} 兜底)
rec.input_tokens = sum((c.get("usage") or {}).get("prompt_tokens") or 0 for c in calls)
rec.output_tokens = sum((c.get("usage") or {}).get("completion_tokens") or 0 for c in calls)
# 本次比价 LLM 成本(元)+ 当时单价快照:按 app_config 现价逐模型算好冻结(services/llm_cost.py)。
rec.llm_cost_yuan, rec.llm_price_snapshot = compute_llm_cost(calls, get_llm_prices(db))
db.commit()
logger.info(
"backfill llm_calls trace=%s n=%d in_tok=%d out_tok=%d",
trace_id, len(calls), rec.input_tokens, rec.output_tokens,
)
except Exception as e: # noqa: BLE001 best-effort
logger.warning("backfill llm_calls failed trace=%s: %s", trace_id, e)
finally:
db.close()
backfill_comparison_llm_cost(record_id, trace_id)
@router.get(
+6
View File
@@ -377,6 +377,12 @@ class Settings(BaseSettings):
# 靠这个共享密钥头(X-Internal-Secret)校验,与 pricebot 侧 INTERNAL_API_SECRET 同值。
# 默认空 = 内部写端点关闭(返 503),启用前两边都要配上同一高熵串。
INTERNAL_API_SECRET: str = ""
# Current compare clients are persisted by server-side harvest. Repair any
# recent terminal rows left without LLM usage by transient upstream/auth failures.
LLM_COST_BACKFILL_ENABLED: bool = True
LLM_COST_BACKFILL_INTERVAL_SEC: int = 300
LLM_COST_BACKFILL_BATCH_SIZE: int = 100
LLM_COST_BACKFILL_LOOKBACK_DAYS: int = 30
# ===== 媒体文件(用户头像上传)=====
# 落盘根目录(data/ 已 gitignore,上传不进库);对外经 StaticFiles 挂在 MEDIA_URL_PREFIX。
+63
View File
@@ -0,0 +1,63 @@
"""Periodic repair worker for comparison records with missing LLM token cost."""
from __future__ import annotations
import asyncio
import contextlib
import logging
from app.core.config import settings
from app.services.comparison_llm_backfill import repair_missing_comparison_llm_costs
from app.services.pricebot_llm_calls import pricebot_llm_auth_ready
logger = logging.getLogger("shagua.llm_cost_backfill_worker")
async def _run_loop() -> None:
interval = max(60, int(settings.LLM_COST_BACKFILL_INTERVAL_SEC))
logger.info(
"LLM cost backfill worker started interval=%ss batch=%s lookback_days=%s",
interval,
settings.LLM_COST_BACKFILL_BATCH_SIZE,
settings.LLM_COST_BACKFILL_LOOKBACK_DAYS,
)
try:
while True:
try:
auth_ready = await asyncio.to_thread(pricebot_llm_auth_ready)
if auth_ready:
result = await asyncio.to_thread(
repair_missing_comparison_llm_costs,
limit=settings.LLM_COST_BACKFILL_BATCH_SIZE,
lookback_days=settings.LLM_COST_BACKFILL_LOOKBACK_DAYS,
)
logger.info("LLM cost backfill batch result=%s", result)
else:
logger.error(
"LLM cost backfill skipped: PriceBot internal auth is not ready"
)
except Exception: # noqa: BLE001
logger.exception("LLM cost backfill batch failed")
await asyncio.sleep(interval)
except asyncio.CancelledError:
logger.info("LLM cost backfill worker stopped")
raise
def start_llm_cost_backfill_worker() -> asyncio.Task | None:
if not settings.LLM_COST_BACKFILL_ENABLED:
logger.info("LLM cost backfill worker disabled")
return None
if not settings.INTERNAL_API_SECRET:
logger.warning(
"LLM cost backfill worker not started: INTERNAL_API_SECRET is empty"
)
return None
return asyncio.create_task(_run_loop(), name="llm-cost-backfill")
async def stop_llm_cost_backfill_worker(task: asyncio.Task | None) -> None:
if task is None:
return
task.cancel()
with contextlib.suppress(asyncio.CancelledError):
await task
+6
View File
@@ -60,6 +60,10 @@ from app.core.inactivity_reset_worker import (
start_inactivity_reset_worker,
stop_inactivity_reset_worker,
)
from app.core.llm_cost_backfill_worker import (
start_llm_cost_backfill_worker,
stop_llm_cost_backfill_worker,
)
from app.core.logging import setup_logging
from app.core.observe import RequestMetricsMiddleware
from app.core.observe_worker import (
@@ -103,6 +107,7 @@ async def lifespan(_: FastAPI) -> AsyncIterator[None]:
daily_exchange_task = start_daily_exchange_worker()
observe_task = start_observe_worker()
inactivity_task = start_inactivity_reset_worker()
llm_cost_backfill_task = start_llm_cost_backfill_worker()
try:
yield
finally:
@@ -112,6 +117,7 @@ async def lifespan(_: FastAPI) -> AsyncIterator[None]:
await stop_daily_exchange_worker(daily_exchange_task)
await stop_observe_worker(observe_task)
await stop_inactivity_reset_worker(inactivity_task)
await stop_llm_cost_backfill_worker(llm_cost_backfill_task)
await aclose_pricebot_client()
mt_meituan.close_client()
logger.info("shutting down")
+8 -1
View File
@@ -11,7 +11,7 @@ from __future__ import annotations
from datetime import datetime, timezone
from sqlalchemy import Boolean, DateTime, Integer, String, false, func
from sqlalchemy import Boolean, DateTime, Integer, String, Text, false, func
from sqlalchemy.orm import Mapped, mapped_column
from app.db.base import Base
@@ -62,6 +62,13 @@ class User(Base):
Boolean, nullable=False, default=False, server_default=false()
)
# 运营人工风险标记。与自动风控分值分开:这里表达人工复核结论,并保留可编辑备注,
# 供邀请提现、其他提现和用户管理三个页面统一展示。
is_high_risk: Mapped[bool] = mapped_column(
Boolean, nullable=False, default=False, server_default=false(), index=True
)
high_risk_note: Mapped[str | None] = mapped_column(Text, nullable=True)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), server_default=func.now(), nullable=False
)
+33 -9
View File
@@ -55,13 +55,22 @@ def _product_names_from_items(items: list | None) -> str | None:
def _derive(payload: ComparisonRecordIn) -> dict:
"""从上报 payload 派生结构化列(best/saved/is_source_best/status)。"""
results = payload.comparison_results
_pr = payload.platform_results or {}
# 最优 = rank 最小的一条;协议已升序,但不信顺序,显式按 rank/price 兜底取最小价。
best = None
def _is_short(r) -> bool:
# 缺菜(漏菜)店: 少买了菜总价虚低, 不参与最优评选。逐平台 skipped 在 platform_results, 行里没有。
# platform_results 内层结构宽松(pricebot/老客户端透传, 可伪造), 值非 dict 时按"不缺菜"处理, 不崩。
info = _pr.get(r.platform_id) if r.platform_id else None
return isinstance(info, dict) and (info.get("skipped_dish_count") or 0) > 0
# 最优 = 非缺菜里 rank 最小(=最便宜)的一条;协议已升序,但不信顺序,显式按 rank/price 取。
# 源平台永远全菜, 故全目标缺菜时回落到源(is_source_best、saved=0), 不把虚低价当最低。
priced = [r for r in results if r.price is not None]
if priced:
clean = [r for r in priced if not _is_short(r)]
best = None
if clean:
best = min(
priced,
clean,
key=lambda r: (r.rank if r.rank is not None else 10**9, r.price),
)
@@ -196,14 +205,29 @@ def upsert_record(
# ============================================================
def _derive_from_results(results: list[dict]) -> dict:
def _derive_from_results(
results: list[dict], platform_results: dict | None = None
) -> dict:
"""从 done 帧 comparison_results(pricebot 原始 dict 列表)派生结构化列。
等价 _derive,但吃原始字段(is_source/price/rank/platform_id/store_name...)而非 pydantic 对象"""
等价 _derive,但吃原始字段(is_source/price/rank/platform_id/store_name...)而非 pydantic 对象
platform_results(done.params.platform_results): 逐平台 skipped_dish_count 在这里(行里没有)
传入则派生 best 时排除缺菜(漏菜) 少买了菜总价虚低, 不能当记录级"最低价"/算虚假省额;
源平台永远全菜, 故全目标缺菜时 best 回落到源(is_source_best不虚报省)不传 rank/price, 行为不变"""
_pr = platform_results or {}
def _is_short(r: dict) -> bool:
# platform_results 内层结构宽松(pricebot/客户端透传), 值非 dict 时按"不缺菜"处理, 不崩。
pid = r.get("platform_id")
info = _pr.get(pid) if pid else None
return isinstance(info, dict) and (info.get("skipped_dish_count") or 0) > 0
priced = [r for r in results if r.get("price") is not None]
clean = [r for r in priced if not _is_short(r)] # 缺菜店排除出最优评选
best = None
if priced:
if clean:
best = min(
priced,
clean,
key=lambda r: (r.get("rank") if r.get("rank") is not None else 10**9, r["price"]),
)
src_row = next((r for r in results if r.get("is_source")), None)
@@ -389,7 +413,7 @@ def harvest_done(
返回 (记录, 是否本次****落成 success)供调用方据此幂等发一次邀请奖
行不存在(理论上帧0已建;防御)则新建"""
results = done_params.get("comparison_results") or []
derived = _derive_from_results(results)
derived = _derive_from_results(results, done_params.get("platform_results"))
# 菜品:pricebot 已把源单菜品塞进 comparison_results[源行].items
items = next((r.get("items") or [] for r in results if r.get("is_source")), [])
fields = dict(
+5
View File
@@ -93,6 +93,11 @@ def get_user_by_id(db: Session, user_id: int) -> User | None:
return db.get(User, user_id)
def user_exists(db: Session, user_id: int) -> bool:
"""只查主键判断用户是否存在,避免只读统计接口依赖完整用户表结构。"""
return db.scalar(select(User.id).where(User.id == user_id)) is not None
def get_user_by_phone(db: Session, phone: str) -> User | None:
stmt = select(User).where(User.phone == phone)
return db.execute(stmt).scalar_one_or_none()
+135 -8
View File
@@ -6,6 +6,7 @@
"""
from __future__ import annotations
import logging
import re
import unicodedata
import uuid
@@ -20,7 +21,6 @@ from app.core.config import settings
from app.core.rewards import COIN_PER_CENT, coins_to_cents
from app.integrations import wxpay
from app.models.user import User
from app.repositories.user import apply_wechat_display_identity
from app.models.wallet import (
CashTransaction,
CoinAccount,
@@ -29,12 +29,55 @@ from app.models.wallet import (
WechatTransferAuthorization,
WithdrawOrder,
)
from app.repositories.user import apply_wechat_display_identity
from app.services import notification_events
logger = logging.getLogger(__name__)
# 微信转账终态:成功 / 失败(失败/取消/关闭都退款)
_WX_STATE_SUCCESS = "SUCCESS"
_WX_STATE_FAILED = {"FAIL", "CANCELLED", "CLOSED"}
_WX_STATE_WAIT_CONFIRM = "WAIT_USER_CONFIRM" # 用户还没在微信确认页确认
_WX_FAIL_REASON_LABELS = {
"ACCOUNT_FROZEN": "用户微信账户被冻结",
"ACCOUNT_NOT_EXIST": "用户微信账户不存在",
"BANK_CARD_ACCOUNT_ABNORMAL": "用户银行卡已销户、冻结、作废或挂失",
"BANK_CARD_BANK_INFO_WRONG": "用户登记的银行或分支行信息有误",
"BANK_CARD_CARD_INFO_WRONG": "用户银行卡户名或卡号有误",
"BANK_CARD_COLLECTIONS_ABOVE_QUOTA": "用户银行卡收款达到限额",
"BANK_CARD_PARAM_ERROR": "用户收款银行卡信息错误",
"BANK_CARD_STATUS_ABNORMAL": "用户银行卡状态异常",
"BLOCK_B2C_USERLIMITAMOUNT_BSRULE_MONTH": "用户本月转账收款已达限额",
"BLOCK_B2C_USERLIMITAMOUNT_MONTH": "用户账户存在风险,本月收款受限",
"DAY_RECEIVED_COUNT_EXCEED": "用户当日收款次数已达上限",
"DAY_RECEIVED_QUOTA_EXCEED": "用户当日收款额度已达上限",
"EXCEEDED_ESTIMATED_AMOUNT": "转账金额超过预约金额范围",
"ID_CARD_NOT_CORRECT": "收款人身份证校验不通过",
"MCH_CANCEL": "商户已撤销付款",
"MERCHANT_REJECT": "商户转账验密人已驳回",
"MERCHANT_NOT_CONFIRM": "商户转账验密人超时未确认",
"NAME_NOT_CORRECT": "收款人姓名校验不通过",
"OPENID_INVALID": "用户 OpenID 无效或不属于当前 AppID",
"OTHER_FAIL_REASON_TYPE": "微信返回其他失败原因",
"OVERDUE_CLOSE": "超过微信系统重试期,订单自动关闭",
"PAYEE_ACCOUNT_ABNORMAL": "用户微信账户收款异常",
"PAYER_ACCOUNT_ABNORMAL": "商户账户付款受限",
"PRODUCT_AUTH_CHECK_FAIL": "商户未开通转账权限或权限已冻结",
"REALNAME_ACCOUNT_RECEIVED_QUOTA_EXCEED": "用户微信实名账户收款受限",
"REAL_NAME_CHECK_FAIL": "用户未完成微信实名认证",
"RECEIVE_ACCOUNT_NOT_CONFIGURE": "商户未配置收款用户列表",
"RESERVATION_INFO_NOT_MATCH": "转账信息与预约信息不一致",
"RESERVATION_SCENE_NOT_MATCH": "转账场景与预约场景不一致",
"RESERVATION_STATE_INVALID": "预约转账单状态异常",
"TRANSFER_QUOTA_EXCEED": "用户单笔收款额度已达上限",
"TRANSFER_REMARK_SET_FAIL": "微信转账备注设置失败",
"TRANSFER_RISK": "该笔转账存在风险,已被微信拦截",
"TRANSFER_SCENE_INVALID": "商户未获取当前转账场景",
"TRANSFER_SCENE_UNAVAILABLE": "当前转账场景暂不可用",
"RELATED_ORDER_TRANSFER_AMOUNT_EXCEED": "关联订单累计付款金额超过上限",
"RELATED_ORDER_TRANSFER_COUNT_EXCEED": "关联订单累计付款次数超过上限",
"BUDGET_NOT_ENOUGH": "商户预算资金不足",
}
# 占用新人档「一次性」资格的提现状态:进行中(reviewing/pending)或成功打款(success)。
# 被拒/转账失败/解绑退回(rejected/failed,均已退款、钱没到手)不在此列 → 新人档恢复可提
# (2026-07-16 修正:此前判定不看状态,解绑微信退回后 0.1 被误判已用、资格永久锁死)。
@@ -479,6 +522,19 @@ def _try_deduct_cash(db: Session, user_id: int, amount_cents: int, source: str =
def _add_cash(db: Session, user_id: int, amount_cents: int, source: str = "coin_cash") -> int:
"""原子增加指定账户余额(退款用),返回加后余额。source 决定退回哪个账户(两账户隔离)。"""
# Normal withdrawals always create the account before deducting funds. This
# fallback covers legacy rows, hand-written fixtures, and broken migrations:
# a missing balance snapshot must not make a legitimate refund fail with 500.
if db.get(CoinAccount, user_id) is None:
logger.error(
"withdraw refund found missing coin_account; recreating empty account: "
"user_id=%s source=%s amount_cents=%s",
user_id,
source,
amount_cents,
)
get_or_create_account(db, user_id, commit=False)
col = _balance_col(source)
db.execute(
update(CoinAccount)
@@ -582,6 +638,32 @@ def _wx_not_found(result: dict) -> bool:
return "NOT_FOUND" in str(code)
def _wechat_api_error_reason(data: object) -> str:
"""Format a non-200 WeChat API response for operator display."""
if not isinstance(data, dict):
return f"微信发起转账失败:{data}"
code = str(data.get("code") or "").strip()
message = str(data.get("message") or "").strip()
if code and message:
return f"微信发起转账失败:{message}{code}"
return f"微信发起转账失败:{message or code or '未知错误'}"
def _wechat_terminal_failure_reason(data: dict, state: str) -> str:
"""Translate WeChat query ``fail_reason`` while preserving unknown codes."""
code = str(data.get("fail_reason") or "").strip()
if code:
label = _WX_FAIL_REASON_LABELS.get(code)
if label:
return f"微信转账失败:{label}{code}"
return f"微信转账失败:{code}"
if state == "CANCELLED":
return "微信转账已撤销(CANCELLED"
if state == "CLOSED":
return "微信转账已关闭(CLOSED"
return f"微信转账失败(状态:{state or 'FAIL'}"
def _settle_after_ambiguous(db: Session, order: WithdrawOrder, reason: str) -> None:
"""转账调用结果不明(超时/异常/非200)时,**先查单再决定**,绝不盲目退款(防退款后又到账)。
- 微信查到 SUCCESS 钱已出, success,不退款
@@ -607,13 +689,17 @@ def _settle_after_ambiguous(db: Session, order: WithdrawOrder, reason: str) -> N
state = q["data"].get("state", "")
order.wechat_state = state
order.transfer_bill_no = q["data"].get("transfer_bill_no") or order.transfer_bill_no
if state == _WX_STATE_SUCCESS:
order.status = "success"
order.transfer_bill_no = q["data"].get("transfer_bill_no")
db.commit()
notification_events.notify_withdraw_success(db, order) # PRD #3 提现到账
elif state in _WX_STATE_FAILED:
_refund_withdraw(db, order, reason=reason)
_refund_withdraw(
db,
order,
reason=_wechat_terminal_failure_reason(q["data"], state),
)
else:
order.package_info = q["data"].get("package_info") or order.package_info
db.commit()
@@ -967,6 +1053,14 @@ def _apply_transfer_result(db: Session, order: WithdrawOrder, data: dict) -> Wit
order.package_info = data.get("package_info") # 免确认转账无此字段(None);确认模式带它供拉确认页
if data.get("state") == _WX_STATE_SUCCESS:
order.status = "success"
elif data.get("state") in _WX_STATE_FAILED:
_refund_withdraw(
db,
order,
reason=_wechat_terminal_failure_reason(data, str(data.get("state") or "")),
)
db.refresh(order)
return order
db.commit()
db.refresh(order)
if order.status == "success": # 免确认转账直接到账 → PRD #3 提现到账
@@ -1011,7 +1105,11 @@ def execute_withdraw_transfer(db: Session, order: WithdrawOrder) -> WithdrawOrde
return order
if result["status_code"] != 200:
# 金额安全:查转账单后定夺,绝不盲退(未创建→退款,已创建→按真实状态)
_settle_after_ambiguous(db, order, reason=str(result["data"].get("message") or result["data"]))
_settle_after_ambiguous(
db,
order,
reason=_wechat_api_error_reason(result["data"]),
)
# 授权有效性:回查授权单,微信侧已失效(用户关闭/风控)→标 closed,下次提现自动回退方式一重新授权
_refresh_active_auth(db, order.user_id)
db.refresh(order)
@@ -1039,7 +1137,11 @@ def execute_withdraw_transfer(db: Session, order: WithdrawOrder) -> WithdrawOrde
return order
if result["status_code"] != 200:
_settle_after_ambiguous(db, order, reason=str(result["data"].get("message") or result["data"]))
_settle_after_ambiguous(
db,
order,
reason=_wechat_api_error_reason(result["data"]),
)
db.refresh(order)
return order
@@ -1096,19 +1198,24 @@ def refresh_withdraw_status(
).scalar_one_or_none()
if order is None:
raise WithdrawOrderNotFound
if order.status != "pending":
return order # 已终态,不再查
if order.status not in {"pending", "failed"}:
return order
enrich_failed_order = order.status == "failed"
try:
result = wxpay.query_transfer(out_bill_no)
except wxpay.WxPayNotConfiguredError:
raise
except Exception as exc: # noqa: BLE001 - 查单失败不能把运营后台打成 500
if enrich_failed_order:
return order
order.fail_reason = f"微信查单异常,保持pending: {exc}"[:256]
db.commit()
db.refresh(order)
return order
if result["status_code"] != 200:
if enrich_failed_order:
return order
if _wx_not_found(result):
# 微信明确无此单 → 转账从未创建(如崩溃在扣款后/调用前),退款安全
_refund_withdraw(db, order, reason="微信无此单,已退回")
@@ -1117,12 +1224,32 @@ def refresh_withdraw_status(
state = result["data"].get("state", "")
order.wechat_state = state
order.transfer_bill_no = (
result["data"].get("transfer_bill_no") or order.transfer_bill_no
)
if enrich_failed_order:
if state in _WX_STATE_FAILED:
order.fail_reason = _wechat_terminal_failure_reason(
result["data"], state
)[:256]
elif state == _WX_STATE_SUCCESS:
order.fail_reason = (
"资金状态异常:本地已退款,但微信查单显示已到账,请人工核查"
)
db.commit()
db.refresh(order)
return order
if state == _WX_STATE_SUCCESS:
order.status = "success"
db.commit()
notification_events.notify_withdraw_success(db, order) # PRD #3 提现到账
elif state in _WX_STATE_FAILED:
_refund_withdraw(db, order, reason=f"微信转账状态 {state}")
_refund_withdraw(
db,
order,
reason=_wechat_terminal_failure_reason(result["data"], state),
)
elif state == _WX_STATE_WAIT_CONFIRM and cancel_if_unconfirmed:
# 用户从确认页回来了却仍未确认 → 视为放弃:撤销微信单(防事后确认导致重复打款)后退款。
# 撤单失败(可能已被确认进 ACCEPTED 的竞态)则保持 pending,等下次查询。
+4
View File
@@ -85,6 +85,10 @@ class ComparisonResultIn(BaseModel):
status: str | None = None
# 门店打烊原因(price 为 None 时带): 与 status="store_closed" 等价的更早信号, 一并落库供前端兜底判打烊。
store_closed: str | None = None
# 该平台缺菜(漏菜)数量(pricebot 冗余进 comparison_results 行): 有价但少买了菜时 >0。
# 记录页三平台网格据此逐格标"缺少 X 个菜品"(网格只拿 comparison_results, 拿不到 platform_results)。
# 必须显式声明: 落库走 model_dump(), 不声明会被 pydantic 静默丢弃 → 记录页拿不到数量。
skipped_dish_count: int | None = None
class ComparisonRecordIn(BaseModel):
+156
View File
@@ -0,0 +1,156 @@
"""Persist and repair comparison-record LLM token costs."""
from __future__ import annotations
import logging
import time
from datetime import UTC, datetime, timedelta
from sqlalchemy import select
from app.core.config import settings
from app.core.rewards import CN_TZ
from app.db.session import SessionLocal
from app.models.app_config import AppConfig
from app.models.comparison import ComparisonRecord
from app.services.llm_cost import compute_llm_cost, get_llm_prices
from app.services.pricebot_llm_calls import fetch_llm_calls
logger = logging.getLogger("shagua.comparison_llm_backfill")
def _utc_to_beijing_naive(value: datetime) -> datetime:
"""Convert a DB UTC timestamp to comparison_record's Beijing wall-clock."""
if value.tzinfo is None:
value = value.replace(tzinfo=UTC)
return value.astimezone(CN_TZ).replace(tzinfo=None)
def _store_calls(record_id: int, trace_id: str, calls: list[dict]) -> bool:
"""Store calls and all derived fields atomically."""
with SessionLocal() as db:
rec = db.get(ComparisonRecord, record_id)
if rec is None or rec.trace_id != trace_id:
logger.warning(
"LLM cost backfill record mismatch record_id=%s trace=%s",
record_id,
trace_id,
)
return False
# Never recalculate a frozen historical cost with a newer price config.
if rec.llm_cost_yuan is not None and rec.llm_calls:
return False
rec.llm_calls = calls
rec.llm_call_count = len(calls)
rec.retry_count = sum(1 for call in calls if call.get("error"))
rec.input_tokens = sum(
(call.get("usage") or {}).get("prompt_tokens") or 0 for call in calls
)
rec.output_tokens = sum(
(call.get("usage") or {}).get("completion_tokens") or 0 for call in calls
)
rec.llm_cost_yuan, rec.llm_price_snapshot = compute_llm_cost(
calls, get_llm_prices(db)
)
db.commit()
logger.info(
"LLM cost backfilled trace=%s calls=%d input_tokens=%d "
"output_tokens=%d cost=%s",
trace_id,
len(calls),
rec.input_tokens,
rec.output_tokens,
rec.llm_cost_yuan,
)
return True
def backfill_comparison_llm_cost(
record_id: int,
trace_id: str,
*,
attempts: int = 3,
retry_delays: tuple[float, ...] = (1.0, 3.0),
) -> bool:
"""Fetch and persist one record, retrying short-lived upstream races."""
if not settings.INTERNAL_API_SECRET or not trace_id:
logger.warning(
"LLM cost backfill skipped trace=%s: INTERNAL_API_SECRET is not configured",
trace_id,
)
return False
total_attempts = max(1, attempts)
for attempt in range(total_attempts):
calls = fetch_llm_calls(trace_id)
if calls:
try:
return _store_calls(record_id, trace_id, calls)
except Exception: # noqa: BLE001 - background repair must stay alive
logger.exception(
"LLM cost store failed trace=%s record_id=%s",
trace_id,
record_id,
)
return False
if attempt + 1 < total_attempts:
delay = retry_delays[min(attempt, len(retry_delays) - 1)] if retry_delays else 0
if delay > 0:
time.sleep(delay)
logger.warning(
"LLM cost backfill has no calls trace=%s record_id=%s attempts=%d",
trace_id,
record_id,
total_attempts,
)
return False
def repair_missing_comparison_llm_costs(
*,
limit: int = 100,
lookback_days: int = 30,
) -> dict[str, int]:
"""Repair a bounded batch of recent terminal records with missing cost."""
cutoff = datetime.now(CN_TZ).replace(tzinfo=None) - timedelta(
days=max(1, lookback_days)
)
with SessionLocal() as db:
# app_config has no price history. Repricing a record from before the
# current config became effective would fabricate a historical cost, so
# only repair records at/after that timestamp.
price_config_updated_at = db.execute(
select(AppConfig.updated_at).where(AppConfig.key == "llm_token_price")
).scalar_one_or_none()
date_conditions = [ComparisonRecord.created_at >= cutoff]
if price_config_updated_at is not None:
date_conditions.append(
ComparisonRecord.created_at
>= _utc_to_beijing_naive(price_config_updated_at)
)
candidates = list(
db.execute(
select(ComparisonRecord.id, ComparisonRecord.trace_id)
.where(
*date_conditions,
ComparisonRecord.status.in_(("success", "failed")),
ComparisonRecord.llm_cost_yuan.is_(None),
)
.order_by(ComparisonRecord.created_at.desc(), ComparisonRecord.id.desc())
.limit(max(1, limit))
).all()
)
repaired = 0
for record_id, trace_id in candidates:
if backfill_comparison_llm_cost(
record_id, trace_id, attempts=1, retry_delays=()
):
repaired += 1
return {
"candidates": len(candidates),
"repaired": repaired,
"unresolved": len(candidates) - repaired,
}
+57 -9
View File
@@ -20,19 +20,67 @@ from app.core.pricebot_router import pick_pricebot
logger = logging.getLogger("shagua.pricebot_llm")
def pricebot_llm_auth_ready() -> bool:
"""Verify every configured PriceBot instance accepts the shared secret."""
secret = settings.INTERNAL_API_SECRET
if not secret:
logger.error("PriceBot LLM auth check failed: INTERNAL_API_SECRET is empty")
return False
for base in settings.pricebot_instances:
url = f"{base.rstrip('/')}/api/internal/llm_calls/__auth_probe__"
try:
resp = httpx.get(
url, headers={"X-Internal-Secret": secret}, timeout=3.0
)
except Exception as exc: # noqa: BLE001
logger.error("PriceBot LLM auth check unavailable base=%s: %s", base, exc)
return False
if resp.status_code != 200:
logger.error(
"PriceBot LLM auth check rejected base=%s status=%s; "
"verify both services use the same INTERNAL_API_SECRET",
base,
resp.status_code,
)
return False
return True
def fetch_llm_calls(trace_id: str) -> list[dict]:
"""返回该次比价的 LLM 调用明细列表(每条 {scene,model,input_messages,output,usage,latency_ms,error});
未配密钥 / trace_id / 拉取失败 []"""
secret = settings.INTERNAL_API_SECRET
if not secret or not trace_id:
return []
base = pick_pricebot(trace_id).rstrip("/")
url = f"{base}/api/internal/llm_calls/{trace_id}"
try:
resp = httpx.get(url, headers={"X-Internal-Secret": secret}, timeout=5.0)
if resp.status_code == 200:
return resp.json().get("calls", []) or []
logger.warning("fetch_llm_calls trace=%s status=%s", trace_id, resp.status_code)
except Exception as e: # noqa: BLE001 — best-effort,任何异常都不该影响上报
logger.warning("fetch_llm_calls trace=%s failed: %s", trace_id, e)
preferred = pick_pricebot(trace_id)
# LLM JSONL is instance-local. If the cluster topology changed after a
# historical trace was created, consistent hashing may now point elsewhere;
# probe the remaining configured instances only when the preferred one is empty.
bases = [preferred, *(base for base in settings.pricebot_instances if base != preferred)]
for base in bases:
url = f"{base.rstrip('/')}/api/internal/llm_calls/{trace_id}"
try:
resp = httpx.get(url, headers={"X-Internal-Secret": secret}, timeout=5.0)
if resp.status_code == 200:
calls = resp.json().get("calls", []) or []
if calls:
return calls
continue
if resp.status_code in (401, 403):
logger.error(
"fetch_llm_calls rejected trace=%s base=%s status=%s; "
"INTERNAL_API_SECRET differs between app-server and PriceBot",
trace_id,
base,
resp.status_code,
)
else:
logger.warning(
"fetch_llm_calls trace=%s base=%s status=%s",
trace_id,
base,
resp.status_code,
)
except Exception as e: # noqa: BLE001 — best-effort
logger.warning("fetch_llm_calls trace=%s base=%s failed: %s", trace_id, base, e)
return []
+40
View File
@@ -0,0 +1,40 @@
# 比价 TOKEN 成本采集与补偿
## 部署前置
App Server 与 PriceBot 使用各自独立的 `.env`,但下面的值必须完全一致:
- `/opt/shaguabijia-app-server/.env`
- `/opt/pricebot-backend/.env`
- 配置项:`INTERNAL_API_SECRET`
不要把密钥原文写入日志、命令历史或 Git。修改后同时重启两个服务。
App Server 启动后会逐个探测 `PRICEBOT_INSTANCES` 的内部读取接口。鉴权不一致时会记录
`PriceBot LLM auth check rejected`,并跳过本轮补偿,避免对所有缺失记录重复发送失败请求。
## 数据链路
1. 当前客户端由 App Server 在 PriceBot 最终 `done` 帧到达时 harvest 比价记录。
2. harvest 成功后立即异步读取同一 `trace_id` 的 LLM 调用,冻结 Token、成本和单价快照。
3. 周期 worker 扫描近期 `success/failed``llm_cost_yuan IS NULL` 的记录进行补偿;
为避免用现价伪造历史成本,只处理当前单价配置生效时间之后的记录。
4. 管理后台顶部“平均 TOKEN 成本”使用筛选范围内已冻结成本的数据库平均值。
## 上线验收(只读 SQL
```sql
SELECT
(created_at AT TIME ZONE 'Asia/Shanghai')::date AS day,
count(*) AS records,
count(llm_cost_yuan) AS cost_records,
round(avg(llm_cost_yuan)::numeric, 6) AS avg_token_cost
FROM comparison_record
WHERE created_at >= now() - interval '3 days'
GROUP BY 1
ORDER BY 1 DESC;
```
新产生的正常终态比价记录应在短时间内写入 `input_tokens``output_tokens`
`llm_cost_yuan`。历史记录只有在 PriceBot 的对应 trace JSONL 仍保留时才能准确回填;
原始调用已经清理的记录不能用估算值冒充真实成本。
+6 -6
View File
@@ -175,22 +175,22 @@ def seed(db) -> list[Feedback]:
# 普通反馈 · 新端(带环境快照)· 无图
fb("13255550001", "签到金币到账有时候会延迟一两分钟,能不能做成实时到账?",
source="profile",
app_version="2.3.1", device_model="PJA110", rom_name="ColorOS", android_version="14",
app_version="2.3.1", device_model="PJA110", rom_name="ColorOS 14", 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",
app_version="2.3.1", device_model="M2012K11AC", rom_name="MIUI 14", 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",
app_version="2.3.0", device_model="V2309A", rom_name="OriginOS 4", 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",
app_version="2.3.1", device_model="23078RKD5C", rom_name="MIUI 14", android_version="14",
created=ago(hours=3)),
# 比价反馈 · scene=比价太慢 · 历史数据(env 全 NULL、contact 有值)
fb("13255550004", "比价转圈太久了,经常要等十几秒才出结果,体验不太好。",
@@ -206,7 +206,7 @@ def seed(db) -> list[Feedback]:
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",
app_version="2.2.8", device_model="PJA110", rom_name="ColorOS 13", android_version="13",
created=ago(days=2)),
# ===== 未采纳 rejected(带原因 + 回复)=====
@@ -214,7 +214,7 @@ def seed(db) -> list[Feedback]:
source="comparison", scene="价格不准",
status="rejected", reject_reason="截图价格为限时活动价且已过期,不满足「长期可复现更低价」条件,暂不采纳。",
admin_reply="感谢参与,本次未通过,欢迎继续上报有效更低价~",
app_version="2.3.0", device_model="M2012K11AC", rom_name="MIUI", android_version="13",
app_version="2.3.0", device_model="M2012K11AC", rom_name="MIUI 14", android_version="13",
created=ago(days=3)),
]
db.add_all(feedbacks)
+163
View File
@@ -8,6 +8,7 @@ from sqlalchemy import delete
from app.admin.repositories import ad_revenue
from app.db.session import SessionLocal
from app.models.ad_ecpm import AdEcpmRecord
from app.models.ad_feed_reward import AdFeedRewardRecord
from app.models.ad_pangle_revenue import AdPangleDailyRevenue
from app.models.ad_reward import AdRewardRecord
from app.models.user import User
@@ -208,3 +209,165 @@ def test_reward_video_incomplete_playback_has_zero_revenue() -> None:
db.execute(delete(User).where(User.phone == phone))
db.commit()
db.close()
def test_business_category_ecpm_merges_draw_feed_and_both_video_types() -> None:
db = SessionLocal()
phone = "18800009993"
category_date = "2040-02-05"
try:
user = User(phone=phone, username="29999999993", register_channel="sms")
db.add(user)
db.flush()
db.add_all([
AdEcpmRecord(
user_id=user.id, ad_type="draw", ad_session_id="category-draw",
app_env="prod", our_code_id="prod-draw", ecpm_raw="10000",
report_date=category_date, created_at=datetime(2040, 2, 5, 1, tzinfo=UTC),
),
AdEcpmRecord(
user_id=user.id, ad_type="feed", ad_session_id="category-feed",
app_env="prod", our_code_id="prod-draw", ecpm_raw="20000",
report_date=category_date, created_at=datetime(2040, 2, 5, 2, tzinfo=UTC),
),
AdEcpmRecord(
user_id=user.id, ad_type="reward_video", ad_session_id="category-reward",
app_env="prod", our_code_id="prod-reward", ecpm_raw="30000",
report_date=category_date, created_at=datetime(2040, 2, 5, 3, tzinfo=UTC),
),
AdEcpmRecord(
user_id=user.id, ad_type="withdrawal_video", ad_session_id="category-withdraw",
app_env="prod", our_code_id="prod-reward", ecpm_raw="50000",
report_date=category_date, created_at=datetime(2040, 2, 5, 4, tzinfo=UTC),
),
])
db.commit()
result = ad_revenue.ad_revenue_report(
db,
date_from=category_date,
date_to=category_date,
user_id=user.id,
app_env="prod",
revenue_scope="all",
)
assert result["category_stats"] == {
"draw": {"impressions": 2, "revenue_yuan": 0.3, "ecpm_yuan": 150.0},
"video": {"impressions": 2, "revenue_yuan": 0.8, "ecpm_yuan": 400.0},
}
finally:
db.rollback()
db.execute(delete(AdEcpmRecord).where(AdEcpmRecord.report_date == category_date))
db.execute(delete(User).where(User.phone == phone))
db.commit()
db.close()
def test_feed_reward_details_keep_each_record_adn() -> None:
db = SessionLocal()
phone = "18800009994"
detail_date = "2040-02-06"
try:
user = User(phone=phone, username="29999999994", register_channel="sms")
db.add(user)
db.flush()
db.add_all([
AdFeedRewardRecord(
client_event_id="detail-adn-pangle", user_id=user.id,
reward_date=detail_date, duration_seconds=20, unit_count=1,
ecpm_raw="12000", adn="pangle", slot_id="rit-pangle",
ad_type="draw", feed_scene="coupon", trace_id="detail-adn-trace",
app_env="prod", our_code_id="104098712", coin=12, status="granted",
created_at=datetime(2040, 2, 6, 1, tzinfo=UTC),
),
AdFeedRewardRecord(
client_event_id="detail-adn-gdt", user_id=user.id,
reward_date=detail_date, duration_seconds=20, unit_count=1,
ecpm_raw="25000", adn="gdt", slot_id="rit-gdt",
ad_type="draw", feed_scene="coupon", trace_id="detail-adn-trace",
app_env="prod", our_code_id="104098712", coin=25, status="granted",
created_at=datetime(2040, 2, 6, 2, tzinfo=UTC),
),
])
db.commit()
result = ad_revenue.ad_revenue_report(
db, date_from=detail_date, date_to=detail_date,
user_id=user.id, app_env="prod", revenue_scope="all",
)
item = next(row for row in result["items"] if row["event_key"].startswith("feedgrp-"))
assert [detail["adn"] for detail in item["sub_rewards"]] == ["pangle", "gdt"]
assert [detail["slot_id"] for detail in item["sub_rewards"]] == ["rit-pangle", "rit-gdt"]
finally:
db.rollback()
db.execute(delete(AdFeedRewardRecord).where(AdFeedRewardRecord.reward_date == detail_date))
db.execute(delete(User).where(User.phone == phone))
db.commit()
db.close()
def test_feed_reward_source_fallback_requires_unique_trace_and_ecpm() -> None:
db = SessionLocal()
phone = "18800009995"
fallback_date = "2040-02-07"
try:
user = User(phone=phone, username="29999999995", register_channel="sms")
db.add(user)
db.flush()
db.add_all([
AdFeedRewardRecord(
client_event_id="source-fallback-unique", user_id=user.id,
reward_date=fallback_date, duration_seconds=3, unit_count=0,
ecpm_raw="4700", ad_session_id="flow-session", trace_id="source-trace",
app_env="prod", our_code_id="104098712", coin=0, status="too_short",
ad_type="draw", feed_scene="comparison",
created_at=datetime(2040, 2, 7, 1, tzinfo=UTC),
),
AdFeedRewardRecord(
client_event_id="source-fallback-ambiguous", user_id=user.id,
reward_date=fallback_date, duration_seconds=3, unit_count=0,
ecpm_raw="4800", ad_session_id="flow-session", trace_id="source-trace",
app_env="prod", our_code_id="104098712", coin=0, status="too_short",
ad_type="draw", feed_scene="comparison",
created_at=datetime(2040, 2, 7, 2, tzinfo=UTC),
),
AdEcpmRecord(
user_id=user.id, ad_type="draw", ad_session_id="impression-unique",
trace_id="source-trace", ecpm_raw="4700", adn="baidu", slot_id="rit-baidu",
app_env="prod", our_code_id="104098712", report_date=fallback_date,
created_at=datetime(2040, 2, 7, 1, tzinfo=UTC),
),
AdEcpmRecord(
user_id=user.id, ad_type="draw", ad_session_id="impression-ambiguous-a",
trace_id="source-trace", ecpm_raw="4800", adn="baidu", slot_id="rit-baidu",
app_env="prod", our_code_id="104098712", report_date=fallback_date,
created_at=datetime(2040, 2, 7, 2, tzinfo=UTC),
),
AdEcpmRecord(
user_id=user.id, ad_type="draw", ad_session_id="impression-ambiguous-b",
trace_id="source-trace", ecpm_raw="4800", adn="ks", slot_id="rit-ks",
app_env="prod", our_code_id="104098712", report_date=fallback_date,
created_at=datetime(2040, 2, 7, 2, 1, tzinfo=UTC),
),
])
db.commit()
result = ad_revenue.ad_revenue_report(
db, date_from=fallback_date, date_to=fallback_date,
user_id=user.id, app_env="prod", revenue_scope="all",
)
item = next(row for row in result["items"] if row["event_key"].startswith("feedgrp-"))
details = {detail["ecpm"]: detail for detail in item["sub_rewards"]}
assert details["4700"]["adn"] == "baidu"
assert details["4700"]["slot_id"] == "rit-baidu"
assert details["4800"]["adn"] is None
assert details["4800"]["slot_id"] is None
finally:
db.rollback()
db.execute(delete(AdFeedRewardRecord).where(AdFeedRewardRecord.reward_date == fallback_date))
db.execute(delete(AdEcpmRecord).where(AdEcpmRecord.report_date == fallback_date))
db.execute(delete(User).where(User.phone == phone))
db.commit()
db.close()
@@ -0,0 +1,71 @@
from datetime import datetime, timedelta, timezone
from app.admin.repositories import queries
from app.db.session import SessionLocal
from app.models.comparison import ComparisonRecord
from app.models.feedback import Feedback
from app.repositories import user as user_repo
def test_feedback_device_details_use_same_user_and_model() -> None:
with SessionLocal() as db:
submitted_at = datetime.now(timezone.utc)
user = user_repo.upsert_user_for_login(
db,
phone="13800009876",
register_channel="sms",
)
db.add_all(
[
ComparisonRecord(
user_id=user.id,
trace_id="feedback-device-v2166ba",
device_model="V2166BA",
device_manufacturer="vivo",
rom_name="OriginOS",
rom_version=13,
created_at=submitted_at - timedelta(minutes=5),
),
ComparisonRecord(
user_id=user.id,
trace_id="feedback-device-other",
device_model="OTHER-CODE",
device_manufacturer="Other",
rom_name="OtherOS",
rom_version=99,
created_at=submitted_at - timedelta(minutes=5),
),
ComparisonRecord(
user_id=user.id,
trace_id="feedback-device-future-upgrade",
device_model="V2166BA",
device_manufacturer="vivo-new",
rom_name="OriginOS",
rom_version=99,
created_at=submitted_at + timedelta(minutes=5),
),
Feedback(
user_id=user.id,
content="设备信息补全测试",
contact="",
status="pending",
device_model="V2166BA",
rom_name="OriginOS",
android_version="13",
created_at=submitted_at,
),
]
)
db.commit()
items, _next_cursor, total = queries.list_feedbacks(
db,
user_id=user.id,
limit=20,
)
assert total == 1
assert items[0].device_model == "V2166BA"
assert items[0].device_model_name == "vivo Y77e"
assert items[0].device_manufacturer == "vivo"
assert items[0].rom_version == 13
+314 -1
View File
@@ -1,7 +1,7 @@
"""Admin M2 读接口测试:大盘聚合 + 用户/流水/提现/反馈列表 + 鉴权拦截。"""
from __future__ import annotations
from datetime import datetime
from datetime import UTC, datetime
import pytest
from fastapi.testclient import TestClient
@@ -9,9 +9,13 @@ from sqlalchemy import event
from app.admin.main import admin_app
from app.admin.repositories import admin_user as admin_repo
from app.admin.repositories import queries
from app.db.session import SessionLocal, engine
from app.models.comparison import ComparisonRecord
from app.models.feedback import Feedback
from app.models.invite import InviteRelation
from app.models.savings import SavingsRecord
from app.models.user import User
from app.models.wallet import CashTransaction, WithdrawOrder
from app.repositories import user as user_repo
from app.repositories import wallet as wallet_repo
@@ -144,6 +148,8 @@ def test_user_reward_detail_does_not_select_unrelated_new_ad_columns(
) -> None:
if "ad_reward_record.boost_round_id" in statement:
raise AssertionError("提现详情不应查询未使用的 boost_round_id")
if "FROM user" in statement and "user.phone" in statement:
raise AssertionError("奖励统计的用户存在性检查不应展开完整 user 表")
event.listen(engine, "before_cursor_execute", reject_full_ad_reward_projection)
try:
@@ -162,6 +168,68 @@ def test_user_reward_detail_does_not_select_unrelated_new_ad_columns(
assert records.status_code == 200, records.text
def test_user_reward_stats_can_scope_withdrawals_by_account(
admin_client: TestClient, admin_token: str
) -> None:
uid = _seed_user_with_data("13800000023")
db = SessionLocal()
try:
account = wallet_repo.get_or_create_account(db, uid)
account.cash_balance_cents = 123
account.invite_cash_balance_cents = 456
db.add_all(
[
WithdrawOrder(
user_id=uid,
out_bill_no="rewardstatsinvite00000001",
amount_cents=250,
source="invite_cash",
status="success",
),
WithdrawOrder(
user_id=uid,
out_bill_no="rewardstatsinvite00000002",
amount_cents=50,
source="invite_cash",
status="reviewing",
),
]
)
db.commit()
finally:
db.close()
coin = admin_client.get(
f"/admin/api/users/{uid}/reward-stats",
params={"withdraw_source": "coin_cash"},
headers=_auth(admin_token),
)
invite = admin_client.get(
f"/admin/api/users/{uid}/reward-stats",
params={"withdraw_source": "invite_cash"},
headers=_auth(admin_token),
)
assert coin.status_code == 200, coin.text
assert invite.status_code == 200, invite.text
assert coin.json()["withdraw_success_cents"] == 100
assert coin.json()["withdraw_total"] == 1
assert coin.json()["cash_balance_cents"] == 123
assert invite.json()["withdraw_success_cents"] == 250
assert invite.json()["withdraw_total"] == 2
assert invite.json()["cash_balance_cents"] == 456
def test_user_coin_record_sort_accepts_mixed_timezone_datetimes() -> None:
"""线上 PostgreSQL 返回 awareSQLite/历史转换可能返回 naive,二者必须可混排。"""
naive = datetime(2038, 1, 1, 8, 0)
aware = datetime(2038, 1, 1, 7, 0, tzinfo=UTC)
rows = [{"created_at": aware}, {"created_at": naive}]
rows.sort(key=queries._coin_record_sort_key, reverse=True)
assert rows == [{"created_at": naive}, {"created_at": aware}]
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))
@@ -307,10 +375,19 @@ def test_withdraw_list_exposes_source_and_filters(
db = SessionLocal()
try:
uid = user_repo.upsert_user_for_login(db, phone="13800009002", register_channel="sms").id
# 直接从 ORM 查询,验证列表富化返回人工风险结论。
user = db.get(User, uid)
assert user is not None
user.is_high_risk = True
user.high_risk_note = "邀请行为异常"
db.add(WithdrawOrder(user_id=uid, out_bill_no=f"src{uid}coin0001",
amount_cents=100, status="success", source="coin_cash"))
db.add(WithdrawOrder(user_id=uid, out_bill_no=f"src{uid}invite001",
amount_cents=200, status="success", source="invite_cash"))
db.add(WithdrawOrder(user_id=uid, out_bill_no=f"src{uid}coinreview",
amount_cents=50, status="reviewing", source="coin_cash"))
db.add(WithdrawOrder(user_id=uid, out_bill_no=f"src{uid}invitereview",
amount_cents=80, status="reviewing", source="invite_cash"))
db.commit()
finally:
db.close()
@@ -325,6 +402,242 @@ def test_withdraw_list_exposes_source_and_filters(
)
items = r.json()["items"]
assert items and all(it["source"] == "invite_cash" for it in items)
assert all(it["is_high_risk"] is True for it in items)
assert all(it["high_risk_note"] == "邀请行为异常" for it in items)
# 累计提现按当前审核页账户隔离,不把普通现金的 100 分串进邀请金。
assert all(it["cumulative_success_cents"] == 200 for it in items)
invite_summary = admin_client.get(
"/admin/api/withdraws/summary",
params={"source": "invite_cash"},
headers=_auth(admin_token),
).json()
other_summary = admin_client.get(
"/admin/api/withdraws/summary",
params={"source": "coin_cash"},
headers=_auth(admin_token),
).json()
assert invite_summary["reviewing_count"] >= 1
assert other_summary["reviewing_count"] >= 1
assert invite_summary["reviewing_amount_cents"] >= 80
assert other_summary["reviewing_amount_cents"] >= 50
def test_invite_withdraw_detail_contains_invitee_first_actions(
admin_client: TestClient, admin_token: str
) -> None:
db = SessionLocal()
try:
inviter = user_repo.upsert_user_for_login(
db, phone="13800009031", register_channel="sms"
)
invitee = user_repo.upsert_user_for_login(
db, phone="13800009032", register_channel="sms"
)
db.add(
InviteRelation(
inviter_user_id=inviter.id,
invitee_user_id=invitee.id,
status="effective",
compare_reward_granted=True,
compare_reward_cents=50,
)
)
db.add(
ComparisonRecord(
user_id=invitee.id,
trace_id="invite-detail-first-compare",
status="success",
store_name="首次比价商家",
product_names="商品甲、商品乙",
)
)
db.add(
SavingsRecord(
user_id=invitee.id,
order_amount_cents=1888,
saved_amount_cents=300,
source="compare",
shop_name="首次下单商家",
title="首次下单商品",
dishes=[],
)
)
bill = "inviteoverviewbill0001"
db.add(
WithdrawOrder(
user_id=inviter.id,
out_bill_no=bill,
amount_cents=100,
status="reviewing",
source="invite_cash",
)
)
db.commit()
finally:
db.close()
response = admin_client.get(
f"/admin/api/withdraws/{bill}", headers=_auth(admin_token)
)
assert response.status_code == 200, response.text
overview = response.json()["invite_overview"]
assert overview["invite_total"] == 1
assert overview["invite_success_total"] == 1
item = overview["items"][0]
assert item["phone"] == "13800009032"
assert item["first_compare_store"] == "首次比价商家"
assert item["first_compare_products"] == "商品甲、商品乙"
assert item["first_order_store"] == "首次下单商家"
assert item["first_order_products"] == "首次下单商品"
assert item["first_order_amount_cents"] == 1888
filtered = admin_client.get(
f"/admin/api/withdraws/{bill}",
params={
"date_from": "2099-01-01T00:00:00Z",
"date_to": "2099-01-02T00:00:00Z",
},
headers=_auth(admin_token),
)
assert filtered.status_code == 200, filtered.text
assert filtered.json()["invite_overview"]["invite_total"] == 0
def test_withdraw_summary_counts_match_status_lists_by_source(
admin_client: TestClient, admin_token: str
) -> None:
"""每个审核页签的汇总数必须等于相同 source、status 的列表总数。"""
db = SessionLocal()
try:
uid = user_repo.upsert_user_for_login(
db, phone="13800009003", register_channel="sms"
).id
for index, status in enumerate(
("reviewing", "pending", "success", "rejected", "failed"), start=1
):
db.add(
WithdrawOrder(
user_id=uid,
out_bill_no=f"summary{uid}invite{index}",
amount_cents=index * 10,
status=status,
source="invite_cash",
)
)
db.add(
WithdrawOrder(
user_id=uid,
out_bill_no=f"summary{uid}coinfailed",
amount_cents=99,
status="failed",
source="coin_cash",
)
)
db.commit()
finally:
db.close()
status_fields = {
"reviewing": "reviewing_count",
"pending": "pending_count",
"success": "success_count",
"rejected": "rejected_count",
"failed": "failed_count",
}
for source in ("coin_cash", "invite_cash"):
summary_response = admin_client.get(
"/admin/api/withdraws/summary",
params={"source": source},
headers=_auth(admin_token),
)
assert summary_response.status_code == 200, summary_response.text
summary = summary_response.json()
for status, field in status_fields.items():
list_response = admin_client.get(
"/admin/api/withdraws",
params={"source": source, "status": status, "limit": 1},
headers=_auth(admin_token),
)
assert list_response.status_code == 200, list_response.text
assert summary[field] == list_response.json()["total"]
reviewing_response = admin_client.get(
"/admin/api/withdraws",
params={"source": source, "status": "reviewing", "limit": 100},
headers=_auth(admin_token),
)
assert reviewing_response.status_code == 200, reviewing_response.text
reviewing_body = reviewing_response.json()
assert reviewing_body["total"] == len(reviewing_body["items"])
assert summary["reviewing_amount_cents"] == sum(
item["amount_cents"] for item in reviewing_body["items"]
)
all_response = admin_client.get(
"/admin/api/withdraws",
params={"source": source, "limit": 1},
headers=_auth(admin_token),
)
assert all_response.status_code == 200, all_response.text
assert summary["total_count"] == all_response.json()["total"]
assert summary["total_count"] == sum(
summary[field] for field in status_fields.values()
)
assert admin_client.get(
"/admin/api/withdraws/summary",
params={"source": "unknown"},
headers=_auth(admin_token),
).status_code == 422
def test_comparison_records_show_readable_device_and_rom_version(
admin_client: TestClient, admin_token: str
) -> None:
db = SessionLocal()
try:
user = user_repo.upsert_user_for_login(
db, phone="13800009041", register_channel="sms"
)
record = ComparisonRecord(
user_id=user.id,
trace_id="comparison-readable-device",
status="success",
device_model="V2166BA",
device_manufacturer="vivo",
rom_vendor="vivo",
rom_name="OriginOS",
rom_version=4,
android_version="14",
)
db.add(record)
db.commit()
record_id = record.id
user_id = user.id
finally:
db.close()
response = admin_client.get(
"/admin/api/comparison-records",
params={"user_id": user_id},
headers=_auth(admin_token),
)
assert response.status_code == 200, response.text
item = response.json()["items"][0]
assert item["device_model"] == "V2166BA"
assert item["device_model_name"] == "vivo Y77e"
assert item["rom_name"] == "OriginOS"
assert item["rom_version"] == 4
detail = admin_client.get(
f"/admin/api/comparison-records/{record_id}",
headers=_auth(admin_token),
)
assert detail.status_code == 200, detail.text
assert detail.json()["device_model_name"] == "vivo Y77e"
assert detail.json()["rom_version"] == 4
def test_ad_coin_audit_full_count_truncate_and_only_mismatch(
+7 -1
View File
@@ -190,7 +190,13 @@ def test_builtin_roles_labels_and_pages(admin_client, super_token) -> None:
assert roles["finance"]["label"] == "财务"
assert roles["tech"]["label"] == "技术"
# 页集对齐 Prototypes/dashboard/permissions.md 的 ROLES
assert set(roles["finance"]["pages"]) == {"dashboard", "ad-revenue-report", "cps", "withdraws"}
assert set(roles["finance"]["pages"]) == {
"dashboard",
"ad-revenue-report",
"cps",
"invite-withdraws",
"withdraws",
}
assert set(roles["tech"]["pages"]) == {
"dashboard", "risk-monitor", "device-liveness", "analytics-health", "config",
"ad-revenue", "huawei-review", "event-logs", "audit-logs",
+94 -7
View File
@@ -103,6 +103,61 @@ def _seed_price_report(phone: str) -> int:
db.close()
def test_set_user_risk_requires_note_and_writes_audit(
admin_client: TestClient, operator_token: str
) -> None:
uid = _seed_user("13900000991")
missing_note = admin_client.post(
f"/admin/api/users/{uid}/risk",
json={"is_high_risk": True, "note": " "},
headers=_auth(operator_token),
)
assert missing_note.status_code == 422
marked = admin_client.post(
f"/admin/api/users/{uid}/risk",
json={"is_high_risk": True, "note": "邀请记录存在批量异常"},
headers=_auth(operator_token),
)
assert marked.status_code == 200, marked.text
listed = admin_client.get(
"/admin/api/users",
params={"phone": "13900000991"},
headers=_auth(operator_token),
).json()["items"][0]
assert listed["is_high_risk"] is True
assert listed["high_risk_note"] == "邀请记录存在批量异常"
db = SessionLocal()
try:
audit = db.execute(
select(AdminAuditLog)
.where(
AdminAuditLog.action == "user.risk.set",
AdminAuditLog.target_id == str(uid),
)
.order_by(AdminAuditLog.id.desc())
).scalars().first()
assert audit is not None
assert audit.detail["after"]["is_high_risk"] is True
finally:
db.close()
cleared = admin_client.post(
f"/admin/api/users/{uid}/risk",
json={"is_high_risk": False, "note": None},
headers=_auth(operator_token),
)
assert cleared.status_code == 200
overview = admin_client.get(
f"/admin/api/users/{uid}", headers=_auth(operator_token)
).json()
assert overview["user"]["is_high_risk"] is False
assert overview["user"]["high_risk_note"] is None
# ===== 调金币 =====
def test_grant_coins_writes_txn_and_audit(admin_client: TestClient, finance_token: str) -> None:
@@ -579,15 +634,47 @@ def test_withdraw_refresh_404(admin_client: TestClient, finance_token: str) -> N
).status_code == 404
def test_withdraw_reconcile(admin_client: TestClient, finance_token: str, monkeypatch) -> None:
def test_withdraw_refresh_enriches_existing_failed_reason(
admin_client: TestClient, finance_token: str, monkeypatch
) -> None:
uid = _seed_user("13900000017")
with SessionLocal() as db:
db.add(
WithdrawOrder(
user_id=uid,
out_bill_no="adminfailedreason01",
amount_cents=100,
status="failed",
wechat_state="FAIL",
fail_reason="微信转账状态 FAIL",
)
)
db.commit()
from app.repositories import wallet as wr
monkeypatch.setattr(
wr.wxpay, "query_transfer",
lambda obn: {"status_code": 200, "data": {"state": "SUCCESS"}},
wr.wxpay,
"query_transfer",
lambda out_bill_no: {
"status_code": 200,
"data": {
"state": "FAIL",
"fail_reason": "TRANSFER_RISK",
"transfer_bill_no": "wx_failed_reason_01",
},
},
)
r = admin_client.post(
"/admin/api/withdraws/reconcile", params={"older_than_minutes": 0},
response = admin_client.post(
"/admin/api/withdraws/adminfailedreason01/refresh",
headers=_auth(finance_token),
)
assert r.status_code == 200, r.text
assert "checked" in r.json() and "resolved" in r.json()
assert response.status_code == 200, response.text
assert response.json()["status"] == "failed"
assert response.json()["wechat_state"] == "FAIL"
assert response.json()["transfer_bill_no"] == "wx_failed_reason_01"
assert response.json()["fail_reason"] == (
"微信转账失败:该笔转账存在风险,已被微信拦截(TRANSFER_RISK"
)
+5 -3
View File
@@ -9,17 +9,16 @@ pricebot 用 httpx mock,不真连(同 test_compare_proxy)。
"""
from __future__ import annotations
import json
import uuid
from unittest.mock import MagicMock, patch
import httpx
from sqlalchemy import select
from app.db.session import SessionLocal
from app.models.comparison import ComparisonRecord
from app.repositories import comparison as crud
from app.schemas.compare_record import ComparisonRecordIn
from sqlalchemy import select
def _tid() -> str:
@@ -216,7 +215,9 @@ def test_price_step_done_harvests_success(client) -> None:
"action": {"command": "done", "params": _done_params()},
"trace_url": "https://price.shaguabijia.com/traces/done2/"}
p, _cap = _mock_pricebot(done_frame)
with p:
with p, patch(
"app.api.v1.compare.backfill_comparison_llm_cost"
) as backfill:
r = client.post("/api/v1/price/step", json=_stub_body(trace_id=tid, step=8))
assert r.status_code == 200
with SessionLocal() as db:
@@ -224,6 +225,7 @@ def test_price_step_done_harvests_success(client) -> None:
assert rec is not None and rec.status == "success"
assert rec.best_platform_id == "meituan"
assert rec.saved_amount_cents == 500
backfill.assert_called_once_with(rec.id, tid)
def test_trace_finalize_harvests_abort(client) -> None:
+161
View File
@@ -0,0 +1,161 @@
from __future__ import annotations
from datetime import UTC, datetime, timedelta
from app.core.rewards import CN_TZ
from app.db.session import SessionLocal
from app.models.app_config import AppConfig
from app.models.comparison import ComparisonRecord
from app.services import comparison_llm_backfill
def _record(
trace_id: str,
*,
status: str = "success",
created_at: datetime | None = None,
) -> int:
with SessionLocal() as db:
rec = ComparisonRecord(
trace_id=trace_id,
status=status,
created_at=created_at or datetime.now(),
)
db.add(rec)
db.commit()
return rec.id
def _delete(record_id: int) -> None:
with SessionLocal() as db:
rec = db.get(ComparisonRecord, record_id)
if rec is not None:
db.delete(rec)
db.commit()
def test_backfill_retries_then_persists_cost(monkeypatch):
record_id = _record("llm-retry-1")
calls = [
{
"model": "unknown-model",
"error": None,
"usage": {"prompt_tokens": 1000, "completion_tokens": 500},
}
]
responses = iter([[], calls])
monkeypatch.setattr(
comparison_llm_backfill,
"fetch_llm_calls",
lambda trace_id: next(responses),
)
sleeps: list[float] = []
monkeypatch.setattr(comparison_llm_backfill.time, "sleep", sleeps.append)
monkeypatch.setattr(
comparison_llm_backfill.settings, "INTERNAL_API_SECRET", "test-secret"
)
try:
assert comparison_llm_backfill.backfill_comparison_llm_cost(
record_id, "llm-retry-1", attempts=2, retry_delays=(0.25,)
)
assert sleeps == [0.25]
with SessionLocal() as db:
rec = db.get(ComparisonRecord, record_id)
assert rec.input_tokens == 1000
assert rec.output_tokens == 500
assert rec.llm_cost_yuan is not None
assert rec.llm_calls == calls
finally:
_delete(record_id)
def test_repair_batch_only_targets_terminal_missing_rows(monkeypatch):
missing_id = _record("llm-repair-missing")
running_id = _record("llm-repair-running", status="running")
calls = [
{
"model": "unknown-model",
"error": None,
"usage": {"prompt_tokens": 100, "completion_tokens": 20},
}
]
seen: list[str] = []
def fetch(trace_id: str) -> list[dict]:
seen.append(trace_id)
return calls
monkeypatch.setattr(comparison_llm_backfill, "fetch_llm_calls", fetch)
monkeypatch.setattr(
comparison_llm_backfill.settings, "INTERNAL_API_SECRET", "test-secret"
)
try:
result = comparison_llm_backfill.repair_missing_comparison_llm_costs(
limit=10, lookback_days=1
)
assert result["repaired"] >= 1
assert "llm-repair-missing" in seen
assert "llm-repair-running" not in seen
with SessionLocal() as db:
assert db.get(ComparisonRecord, missing_id).llm_cost_yuan is not None
assert db.get(ComparisonRecord, running_id).llm_cost_yuan is None
finally:
_delete(missing_id)
_delete(running_id)
def test_repair_excludes_records_before_current_price_config(monkeypatch):
price_changed_at = datetime.now(UTC) - timedelta(hours=1)
before_change = (price_changed_at - timedelta(minutes=30)).astimezone(CN_TZ)
after_change = (price_changed_at + timedelta(minutes=30)).astimezone(CN_TZ)
before_id = _record(
"llm-before-price-change",
created_at=before_change.replace(tzinfo=None),
)
after_id = _record(
"llm-after-price-change",
created_at=after_change.replace(tzinfo=None),
)
with SessionLocal() as db:
existing = db.get(AppConfig, "llm_token_price")
if existing is not None:
db.delete(existing)
db.flush()
db.add(
AppConfig(
key="llm_token_price",
value={"default": {"input_per_1m": 1, "output_per_1m": 1}},
updated_at=price_changed_at,
)
)
db.commit()
seen: list[str] = []
def backfill(record_id: int, trace_id: str, **kwargs) -> bool:
seen.append(trace_id)
return True
monkeypatch.setattr(
comparison_llm_backfill,
"backfill_comparison_llm_cost",
backfill,
)
try:
result = comparison_llm_backfill.repair_missing_comparison_llm_costs(
limit=10_000,
lookback_days=1,
)
assert "llm-after-price-change" in seen
assert "llm-before-price-change" not in seen
assert result["repaired"] == len(seen)
assert result["unresolved"] == 0
finally:
_delete(before_id)
_delete(after_id)
with SessionLocal() as db:
config = db.get(AppConfig, "llm_token_price")
if config is not None:
db.delete(config)
db.commit()
@@ -0,0 +1,17 @@
"""ComparisonResultIn 保留逐平台 skipped_dish_count(落库/回读用)。
记录页三平台网格要按逐平台缺菜数标"缺少 X 个菜品"; pricebot 现把 skipped_dish_count 冗余进
comparison_results 上报边界必须显式声明该字段, 否则 model_dump() 会静默丢弃(POST 路径),
记录页拿不到 schema, 不碰 DB
"""
from app.schemas.compare_record import ComparisonResultIn
def test_comparison_result_in_preserves_skipped_dish_count():
r = ComparisonResultIn(platform_id="jd_waimai", price=25.0, skipped_dish_count=2)
assert r.model_dump().get("skipped_dish_count") == 2
def test_comparison_result_in_skipped_defaults_none():
r = ComparisonResultIn(platform_id="meituan", price=42.0, is_source=True)
assert r.model_dump().get("skipped_dish_count") is None
+102
View File
@@ -0,0 +1,102 @@
"""_derive_from_results / _derive: 缺菜(漏菜)店总价虚低, 不当记录级"最低价"
回归: pricebot comparison_results[].rank 是纯价格排序(含缺菜), server 派生 best 若照单全收,
会把缺菜店的虚低价当 best_price 记录页戴"最低"红框 + 虚假省额
修复: 派生 best 时按 platform_results[pid].skipped_dish_count 排除缺菜店(源平台永远全菜, 仍可当 best)
纯函数, 不碰 DB
"""
from app.repositories.comparison import _derive, _derive_from_results
from app.schemas.compare_record import ComparisonRecordIn, ComparisonResultIn
def test_derive_from_results_excludes_short_ordered_from_best():
# jd 缺 2 道菜 → 虚低 ¥25(rank=1); tb 全有 ¥38.5; 源美团 ¥42。best 应是 tb(干净最便宜), 不是 jd。
results = [
{"platform_id": "meituan", "platform_name": "美团", "price": 42.0, "is_source": True, "rank": 3},
{"platform_id": "jd_waimai", "platform_name": "京东外卖", "price": 25.0, "is_source": False, "rank": 1},
{"platform_id": "taobao_flash", "platform_name": "淘宝闪购", "price": 38.5, "is_source": False, "rank": 2},
]
platform_results = {
"jd_waimai": {"skipped_dish_count": 2},
"taobao_flash": {"skipped_dish_count": 0},
"meituan": {"is_source": True},
}
d = _derive_from_results(results, platform_results)
assert d["best_platform_id"] == "taobao_flash"
assert d["best_price_cents"] == 3850
assert d["saved_amount_cents"] == 4200 - 3850 # 350, 用干净店算省额, 不是缺菜虚低价 42-25=17元
assert d["is_source_best"] is False
def test_derive_from_results_all_targets_short_falls_back_to_source():
# 唯一比源便宜的都是缺菜 → 不crown缺菜店; 源全菜 → best=源, is_source_best, 不虚报省额。
results = [
{"platform_id": "meituan", "price": 42.0, "is_source": True, "rank": 2},
{"platform_id": "jd_waimai", "price": 25.0, "is_source": False, "rank": 1},
]
platform_results = {"jd_waimai": {"skipped_dish_count": 3}}
d = _derive_from_results(results, platform_results)
assert d["best_platform_id"] == "meituan"
assert d["is_source_best"] is True
assert d["saved_amount_cents"] == 0
def test_derive_from_results_no_platform_results_keeps_old_behavior():
# 不传 platform_results(老 harvest / 无缺菜信息)→ 行为不变: 纯 rank/price 选 best。
results = [
{"platform_id": "meituan", "price": 42.0, "is_source": True, "rank": 2},
{"platform_id": "jd_waimai", "price": 25.0, "is_source": False, "rank": 1},
]
d = _derive_from_results(results)
assert d["best_platform_id"] == "jd_waimai"
assert d["best_price_cents"] == 2500
def test_derive_from_results_malformed_platform_results_no_crash():
# 内层值非 dict(异常/伪造上报)→ 不抛 AttributeError, 按"不缺菜"处理, 照常选最便宜。
results = [
{"platform_id": "meituan", "price": 42.0, "is_source": True, "rank": 2},
{"platform_id": "jd_waimai", "price": 25.0, "is_source": False, "rank": 1},
]
d = _derive_from_results(results, {"jd_waimai": "oops"})
assert d["best_platform_id"] == "jd_waimai"
assert d["best_price_cents"] == 2500
def test_derive_pydantic_excludes_short():
# _derive(老客户端 POST 路径)同样排除缺菜店: jd 缺菜虚低 ¥25 不当 best, 取干净的淘宝 ¥38.5。
payload = ComparisonRecordIn(
trace_id="t-short-pyd",
source_price=42.0,
source_platform_id="meituan",
comparison_results=[
ComparisonResultIn(platform_id="meituan", platform_name="美团", price=42.0, is_source=True, rank=3),
ComparisonResultIn(platform_id="jd_waimai", platform_name="京东外卖", price=25.0, is_source=False, rank=1),
ComparisonResultIn(platform_id="taobao_flash", platform_name="淘宝闪购", price=38.5, is_source=False, rank=2),
],
platform_results={
"jd_waimai": {"skipped_dish_count": 2},
"taobao_flash": {"skipped_dish_count": 0},
},
)
d = _derive(payload)
assert d["best_platform_id"] == "taobao_flash"
assert d["best_price_cents"] == 3850
assert d["saved_amount_cents"] == 4200 - 3850
assert d["is_source_best"] is False
def test_derive_pydantic_malformed_platform_results_no_crash():
# _derive 的 platform_results 来自老客户端透传(可伪造): 内层非 dict 不应打 500。
payload = ComparisonRecordIn(
trace_id="t-malformed-pyd",
source_price=42.0,
comparison_results=[
ComparisonResultIn(platform_id="meituan", price=42.0, is_source=True, rank=2),
ComparisonResultIn(platform_id="jd_waimai", price=25.0, is_source=False, rank=1),
],
platform_results={"jd_waimai": "oops"},
)
d = _derive(payload) # 不抛 AttributeError
assert d["best_platform_id"] == "jd_waimai"
assert d["best_price_cents"] == 2500
+45
View File
@@ -127,6 +127,51 @@ def test_invite_cash_reject_refunds_invite_account(client, monkeypatch) -> None:
db.close()
def test_invite_cash_reject_recreates_missing_account(client, monkeypatch) -> None:
"""A legacy or hand-written order without CoinAccount can still be refunded."""
_patch_userinfo(monkeypatch, "openid_ic_missing_account")
token = _login(client, "13800004012")
_seed_balances(client, token, "13800004012", cash=0, invite_cash=200)
client.post("/api/v1/wallet/bind-wechat", json={"code": "c"}, headers=_auth(token))
response = client.post(
"/api/v1/wallet/withdraw",
json={"amount_cents": 200, "source": "invite_cash"},
headers=_auth(token),
)
assert response.status_code == 200, response.text
bill = response.json()["out_bill_no"]
with SessionLocal() as db:
user = db.execute(
select(User).where(User.phone == "13800004012")
).scalar_one()
db.delete(db.get(CoinAccount, user.id))
db.commit()
_reject(bill, "missing account fixture")
with SessionLocal() as db:
user = db.execute(
select(User).where(User.phone == "13800004012")
).scalar_one()
account = db.get(CoinAccount, user.id)
order = crud_wallet._get_withdraw_or_raise(db, bill)
refunds = db.execute(
select(InviteCashTransaction).where(
InviteCashTransaction.user_id == user.id,
InviteCashTransaction.biz_type == "invite_withdraw_refund",
InviteCashTransaction.ref_id == bill,
)
).scalars().all()
assert account is not None
assert account.invite_cash_balance_cents == 200
assert order.status == "rejected"
assert order.fail_reason == "missing account fixture"
assert len(refunds) == 1 and refunds[0].amount_cents == 200
def test_two_accounts_withdraw_independent(client, monkeypatch) -> None:
"""两账户各提各的不串:提 invite_cash(拒绝→验证退款回原账户),再提 cash,各扣各账户互不串。"""
_patch_userinfo(monkeypatch, "openid_ic_3")
+7 -1
View File
@@ -132,6 +132,7 @@ def test_backfill_llm_calls_stores_cost_and_snapshot(monkeypatch):
from app.models.app_config import AppConfig
from app.models.comparison import ComparisonRecord
from app.repositories import app_config
from app.services import comparison_llm_backfill
sample = [
{"model": "qwen3.5-flash", "error": None, "usage": {"prompt_tokens": 1512, "completion_tokens": 22}},
@@ -139,7 +140,12 @@ def test_backfill_llm_calls_stores_cost_and_snapshot(monkeypatch):
{"model": "qwen3.5-flash", "error": None, "usage": {"prompt_tokens": 1940, "completion_tokens": 142}},
{"model": "qwen3.5-flash", "error": None, "usage": {"prompt_tokens": 1325, "completion_tokens": 13}},
]
monkeypatch.setattr(compare_record, "fetch_llm_calls", lambda trace_id: sample)
monkeypatch.setattr(
comparison_llm_backfill, "fetch_llm_calls", lambda trace_id: sample
)
monkeypatch.setattr(
comparison_llm_backfill.settings, "INTERNAL_API_SECRET", "test-secret"
)
db = SessionLocal()
try:
+55
View File
@@ -0,0 +1,55 @@
from unittest.mock import MagicMock
from app.core.config import settings
from app.services import pricebot_llm_calls
def test_auth_probe_accepts_matching_secret(monkeypatch):
monkeypatch.setattr(settings, "INTERNAL_API_SECRET", "same-secret")
response = MagicMock(status_code=200)
get = MagicMock(return_value=response)
monkeypatch.setattr(pricebot_llm_calls.httpx, "get", get)
assert pricebot_llm_calls.pricebot_llm_auth_ready() is True
assert get.call_count == len(settings.pricebot_instances)
assert all(
call.kwargs["headers"]["X-Internal-Secret"] == "same-secret"
for call in get.call_args_list
)
def test_auth_probe_rejects_mismatched_secret(monkeypatch):
monkeypatch.setattr(settings, "INTERNAL_API_SECRET", "app-server-secret")
monkeypatch.setattr(
pricebot_llm_calls.httpx,
"get",
MagicMock(return_value=MagicMock(status_code=403)),
)
assert pricebot_llm_calls.pricebot_llm_auth_ready() is False
def test_fetch_falls_back_to_other_instance_when_hash_target_is_empty(monkeypatch):
monkeypatch.setattr(settings, "INTERNAL_API_SECRET", "same-secret")
monkeypatch.setattr(
settings,
"PRICEBOT_INSTANCES",
"http://pricebot-1:8000,http://pricebot-2:8000",
)
monkeypatch.setattr(
pricebot_llm_calls,
"pick_pricebot",
lambda trace_id: "http://pricebot-1:8000",
)
calls = [{"model": "qwen", "usage": {"prompt_tokens": 1}}]
def get(url, **kwargs):
response = MagicMock(status_code=200)
response.json.return_value = {
"calls": [] if "pricebot-1" in url else calls
}
return response
monkeypatch.setattr(pricebot_llm_calls.httpx, "get", get)
assert pricebot_llm_calls.fetch_llm_calls("trace-after-rescale") == calls
+73 -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, WechatTransferAuthorization
from app.models.wallet import CoinAccount, WechatTransferAuthorization, WithdrawOrder
from app.repositories import wallet as crud_wallet
@@ -232,6 +232,78 @@ def test_withdraw_approve_transfer_fail_refunds(client, monkeypatch) -> None:
assert r.json()["items"][0]["status"] == "failed"
def test_approved_withdraw_query_fail_exposes_wechat_reason(client, monkeypatch) -> None:
"""审核通过后微信终态失败时,保存官方失败码对应原因并退回余额。"""
monkeypatch.setattr(
"app.integrations.wxpay.code_to_userinfo",
lambda code: {
"openid": "openid_query_fail",
"nickname": None,
"avatar_url": None,
"raw": {},
},
)
monkeypatch.setattr(
"app.integrations.wxpay.create_transfer",
lambda openid, amount_fen, out_bill_no, user_name=None: {
"status_code": 200,
"data": {
"state": "WAIT_USER_CONFIRM",
"package_info": "pkg_query_fail",
"transfer_bill_no": "tb_query_fail",
},
},
)
token = _login(client, "13800002025")
_seed_cash(client, token, "13800002025", 100)
client.post("/api/v1/wallet/bind-wechat", json={"code": "c"}, headers=_auth(token))
response = client.post(
"/api/v1/wallet/withdraw",
json={"amount_cents": 50},
headers=_auth(token),
)
bill = response.json()["out_bill_no"]
_approve(bill)
monkeypatch.setattr(
"app.integrations.wxpay.query_transfer",
lambda out_bill_no: {
"status_code": 200,
"data": {
"state": "FAIL",
"fail_reason": "PAYEE_ACCOUNT_ABNORMAL",
"transfer_bill_no": "tb_query_fail",
},
},
)
response = client.get(
"/api/v1/wallet/withdraw/status",
params={"out_bill_no": bill},
headers=_auth(token),
)
assert response.status_code == 200, response.text
assert response.json()["status"] == "failed"
assert response.json()["wechat_state"] == "FAIL"
assert response.json()["fail_reason"] == (
"微信转账失败:用户微信账户收款异常(PAYEE_ACCOUNT_ABNORMAL"
)
with SessionLocal() as db:
order = db.execute(
select(WithdrawOrder).where(WithdrawOrder.out_bill_no == bill)
).scalar_one()
assert order.transfer_bill_no == "tb_query_fail"
account = client.get("/api/v1/wallet/account", headers=_auth(token))
assert account.json()["cash_balance_cents"] == 100
def test_unknown_wechat_fail_reason_code_is_preserved() -> None:
assert crud_wallet._wechat_terminal_failure_reason(
{"fail_reason": "NEW_WECHAT_REASON"}, "FAIL"
) == "微信转账失败:NEW_WECHAT_REASON"
def test_withdraw_idempotent_same_bill_no(client, monkeypatch) -> None:
"""#2 同 out_bill_no 重试:只转一次,第二次返回同一单,余额只扣一次。"""
monkeypatch.setattr("app.integrations.wxpay.code_to_userinfo", lambda code: {"openid": "openid_idem", "nickname": None, "avatar_url": None, "raw": {}})
-151
View File
@@ -1,151 +0,0 @@
"""提现现金账本校验(admin ledger-check)测试:两本物理隔离的账各自对账。
历史盲区:`withdraw_ledger_check` 曾拿全部提现单去和**普通现金流水**(cash_transaction)比对,
source=invite_cash 的提现单流水其实在 invite_cash_transaction ,导致每笔邀请提现单都被
误报缺扣款/缺退款流水这里用真实提现 API 造单 + before/after 差值断言锁定修复:
1) 邀请提现单不再污染普通现金账的缺流水计数;
2) 邀请账户已被纳入对账(能抓到它自己的缺流水);
3) 普通现金账的原有对账未被改坏
conftest 的库是 session 级共享测试间不清,故一律用 before/after 差值,只反映本用例造的数据
"""
from __future__ import annotations
from sqlalchemy import delete, select
from app.admin.repositories.queries import withdraw_ledger_check
from app.db.session import SessionLocal
from app.models.user import User
from app.models.wallet import CoinAccount, InviteCashTransaction
from app.repositories import wallet as crud_wallet
def _login(client, phone: str) -> str:
client.post("/api/v1/auth/sms/send", json={"phone": phone})
r = client.post("/api/v1/auth/sms/login", json={"phone": phone, "code": "123456"})
assert r.status_code == 200, r.text
return r.json()["access_token"]
def _auth(token: str) -> dict[str, str]:
return {"Authorization": f"Bearer {token}"}
def _patch_userinfo(monkeypatch, openid: str) -> None:
monkeypatch.setattr(
"app.integrations.wxpay.code_to_userinfo",
lambda code: {"openid": openid, "nickname": None, "avatar_url": None, "raw": {}},
)
def _seed_balances(client, token: str, phone: str, *, cash: int = 0, invite_cash: int = 0) -> None:
client.get("/api/v1/wallet/account", headers=_auth(token))
db = SessionLocal()
try:
user = db.execute(select(User).where(User.phone == phone)).scalar_one()
acc = db.get(CoinAccount, user.id)
acc.cash_balance_cents = cash
acc.invite_cash_balance_cents = invite_cash
db.commit()
finally:
db.close()
def _reject(bill: str, reason: str = "测试拒绝") -> None:
db = SessionLocal()
try:
crud_wallet.reject_withdraw(db, bill, reason)
finally:
db.close()
def _ledger() -> dict:
db = SessionLocal()
try:
return withdraw_ledger_check(db)
finally:
db.close()
def test_rejected_invite_withdraw_not_flagged_missing(client, monkeypatch) -> None:
"""核心回归:一笔被拒绝的 invite_cash 提现单,扣款/退款流水都在 invite_cash_transaction,
不应让普通现金账的缺扣款/缺退款计数增加(修复前每笔会各 +1)"""
before = _ledger()
_patch_userinfo(monkeypatch, "openid_lc_1")
token = _login(client, "13800005001")
_seed_balances(client, token, "13800005001", cash=0, invite_cash=500)
client.post("/api/v1/wallet/bind-wechat", json={"code": "c"}, headers=_auth(token))
r = client.post(
"/api/v1/wallet/withdraw",
json={"amount_cents": 200, "source": "invite_cash"},
headers=_auth(token),
)
assert r.status_code == 200, r.text
_reject(r.json()["out_bill_no"]) # rejected + 退款流水落 invite_cash_transaction
after = _ledger()
# 普通现金账不该因这笔 invite 单产生缺流水(修复前会各 +1 → 就是页面上看到的误报)
assert after["missing_withdraw_txn_count"] == before["missing_withdraw_txn_count"]
assert after["missing_refund_txn_count"] == before["missing_refund_txn_count"]
# 邀请账扣款 + 退款流水齐全,邀请账自身也不该缺
assert after["invite_missing_withdraw_txn_count"] == before["invite_missing_withdraw_txn_count"]
assert after["invite_missing_refund_txn_count"] == before["invite_missing_refund_txn_count"]
def test_invite_ledger_detects_missing_withdraw_txn(client, monkeypatch) -> None:
"""删掉一笔 invite 提现单的扣款流水 → 邀请账缺扣款计数 +1、ok=False,
证明邀请账户已真正纳入对账(修复前邀请账完全不校验永远报不出问题)"""
_patch_userinfo(monkeypatch, "openid_lc_2")
token = _login(client, "13800005002")
_seed_balances(client, token, "13800005002", cash=0, invite_cash=500)
client.post("/api/v1/wallet/bind-wechat", json={"code": "c"}, headers=_auth(token))
r = client.post(
"/api/v1/wallet/withdraw",
json={"amount_cents": 200, "source": "invite_cash"},
headers=_auth(token),
)
bill = r.json()["out_bill_no"]
before = _ledger()
db = SessionLocal()
try:
db.execute(
delete(InviteCashTransaction).where(
InviteCashTransaction.ref_id == bill,
InviteCashTransaction.biz_type == "invite_withdraw",
)
)
db.commit()
finally:
db.close()
after = _ledger()
assert (
after["invite_missing_withdraw_txn_count"]
== before["invite_missing_withdraw_txn_count"] + 1
)
assert after["ok"] is False
def test_coin_cash_withdraw_still_reconciled(client, monkeypatch) -> None:
"""普通现金 coin_cash 提现单齐全时不新增缺流水(确保分账改造没弄坏原有普通现金对账)。"""
_patch_userinfo(monkeypatch, "openid_lc_3")
token = _login(client, "13800005003")
_seed_balances(client, token, "13800005003", cash=500, invite_cash=0)
client.post("/api/v1/wallet/bind-wechat", json={"code": "c"}, headers=_auth(token))
before = _ledger()
r = client.post(
"/api/v1/wallet/withdraw",
# 50 分 = 0.5 元档(7-9 起 coin_cash 只能提预设档位)
json={"amount_cents": 50, "source": "coin_cash"},
headers=_auth(token),
)
assert r.status_code == 200, r.text
after = _ledger()
# 普通现金提现扣款流水随单写入 cash_transaction,缺扣款计数不变;邀请账更不受影响
assert after["missing_withdraw_txn_count"] == before["missing_withdraw_txn_count"]
assert after["invite_missing_withdraw_txn_count"] == before["invite_missing_withdraw_txn_count"]