Compare commits

...

5 Commits

Author SHA1 Message Date
guke 78d8951e93 perf(pricebot): 透传复用共享 httpx 单例,免每请求重建 SSL 上下文 + 绕过进程代理
coupon/compare 透传原先每请求 async with httpx.AsyncClient(...) 新建:
每次重建 SSL 上下文(加载 certifi CA)实测 ~1s+,而 pricebot 是纯 HTTP
透传根本用不到 TLS;且 trust_env 默认 True 会把 http://localhost:8000
经进程代理(Clash)再绕几秒。

抽 app/core/pricebot_client.py 共享单例:trust_env=False 直连,lifespan
启动预热(SSL 一次性成本付在启动)、关停 aclose;timeout 下放到 .post()
保留 coupon 30s / compare 60s 差异。keep-alive 复用 TCP,每帧降到个位数 ms。
2026-06-28 09:28:46 +08:00
wuqi f3cd97a190 feat: 信息流广告改"所见即所得"发奖——直接发客户端小球显示金币 (#86)
Reviewed-on: #86
Co-authored-by: wuqi <wuqi@wonderable.ai>
Co-committed-by: wuqi <wuqi@wonderable.ai>
2026-06-27 23:49:51 +08:00
zhuzihao b4c27f4d88 fix(ad): eCPM 上报撞会话唯一约束不再抛 500,按全局口径兜底去重 (#85)
create_ecpm_record 撞 uq_ad_ecpm_record_session(只含 ad_session_id、全局唯一)
时,改用与约束同口径的全局查找 _find_by_session_global(仅按 ad_session_id)兜底,
并去掉末尾的 raise。原逻辑兜底用带 user_id 的 find_by_session:不同 user 上报了
同一 ad_session_id 时二次查找仍为 None → raise → 接口 500,违背该接口
best-effort / fire-and-forget 约定(丢一两条不影响业务,穿山甲后台才是结算权威)。
find_by_session 签名与行为不变,另两个调用方(ad_reward、ad.py)不受影响。

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

---------

Co-authored-by: zzhyyyyy <2685922758@qq.com>
Reviewed-on: #85
Co-authored-by: zhuzihao <zhuzihao@wonderable.ai>
Co-committed-by: zhuzihao <zhuzihao@wonderable.ai>
2026-06-27 22:42:58 +08:00
zhuzihao e38120ad49 feat(ad-revenue): 收益报表分页/场景筛选/倒序,并修复信息流金币审计复算口径+ 大盘改版 (#84)
收益报表(/admin/api/ad-revenue-report):
- 明细改按时间倒序;新增 offset 真分页(limit 作每页大小、total 为全量),可翻页看当前筛选下全部数据,突破原 1000 条上限。
- 「场景」(feed_scene)下推后端做全局筛选,同时作用于明细/合计/daily·hourly 趋势(原为前端仅过滤明细)。
- 新增全量 hourly 序列,按小时趋势改用它,不再受分页截断影响。

修复信息流金币审计复算口径漂移(ad_audit):
- 发奖侧 grant_feed_reward 早已是「一条广告=1份、LT 按账号累计条数(COUNT)」,但审计仍按 unit_count 逐份累加 + SUM(unit_count) 做 LT 基线,导致单条停留>20s(份数>1)时应发虚高、必然「✗ 不符」。
- 审计改为每条 granted 按 1 份复算、LT 基线用 COUNT,与发奖对齐;金币审计页与收益报表(复用同一复算)一并恢复正确。纯复算口径修正,不改实际发奖、不动钱。

文档同步更新 admin-ad-revenue-report.md。

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

---------

Co-authored-by: zzhyyyyy <2685922758@qq.com>
Reviewed-on: #84
Co-authored-by: zhuzihao <zhuzihao@wonderable.ai>
Co-committed-by: zhuzihao <zhuzihao@wonderable.ai>
2026-06-27 22:42:54 +08:00
marco 3d67749101 fix(compare-record): specs 兼容 pricebot 嵌套规格对象
pricebot 嵌套规格统一(2026-06-27)后 calibration.specs 可能是规格对象
[{name, qty, sub_specs}] 而非字符串。ComparisonItemIn.specs 加 before-validator
把对象拍平成可读字符串"主项(子1,子2)",兼容新旧输入、保持 list[str] 契约(下游零改动)。
防两坑:① 直接声明 list[str] → 对象 specs 整条 422 被拒、不入库;
② 仅放宽成裸 list → 下游 join 出"[object Object]"乱码。

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-27 09:26:00 +08:00
15 changed files with 326 additions and 73 deletions
+27 -25
View File
@@ -4,9 +4,9 @@
- 看视频:每条 granted = 1 份,第 N 份 = 该用户 granted 的 reward_video **账号累计**顺序号
(与 ad_reward.grant_ad_reward 里 `_granted_cumulative + 1` 一致;LT 因子不按天重置,
故复算时要把当日序号叠加上该用户在本日**之前**的累计已发份数)。
- 信息流:每条按 unit_count 份逐份累加,LT 序号 = 该用户 granted 份数**账号累计**
(与 ad_feed_reward._unit_reward_total 的 existing_units 一致;同样不按天重置,
复算需叠加本日之前的累计数)。
- 信息流:**每条 granted = 1 份**(与 ad_feed_reward.grant_feed_reward 同口径:看满一份即发该条
满额,**不按 unit_count 逐份累加**),LT 序号 = 该用户 granted **条数**账号累计
(与 ad_feed_reward.granted_unit_total 的 COUNT 一致;不按天重置,复算需叠加本日之前的累计数)。
非 granted(capped/ecpm_missing)不占用份序号、应发恒 0,据此校验闸口是否确实没发。
"""
@@ -108,14 +108,18 @@ def _reward_video_rows(
return rows
def _feed_prior_granted_units(
def _feed_prior_granted_count(
db: Session, *, date: str, user_id: int | None
) -> dict[int, int]:
"""各用户在 date **之前** granted 的信息流份数累计,作为当日复算的 LT 序号起点。"""
"""各用户在 date **之前** granted 的信息流**条数**累计,作为当日复算的 LT 序号起点。
与发奖侧 ad_feed_reward.granted_unit_total(COUNT status=granted)对齐:一条广告 = 1 份,
LT 按账号累计**条数**递进。**不再用 SUM(unit_count)**——那是「一条按时长折多份」的过时口径,
与现行发奖(每条 1 份)漂移,会让 unit_count>1 的记录复算虚高、对账恒「不符」。"""
stmt = (
select(
AdFeedRewardRecord.user_id,
func.coalesce(func.sum(AdFeedRewardRecord.unit_count), 0),
func.count(),
)
.where(
AdFeedRewardRecord.reward_date < date,
@@ -144,10 +148,11 @@ def _feed_scene_matches(rec: AdFeedRewardRecord, scene: str | None) -> bool:
def _feed_rows(
db: Session, *, date: str, user_id: int | None, scene: str | None = None
) -> list[dict]:
"""信息流记录复算。granted 记录逐份累加,LT 序号沿用账号累计份数(含本日之前)。
"""信息流记录复算。**每条 granted = 1 份**(与发奖同口径,不按 unit_count 累加),
LT 序号沿用账号累计**条数**(含本日之前)。
**关键:LT 因子账号累计按全表 unit 累计(feed+draw 共享同一发奖池/上限),不按 ad_type 拆分**——
故无论 scene 怎么筛展示,这里都遍历当日**全部**信息流记录维持 granted_units 累加;scene 只决定
**关键:LT 因子账号累计按全表 granted 条数累计(feed+draw 共享同一发奖池/上限),不按 ad_type 拆分**——
故无论 scene 怎么筛展示,这里都遍历当日**全部**信息流记录维持 granted_count 累加;scene 只决定
哪些行被**留下展示**(由 _feed_scene_matches 判断),不影响累计基线,保证复算序号与正式发奖一致。
"""
stmt = (
@@ -158,23 +163,20 @@ def _feed_rows(
if user_id is not None:
stmt = stmt.where(AdFeedRewardRecord.user_id == user_id)
# 本日之前的累计份数做起点,与 _unit_reward_total 的 existing_units(累计)对齐
granted_units: dict[int, int] = _feed_prior_granted_units(db, date=date, user_id=user_id)
# 本日之前的累计**条数**做起点,与发奖侧 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():
keep = _feed_scene_matches(rec, scene) # 累计照常推进,这里只决定是否展示本行
if rec.status == "granted":
existing = granted_units.get(rec.user_id, 0)
units = rec.unit_count
granted_units[rec.user_id] = existing + units
# 一条广告 = 1 份(与 grant_feed_reward 同口径:看满一份即发该条满额,不按 unit_count 累加)。
# nth = 账号累计第几**条**(含本日之前),与发奖侧 granted_unit_total+1 对齐;累计照常推进
# (即便 scene 不匹配不展示也要 +1,保证序号与正式发奖一致)。
nth = granted_count.get(rec.user_id, 0) + 1
granted_count[rec.user_id] = nth
if not keep:
continue
expected = sum(
rewards.calculate_ad_reward_coin(rec.ecpm_raw, existing + offset)
for offset in range(1, units + 1)
)
start = existing + 1 if units > 0 else None
end = existing + units if units > 0 else None
expected = rewards.calculate_ad_reward_coin(rec.ecpm_raw, nth)
rows.append({
"scene": "feed",
"ad_type": rec.ad_type or "feed",
@@ -188,11 +190,11 @@ def _feed_rows(
"status": rec.status,
"ecpm": rec.ecpm_raw,
"ecpm_factor": rewards.ad_ecpm_factor(rewards.parse_ecpm_yuan(rec.ecpm_raw)),
"units": units,
"lt_index_start": start,
"lt_index_end": end,
"lt_factor_start": rewards.ad_lt_factor(start) if start else None,
"lt_factor_end": rewards.ad_lt_factor(end) if end else None,
"units": 1,
"lt_index_start": nth,
"lt_index_end": nth,
"lt_factor_start": rewards.ad_lt_factor(nth),
"lt_factor_end": rewards.ad_lt_factor(nth),
"expected_coin": expected,
"actual_coin": rec.coin,
"matched": expected == rec.coin,
+68 -5
View File
@@ -27,6 +27,7 @@ from sqlalchemy import select
from sqlalchemy.orm import Session
from app.admin.repositories import ad_audit
from app.admin.repositories import stats as admin_stats
from app.core import rewards
from app.models.ad_ecpm import AdEcpmRecord
from app.models.user import User
@@ -84,14 +85,20 @@ def ad_revenue_report(
date_to: str,
user_id: int | None = None,
ad_type: str | None = None,
feed_scene: str | None = None,
granularity: str = "day",
limit: int = 500,
offset: int = 0,
sort: str = "time",
) -> dict:
"""日期区间(北京时间,闭区间)**逐条广告事件**列表 + 发奖对账。单日时 date_from==date_to。
每个 item = 一次广告事件(展示与发奖按 ad_session_id 合并;信息流展示 / 发奖各自成行)。
ad_type: None=全部 / reward_video / feed / draw。granularity=hour 时每行带北京小时(由各自时间算)。
limit 只截断 items(事件明细),total 与 total_* / daily 在全量上统计,数字始终可信
ad_type: None=全部 / reward_video / feed / draw。feed_scene: None=全部 /
comparison / coupon / welfare,作为全局筛选(同时作用于明细、合计与 daily/hourly 趋势)
granularity=hour 时每行带北京小时(由各自时间算),并额外返回全量 hourly 序列。
事件按时间倒序(新→旧)排列;limit/offset 对排序后的全量做分页切片(items 为当前页),
total 与 total_* / daily / hourly 在全量上统计,不受分页影响。
"""
by_hour = granularity == "hour"
@@ -200,7 +207,17 @@ def ad_revenue_report(
"reward_detail": _reward_detail(row),
})
events.sort(key=lambda e: (e["report_date"], e["user_id"], e["created_at"]))
# 「场景」作为全局筛选(与 user_id/ad_type 一致):同时作用于明细、合计与 daily/hourly 趋势。
# feed_scene 仅信息流 / Draw 有值,激励视频与旧数据为 None;选中后只保留该场景事件。
if feed_scene is not None:
events = [e for e in events if e.get("feed_scene") == feed_scene]
# 排序:time=按时间倒序(新→旧);ecpm=按 eCPM 数值倒序(eCPM 原值是字符串「分」,转数值排;
# 纯发奖行用其发奖采用的 eCPM,缺失/非法计 0 排末尾)。
if sort == "ecpm":
events.sort(key=lambda e: rewards.parse_ecpm_fen(e["ecpm"]), reverse=True)
else:
events.sort(key=lambda e: (e["report_date"], e["created_at"]), reverse=True)
# 补手机号(admin 展示用,完整不脱敏,与用户 / 钱包 / 比价记录页一致):批量一次查,避免 N+1。
uids = {e["user_id"] for e in events}
@@ -238,14 +255,60 @@ def ad_revenue_report(
for d in sorted(daily_map.values(), key=lambda x: x["date"])
]
# 按小时汇总(全量,不受分页 limit/offset 影响):供前端按小时趋势图(单日 granularity=hour 时用)。
# 只在 by_hour 下聚合(此时每个 event 带 hour);否则空。前端按天趋势仍用 daily。
hourly: list[dict] = []
if by_hour:
hour_map: dict[int, dict] = {}
for e in events:
h = e["hour"]
if h is None:
continue
hd = hour_map.get(h)
if hd is None:
hd = {"hour": h, "impressions": 0, "revenue_yuan": 0.0,
"expected_coin": 0, "actual_coin": 0}
hour_map[h] = hd
hd["impressions"] += e["impressions"]
hd["revenue_yuan"] += e["revenue_yuan"]
hd["expected_coin"] += e["expected_coin"]
hd["actual_coin"] += e["actual_coin"]
hourly = [
{**hd, "revenue_yuan": round(hd["revenue_yuan"], 6)}
for hd in sorted(hour_map.values(), key=lambda x: x["hour"])
]
# 分广告类型小计(按 ad_type:展示条数 + 预估收益;eCPM 由前端用 收益÷展示×1000 算)。
# 基于全量(已按 feed_scene 过滤)events;前端只取 draw / reward_video 两类展示。
type_map: dict[str, dict] = {}
for e in events:
t = type_map.get(e["ad_type"])
if t is None:
t = {"impressions": 0, "revenue_yuan": 0.0}
type_map[e["ad_type"]] = t
t["impressions"] += e["impressions"]
t["revenue_yuan"] += e["revenue_yuan"]
type_stats = {
k: {"impressions": v["impressions"], "revenue_yuan": round(v["revenue_yuan"], 6)}
for k, v in type_map.items()
}
# DAU:复用大盘「今日活跃」口径(stats.today_dau,last_login_at)。该口径只能算今日,
# 故仅当查询=今日单天时给值;历史 / 多天区间返回 None,前端显示「-」。
is_today = date_from == date_to == rewards.cn_today().isoformat()
dau = admin_stats.today_dau(db) if is_today else None
return {
"total": len(events),
"truncated": len(events) > limit,
"truncated": len(events) > offset + limit,
"total_impressions": total_impressions,
"total_revenue_yuan": total_revenue_yuan,
"total_expected_coin": total_expected_coin,
"total_actual_coin": total_actual_coin,
"mismatch_count": mismatch_count,
"daily": daily,
"items": events[:limit],
"hourly": hourly,
"type_stats": type_stats,
"dau": dau,
"items": events[offset:offset + limit],
}
+14 -1
View File
@@ -28,6 +28,19 @@ def _beijing_today_start_utc() -> datetime:
return start_bj.astimezone(timezone.utc)
def today_dau(db: Session) -> int:
"""今日活跃用户数(DAU):北京时区今天 0 点后登录过(last_login_at)。
大盘与广告收益报表共用此口径,单一来源避免漂移。
⚠️ last_login_at 是单值字段(只存最后一次登录时刻),故只能算「今日」,
无法回溯历史某天的 DAU——调用方按此约束决定历史区间是否展示。
"""
today_start = _beijing_today_start_utc()
return db.execute(
select(func.count(User.id)).where(User.last_login_at >= today_start)
).scalar_one()
def dashboard_overview(db: Session) -> dict:
today_start = _beijing_today_start_utc()
@@ -69,7 +82,7 @@ def dashboard_overview(db: Session) -> dict:
"disabled": by_status.get("disabled", 0),
"deleted": by_status.get("deleted", 0),
"new_today": _count(User, User.created_at >= today_start),
"dau": _count(User, User.last_login_at >= today_start),
"dau": today_dau(db),
},
"coins": {
# 累计发放金币(coin_transaction 里所有 amount>0 之和;负数是兑换/扣减不计)
+24 -3
View File
@@ -11,7 +11,13 @@ from fastapi import APIRouter, Depends, HTTPException, Query
from app.admin.deps import AdminDb, get_current_admin
from app.admin.repositories import ad_revenue
from app.admin.schemas.ad_revenue import AdRevenueDaily, AdRevenueReportOut, AdRevenueRow
from app.admin.schemas.ad_revenue import (
AdRevenueDaily,
AdRevenueHourly,
AdRevenueReportOut,
AdRevenueRow,
AdRevenueTypeStat,
)
from app.core.rewards import cn_today
router = APIRouter(
@@ -43,10 +49,21 @@ def get_ad_revenue_report(
str | None,
Query(description="reward_video / feed / draw;不传=全部类型"),
] = None,
feed_scene: Annotated[
str | None,
Query(
description="comparison(比价) / coupon(领券) / welfare(福利);不传=全部场景。"
"全局筛选,同时影响明细 / 合计 / 趋势"
),
] = None,
granularity: Annotated[
str, Query(description="day=按天 / hour=按小时(北京时间);区间>1 天建议用 day")
] = "day",
limit: Annotated[int, Query(ge=1, le=1000)] = 500,
limit: Annotated[int, Query(ge=1, le=1000, description="每页条数(分页大小)")] = 500,
offset: Annotated[int, Query(ge=0, description="分页偏移(已跳过的条数)=(页码-1)×每页条数")] = 0,
sort: Annotated[
str, Query(description="排序:time=时间倒序(默认) / ecpm=按 eCPM 数值倒序")
] = "time",
) -> AdRevenueReportOut:
today = cn_today()
d_from = _parse_day(date_from, field="date_from", default=today)
@@ -58,12 +75,16 @@ def get_ad_revenue_report(
result = ad_revenue.ad_revenue_report(
db, date_from=d_from.isoformat(), date_to=d_to.isoformat(),
user_id=user_id, ad_type=ad_type, granularity=granularity, limit=limit,
user_id=user_id, ad_type=ad_type, feed_scene=feed_scene,
granularity=granularity, limit=limit, offset=offset, sort=sort,
)
return AdRevenueReportOut(
date_from=d_from.isoformat(),
date_to=d_to.isoformat(),
daily=[AdRevenueDaily(**d) for d in result["daily"]],
hourly=[AdRevenueHourly(**h) for h in result["hourly"]],
type_stats={k: AdRevenueTypeStat(**v) for k, v in result["type_stats"].items()},
dau=result["dau"],
total=result["total"],
truncated=result["truncated"],
total_impressions=result["total_impressions"],
+32 -3
View File
@@ -40,7 +40,7 @@ class AdRevenueRecord(BaseModel):
class AdRevenueDaily(BaseModel):
"""按日期汇总的一天(供前端按天趋势图;全量,不受 limit 影响)。"""
"""按日期汇总的一天(供前端按天趋势图;全量,不受分页影响)。"""
date: str = Field(..., description="北京时间 YYYY-MM-DD")
impressions: int = Field(..., description="当天展示条数合计")
@@ -49,6 +49,23 @@ class AdRevenueDaily(BaseModel):
actual_coin: int = Field(..., description="当天实发金币合计")
class AdRevenueHourly(BaseModel):
"""按北京小时(0–23)汇总的一小时(供前端按小时趋势图;全量,不受分页影响,单日 granularity=hour 时非空)。"""
hour: int = Field(..., description="北京时间小时 023")
impressions: int = Field(..., description="该小时展示条数合计")
revenue_yuan: float = Field(..., description="该小时预估收益合计(元)")
expected_coin: int = Field(..., description="该小时应发金币合计")
actual_coin: int = Field(..., description="该小时实发金币合计")
class AdRevenueTypeStat(BaseModel):
"""按广告类型(ad_type)的小计:展示条数 + 预估收益(eCPM 由前端用 收益÷展示×1000 算)。"""
impressions: int = Field(..., description="该类型展示条数合计")
revenue_yuan: float = Field(..., description="该类型预估收益合计(元)")
class AdRevenueRow(BaseModel):
"""一次广告事件(逐条一行):激励视频展示与发奖按 ad_session_id 合并;信息流展示 / 发奖各自成行。"""
@@ -91,8 +108,20 @@ class AdRevenueReportOut(BaseModel):
date_from: str = Field(..., description="报表起始日期(北京时间 YYYY-MM-DD)")
date_to: str = Field(..., description="报表结束日期(北京时间 YYYY-MM-DD,闭区间;单日时与 date_from 相同)")
daily: list[AdRevenueDaily] = Field(..., description="按日期汇总序列(全量,供按天趋势图)")
total: int = Field(..., description="广告事件总数(全量,不受 limit 影响)")
truncated: bool = Field(..., description="明细是否被 limit 截断")
hourly: list[AdRevenueHourly] = Field(
default_factory=list,
description="按小时汇总序列(全量,供按小时趋势图;按天查询时为空)",
)
type_stats: dict[str, AdRevenueTypeStat] = Field(
default_factory=dict,
description="按广告类型(ad_type)小计 {ad_type: {impressions, revenue_yuan}};前端取 draw / reward_video 做分类大盘",
)
dau: int | None = Field(
None,
description="今日活跃用户数(复用大盘口径,last_login_at);**仅查询=今日单天时有值**,历史/多天为 null",
)
total: int = Field(..., description="广告事件总数(全量,不受分页影响;= 当前筛选下的分页总条数)")
truncated: bool = Field(..., description="当前页之后是否还有更多事件(len(events) > offset + limit)")
total_impressions: int = Field(..., description="全量展示条数合计")
total_revenue_yuan: float = Field(..., description="全量收益合计(元)")
total_expected_coin: int = Field(..., description="全量应发金币合计")
+1
View File
@@ -411,6 +411,7 @@ def feed_reward(payload: FeedRewardIn, user: CurrentUser, db: DbSession) -> Feed
app_env=payload.app_env,
our_code_id=payload.our_code_id,
aborted=payload.aborted,
display_coin=payload.display_coin,
)
logger.info(
"feed ad reward user_id=%d event=%s status=%s units=%d coin=%d",
+5 -4
View File
@@ -26,6 +26,7 @@ import httpx
from fastapi import APIRouter, HTTPException, Request, status
from app.core.config import settings
from app.core.pricebot_client import get_pricebot_client
from app.core.pricebot_router import pick_pricebot
logger = logging.getLogger("shagua.compare")
@@ -65,10 +66,10 @@ async def _passthrough(request: Request, upstream_path: str) -> dict[str, Any]:
)
try:
async with httpx.AsyncClient(timeout=timeout) as client:
resp = await client.post(
url, content=raw, headers={"Content-Type": "application/json"}
)
client = get_pricebot_client()
resp = await client.post(
url, content=raw, headers={"Content-Type": "application/json"}, timeout=timeout
)
except httpx.RequestError as e:
logger.error("[pricebot] request failed: %s", e)
raise HTTPException(
+5 -4
View File
@@ -20,6 +20,7 @@ from fastapi.concurrency import run_in_threadpool
from app.api.deps import CurrentUser, DbSession
from app.core.config import settings
from app.core.pricebot_client import get_pricebot_client
from app.core.pricebot_router import pick_pricebot
from app.db.session import SessionLocal
from app.repositories import coupon_state as coupon_repo
@@ -141,10 +142,10 @@ async def coupon_step(
)
try:
async with httpx.AsyncClient(timeout=timeout) as client:
resp = await client.post(
url, content=raw, headers={"Content-Type": "application/json"}
)
client = get_pricebot_client()
resp = await client.post(
url, content=raw, headers={"Content-Type": "application/json"}, timeout=timeout
)
except httpx.RequestError as e:
logger.error("[pricebot] request failed: %s", e)
raise HTTPException(
+35
View File
@@ -0,0 +1,35 @@
"""透传到 pricebot 的共享 httpx.AsyncClient 单例。
为什么不能每请求新建(coupon.py / compare.py 老写法 async with httpx.AsyncClient(...)):
① 每次构造都重建一套 SSL 上下文(httpx.create_ssl_context 加载 certifi CA),实测
~1s+/次;而 pricebot 是纯 http 透传,根本用不到 TLS → 纯浪费,且每帧重交一次。
② trust_env 默认 True 会读进程 HTTP_PROXY,把 http://localhost:8000 这条本地透传整个
塞进本机代理(如 Clash 7897),恒定再多几秒。
单例:启动只建一次(SSL/连接池一次性),keep-alive 复用 TCP,每帧降到个位数 ms。
trust_env=False:对齐 integrations/meituan.py 的既有约定,不被进程代理误导,直连 pricebot。
"""
from __future__ import annotations
import httpx
_client: httpx.AsyncClient | None = None
def get_pricebot_client() -> httpx.AsyncClient:
"""取透传单例。lifespan 启动会预热;未预热(如测试态)懒建兜底。
超时不在此固化(coupon 30s / compare 60s 不同),由调用点 client.post(timeout=...) 传。
懒建无 await,asyncio 单线程下不会有并发竞态。
"""
global _client
if _client is None:
_client = httpx.AsyncClient(trust_env=False)
return _client
async def aclose_pricebot_client() -> None:
"""lifespan 关停时调,优雅关连接池。"""
global _client
if _client is not None:
await _client.aclose()
_client = None
+3
View File
@@ -49,6 +49,7 @@ from app.core.daily_exchange_worker import (
stop_daily_exchange_worker,
)
from app.core.logging import setup_logging
from app.core.pricebot_client import aclose_pricebot_client, get_pricebot_client
from app.core.withdraw_reconcile_worker import (
start_withdraw_reconcile_worker,
stop_withdraw_reconcile_worker,
@@ -68,6 +69,7 @@ async def lifespan(_: FastAPI) -> AsyncIterator[None]:
settings.APP_DEBUG,
settings.DATABASE_URL.split("://", 1)[0],
)
get_pricebot_client() # 预热透传 client:把建 SSL 上下文的一次性成本付在启动,首个领券请求即热
reconcile_task = start_withdraw_reconcile_worker()
heartbeat_task = start_heartbeat_monitor()
daily_exchange_task = start_daily_exchange_worker()
@@ -77,6 +79,7 @@ async def lifespan(_: FastAPI) -> AsyncIterator[None]:
await stop_heartbeat_monitor(heartbeat_task)
await stop_withdraw_reconcile_worker(reconcile_task)
await stop_daily_exchange_worker(daily_exchange_task)
await aclose_pricebot_client()
logger.info("shutting down")
+24 -5
View File
@@ -53,11 +53,16 @@ def create_ecpm_record(
db.commit()
except IntegrityError:
db.rollback()
if ad_session_id:
existing = find_by_session(db, user_id=user_id, ad_session_id=ad_session_id)
if existing is not None:
return existing
raise
# 撞唯一约束 uq_ad_ecpm_record_session(全局按 ad_session_id、不含 user_id):并发同会话重复上报,
# 或同一 ad_session_id 已被先到的上报占用。本接口 fire-and-forget、best-effort —— 丢一条不影响业务
# (穿山甲后台才是结算权威),绝不向客户端抛 500。兜底查找须与唯一约束**同口径**(只按 ad_session_id、
# 不带 user_id):否则不同 user 上报了同一 ad_session_id 时,带 user_id 的查找会漏掉那条别人的记录 →
# 旧逻辑在此 raise 成 500(本应静默吞掉)。
existing = _find_by_session_global(db, ad_session_id)
if existing is not None:
return existing
# 极少:rollback 后既存记录又查不到(并发删除 / 竞态)。吞掉、返回未入库的内存对象(调用方不读返回值)。
return rec
db.refresh(rec)
return rec
@@ -76,6 +81,20 @@ def find_by_session(
).scalar_one_or_none()
def _find_by_session_global(db: Session, ad_session_id: str | None) -> AdEcpmRecord | None:
"""按 ad_session_id **全局**查找(与唯一约束 uq_ad_ecpm_record_session 同口径,不含 user_id)。
仅 create_ecpm_record 撞约束后兜底用:此时撞的是全局会话约束,既存记录可能属于**另一个 user**,
带 user_id 的 find_by_session 会漏掉它、导致误判「查无 → raise 500」。其它业务查「某 user 的某次
展示 eCPM」仍用 find_by_session(带 user_id,语义更准),不走这里。
"""
if not ad_session_id:
return None
return db.execute(
select(AdEcpmRecord).where(AdEcpmRecord.ad_session_id == ad_session_id)
).scalar_one_or_none()
def count_today(db: Session, user_id: int) -> int:
"""该用户今日(北京时间)上报的 eCPM 条数,排查/对账辅助用。"""
return db.execute(
+28 -17
View File
@@ -73,16 +73,19 @@ def grant_feed_reward(
app_env: str | None = None,
our_code_id: str | None = None,
aborted: bool = False,
display_coin: int = 0,
) -> AdFeedRewardRecord:
"""**每条**信息流广告(客户端每条各上报一次)结算奖励。client_event_id 幂等,同号重试不重复发。
发奖规则:**一条广告 = 一个单次公式值**(rewards.calculate_ad_reward_coin),因子2(LT)按账号累计
**条**数递进;看满一份时长(unit_count>=1, 即 ≥10 秒)才发,**不逐份累加**。
发奖规则(所见即所得, 2026-06-27 用户拍板「显示多少给多少」):优先**直接发客户端小球显示的金币
display_coin**;防刷钳到本条「1 份满额」(eCPM 已钳 AD_ECPM_MAX_FEN, 因子2 按账号累计已发条数取档),
合法显示(实际因子2 × 进度 p ≤ 1 份)不被砍, 只挡伪造天价值。旧客户端不传 display_coin 时退回
「看满 10 秒发整份」(兼容不断币)。因子2(LT)由**客户端**按 granted 行 COUNT(拉自 /feed-reward/units)
算进 display_coin, 后端只记 granted 行让该计数自增, 不再服务端重算份值。
- aborted=True(用户中途 ✕ 关闭这条):本条不发,记 status='closed_early'
- 时长不足 10 秒(unit_count==0):记 status='too_short' 不发。
- display_coin 为 0 且时长不足一份:记 status='too_short' 不发(不计 LT / 当日上限)
- 命中当日条数上限:记 status='capped' 不发。
duration_seconds 是**这一条**的观看秒数。服务端两道硬闸防刷:时长钳到 FEED_MAX_DURATION_SECONDS、
eCPM 在 calculate_ad_reward_coin 内钳到 AD_ECPM_MAX_FEN;叠加每日 get_ad_daily_limit 条数上限。
duration_seconds 落库留痕(unit_count 字段), 旧端兼容路径据它判是否满 1 份。
feed_scene:点位场景(comparison/coupon/welfare),仅归类落库,不参与计算。
ad_type:广告形态(feed 信息流 / draw Draw 信息流),仅归类落库;**每日上限与因子2(LT)仍按本表
全表 unit 累计(feed+draw 共享同一发奖池/上限),不按 ad_type 拆分**。
@@ -139,14 +142,26 @@ def grant_feed_reward(
)
return _commit_record(db, rec, client_event_id)
# 整场总时长不足 10 秒,凑不满一份 → 不发,记 too_short 留痕。
if unit_count == 0:
# 所见即所得(用户 2026-06-27「显示多少给多少」): 优先发**客户端小球显示**的金币 display_coin,
# 钳到本条「1 份满额」防刷(eCPM 已钳 AD_ECPM_MAX_FEN; 合法显示=因子2×p≤1份, 不会被砍)。
# 因子2(LT)按账号累计已发条数(granted 行 COUNT), 第 existing_ads+1 条。
existing_ads = granted_unit_total(db, user_id)
unit_cap = rewards.calculate_ad_reward_coin(ecpm, existing_ads + 1)
if display_coin > 0:
coin = min(display_coin, unit_cap) # 新端: 所见即所得(直接发小球显示金币)
elif unit_count >= 1:
coin = unit_cap # 旧端没传 display_coin: 退回「看满 1 份发整份」(兼容)
else:
coin = 0
# 显示金币为 0 且没满一份 → 不发, 记 too_short 留痕(不写 granted 行 → 不计 LT / 当日上限)。
if coin <= 0:
rec = AdFeedRewardRecord(
client_event_id=client_event_id,
user_id=user_id,
reward_date=today,
duration_seconds=safe_duration,
unit_count=0,
unit_count=unit_count,
ad_session_id=ad_session_id,
ecpm_raw=ecpm,
adn=adn,
@@ -161,15 +176,11 @@ def grant_feed_reward(
)
return _commit_record(db, rec, client_event_id)
# 一条广告 = 一个「单次公式值」(因子2 按账号累计**条**数, 即第 existing_ads+1 条);看满一份(unit_count>=1)即发,不逐份累加。
existing_ads = granted_unit_total(db, user_id)
coin = rewards.calculate_ad_reward_coin(ecpm, existing_ads + 1)
if coin > 0:
crud_wallet.grant_coins(
db, user_id, coin,
biz_type="feed_ad_reward", ref_id=client_event_id,
remark="信息流广告奖励",
)
crud_wallet.grant_coins(
db, user_id, coin,
biz_type="feed_ad_reward", ref_id=client_event_id,
remark="信息流广告奖励",
)
rec = AdFeedRewardRecord(
client_event_id=client_event_id,
user_id=user_id,
+5
View File
@@ -167,6 +167,11 @@ class FeedRewardIn(BaseModel):
aborted: bool = Field(
False, description="用户中途 ✕ 关闭广告(未走完比价):整场不发,记 closed_early"
)
display_coin: int = Field(
0, ge=0,
description="客户端金币小球**本条显示**的金币(所见即所得):后端直接发这个数,钳到本条最大 1 份"
"满额防刷。缺省 0 = 旧客户端不传,退回服务端「看满 10 秒发整份」",
)
class FeedRewardOut(BaseModel):
+29 -1
View File
@@ -11,7 +11,7 @@ from __future__ import annotations
from datetime import datetime
from pydantic import BaseModel, ConfigDict, Field
from pydantic import BaseModel, ConfigDict, Field, field_validator
# ===== 上报请求 =====
@@ -21,8 +21,36 @@ class ComparisonItemIn(BaseModel):
name: str
qty: int = 1
# specs 仅供比价记录展示(admin-web 详情 / app 记录页都按字符串数组渲染并 join)。
# pricebot(2026-06-27 嵌套规格统一)起 calibration 的 specs 可能是规格对象
# [{name, qty, sub_specs}] 而非字符串 → 下面的 before-validator 统一拍平成可读字符串,
# 兼容新旧两种输入、保持 list[str] 契约不变(下游零改动)。
# ⚠️ 两个坑都踩过, 必须"拍平"而非别的: ① 直接声明 list[str] 不拍平 → 对象 specs 整条
# 422 被拒、不入库(同下方 platform_results list→dict 同类事故); ② 仅放宽成裸 list 又会让
# 下游 join 出 "[object Object]"/对象 toString 的乱码。
specs: list[str] | None = None
@field_validator("specs", mode="before")
@classmethod
def _flatten_specs(cls, v: object) -> object:
"""pricebot 规格对象 [{name, qty, sub_specs}] → 可读字符串数组; 字符串元素原样保留;
非 list 原样交还(让 pydantic 照常报类型错)。嵌套规格拼成 '主项(子1,子2)'"""
if not isinstance(v, list):
return v
out: list[str] = []
for it in v:
if isinstance(it, str):
s = it
elif isinstance(it, dict):
name = str(it.get("name") or "").strip()
subs = [str(x).strip() for x in (it.get("sub_specs") or []) if str(x).strip()]
s = f"{name}({','.join(subs)})" if name and subs else (name or ",".join(subs))
else:
continue
if s:
out.append(s)
return out
class AppliedCouponIn(BaseModel):
"""单笔已用优惠(来自 comparison_results[].applied_coupons)。amount 单位:元、正数。"""
+26 -5
View File
@@ -27,8 +27,11 @@
| `date_to` | string | =`date_from` | 结束日 北京时间 `YYYY-MM-DD`,**闭区间**;单日时与 `date_from` 相同 |
| `user_id` | int | 全部 | 只看某用户;不传=所有用户 |
| `ad_type` | string | 全部 | `reward_video` / `feed` / `draw`;不传=全部类型 |
| `feed_scene` | string | 全部 | `comparison`(比价)/ `coupon`(领券)/ `welfare`(福利);**全局筛选**,同时作用于明细 / 合计 / `daily`·`hourly` 趋势;不传=全部场景 |
| `granularity` | string | `day` | `day`=按天 / `hour`=按小时(聚合键再加北京时间小时 0–23);**区间>1 天建议用 day** |
| `limit` | int(1~1000) | 500 | **展示**明细组数(截断;`total`/`total_*`/`daily` 按全量统计不受影响) |
| `limit` | int(1~1000) | 500 | **每页条数**(分页大小);`total`/`total_*`/`daily`/`hourly` 按全量统计不受分页影响 |
| `offset` | int(≥0) | 0 | 分页偏移(已跳过条数)=(页码−1)×`limit` |
| `sort` | string | `time` | 明细排序:`time`=按时间倒序(新→旧) / `ecpm`=按 eCPM 数值倒序 |
约束:`date_to` 不早于 `date_from`、区间最长 **92 天**、日期须 `YYYY-MM-DD`,否则 `422`
@@ -36,15 +39,18 @@
| 字段 | 类型 | 说明 |
|---|---|---|
| `date_from` / `date_to` | string | 报表起止日期(闭区间) |
| `daily` | `AdRevenueDaily[]` | 按日期汇总序列(全量,供按天趋势图;不受 `limit` 影响) |
| `total` | int | 聚合组**总数**(全量,不受 `limit` 影响) |
| `truncated` | bool | 明细是否被 `limit` 截断 |
| `daily` | `AdRevenueDaily[]` | 按日期汇总序列(全量,供按天趋势图;不受分页影响) |
| `hourly` | `AdRevenueHourly[]` | 按小时汇总序列(全量,供按小时趋势图;**仅 `granularity=hour` 时非空**;不受分页影响) |
| `type_stats` | `{[ad_type]: AdRevenueTypeStat}` | 按广告类型(`ad_type`)小计(全量);前端取 `draw` / `reward_video` 做分类大盘 |
| `dau` | int \| null | 今日活跃用户数(复用大盘口径 `last_login_at`,今日登录过);**仅查询=今日单天时有值**,历史/多天为 `null` |
| `total` | int | 当前筛选下的**分页总条数**(全量,不受分页影响;= 前端分页器 total) |
| `truncated` | bool | 当前页之后是否还有更多事件(`len(events) > offset + limit`) |
| `total_impressions` | int | 全量展示条数合计 |
| `total_revenue_yuan` | float | 全量收益合计(元) |
| `total_expected_coin` | int | 全量应发金币合计 |
| `total_actual_coin` | int | 全量实发金币合计 |
| `mismatch_count` | int | 应发≠实发的组数(=0 说明全部按公式发放) |
| `items` | `AdRevenueRow[]` | 聚合明细(按 日期→用户→类型→代码位 排序) |
| `items` | `AdRevenueRow[]` | 逐条广告事件(**按时间倒序:新→旧**);`limit`/`offset` 对全量做分页切片,返回当前页 |
### AdRevenueDaily(`daily[]` — 按天趋势)
| 字段 | 类型 | 说明 |
@@ -55,6 +61,21 @@
| `expected_coin` | int | 当天应发金币合计 |
| `actual_coin` | int | 当天实发金币合计 |
### AdRevenueHourly(`hourly[]` — 按小时趋势,仅 `granularity=hour` 时非空)
| 字段 | 类型 | 说明 |
|---|---|---|
| `hour` | int | 北京时间小时 023 |
| `impressions` | int | 该小时展示条数合计 |
| `revenue_yuan` | float | 该小时预估收益合计(元) |
| `expected_coin` | int | 该小时应发金币合计 |
| `actual_coin` | int | 该小时实发金币合计 |
### AdRevenueTypeStat(`type_stats[ad_type]` — 分广告类型小计,供大盘第二行)
| 字段 | 类型 | 说明 |
|---|---|---|
| `impressions` | int | 该类型展示条数合计 |
| `revenue_yuan` | float | 该类型预估收益合计(元);eCPM 由前端用 收益÷展示×1000 算 |
### AdRevenueRow(`items[]`)
| 字段 | 类型 | 说明 |
|---|---|---|