Compare commits
10 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| ab034d3c15 | |||
| 58d609e5d2 | |||
| 8e01ae0bf9 | |||
| 7e17df4130 | |||
| 46f68b88e3 | |||
| 1226bc8365 | |||
| 973c440a8a | |||
| 22a1105000 | |||
| b2a528eba1 | |||
| cdd49c6421 |
@@ -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")
|
||||
@@ -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",
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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],
|
||||
|
||||
@@ -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:
|
||||
|
||||
+261
-168
@@ -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。
|
||||
|
||||
@@ -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 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,
|
||||
|
||||
@@ -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(
|
||||
|
||||
@@ -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+历史 feed,video=reward_video+withdrawal_video",
|
||||
)
|
||||
scene_stats: dict[str, AdRevenueTypeStat] = Field(
|
||||
default_factory=dict,
|
||||
description="按信息流场景(feed_scene)小计 {comparison/coupon/welfare: {impressions, revenue_yuan}};"
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
@@ -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
@@ -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
|
||||
|
||||
@@ -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(
|
||||
|
||||
@@ -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。
|
||||
|
||||
@@ -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
|
||||
@@ -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
@@ -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
|
||||
)
|
||||
|
||||
@@ -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(
|
||||
|
||||
+135
-8
@@ -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,等下次查询。
|
||||
|
||||
@@ -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):
|
||||
|
||||
@@ -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,
|
||||
}
|
||||
@@ -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 []
|
||||
|
||||
@@ -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 仍保留时才能准确回填;
|
||||
原始调用已经清理的记录不能用估算值冒充真实成本。
|
||||
@@ -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)
|
||||
|
||||
@@ -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
|
||||
@@ -13,6 +13,9 @@ 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
|
||||
@@ -165,6 +168,57 @@ 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 返回 aware,SQLite/历史转换可能返回 naive,二者必须可混排。"""
|
||||
naive = datetime(2038, 1, 1, 8, 0)
|
||||
@@ -321,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()
|
||||
@@ -339,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(
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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)"
|
||||
)
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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
|
||||
@@ -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
|
||||
@@ -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")
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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
@@ -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": {}})
|
||||
|
||||
@@ -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"]
|
||||
Reference in New Issue
Block a user