Compare commits

...

5 Commits

Author SHA1 Message Date
zhuzihao e9fd51d119 fix(auth): 发码防刷改为按成功计数 + 新增每设备每日发码上限 (#136)
- 修复:发码限流原为原子「判+记」,被单号 60s 冷却挡下的重发也占设备额度
  → 正常用户连点重发可能被误锁 1 小时。改为「先判后记、只对成功发码计数」:
  check 判在真发之前(超限直接 429、不真发),record 只在 send_code 成功后调;
  被单号冷却 / 供应商失败抛 429 时直接返回、不计数。
- 新增:同一设备(device_id)+ IP 每天最多 20 次发码上限,与原每小时 5 次两道闸并存,
  均按成功计数,叠一层日封顶挡低频长时间轰炸。
- 基建:ratelimit.py 新增 RateLimitRule + check_rate_limits / record_rate_limits
  (peek/commit 拆分);原子的 enforce_rate_limit 仍保留给登录爆破(失败也计)不变。
- 测试:补 2 个用例(冷却挡下不占额度 / 每日上限)。

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

---------

Co-authored-by: zzhyyyyy <2685922758@qq.com>
Reviewed-on: #136
Co-authored-by: zhuzihao <zhuzihao@wonderable.ai>
Co-committed-by: zhuzihao <zhuzihao@wonderable.ai>
2026-07-16 09:40:56 +08:00
guke 5c6840dd71 feat(compare): 比价记录 LLM token 成本落库与展示(按当时价冻结) (#133)
- comparison_record 加 llm_cost_yuan(元/float)+ llm_price_snapshot(JSON)两列
- _backfill_llm_calls 回填时按 app_config 当时单价逐模型算成本、冻结成本+快照到记录
- app_config 新增 llm_token_price 配置(per_model + default 兜底,运营在系统配置页可改)
- services/llm_cost.py:compute_llm_cost 纯函数(按 model 分桶、error/无 usage 跳过、
  脏价格当 unpriced 不抛异常以免连累 token 回填)+ get_llm_prices reader
- admin schema 暴露成本:列表项带 llm_cost_yuan,详情另带价格快照
- tests/test_llm_cost.py(10 测试);scripts/seed_mock_llm_cost.py(mock seeder)

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

---------

Co-authored-by: guke <guke@autohome.com.cn>
Reviewed-on: #133
2026-07-13 17:46:11 +08:00
marco 824045dd19 fix迁移问题 2026-07-11 13:47:12 +08:00
guke 930eff822c feat(ad-revenue): 领券/比价看板逐次广告收益(ad_ecpm.trace_id + 逐页聚合) (#131)
## 背景
admin「领券数据」「比价记录」两个看板此前只能看到场景级(所有领券/比价)的广告收益,
无法定位「这一次领券/比价具体赚了多少」。根因:收益表 `ad_ecpm_record` 缺 `trace_id`,
无法与领券会话 / 比价记录按 trace 关联。

## 改动
- **模型/迁移**:`ad_ecpm_record` 新增 `trace_id`(String(64), index, nullable);
  迁移 `ad_ecpm_trace_id` 加列 + 索引 `ix_ad_ecpm_record_trace_id`,并**收敛当前两个
  alembic head**(`11c44afbea58` selfstat + `merge_pages_override_coupon_slot`)为单 head。
- **上报链路**:`EcpmReportIn` 增 `trace_id` 字段;`/api/v1/ad/ecpm-report` 透传;
  `create_ecpm_record` 落库。
- **收益聚合**:新增 `revenue_yuan_by_trace(db, trace_ids)`——按 trace_id 聚合展示收益,
  单条 = `min(eCPM元, ¥500钳顶)/1000`,与广告收益报表 `ad_revenue.py` **同口径**;
  只吃当前页的 trace_id(逐页批量,索引命中,无 N+1)。
- **两个看板**:`CouponDataRow` / `AdminComparisonListItem` 增 `ad_revenue_yuan`;
  `coupon_data_report`(主表 + 用户抽屉)与 `list_comparison_records` 分页后逐页补该字段。
- **测试**:`tests/test_ad_ecpm_trace_revenue.py`(聚合/钳顶/落库)、
  `tests/test_board_ad_revenue.py`(两看板 + 抽屉)。

---------

Co-authored-by: guke <guke@autohome.com.cn>
Reviewed-on: #131
2026-07-10 22:14:00 +08:00
marco 285e46ebaf 迁移处理 2026-07-10 19:23:12 +08:00
26 changed files with 1905 additions and 27 deletions
+44
View File
@@ -0,0 +1,44 @@
"""ad_ecpm_record.trace_id(展示收益归属到比价/领券 trace)
信息流(Draw)展示 eCPM 上报时带上本场比价/领券 trace_id,落此列;领券数据 / 比价记录看板
按 trace_id 聚合"本次广告收益"。激励视频/福利/旧客户端为 NULL。
本迁移原以 (11c44afbea58, merge_pages_override_coupon_slot) 为双亲、顺带收敛双 head,
但与它并行落 main 的 merge_selfstat_coupon_slot 已用同一对双亲做了纯收敛 → 同一对
父节点出现两个收敛点、main 上又成双 head。故重挂到该 merge 之后成单链(仅改链接、
schema 改动不变;两文件都保留,已 stamp 在 merge 上的库可直接线性升级)。
Revision ID: ad_ecpm_trace_id
Revises: merge_selfstat_coupon_slot
Create Date: 2026-07-10
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
revision: str = "ad_ecpm_trace_id"
down_revision: Union[str, Sequence[str], None] = "merge_selfstat_coupon_slot"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
# SQLite 下 ADD COLUMN(可空)与 CREATE INDEX 均原生支持,无需 batch_alter_table
# (同 ad_feed_reward_trace_id 迁移)。
op.add_column(
"ad_ecpm_record",
sa.Column("trace_id", sa.String(length=64), nullable=True),
)
op.create_index(
op.f("ix_ad_ecpm_record_trace_id"),
"ad_ecpm_record",
["trace_id"],
unique=False,
)
def downgrade() -> None:
op.drop_index(op.f("ix_ad_ecpm_record_trace_id"), table_name="ad_ecpm_record")
op.drop_column("ad_ecpm_record", "trace_id")
+33
View File
@@ -0,0 +1,33 @@
"""comparison_record: llm_cost_yuan + llm_price_snapshot(比价 LLM 调用成本 + 当时单价快照)
回填 llm_calls 时按「当时的价」逐模型算出本次比价 LLM 总成本(元),连同所用单价快照一起冻结到
记录上;admin 比价记录详情展示实际成本(旧记录 NULL → 前端回退估算)。见 services/llm_cost.py。
Revision ID: comparison_llm_cost
Revises: ad_ecpm_trace_id
Create Date: 2026-07-13
"""
from collections.abc import Sequence
import sqlalchemy as sa
from sqlalchemy.dialects import postgresql
from alembic import op
revision: str = "comparison_llm_cost"
down_revision: str | Sequence[str] | None = "ad_ecpm_trace_id"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
_JSONB = sa.JSON().with_variant(postgresql.JSONB(), "postgresql")
def upgrade() -> None:
# 均可空、无索引;SQLite 原生支持 ADD COLUMN,无需 batch_alter_table(同 comparison_debug_fields)。
op.add_column("comparison_record", sa.Column("llm_cost_yuan", sa.Float(), nullable=True))
op.add_column("comparison_record", sa.Column("llm_price_snapshot", _JSONB, nullable=True))
def downgrade() -> None:
op.drop_column("comparison_record", "llm_price_snapshot")
op.drop_column("comparison_record", "llm_cost_yuan")
@@ -0,0 +1,29 @@
"""合并两个 alembic head:11c44afbea58(#127 埋点健康度 selfstat)+ merge_pages_override_coupon_slot(#130 自带的合并迁移)。
三条分支都从 admin_user_plain_password 分叉(#126 权限 / #127 selfstat / #130 领券成功率)。
#130 自带的 merge 创建时本地 main 尚无 #127 的 11c44afbea58,只收敛了 #126 + 自身两条,
#130 合入后 main 上仍留两个 head → `alembic upgrade head`(单数,部署/run.sh 用)直接报错、服务起不来。
本迁移仅把二者收敛成单 head;**不含任何表结构 / 数据改动**(纯 merge)。
Revision ID: merge_selfstat_coupon_slot
Revises: 11c44afbea58, merge_pages_override_coupon_slot
Create Date: 2026-07-10 00:00:00.000000
"""
from collections.abc import Sequence
revision: str = "merge_selfstat_coupon_slot"
down_revision: str | Sequence[str] | None = (
"11c44afbea58",
"merge_pages_override_coupon_slot",
)
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
def upgrade() -> None:
"""纯合并 head,无 schema 改动。"""
def downgrade() -> None:
"""拆回两个 head,无 schema 改动。"""
+10 -3
View File
@@ -18,6 +18,7 @@ from sqlalchemy.orm import Session
from app.core import rewards
from app.models.coupon_state import CouponClaimRecord, CouponSession
from app.models.user import User
from app.repositories import ad_ecpm as crud_ecpm
from app.repositories.coupon_state import DEFAULT_PLATFORMS, coupon_id_to_platform
@@ -85,7 +86,7 @@ def _success_rates(rows: list) -> dict:
}
def _session_to_row(r, phone: str | None = None, nickname: str | None = None) -> dict:
def _session_to_row(r, phone: str | None = None, nickname: str | None = None, ad_revenue_yuan: float = 0.0) -> dict:
"""CouponSession ORM → 明细行 dict(主表「领券数据」与「用户全部领券」抽屉共用)。"""
return {
"id": r.id,
@@ -104,6 +105,7 @@ def _session_to_row(r, phone: str | None = None, nickname: str | None = None) ->
"started_at": r.started_at,
"claimed_count": r.claimed_count,
"trace_url": r.trace_url,
"ad_revenue_yuan": ad_revenue_yuan,
}
@@ -246,10 +248,11 @@ def coupon_data_report(
select(User.id, User.phone, User.nickname).where(User.id.in_(uids))
).all()
}
rev_map = crud_ecpm.revenue_yuan_by_trace(db, [r.trace_id for r in page])
items = []
for r in page:
phone, nickname = user_map.get(r.user_id, (None, None)) if r.user_id is not None else (None, None)
items.append(_session_to_row(r, phone, nickname))
items.append(_session_to_row(r, phone, nickname, ad_revenue_yuan=rev_map.get(r.trace_id, 0.0)))
return {
"summary": summary,
@@ -271,7 +274,11 @@ def coupon_user_records(db: Session, *, user_id: int, limit: int = 100) -> dict:
total = db.execute(
select(func.count()).select_from(CouponSession).where(CouponSession.user_id == user_id)
).scalar_one()
return {"items": [_session_to_row(r) for r in rows], "total": int(total)}
rev_map = crud_ecpm.revenue_yuan_by_trace(db, [r.trace_id for r in rows])
return {
"items": [_session_to_row(r, ad_revenue_yuan=rev_map.get(r.trace_id, 0.0)) for r in rows],
"total": int(total),
}
_SLOT_OK = ("success", "already_claimed")
+6
View File
@@ -32,6 +32,7 @@ from app.models.wallet import (
InviteCashTransaction,
WithdrawOrder,
)
from app.repositories import ad_ecpm
# 「最近活跃」计入的行为事件(与大盘 DAU/留存活跃口径一致:开始比价 + 开始领券)
_ACTIVE_EVENTS = (COMPARE_START_EVENT, COUPON_START_EVENT)
@@ -298,6 +299,11 @@ def list_comparison_records(
limit=limit, cursor=cursor,
)
_attach_user_info(db, 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])
for it in items:
it.ad_revenue_yuan = rev.get(it.trace_id, 0.0)
return items, next_cursor, total
+5
View File
@@ -36,11 +36,14 @@ class AdminComparisonListItem(BaseModel):
retry_count: int | None = None
input_tokens: int | None = None # Σ usage.prompt_tokens(server 派生)
output_tokens: int | None = None # Σ usage.completion_tokens(server 派生)
# 本次比价 LLM 总成本(元,按当时价冻结);旧记录/未回填为 None → 前端「成本」列回退估算。见 services/llm_cost.py。
llm_cost_yuan: float | None = None
device_model: str | None = None
rom_vendor: str | None = None
rom_name: str | None = None
android_version: str | None = None
app_version: str | None = None
ad_revenue_yuan: float = 0.0 # 本次比价看的信息流广告预估收益(元),queries 瞬态挂 ORM 实例上
created_at: datetime
@@ -71,3 +74,5 @@ class AdminComparisonDetail(AdminComparisonListItem):
# 原始上报全量;「卡在哪一步」从 raw_payload.platform_results[*].status 读
# (store_not_found/items_not_found/below_minimum/unsupported = 卡在 找店/加菜/起送/读价)。
raw_payload: dict | None = None
# 算成本所用单价快照 {mode, prices:{model:{...}}}(llm_cost_yuan 继承自列表项)。见 services/llm_cost.py。
llm_price_snapshot: dict | None = None
+3
View File
@@ -70,6 +70,9 @@ class CouponDataRow(BaseModel):
started_at: datetime = Field(..., description="发起时刻(明细「时间」列)")
claimed_count: int | None = None
trace_url: str | None = Field(None, description="pricebot 公网 trace 链接(仅 completed 有);admin 渲染可点链接,无则显示可复制 trace_id")
ad_revenue_yuan: float = Field(
0.0, description="本次领券看的信息流广告预估收益(元);按 trace_id 聚合 ad_ecpm_record"
)
class CouponDataOut(BaseModel):
+1
View File
@@ -286,6 +286,7 @@ def ecpm_report(payload: EcpmReportIn, user: CurrentUser, db: DbSession) -> Ecpm
ad_session_id=payload.ad_session_id,
adn=payload.adn, slot_id=payload.slot_id,
feed_scene=payload.feed_scene,
trace_id=payload.trace_id,
app_env=payload.app_env, our_code_id=payload.our_code_id,
)
logger.info(
+24 -15
View File
@@ -12,11 +12,16 @@ from __future__ import annotations
import logging
from fastapi import APIRouter, HTTPException, Request, status
from fastapi import APIRouter, HTTPException, Request
from app.api.deps import CurrentUser, DbSession
from app.core import test_account
from app.core.ratelimit import enforce_rate_limit
from app.core.ratelimit import (
RateLimitRule,
check_rate_limits,
enforce_rate_limit,
record_rate_limits,
)
from app.core.security import TokenError, decode_token, issue_token_pair
from app.integrations.jiguang import JiguangError, mask_phone, verify_and_get_phone
from app.integrations.sms import SmsError, send_code, verify_code
@@ -40,9 +45,10 @@ router = APIRouter(prefix="/api/v1/auth", tags=["auth"])
# 手机号登录防刷:同一设备(device_id) + 同一 IP 每小时最多的登录尝试次数(成功/失败都计)。
SMS_LOGIN_MAX_PER_HOUR = 5
# 发码防刷:同一设备(device_id) + 同一 IP 每小时最多的发码次数。
# 发码防刷(同一设备 device_id + 同一 IP,**只按成功发码计数**;被单号 60s 冷却挡下的重发不占额度):
# 堵「换手机号绕开单号 60s 冷却」的洞 —— 冷却是单号维度,一机换号能绕开。
SMS_SEND_MAX_PER_HOUR_PER_DEVICE = 5
SMS_SEND_MAX_PER_HOUR_PER_DEVICE = 5 # 每小时上限
SMS_SEND_MAX_PER_DAY_PER_DEVICE = 20 # 每天上限(再叠一层日封顶,挡低频长时间轰炸)
def _login_response(
@@ -99,23 +105,26 @@ def sms_send(req: SmsSendRequest, request: Request) -> SmsSendResponse:
logger.info("test_account sms_send short-circuit (不真发)")
return SmsSendResponse(sent=True, mock=True, cooldown_sec=0)
# 防刷:同一设备(device_id) + 同一 IP 每小时最多 SMS_SEND_MAX_PER_HOUR_PER_DEVICE 次发码
# 补「换手机号绕开单号 60s 冷却」的洞(冷却是单号维度,一机换号能绕);设备维度按机器封顶,
# 挡短信轰炸/烧钱。放在真发(send_code)之前 → 超限直接拦下、不真发短信。
enforce_rate_limit(
request,
scope="sms-send-device",
subject=req.device_id,
limit=SMS_SEND_MAX_PER_HOUR_PER_DEVICE,
window_sec=3600,
detail="操作过于频繁,请稍后再试",
)
# 发码防刷:同一设备(device_id) + 同一 IP,每小时 / 每天两道闸,**均只按成功发码计数**
# 补「换手机号绕开单号 60s 冷却」的洞(冷却是单号维度,一机换号能绕);设备维度按机器封顶,挡短信轰炸/烧钱。
# 关键:被单号 60s 冷却挡下的重发是「没真发、没烧钱」→ 不该占额度。故 check(先判)放在真发之前
# (超限直接 429、不真发),record(计数)只在 send_code 成功后调 —— 冷却/供应商失败抛 429 时直接返回、不计数。
send_rules = [
RateLimitRule("sms-send-device", SMS_SEND_MAX_PER_HOUR_PER_DEVICE, 3600,
"操作过于频繁,请稍后再试"),
RateLimitRule("sms-send-device-daily", SMS_SEND_MAX_PER_DAY_PER_DEVICE, 86400,
"今日验证码发送次数过多,请明天再试"),
]
check_rate_limits(request, subject=req.device_id, rules=send_rules)
try:
cooldown = send_code(req.phone)
except SmsError as e:
raise HTTPException(status_code=e.status_code, detail=str(e)) from e
# 发码成功 → 两道闸各 +1(被单号冷却挡下的重发走不到这里,故不占额度)
record_rate_limits(request, subject=req.device_id, rules=send_rules)
from app.core.config import settings # 局部 import 避免循环
return SmsSendResponse(sent=True, mock=settings.SMS_MOCK, cooldown_sec=cooldown)
+3
View File
@@ -27,6 +27,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
logger = logging.getLogger("shagua.compare_record")
@@ -81,6 +82,8 @@ def _backfill_llm_calls(record_id: int, trace_id: str) -> None:
# 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",
+15
View File
@@ -96,4 +96,19 @@ CONFIG_DEFS: dict[str, dict[str, Any]] = {
"group": "首页轮播", "type": "enum", "hidden": True,
"help": "mixed=真实优先+种子补位(默认);real=只用真实比价记录;seed=只用种子/合成(演示)。",
},
# 比价 LLM 调用成本计价。值是嵌套 JSON(非 str→int),借 dict_str_int 类型在配置页走原始 JSON
# 编辑框;set_value 不校验类型,嵌套 JSON 照存。
"llm_token_price": {
"default": {
"per_model": {"qwen3.5-flash": {"input_per_1m": 0.8, "output_per_1m": 2.0}},
"default": {"input_per_1m": 3.0, "output_per_1m": 15.0},
"currency": "CNY", "unit": "per_1m_tokens",
},
"label": "LLM 模型单价(元/百万 token)",
"group": "LLM 成本", "type": "dict_str_int",
"help": (
"比价 LLM 调用成本计价。JSON:per_model 按模型配 input/output 单价(元/1M token),"
"default 兜底未登记的模型。改价只影响之后回填的新记录,历史记录用当时价格快照。"
),
},
}
+93 -8
View File
@@ -9,29 +9,41 @@ from __future__ import annotations
import threading
import time
from typing import NamedTuple
from fastapi import HTTPException, Request, status
from app.core.config import settings
# key -> (window_start_ts, count)
_buckets: dict[str, tuple[float, int]] = {}
# key -> (window_start_ts, count, window_sec)
# 存每个 key 自己的 window_sec:_buckets 混着不同窗口(60s 广告 / 3600s 登录 / 86400s 日闸)的 key,
# GC 必须按各 key 自己的窗口判过期(见 [_purge_expired]),否则短窗口调用触发的 GC 会误删长窗口 key。
_buckets: dict[str, tuple[float, int, float]] = {}
_lock = threading.Lock()
_GC_THRESHOLD = 10000 # _buckets 超此阈值才顺手清过期 key(仿 sms.py;测试可 monkeypatch 调小强制每次扫)
def _purge_expired(now: float) -> None:
"""清过期 key(**仅在持有 _lock 时调用**)。按每个 key 自己存的 window_sec 判过期,而非调用方的窗口
_buckets 是全局共享混着 60s(广告)/3600s(登录)/86400s(日闸)不同窗口的 key;若用调用方窗口,
高频的 60s 广告端点触发 GC 时会把本该活 3600s/86400s 的登录/日闸计数一并删掉,使其在规模上(超阈值才
触发本清理)被反复清零而失效仅在超阈值时扫,低频开销可忽略"""
if len(_buckets) <= _GC_THRESHOLD:
return
for k in [k for k, (s, _, w) in _buckets.items() if now - s >= w]:
_buckets.pop(k, None)
def _hit(key: str, limit: int, window_sec: float) -> bool:
"""记一次访问。返回 True=放行,False=超限。"""
now = time.monotonic()
with _lock:
start, count = _buckets.get(key, (now, 0))
start, count, _ = _buckets.get(key, (now, 0, window_sec))
if now - start >= window_sec: # 窗口过期,重置
start, count = now, 0
count += 1
_buckets[key] = (start, count)
# 顺手清过期 key,防内存无限涨(低频访问足够)
if len(_buckets) > 10000:
for k in [k for k, (s, _) in _buckets.items() if now - s >= window_sec]:
_buckets.pop(k, None)
_buckets[key] = (start, count, window_sec)
_purge_expired(now) # 顺手清过期 key(按各自窗口),防内存无限涨
return count <= limit
@@ -83,3 +95,76 @@ def enforce_rate_limit(
status_code=status.HTTP_429_TOO_MANY_REQUESTS,
detail=detail,
)
# ===================== 先判 / 后记(只按「成功」计数)=====================
# _hit 是原子「判+记」:一调用就 +1,适合登录爆破(失败尝试也要计)。但对「短信发码」这类
# **只想给成功动作计数**的场景不合适 —— 被单号冷却挡下的重发没真发、没烧钱,不该占额度。
# 故拆成 _peek(只判不记)+ _commit(只记):check_rate_limits 先判 → 动作 → 成功后 record。
class RateLimitRule(NamedTuple):
"""一条限流规则。scope 区分不同闸(不同 key 前缀);同一 (subject, IP) 在 window_sec
内最多 limit ,超限抛 429 detail 文案
(scope, window_sec) 成对绑在一条规则里 check(先判) record(计数)复用同一条,
避免两处把窗口/scope 写歪导致 key 对不上
"""
scope: str
limit: int
window_sec: float
detail: str = "操作过于频繁,请稍后再试"
def _peek(key: str, limit: int, window_sec: float) -> bool:
"""只读:当前窗口内是否还没到上限(count < limit)。**不改计数**。
[_commit] 配对实现先判后记只在动作成功后才 _commit"""
now = time.monotonic()
with _lock:
start, count, _ = _buckets.get(key, (now, 0, window_sec))
if now - start >= window_sec: # 窗口已过期 → 视作已重置(count 归零)
count = 0
return count < limit
def _commit(key: str, window_sec: float) -> None:
"""记一次访问(+1)。窗口过期则以本次为起点重置。仅在动作成功后调用。"""
now = time.monotonic()
with _lock:
start, count, _ = _buckets.get(key, (now, 0, window_sec))
if now - start >= window_sec: # 窗口过期,重置
start, count = now, 0
_buckets[key] = (start, count + 1, window_sec)
_purge_expired(now) # 顺手清过期 key(按各自窗口,同 [_hit])
def check_rate_limits(request: Request, subject: str, rules: list[RateLimitRule]) -> None:
"""【先判】一组限流:任一规则已达上限即抛 429,且**不改计数**。
配合 [record_rate_limits] 实现只按成功计数: check 所有闸(全未超才继续) 执行动作
动作**成功后** record动作被下游挡下(如短信单号冷却)没真正发生时不 record 不占额度
key = `scope:subject:client_ip`( [enforce_rate_limit] 同款)
"""
if not settings.RATE_LIMIT_ENABLED:
return
ip = _client_ip(request)
for rule in rules:
if not _peek(f"{rule.scope}:{subject}:{ip}", rule.limit, rule.window_sec):
raise HTTPException(
status_code=status.HTTP_429_TOO_MANY_REQUESTS,
detail=rule.detail,
)
def record_rate_limits(request: Request, subject: str, rules: list[RateLimitRule]) -> None:
"""【记一次】一组限流(每条规则 +1)。仅在动作成功后调用,与 [check_rate_limits] 配对。
check动作record 非原子:并发突发下计数可能略超 limit(每个在途请求各 +1)
防脚本/防轰炸的安全网定位可接受;要精确配额需迁 Redis(见模块 docstring)
"""
if not settings.RATE_LIMIT_ENABLED:
return
ip = _client_ip(request)
for rule in rules:
_commit(f"{rule.scope}:{subject}:{ip}", rule.window_sec)
+2 -1
View File
@@ -13,7 +13,8 @@ worker / 多机时内存不共享 → 冷却、校验都会失效,届时迁移
防刷两层(短信花钱 + `/sms/send` 在登录前无法 JWT 鉴权):
1. 单号 `SMS_SEND_INTERVAL_SEC` 冷却(本文件)
2. 单设备(device_id)每小时频控(api auth.sms_send enforce_rate_limit)+ 极光控制台 IP 白名单/防轰炸(运维侧)
2. 单设备(device_id)+ IP 每小时 / 每天频控(api auth.sms_send check/record_rate_limits,
**只按成功发码计数** 被本文件单号冷却挡下的重发不占额度)+ 极光控制台 IP 白名单/防轰炸(运维侧)
IP 频控(rate_limit 依赖)2026-06-26 按产品要求删除改设备维度; device_id 客户端可伪造/轮换,
脚本轮换 id 能绕过本层 挡脚本狂发主要靠极光控制台侧(+ 可选 nginx 限流)
单号每日上限2026-07-03 按精简要求删除(mentor :登录风控只留单号冷却 + 单设备频控);
+3
View File
@@ -32,6 +32,9 @@ class AdEcpmRecord(Base):
# 点位场景:comparison(比价) / coupon(领券) / welfare(福利),供收益报表区分比价/领券 Draw 收益;
# 仅信息流/Draw 上报(比价与领券共用同一代码位,只能客户端各调用点显式打标),激励视频为 NULL。
feed_scene: Mapped[str | None] = mapped_column(String(16), nullable=True)
# 本次比价/领券 trace_id(信息流场景客户端带上):把这条展示收益归属到对应比价/领券记录。
# 领券数据 / 比价记录看板按 trace_id 聚合"本次广告收益"。激励视频/福利/旧客户端 = NULL。
trace_id: Mapped[str | None] = mapped_column(String(64), index=True, nullable=True)
# 客户端生成的一次广告会话 id;激励视频 S2S 回调 extra 会透传同值
ad_session_id: Mapped[str | None] = mapped_column(String(64), index=True, nullable=True)
# 实际投放的 ADN(穿山甲 getShowEcpm().getSdkName(),如 pangle / gdt)
+6
View File
@@ -137,6 +137,12 @@ class ComparisonRecord(Base):
# 每次 LLM 调用明细 [{scene,model,input_messages,output,usage,latency_ms,error}];
# server 收上报后按 trace_id 同机拉 pricebot 落库(见 compare_record 端点)。旧记录/未采集为 None。
llm_calls: Mapped[list | None] = mapped_column(_JSON, nullable=True)
# 本次比价 LLM 总成本(元):回填时按「当时的价」逐模型算好冻结(见 services/llm_cost.py)。
# 单次亚分级 → float「元」(不用 *_cents)。旧记录/未回填为 None,前端回退「估算成本」。
llm_cost_yuan: Mapped[float | None] = mapped_column(Float, nullable=True)
# 算成本所用单价快照 {mode, prices:{model:{input_per_1m,output_per_1m,_source}}}:app_config 只存
# 当前价、不留历史,故把当时价冻结进来供审计/复算。
llm_price_snapshot: Mapped[dict | None] = mapped_column(_JSON, nullable=True)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), server_default=func.now(), index=True, nullable=False
+27
View File
@@ -10,6 +10,7 @@ from sqlalchemy import func, select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session
from app.core import rewards
from app.core.rewards import cn_today
from app.models.ad_ecpm import AdEcpmRecord
@@ -24,6 +25,7 @@ def create_ecpm_record(
adn: str | None = None,
slot_id: str | None = None,
feed_scene: str | None = None,
trace_id: str | None = None,
app_env: str | None = None,
our_code_id: str | None = None,
) -> AdEcpmRecord:
@@ -43,6 +45,7 @@ def create_ecpm_record(
adn=adn,
slot_id=slot_id,
feed_scene=feed_scene,
trace_id=trace_id,
app_env=app_env,
our_code_id=our_code_id,
ecpm_raw=ecpm_raw,
@@ -105,3 +108,27 @@ def count_today(db: Session, user_id: int) -> int:
AdEcpmRecord.report_date == cn_today().isoformat(),
)
).scalar_one()
def revenue_yuan_by_trace(db: Session, trace_ids: list[str]) -> dict[str, float]:
"""各 trace_id 的广告预估收益(元):按 trace_id 聚合 ad_ecpm_record 的展示收益。
单条展示收益 = min(eCPM元, AD_ECPM_MAX_FEN/100) / 1000( admin 广告收益报表同口径)
ecpm_raw 是字符串且需逐条钳顶,故取回后 Python 求和(行数=本页各 trace 的展示条数,很小)
trace_id 仅信息流(比价/领券)场景客户端带,激励视频/旧数据为 NULL, trace_id 过滤天然只算对应场景
只喂**当前页** trace_id( 一页条数);空集合直接返回(避免 IN () 非法)
"""
if not trace_ids:
return {}
rows = db.execute(
select(AdEcpmRecord.trace_id, AdEcpmRecord.ecpm_raw).where(
AdEcpmRecord.trace_id.in_(trace_ids),
)
).all()
cap_yuan = rewards.AD_ECPM_MAX_FEN / 100.0
out: dict[str, float] = {}
for tid, ecpm_raw in rows:
if not tid:
continue
out[tid] = out.get(tid, 0.0) + min(rewards.parse_ecpm_yuan(ecpm_raw), cap_yuan) / 1000.0
return {tid: round(v, 6) for tid, v in out.items()}
+6
View File
@@ -63,6 +63,12 @@ class EcpmReportIn(BaseModel):
description="点位场景:comparison(比价等待) / coupon(领券) / welfare(福利页);"
"比价与领券共用同一 Draw 代码位,需客户端在各调用点显式标注,供收益报表区分比价/领券;激励视频为空",
)
trace_id: str | None = Field(
None,
max_length=64,
description="本次比价/领券 trace_id(信息流场景带上):把这条展示收益归属到对应比价/领券,"
"供领券数据/比价记录看板聚合本场广告收益;激励视频/福利为空",
)
app_env: str | None = Field(
None, max_length=16, description="我们的穿山甲应用环境:prod(傻瓜比价正式) / test(测试应用)"
)
+53
View File
@@ -0,0 +1,53 @@
"""LLM 调用成本计算(纯逻辑,无 DB):按 model 分桶累加 token × 单价,返回总成本(元)+ 价格快照。
用量取自 comparison_record.llm_calls[].usage(pricebot 已归一为 prompt/completion_tokens);
error / usage 的调用跳过price_cfg = {per_model:{model:{input_per_1m,output_per_1m}}, default:{...}}
成本单位单次亚分级, float(不用 *_cents);snapshot 只含本次用到的模型的价(审计用,
不存整张价表)用到但没配价(既无 per_model 又无 default)的模型 快照标 unpriced,成本按 0
"""
from __future__ import annotations
_PRICE_KEY = "llm_token_price"
def get_llm_prices(db) -> dict:
"""读 LLM 单价配置(app_config;表内无则回退 CONFIG_DEFS 默认)。返回 compute_llm_cost 的 price_cfg。"""
from app.repositories import app_config # 延迟 import:compute_llm_cost 纯逻辑不牵连 DB 层
return app_config.get_value(db, _PRICE_KEY)
def compute_llm_cost(calls: list[dict], price_cfg: dict) -> tuple[float | None, dict | None]:
"""遍历 calls 按 model 分桶,cost = Σ(入/1e6*入价 + 出/1e6*出价);无有效调用 → (None, None)。"""
if not calls:
return None, None
per_model = price_cfg.get("per_model") or {}
default = price_cfg.get("default")
buckets: dict[str, list[int]] = {} # model -> [Σprompt_tokens, Σcompletion_tokens]
for c in calls:
if c.get("error"):
continue
usage = c.get("usage") or {}
model = c.get("model") or "unknown"
b = buckets.setdefault(model, [0, 0])
b[0] += usage.get("prompt_tokens") or 0
b[1] += usage.get("completion_tokens") or 0
if not buckets: # 全是 error / 无 usage
return None, None
total = 0.0
prices: dict[str, dict] = {}
for model, (tin, tout) in buckets.items():
price = per_model.get(model, default)
in_p = price.get("input_per_1m") if isinstance(price, dict) else None
out_p = price.get("output_per_1m") if isinstance(price, dict) else None
# 没配价 / 无 default / 单价残缺或非法(配置页手改 JSON 可能存出脏数据)→ 标记待补价、
# 不计入成本;绝不抛异常,以免连累同一回填里的 token/llm_calls 落库。
if not isinstance(in_p, (int, float)) or not isinstance(out_p, (int, float)):
prices[model] = {"input_per_1m": in_p, "output_per_1m": out_p, "unpriced": True}
continue
total += tin / 1e6 * in_p + tout / 1e6 * out_p
prices[model] = {
"input_per_1m": in_p,
"output_per_1m": out_p,
"_source": "per_model" if model in per_model else "default",
}
return round(total, 6), {"mode": "per_model", "prices": prices}
+2
View File
@@ -40,6 +40,8 @@
| `raw_payload` | JSON(PG: JSONB) | nullable | 客户端原始上报全量(calibration + done.params),取数兜底 |
| `input_tokens` | Integer | nullable | 本次 LLM 累计输入 token = Σ `llm_calls[].usage.prompt_tokens`(server 收上报后从 `llm_calls` 累加;旧记录/未采集为 null) |
| `output_tokens` | Integer | nullable | 本次 LLM 累计输出 token = Σ `llm_calls[].usage.completion_tokens`(同上) |
| `llm_cost_yuan` | Float | nullable | 本次比价 LLM 总成本(元),回填时按「当时价」逐模型算好冻结(见 `services/llm_cost.py`);旧记录/未回填为 null → 前端回退「估算成本」 |
| `llm_price_snapshot` | JSON(PG: JSONB) | nullable | 算成本所用单价快照 `{mode, prices:{model:{input_per_1m,output_per_1m,_source}}}`;`app_config` 只存当前价、不留历史,故冻结当时价供审计/复算 |
| `created_at` | DateTime(tz) | server_default now(), index | 时间 |
> `ordered`(已下单)是**瞬态字段**,不在表里:`list_records` 读取时按 `store_name ∈ 该用户 source='compare' 的 savings_record.shop_name 集合` 现挂到实例上供出参用。
@@ -0,0 +1,830 @@
# 逐次比价/领券广告收益 Implementation Plan
> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
**Goal:** 让 admin「领券数据」和「比价记录」两个看板的 table 每一行显示这一次领券/比价产生的广告收益(预估元)。
**Architecture:** 客户端在信息流(Draw)展示上报 eCPM 时带上本场 `trace_id`,后端落到 `ad_ecpm_record.trace_id`(新列 + 索引)。两个看板在分页后,对**当前页**的 trace_id 批量聚合一次 `ad_ecpm_record` 的展示收益(单条收益 = min(eCPM元,¥500)/1000,与广告收益报表同口径),挂到每行。查询是按 trace_id 索引的单条聚合,与已上线的 `_ad_coins_by_trace`(比价记录页「比价赚N金币」)同一性能剖面。
**Tech Stack:** 后端 FastAPI + SQLAlchemy 2.0 + Alembic;客户端 Android(Kotlin/OkHttp);admin 前端 Next.js + React + Ant Design(shaguabijia-admin-web)。
---
## 背景与约束(执行前必读)
- **收益 ≠ 金币**。本功能查的是「我们赚的广告收益」(数据源 `ad_ecpm_record`,客户端自报 eCPM 折算的预估),不是发给用户的金币(那是 `ad_feed_reward_record`,已有 `trace_id`)。
- **只能到「场景 + 单次」粒度**`trace_id` 由客户端在比价(comparisonTraceId)/领券(sessionTraceId)全流程保持不变。激励视频、福利页、旧客户端不带 trace_id → 该列为 NULL,历史数据无法回填,只对升级后新数据生效。
- **收益口径**(与 `app/admin/repositories/ad_revenue.py:165-167` 完全一致):
单条展示收益(元) = `min(parse_ecpm_yuan(ecpm_raw), AD_ECPM_MAX_FEN/100) / 1000`
其中 `AD_ECPM_MAX_FEN = 50000`(分)= ¥500 CPM 封顶,`parse_ecpm_yuan(x) = parse_ecpm_fen(x)/100`
- **性能前提**:`ad_ecpm_record` 是全库写入量最大的表。必须有 `ix_ad_ecpm_record_trace_id` 索引(本计划 Task A2 建),且只对**当前页**的 trace_id 聚合,绝不对整个日期区间聚合。
- **跨仓库**:本计划涉及三个仓库,路径前缀:
- 后端 `e:\project\shaguabijia-app-server`(相对路径即以此为根)
- 客户端 `E:\project\shaguabijia-app-android`
- admin 前端 `e:\project\shaguabijia-admin-web`
---
## File Structure
### 后端(shaguabijia-app-server)
- Modify `app/models/ad_ecpm.py``AdEcpmRecord``trace_id` 列(索引)
- Create `alembic/versions/ad_ecpm_trace_id.py` — 加列 + 索引,并收敛当前双 head
- Modify `app/schemas/ad.py``EcpmReportIn``trace_id` 字段
- Modify `app/repositories/ad_ecpm.py``create_ecpm_record` 持久化 `trace_id`;新增 `revenue_yuan_by_trace` 聚合器
- Modify `app/api/v1/ad.py``ecpm_report` 透传 `trace_id`
- Modify `app/admin/schemas/coupon_data.py``CouponDataRow``ad_revenue_yuan`
- Modify `app/admin/repositories/coupon_data.py` — 逐页补 `ad_revenue_yuan`
- Modify `app/admin/schemas/comparison.py``AdminComparisonListItem``ad_revenue_yuan`
- Modify `app/admin/repositories/queries.py``list_comparison_records` 逐页补 `ad_revenue_yuan`
- Create `tests/test_ad_ecpm_trace_revenue.py` — 聚合器 + 落库单测
- Create `tests/test_board_ad_revenue.py` — 两个看板收益列单测
> `app/models/__init__.py` **不需改**:`AdEcpmRecord` 已注册,只是加列。
### 客户端(shaguabijia-app-android)
- Modify `app/src/main/java/com/jishisongfu/shaguabijia/agent/network/ApiClient.kt``reportAdImpression``traceId` 参数
- Modify `app/src/main/java/com/jishisongfu/shaguabijia/agent/service/ad/CompareAdController.kt` — 比价展示上报带 `traceId`
- Modify `app/src/main/java/com/jishisongfu/shaguabijia/service/CouponForegroundService.kt` — 领券展示上报带 `traceId`
### admin 前端(shaguabijia-admin-web)
- Modify `src/lib/types.ts``ComparisonRecordListItem``ad_revenue_yuan`
- Modify `src/app/(main)/comparison-records/page.tsx` — 加「广告收益」列
- Modify `src/app/(main)/coupon-data/page.tsx``CouponDataRow` 加字段 + 加「广告收益」列
---
## Phase A — 后端数据打通(落 trace_id + 收益聚合器)
### Task A1: `AdEcpmRecord``trace_id`
**Files:**
- Modify: `app/models/ad_ecpm.py`
- [ ] **Step 1: 加列**
`app/models/ad_ecpm.py` 中,找到 `feed_scene` 这一行:
```python
feed_scene: Mapped[str | None] = mapped_column(String(16), nullable=True)
```
在其**下方**插入:
```python
# 本次比价/领券 trace_id(信息流场景客户端带上):把这条展示收益归属到对应比价/领券记录。
# 领券数据 / 比价记录看板按 trace_id 聚合"本次广告收益"。激励视频/福利/旧客户端 = NULL。
trace_id: Mapped[str | None] = mapped_column(String(64), index=True, nullable=True)
```
- [ ] **Step 2: 提交**
```bash
git add app/models/ad_ecpm.py
git commit -m "feat(ad-ecpm): add trace_id column to AdEcpmRecord model"
```
---
### Task A2: 迁移 — 加列 + 索引,并收敛双 head
**Files:**
- Create: `alembic/versions/ad_ecpm_trace_id.py`
> ⚠️ 当前 `alembic heads` 有**两个 head**:`11c44afbea58`(selfstat 表)与 `merge_pages_override_coupon_slot`(#126+领券合并)。本迁移用元组 `down_revision` 把二者收敛成单 head,同时加列,让 `alembic upgrade head`(单数,run.sh 用)恢复正常。
- [ ] **Step 1: 确认当前 heads 未漂移**
Run: `alembic heads`
Expected: 恰好两行 —
```
11c44afbea58 (head)
merge_pages_override_coupon_slot (head)
```
若不同(他人已合并/新增),把下面 `down_revision` 改成此刻实际的 head 列表。
- [ ] **Step 2: 建迁移文件**
Create `alembic/versions/ad_ecpm_trace_id.py`:
```python
"""ad_ecpm_record.trace_id(展示收益归属到比价/领券 trace)+ 收敛双 head
信息流(Draw)展示 eCPM 上报时带上本场比价/领券 trace_id,落此列;领券数据 / 比价记录看板
按 trace_id 聚合"本次广告收益"。激励视频/福利/旧客户端为 NULL。
顺带把当前两个 head(11c44afbea58 selfstat 表 + merge_pages_override_coupon_slot)收敛成
单 head,让 `alembic upgrade head`(单数,部署/run.sh 用)恢复正常。
Revision ID: ad_ecpm_trace_id
Revises: 11c44afbea58, merge_pages_override_coupon_slot
Create Date: 2026-07-10
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
revision: str = "ad_ecpm_trace_id"
down_revision: Union[str, Sequence[str], None] = (
"11c44afbea58",
"merge_pages_override_coupon_slot",
)
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
# SQLite 下 ADD COLUMN(可空)与 CREATE INDEX 均原生支持,无需 batch_alter_table
# (同 ad_feed_reward_trace_id 迁移)。
op.add_column(
"ad_ecpm_record",
sa.Column("trace_id", sa.String(length=64), nullable=True),
)
op.create_index(
op.f("ix_ad_ecpm_record_trace_id"),
"ad_ecpm_record",
["trace_id"],
unique=False,
)
def downgrade() -> None:
op.drop_index(op.f("ix_ad_ecpm_record_trace_id"), table_name="ad_ecpm_record")
op.drop_column("ad_ecpm_record", "trace_id")
```
- [ ] **Step 3: 应用迁移**
Run: `alembic upgrade head`
Expected: 无报错(不再报 "multiple heads")。
- [ ] **Step 4: 验证单 head + 列存在**
Run: `alembic heads`
Expected: 只有一行 `ad_ecpm_trace_id (head)`
Run: `python -c "from sqlalchemy import inspect; from app.db.session import engine; print([c['name'] for c in inspect(engine).get_columns('ad_ecpm_record')])"`
Expected: 输出的列名列表包含 `trace_id`
- [ ] **Step 5: 提交**
```bash
git add alembic/versions/ad_ecpm_trace_id.py
git commit -m "feat(migration): add ad_ecpm_record.trace_id + index, converge heads"
```
---
### Task A3: `EcpmReportIn``trace_id` 字段
**Files:**
- Modify: `app/schemas/ad.py`
- [ ] **Step 1: 加字段**
`app/schemas/ad.py``EcpmReportIn` 里,找到 `feed_scene` 字段定义(以 `feed_scene: str | None = Field(` 开头的那段)。在该字段**之后**插入:
```python
trace_id: str | None = Field(
None,
max_length=64,
description="本次比价/领券 trace_id(信息流场景带上):把这条展示收益归属到对应比价/领券,"
"供领券数据/比价记录看板聚合本场广告收益;激励视频/福利为空",
)
```
- [ ] **Step 2: 提交**
```bash
git add app/schemas/ad.py
git commit -m "feat(ad-schema): EcpmReportIn accepts trace_id"
```
---
### Task A4: `create_ecpm_record` 持久化 trace_id + 新增 `revenue_yuan_by_trace`
**Files:**
- Modify: `app/repositories/ad_ecpm.py`
- Test: `tests/test_ad_ecpm_trace_revenue.py`
- [ ] **Step 1: 写失败测试**
Create `tests/test_ad_ecpm_trace_revenue.py`:
```python
"""ad_ecpm_record.trace_id 落库 + 按 trace 聚合广告收益(元)。"""
from __future__ import annotations
from datetime import UTC, datetime
from sqlalchemy import delete
from app.db.session import SessionLocal
from app.models.ad_ecpm import AdEcpmRecord
from app.repositories import ad_ecpm as crud_ecpm
def _ecpm(trace_id: str, ecpm_raw: str, session_id: str) -> AdEcpmRecord:
"""构造一条 Draw 展示 eCPM(不 commit;ad_session_id 全局唯一,须各不相同)。"""
return AdEcpmRecord(
user_id=1,
ad_type="draw",
feed_scene="comparison",
ad_session_id=session_id,
ecpm_raw=ecpm_raw,
trace_id=trace_id,
report_date="2020-01-02",
created_at=datetime(2020, 1, 2, tzinfo=UTC),
)
def test_revenue_yuan_by_trace_sums_and_clamps() -> None:
"""同一 trace 多条展示求和;收益=min(eCPM元,¥500)/1000;无展示的 trace 不出现。
ecpm 200 分→2.0 元/千次→0.002 元/次;300 分→0.003;合计 0.005。
"""
db = SessionLocal()
try:
db.add_all([
_ecpm("t1", "200", "sess-t1-a"),
_ecpm("t1", "300", "sess-t1-b"),
_ecpm("t2", "0", "sess-t2-a"),
])
db.flush()
rev = crud_ecpm.revenue_yuan_by_trace(db, ["t1", "t2", "t3"])
assert rev["t1"] == 0.005
assert rev.get("t2", 0.0) == 0.0
assert "t3" not in rev # 无展示的 trace 不出现在结果里
finally:
db.rollback()
db.close()
def test_revenue_yuan_by_trace_empty() -> None:
"""空 trace 列表直接返回 {}(避免 IN () 非法)。"""
db = SessionLocal()
try:
assert crud_ecpm.revenue_yuan_by_trace(db, []) == {}
finally:
db.close()
def test_create_ecpm_record_persists_trace_id() -> None:
"""create_ecpm_record 落 trace_id。"""
db = SessionLocal()
try:
rec = crud_ecpm.create_ecpm_record(
db, 1, ad_type="draw", ecpm_raw="150",
ad_session_id="sess-trace-persist", feed_scene="coupon",
trace_id="trace-xyz",
)
assert rec.trace_id == "trace-xyz"
finally:
db.execute(delete(AdEcpmRecord).where(AdEcpmRecord.ad_session_id == "sess-trace-persist"))
db.commit()
db.close()
```
- [ ] **Step 2: 跑测试确认失败**
Run: `pytest tests/test_ad_ecpm_trace_revenue.py -q`
Expected: FAIL — `create_ecpm_record``trace_id` 参数(TypeError)/ `revenue_yuan_by_trace` 不存在(AttributeError)。
- [ ] **Step 3: 实现**
`app/repositories/ad_ecpm.py`:
(a) 顶部 import 区加(与现有 `from app.core.rewards import cn_today` 并列):
```python
from app.core import rewards
```
(b) `create_ecpm_record` 的签名里,在 `feed_scene: str | None = None,` 之后加一行参数:
```python
trace_id: str | None = None,
```
(c) 同函数体内构造 `AdEcpmRecord(...)` 处,在 `feed_scene=feed_scene,` 之后加一行:
```python
trace_id=trace_id,
```
(d) 文件末尾新增聚合器:
```python
def revenue_yuan_by_trace(db: Session, trace_ids: list[str]) -> dict[str, float]:
"""各 trace_id 的广告预估收益(元):按 trace_id 聚合 ad_ecpm_record 的展示收益。
单条展示收益 = min(eCPM元, AD_ECPM_MAX_FEN/100) / 1000(与 admin 广告收益报表同口径)。
ecpm_raw 是字符串且需逐条钳顶,故取回后 Python 求和(行数=本页各 trace 的展示条数,很小)。
trace_id 仅信息流(比价/领券)场景客户端带,激励视频/旧数据为 NULL,按 trace_id 过滤天然只算对应场景。
只喂**当前页**的 trace_id(≤ 一页条数);空集合直接返回(避免 IN () 非法)。
"""
if not trace_ids:
return {}
rows = db.execute(
select(AdEcpmRecord.trace_id, AdEcpmRecord.ecpm_raw).where(
AdEcpmRecord.trace_id.in_(trace_ids),
)
).all()
cap_yuan = rewards.AD_ECPM_MAX_FEN / 100.0
out: dict[str, float] = {}
for tid, ecpm_raw in rows:
if not tid:
continue
out[tid] = out.get(tid, 0.0) + min(rewards.parse_ecpm_yuan(ecpm_raw), cap_yuan) / 1000.0
return {tid: round(v, 6) for tid, v in out.items()}
```
- [ ] **Step 4: 跑测试确认通过**
Run: `pytest tests/test_ad_ecpm_trace_revenue.py -q`
Expected: PASS(3 passed)。
- [ ] **Step 5: 提交**
```bash
git add app/repositories/ad_ecpm.py tests/test_ad_ecpm_trace_revenue.py
git commit -m "feat(ad-ecpm): persist trace_id + revenue_yuan_by_trace aggregator"
```
---
### Task A5: `ecpm_report` 端点透传 trace_id
**Files:**
- Modify: `app/api/v1/ad.py`
- [ ] **Step 1: 透传字段**
`app/api/v1/ad.py``ecpm_report` 函数里,找到 `crud_ecpm.create_ecpm_record(` 调用,在 `feed_scene=payload.feed_scene,` 之后加一行:
```python
trace_id=payload.trace_id,
```
- [ ] **Step 2: 冒烟验证(手动,可选)**
启动后端(`./run.sh`),用一个有效用户 JWT 调:
Run:
```bash
curl -s -X POST http://127.0.0.1:8770/api/v1/ad/ecpm-report \
-H "Authorization: Bearer <USER_JWT>" -H "Content-Type: application/json" \
-d '{"ad_type":"draw","ecpm":"200","ad_session_id":"smoke-sess-1","feed_scene":"comparison","trace_id":"smoke-trace-1"}'
```
Expected: `{"ok":true}`;库里 `ad_ecpm_record` 出现一条 `trace_id='smoke-trace-1'` 的记录。
> 端点逻辑是纯透传,已由 A4 的 schema/repo 测试覆盖;此步仅人工确认接线。
- [ ] **Step 3: 提交**
```bash
git add app/api/v1/ad.py
git commit -m "feat(ad-api): ecpm-report forwards trace_id to record"
```
---
## Phase B — 后端两个看板补收益列
### Task B1: 领券数据看板逐行补 `ad_revenue_yuan`
**Files:**
- Modify: `app/admin/schemas/coupon_data.py`
- Modify: `app/admin/repositories/coupon_data.py`
- Test: `tests/test_board_ad_revenue.py`
- [ ] **Step 1: 写失败测试**
Create `tests/test_board_ad_revenue.py`:
```python
"""两个看板逐行「本次广告收益」(元):按 trace_id 聚合 ad_ecpm_record。"""
from __future__ import annotations
from datetime import UTC, date, datetime
from sqlalchemy import delete
from app.admin.repositories import queries
from app.admin.repositories.coupon_data import coupon_data_report
from app.db.session import SessionLocal
from app.models.ad_ecpm import AdEcpmRecord
from app.models.comparison import ComparisonRecord
from app.models.coupon_state import CouponSession
def _ecpm(trace_id: str, ecpm_raw: str, session_id: str, scene: str) -> AdEcpmRecord:
return AdEcpmRecord(
user_id=1, ad_type="draw", feed_scene=scene, ad_session_id=session_id,
ecpm_raw=ecpm_raw, trace_id=trace_id, report_date="2020-01-02",
created_at=datetime(2020, 1, 2, tzinfo=UTC),
)
def test_coupon_data_report_includes_ad_revenue() -> None:
"""领券看板明细行带本次广告收益;200+300 分 → 0.005 元。"""
db = SessionLocal()
try:
db.add(CouponSession(
trace_id="rev-cp-1", device_id="d1", status="completed", app_env="prod",
platforms=["meituan-waimai"], platform_success=["meituan-waimai"],
started_at=datetime(2020, 1, 2, tzinfo=UTC), started_date=date(2020, 1, 2),
))
db.add_all([
_ecpm("rev-cp-1", "200", "cp-sess-a", "coupon"),
_ecpm("rev-cp-1", "300", "cp-sess-b", "coupon"),
])
db.flush()
res = coupon_data_report(db, date_from="2020-01-02", date_to="2020-01-02", app_env="prod")
row = next(r for r in res["items"] if r["trace_id"] == "rev-cp-1")
assert row["ad_revenue_yuan"] == 0.005
finally:
db.rollback()
db.close()
def test_comparison_list_includes_ad_revenue() -> None:
"""比价记录列表项带本次广告收益;200 分 → 0.002 元。
用独有 user_id 过滤,确保本行必落在第一页(避免共享测试库里同 user 记录多、分页把它挤掉)。
"""
db = SessionLocal()
try:
db.add(ComparisonRecord(
trace_id="rev-cmp-1", user_id=987654, status="success", business_type="food",
created_at=datetime(2020, 1, 2, tzinfo=UTC),
))
db.add(_ecpm("rev-cmp-1", "200", "cmp-sess-a", "comparison"))
db.commit()
items, _next, _total = queries.list_comparison_records(db, user_id=987654)
row = next(it for it in items if it.trace_id == "rev-cmp-1")
assert row.ad_revenue_yuan == 0.002
finally:
db.execute(delete(AdEcpmRecord).where(AdEcpmRecord.trace_id.in_(["rev-cmp-1"])))
db.execute(delete(ComparisonRecord).where(ComparisonRecord.trace_id == "rev-cmp-1"))
db.commit()
db.close()
```
- [ ] **Step 2: 跑测试确认失败**
Run: `pytest tests/test_board_ad_revenue.py -q`
Expected: FAIL — `coupon_data_report` 明细行无 `ad_revenue_yuan` 键(KeyError);`ComparisonRecord``ad_revenue_yuan`(AttributeError)。
- [ ] **Step 3: 领券 schema 加字段**
`app/admin/schemas/coupon_data.py``CouponDataRow` 里,`trace_url` 字段**之后**加:
```python
ad_revenue_yuan: float = Field(
0.0, description="本次领券看的信息流广告预估收益(元);按 trace_id 聚合 ad_ecpm_record"
)
```
- [ ] **Step 4: 领券 repo 逐页补收益**
`app/admin/repositories/coupon_data.py`:
(a) import 区(现有 `from app.models.user import User` 附近)加:
```python
from app.repositories import ad_ecpm as crud_ecpm
```
(b) `_session_to_row` 签名改为(加末位参数):
```python
def _session_to_row(r, phone: str | None = None, nickname: str | None = None, ad_revenue_yuan: float = 0.0) -> dict:
```
并在其返回的 dict 里,`"trace_url": r.trace_url,` 之后加一行:
```python
"ad_revenue_yuan": ad_revenue_yuan,
```
(c) 在 `coupon_data_report` 里,找到构造明细的这段:
```python
items = []
for r in page:
phone, nickname = user_map.get(r.user_id, (None, None)) if r.user_id is not None else (None, None)
items.append(_session_to_row(r, phone, nickname))
```
替换为(新增 `rev_map` + 传入):
```python
rev_map = crud_ecpm.revenue_yuan_by_trace(db, [r.trace_id for r in page])
items = []
for r in page:
phone, nickname = user_map.get(r.user_id, (None, None)) if r.user_id is not None else (None, None)
items.append(_session_to_row(r, phone, nickname, ad_revenue_yuan=rev_map.get(r.trace_id, 0.0)))
```
> `coupon_user_records`(手机号抽屉)仍走 `_session_to_row(r)`,`ad_revenue_yuan` 取默认 0.0——抽屉不展示收益列,无需补;字段有默认值故 schema 校验不受影响。
- [ ] **Step 5: 跑领券用例确认通过**
Run: `pytest tests/test_board_ad_revenue.py::test_coupon_data_report_includes_ad_revenue -q`
Expected: PASS。
- [ ] **Step 6: 提交**
```bash
git add app/admin/schemas/coupon_data.py app/admin/repositories/coupon_data.py tests/test_board_ad_revenue.py
git commit -m "feat(admin-coupon-data): per-session ad revenue column"
```
---
### Task B2: 比价记录看板逐行补 `ad_revenue_yuan`
**Files:**
- Modify: `app/admin/schemas/comparison.py`
- Modify: `app/admin/repositories/queries.py`
- [ ] **Step 1: 比价 schema 加字段**
`app/admin/schemas/comparison.py``AdminComparisonListItem` 里,`created_at: datetime` **之前**加:
```python
ad_revenue_yuan: float = 0.0 # 本次比价看的信息流广告预估收益(元),queries 瞬态挂 ORM 实例上
```
- [ ] **Step 2: 比价 repo 逐页补收益**
`app/admin/repositories/queries.py`:
(a) import 区加:
```python
from app.repositories import ad_ecpm
```
(b) 在 `list_comparison_records` 里,找到:
```python
_attach_user_info(db, items)
return items, next_cursor, total
```
替换为:
```python
_attach_user_info(db, 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])
for it in items:
it.ad_revenue_yuan = rev.get(it.trace_id, 0.0)
return items, next_cursor, total
```
- [ ] **Step 3: 跑比价用例确认通过**
Run: `pytest tests/test_board_ad_revenue.py::test_comparison_list_includes_ad_revenue -q`
Expected: PASS。
- [ ] **Step 4: 跑全量后端测试(确认无回归)**
Run: `pytest -q`
Expected: 全绿(新增用例通过,原有用例不受影响)。
- [ ] **Step 5: Lint**
Run: `ruff check app/ tests/`
Expected: 无新增告警。
- [ ] **Step 6: 提交**
```bash
git add app/admin/schemas/comparison.py app/admin/repositories/queries.py
git commit -m "feat(admin-comparison): per-comparison ad revenue column"
```
---
## Phase C — 客户端上报带 trace_id(Android)
> 三处都在 `E:\project\shaguabijia-app-android`。改动极小:eCPM 上报点的 trace_id 已在作用域内(比价是 `showAd(traceId)` 参数,领券是 `sessionTraceId` 类字段),只是当前没往上带。发奖(`reportFeedReward`)已在带 traceId,可作参照。
### Task C1: `ApiClient.reportAdImpression``traceId` 参数
**Files:**
- Modify: `app/src/main/java/com/jishisongfu/shaguabijia/agent/network/ApiClient.kt`
- [ ] **Step 1: 加参数**
找到 `reportAdImpression` 的参数列表,末尾 `feedScene: String? = null,` 之后加一行:
```kotlin
traceId: String? = null,
```
- [ ] **Step 2: 写进 payload**
同函数内,找到:
```kotlin
if (!feedScene.isNullOrBlank()) payload.put("feed_scene", feedScene)
```
在其**下方**加一行:
```kotlin
if (!traceId.isNullOrBlank()) payload.put("trace_id", traceId)
```
- [ ] **Step 3: 提交**
```bash
git add app/src/main/java/com/jishisongfu/shaguabijia/agent/network/ApiClient.kt
git commit -m "feat(ad-report): reportAdImpression carries trace_id"
```
---
### Task C2: 比价展示上报带 traceId
**Files:**
- Modify: `app/src/main/java/com/jishisongfu/shaguabijia/agent/service/ad/CompareAdController.kt`
- [ ] **Step 1: 传 traceId**
`showAd(traceId: String)` 内的 `onAdImpression` 回调里,找到 `apiClient.reportAdImpression(` 调用,其中 `feedScene = "comparison",` 之后加一行(`traceId``showAd` 的入参):
```kotlin
traceId = traceId, // 本场比价 trace → 展示收益归属到该次比价
```
- [ ] **Step 2: 提交**
```bash
git add app/src/main/java/com/jishisongfu/shaguabijia/agent/service/ad/CompareAdController.kt
git commit -m "feat(compare-ad): report comparison impression with trace_id"
```
---
### Task C3: 领券展示上报带 traceId
**Files:**
- Modify: `app/src/main/java/com/jishisongfu/shaguabijia/service/CouponForegroundService.kt`
- [ ] **Step 1: 传 traceId**
`onAdImpression` 回调里,找到 `apiClient.reportAdImpression(` 调用,其中 `feedScene = "coupon",` 之后加一行(`sessionTraceId` 为本类字段,整场领券不变):
```kotlin
traceId = sessionTraceId, // 本场领券 trace → 展示收益归属到该次领券
```
- [ ] **Step 2: 编译验证**
Run(在 `E:\project\shaguabijia-app-android`): `./gradlew :app:compileDebugKotlin`
Expected: BUILD SUCCESSFUL。
- [ ] **Step 3: 提交**
```bash
git add app/src/main/java/com/jishisongfu/shaguabijia/service/CouponForegroundService.kt
git commit -m "feat(coupon-ad): report coupon impression with trace_id"
```
---
## Phase D — admin 前端加「广告收益」列
> 两个页面都在 `e:\project\shaguabijia-admin-web`。收益值单位是**元**(小数,单次很小如 ¥0.0050),用 `.toFixed(4)` 展示。
### Task D1: 比价记录页加列
**Files:**
- Modify: `src/lib/types.ts`
- Modify: `src/app/(main)/comparison-records/page.tsx`
- [ ] **Step 1: 类型加字段**
`src/lib/types.ts``ComparisonRecordListItem` 接口里,`created_at: string;` **之前**加:
```typescript
ad_revenue_yuan: number; // 本次比价看的信息流广告预估收益(元)
```
- [ ] **Step 2: 表格加列**
`src/app/(main)/comparison-records/page.tsx``columns` 数组里,找到「省」这一列(以 `title: '省',` 开头的对象),在其**之后**插入一列:
```tsx
{
title: '广告收益',
key: 'ad_revenue',
width: 96,
align: 'right',
render: (_, r) =>
r.ad_revenue_yuan > 0 ? (
<span style={{ color: '#3f8600' }}>¥{r.ad_revenue_yuan.toFixed(4)}</span>
) : (
<span style={{ color: '#ccc' }}>-</span>
),
},
```
- [ ] **Step 3: 加宽横向滚动**
同文件找到比价记录主表的 `scroll={{ x: 1820 }}`,改为:
```tsx
scroll={{ x: 1920 }}
```
- [ ] **Step 4: 提交**
```bash
git add src/lib/types.ts "src/app/(main)/comparison-records/page.tsx"
git commit -m "feat(admin-web): ad revenue column in comparison records table"
```
---
### Task D2: 领券数据页加列
**Files:**
- Modify: `src/app/(main)/coupon-data/page.tsx`
- [ ] **Step 1: 接口加字段**
`src/app/(main)/coupon-data/page.tsx``interface CouponDataRow` 里,`trace_url: string | null;` **之后**加:
```typescript
ad_revenue_yuan: number; // 本次领券看的信息流广告预估收益(元)
```
- [ ] **Step 2: 加金额格式化辅助**
`fmtPct` 定义(以 `const fmtPct =` 开头)之后加:
```typescript
// 元(小数)→ "¥0.0050"(空/≤0 显示 -)。单次广告收益很小,保留 4 位。
const fmtYuan = (v: number | null | undefined): string =>
v == null || v <= 0 ? '-' : `¥${v.toFixed(4)}`;
```
- [ ] **Step 3: 主表加列**
找到主表 `columns`(`CouponDataPage` 组件内的 `const columns: ColumnsType<CouponDataRow> = [`)。在「耗时」列(`dataIndex: 'elapsed_ms'` 的对象)**之后**插入:
```tsx
{
title: '广告收益',
dataIndex: 'ad_revenue_yuan',
width: 100,
align: 'right',
render: (v: number) => fmtYuan(v),
},
```
- [ ] **Step 4: 加宽横向滚动**
找到主表 `scroll={{ x: 1450 }}`,改为:
```tsx
scroll={{ x: 1560 }}
```
- [ ] **Step 5: 前端类型检查 / 构建**
Run(在 `e:\project\shaguabijia-admin-web`): `npm run build`
Expected: 构建成功、无 TS 类型错误。
- [ ] **Step 6: 提交**
```bash
git add "src/app/(main)/coupon-data/page.tsx"
git commit -m "feat(admin-web): ad revenue column in coupon data table"
```
---
## 端到端验证(全部任务完成后)
- [ ] **后端**:`pytest -q` 全绿;`alembic heads` 只有 `ad_ecpm_trace_id` 单 head。
- [ ] **客户端**:装 debug 包,跑一次比价 + 一次领券(要有信息流广告展示);后端库 `ad_ecpm_record` 出现带 `trace_id``feed_scene in (comparison, coupon)` 的记录。
- [ ] **看板**:admin 打开「比价记录」「领券数据」两页,新「广告收益」列对刚才那两次显示 > ¥0 的金额;旧数据(无 trace_id 上报)显示 `-`
- [ ] **口径核对**:任取一行,手动核对 `ad_ecpm_record` 中该 `trace_id` 的各条 `ecpm_raw`,按 `Σ min(ecpm/100, 500)/1000` 算出的值与页面一致。
---
## 回滚
- 前端/客户端:回退对应 commit 即可(纯展示/上报,无副作用)。
- 后端:`alembic downgrade -1``ad_ecpm_trace_id`(会拆回两个 head——与本计划实施前状态一致);看板收益列在无 trace_id 列时会因查询报错,故 downgrade 迁移前需先回退 Phase A/B 的 commit。正常情况不需回滚。
+248
View File
@@ -0,0 +1,248 @@
"""一次性 mock:造带 LLM token 成本的比价记录 + 配好 app_config 模型单价,用于测「管理后端」LLM 成本展示。
覆盖 admin比价记录详情抽屉的LLM 成本展示分支:
app_config.llm_token_price 写一条多模型单价(= 配置页LLM 成本卡片已改,get_llm_prices 读它)
comparison_record 5 ,逐条**复用生产的 compute_llm_cost + _backfill_llm_calls 同款派生**
(llm_call_count/retry_count/input_tokens/output_tokens/llm_cost_yuan/llm_price_snapshot),
确保 mock = 真实回填产出5 条刻意覆盖:
单模型真实样本(qwen3.5-flash ×4) ¥0.006184(核对精确值)
多模型(flash + plus) 快照含两个模型各自 _source=per_model
未登记模型(deepseek-v3) default,快照 _source=default
旧记录( token cost) llm_cost_yuan=NULL 前端回退估算成本
error 调用 error 那次跳过计费retry_count+1
记录挂到库里第一个真实用户(admin 列表能显示手机号);无用户则 user_id=NULL(孤儿行,admin 照样全看)
created_at 用北京 naive最近几分钟内错开,详情列表倒序即 置顶
幂等:重跑先按 trace_id 前缀MOCKLLM-清旧再建app_config 单价是 upsert(不随 --clean-only ,
因该 key 本就是本需求新增无历史真实值;要改价直接去配置页或重跑本脚本)
python -m scripts.seed_mock_llm_cost # 造价格 + 5 条记录
python -m scripts.seed_mock_llm_cost --clean-only # 只清 MOCKLLM- 记录(保留单价)
验收:admin比价记录 traceMOCKLLM- 5 点开详情看LLM 成本:
显示实际·当时价+ 价格快照; 显示估算
"""
from __future__ import annotations
import argparse
import sys
from datetime import datetime, timedelta, timezone
from sqlalchemy import delete, select
from app.db.session import SessionLocal
from app.models.comparison import ComparisonRecord
from app.models.user import User
from app.repositories import app_config
from app.services.llm_cost import compute_llm_cost
if hasattr(sys.stdout, "reconfigure"):
sys.stdout.reconfigure(encoding="utf-8") # Windows 控制台输出中文/¥
_BJ = timezone(timedelta(hours=8))
ID_PREFIX = "MOCKLLM-"
# ── 写进 app_config 的模型单价(get_llm_prices 读它;配置页「LLM 成本」卡片可再改)──
PRICE_CFG = {
"per_model": {
"qwen3.5-flash": {"input_per_1m": 0.8, "output_per_1m": 2.0},
"qwen3.5-plus": {"input_per_1m": 4.0, "output_per_1m": 12.0},
},
"default": {"input_per_1m": 3.0, "output_per_1m": 15.0},
"currency": "CNY",
"unit": "per_1m_tokens",
}
def _c(scene: str, model: str, pin: int, cout: int, error: str | None = None) -> dict:
"""一条 llm_calls 明细,结构对齐真实 pricebot 归一后契约:
{scene, model, input_messages:[{role,content}], output, usage:{prompt/completion/total_tokens},
latency_ms, error}(详情抽屉会遍历 input_messages,缺了会崩)error 的调用无 usage/output"""
return {
"scene": scene,
"model": model,
"error": error,
"input_messages": [
{"role": "system", "content": f"你是比价助手,负责 {scene} 环节。"},
{"role": "user", "content": f"[mock] 请处理本次比价的 {scene} 任务。"},
],
"output": None if error else f"[mock] {scene} 环节完成。",
"usage": None if error else {
"prompt_tokens": pin, "completion_tokens": cout, "total_tokens": pin + cout,
},
"latency_ms": 780,
}
# ── 5 条记录蓝本:calls 决定成本;freeze=False 模拟旧记录(有 token 无 cost)──
RECORDS = [
{
"label": "①单模型·真实样本",
"source": ("美团外卖", 4280), "best": ("京东秒送", 3680),
"store": "肯德基(建国路店)", "product": "疯狂星期四全家桶",
"info": "在京东秒送找到同款,到手价 ¥36.80,省 ¥6.00",
"freeze": True,
"calls": [
_c("store_match", "qwen3.5-flash", 1512, 22),
_c("dish_match", "qwen3.5-flash", 2111, 160),
_c("dish_match", "qwen3.5-flash", 1940, 142),
_c("summary", "qwen3.5-flash", 1325, 13),
],
},
{
"label": "②多模型·flash+plus",
"source": ("淘宝闪购", 5900), "best": ("美团外卖", 5200),
"store": "瑞幸咖啡(国贸店)", "product": "生椰拿铁×2、丝绒拿铁",
"info": "在美团外卖找到同款,到手价 ¥52.00,省 ¥7.00",
"freeze": True,
"calls": [
_c("store_match", "qwen3.5-flash", 2000, 50),
_c("dish_match", "qwen3.5-flash", 1800, 40),
_c("reasoning", "qwen3.5-plus", 3000, 500),
],
},
{
"label": "③未登记模型走 default",
"source": ("京东秒送", 3100), "best": ("美团外卖", 2650),
"store": "麦当劳(soho店)", "product": "麦辣鸡腿堡套餐",
"info": "在美团外卖找到同款,到手价 ¥26.50,省 ¥4.50",
"freeze": True,
"calls": [
_c("store_match", "deepseek-v3", 5000, 800),
],
},
{
"label": "④旧记录·有token无成本(回退估算)",
"source": ("美团外卖", 3600), "best": ("淘宝闪购", 3200),
"store": "华莱士(双井店)", "product": "全鸡汉堡套餐",
"info": "在淘宝闪购找到同款,到手价 ¥32.00,省 ¥4.00",
"freeze": False, # 模拟本需求上线前的老记录:llm_cost_yuan=NULL → 前端回退估算
"calls": [
_c("store_match", "qwen3.5-flash", 2000, 100),
],
},
{
"label": "⑤含 error 调用(跳过计费)",
"source": ("淘宝闪购", 4100), "best": ("京东秒送", 3750),
"store": "海底捞(合生汇店)", "product": "番茄锅底、肥牛卷",
"info": "在京东秒送找到同款,到手价 ¥37.50,省 ¥3.50",
"freeze": True,
"calls": [
_c("store_match", "qwen3.5-flash", 0, 0, error="timeout"),
_c("store_match", "qwen3.5-flash", 1500, 30),
],
},
]
_PLATFORM_ID = { # 展示名 → 平台代号(comparison_results / source/best 列用)
"美团外卖": "meituan", "京东秒送": "jd", "淘宝闪购": "taobao",
}
def _naive_bj_now() -> datetime:
return datetime.now(_BJ).replace(tzinfo=None)
def clean(db) -> int:
n = db.execute(
delete(ComparisonRecord).where(ComparisonRecord.trace_id.like(f"{ID_PREFIX}%"))
).rowcount or 0
db.commit()
return n
def _build_record(spec: dict, owner_id: int | None, created_at: datetime) -> tuple[ComparisonRecord, float | None]:
"""按蓝本造一条记录,LLM 派生完全对齐 _backfill_llm_calls;返回 (记录, 冻结成本或 None)。"""
calls = spec["calls"]
src_name, src_cents = spec["source"]
best_name, best_cents = spec["best"]
# —— 与 _backfill_llm_calls 同款派生 ——
llm_call_count = len(calls)
retry_count = sum(1 for c in calls if c.get("error"))
input_tokens = sum((c.get("usage") or {}).get("prompt_tokens") or 0 for c in calls)
output_tokens = sum((c.get("usage") or {}).get("completion_tokens") or 0 for c in calls)
if spec["freeze"]:
cost, snapshot = compute_llm_cost(calls, PRICE_CFG) # 复用生产纯函数
else:
cost, snapshot = None, None # 旧记录:回填这段代码上线前就有,只有 token 没成本
rec = ComparisonRecord(
user_id=owner_id,
device_id=f"{ID_PREFIX.lower()}dev",
business_type="food",
trace_id=f"{ID_PREFIX}{spec['label'][0]}", # ①..⑤ 各一,唯一
status="success",
source_platform_id=_PLATFORM_ID.get(src_name), source_platform_name=src_name,
source_price_cents=src_cents,
best_platform_id=_PLATFORM_ID.get(best_name), best_platform_name=best_name,
best_price_cents=best_cents,
saved_amount_cents=src_cents - best_cents,
is_source_best=False,
store_name=spec["store"],
product_names=spec["product"],
information=spec["info"],
items=[{"name": spec["product"], "qty": 1}],
comparison_results=[
{"platform_id": _PLATFORM_ID.get(src_name), "platform_name": src_name,
"price": src_cents / 100, "is_source": True, "rank": 2},
{"platform_id": _PLATFORM_ID.get(best_name), "platform_name": best_name,
"price": best_cents / 100, "is_source": False, "rank": 1},
],
total_ms=90_000 + llm_call_count * 1000,
step_count=llm_call_count * 3,
llm_call_count=llm_call_count,
retry_count=retry_count,
input_tokens=input_tokens,
output_tokens=output_tokens,
llm_calls=calls,
llm_cost_yuan=cost,
llm_price_snapshot=snapshot,
created_at=created_at,
)
return rec, cost
def seed(db) -> list[tuple[str, float | None]]:
app_config.set_value(db, "llm_token_price", PRICE_CFG, admin_id=None) # upsert 单价
owner_id = db.execute(select(User.id).order_by(User.id).limit(1)).scalar()
base = _naive_bj_now()
out: list[tuple[str, float | None]] = []
for i, spec in enumerate(RECORDS):
rec, cost = _build_record(spec, owner_id, base - timedelta(minutes=i * 3))
db.add(rec)
out.append((spec["label"], cost))
db.commit()
return out, owner_id
def main() -> None:
parser = argparse.ArgumentParser(description="造带 LLM 成本的比价记录 + app_config 模型单价(测管理后端)")
parser.add_argument("--clean-only", action="store_true", help="只清 MOCKLLM- 记录,不重建(保留单价)")
args = parser.parse_args()
db = SessionLocal()
try:
removed = clean(db)
if removed:
print(f"🧹 已清理旧 mock 记录 {removed}")
if args.clean_only:
print("✅ 仅清理,已完成(app_config 单价保留)。")
return
results, owner_id = seed(db)
print(f"\n✅ 已写入 app_config.llm_token_price(单价)+ {len(results)} 条比价记录"
f"(挂 user_id={owner_id or 'NULL(孤儿行)'})")
print("\n📋 每条冻结成本(admin 详情「LLM 成本」应显示):")
for label, cost in results:
shown = "NULL → 前端回退「估算」" if cost is None else f"¥{cost}"
print(f" {label:<20} {shown}")
print("\n👉 验收:admin「比价记录」→ trace 搜「MOCKLLM-」→ 点开详情核对 LLM 成本 + 价格快照。")
print(" 配置页「系统配置」→「福利页」Tab →「LLM 成本」卡片,单价应为「已改」态。")
finally:
db.close()
if __name__ == "__main__":
main()
+73
View File
@@ -0,0 +1,73 @@
"""ad_ecpm_record.trace_id 落库 + 按 trace 聚合广告收益(元)。"""
from __future__ import annotations
from datetime import UTC, datetime
from sqlalchemy import delete
from app.db.session import SessionLocal
from app.models.ad_ecpm import AdEcpmRecord
from app.repositories import ad_ecpm as crud_ecpm
def _ecpm(trace_id: str, ecpm_raw: str, session_id: str) -> AdEcpmRecord:
"""构造一条 Draw 展示 eCPM(不 commit;ad_session_id 全局唯一,须各不相同)。"""
return AdEcpmRecord(
user_id=1,
ad_type="draw",
feed_scene="comparison",
ad_session_id=session_id,
ecpm_raw=ecpm_raw,
trace_id=trace_id,
report_date="2020-01-02",
created_at=datetime(2020, 1, 2, tzinfo=UTC),
)
def test_revenue_yuan_by_trace_sums_and_clamps() -> None:
"""同一 trace 多条展示求和;收益=min(eCPM元,¥500)/1000;无展示的 trace 不出现。
ecpm 200 2.0 /千次0.002 /;300 0.003;合计 0.005
"""
db = SessionLocal()
try:
db.add_all([
_ecpm("t1", "200", "sess-t1-a"),
_ecpm("t1", "300", "sess-t1-b"),
_ecpm("t2", "0", "sess-t2-a"),
_ecpm("t_cap", "60000", "sess-cap-a"), # 600 元 CPM > ¥500 钳顶
])
db.flush()
rev = crud_ecpm.revenue_yuan_by_trace(db, ["t1", "t2", "t3", "t_cap"])
assert rev["t1"] == 0.005
assert rev["t2"] == 0.0 # 有展示但 eCPM=0 → 0 元(仍在结果里)
assert "t3" not in rev # 无展示的 trace 不出现在结果里
assert rev["t_cap"] == 0.5 # min(¥600, ¥500)/1000,证明钳顶生效(非 0.6)
finally:
db.rollback()
db.close()
def test_revenue_yuan_by_trace_empty() -> None:
"""空 trace 列表直接返回 {}(避免 IN () 非法)。"""
db = SessionLocal()
try:
assert crud_ecpm.revenue_yuan_by_trace(db, []) == {}
finally:
db.close()
def test_create_ecpm_record_persists_trace_id() -> None:
"""create_ecpm_record 落 trace_id。"""
db = SessionLocal()
try:
rec = crud_ecpm.create_ecpm_record(
db, 1, ad_type="draw", ecpm_raw="150",
ad_session_id="sess-trace-persist", feed_scene="coupon",
trace_id="trace-xyz",
)
assert rec.trace_id == "trace-xyz"
finally:
db.execute(delete(AdEcpmRecord).where(AdEcpmRecord.ad_session_id == "sess-trace-persist"))
db.commit()
db.close()
+61
View File
@@ -107,6 +107,67 @@ def test_sms_send_device_ip_rate_limit(client, monkeypatch) -> None:
assert r.status_code == 200, r.text
def test_sms_send_cooldown_reject_not_counted(client, monkeypatch) -> None:
"""发码额度只算「成功发码」:被单号 60s 冷却挡下的重发(429)不占设备额度。
做法:同号狂发只成功 1 其余被冷却挡下;把小时额度设 2,证明换号后仍能再成功发 1
若冷却重发也计数,额度早被那几次耗尽"""
from app.api.v1 import auth
from app.core import ratelimit
monkeypatch.setattr(ratelimit.settings, "RATE_LIMIT_ENABLED", True)
monkeypatch.setattr(auth, "SMS_SEND_MAX_PER_HOUR_PER_DEVICE", 2)
ratelimit._buckets.clear()
device = "dev-cooldown"
phone_a = "13710137000"
# 首发成功(小时闸计 1/2)
assert client.post(
"/api/v1/auth/sms/send", json={"phone": phone_a, "device_id": device}
).status_code == 200
# 同号连发 3 次:都被单号 60s 冷却挡下 → 429,且**不占**设备额度
for _ in range(3):
r = client.post(
"/api/v1/auth/sms/send", json={"phone": phone_a, "device_id": device}
)
assert r.status_code == 429, r.text
# 换号再发:设备额度只用了 1/2(冷却那几次没算)→ 仍放行(计到 2/2)
assert client.post(
"/api/v1/auth/sms/send", json={"phone": "13710137001", "device_id": device}
).status_code == 200
# 又换号:此时小时闸已 2/2 → 429(反证成功发码确实各计了 1)
r = client.post(
"/api/v1/auth/sms/send", json={"phone": "13710137002", "device_id": device}
)
assert r.status_code == 429, r.text
def test_sms_send_daily_cap(client, monkeypatch) -> None:
"""每天发码上限(设备 + IP):成功发码累计到日上限即 429(用不同手机号绕开单号冷却)。
抬高小时闸单独测日闸;超限文案含今日以便前端提示明天再来"""
from app.api.v1 import auth
from app.core import ratelimit
monkeypatch.setattr(ratelimit.settings, "RATE_LIMIT_ENABLED", True)
monkeypatch.setattr(auth, "SMS_SEND_MAX_PER_HOUR_PER_DEVICE", 100) # 抬高小时闸,不干扰
monkeypatch.setattr(auth, "SMS_SEND_MAX_PER_DAY_PER_DEVICE", 3)
ratelimit._buckets.clear()
device = "dev-daily"
for i in range(3):
r = client.post(
"/api/v1/auth/sms/send",
json={"phone": f"13720137{i:03d}", "device_id": device},
)
assert r.status_code == 200, f"{i + 1} 次应放行: {r.text}"
# 第 4 次:同设备同 IP 当日超限 → 429
r = client.post(
"/api/v1/auth/sms/send",
json={"phone": "13720137999", "device_id": device},
)
assert r.status_code == 429, r.text
assert "今日" in r.json()["detail"]
def test_sms_login_device_ip_rate_limit(client, monkeypatch) -> None:
"""防刷:同一设备(device_id) + 同一 IP 每小时最多 SMS_LOGIN_MAX_PER_HOUR 次登录尝试,超出 429。
conftest 默认 RATE_LIMIT_ENABLED=false(内存计数跨用例累加),本用例临时打开并清空计数隔离"""
+87
View File
@@ -0,0 +1,87 @@
"""两个看板逐行「本次广告收益」(元):按 trace_id 聚合 ad_ecpm_record。"""
from __future__ import annotations
from datetime import UTC, date, datetime
from sqlalchemy import delete
from app.admin.repositories import queries
from app.admin.repositories.coupon_data import coupon_data_report, coupon_user_records
from app.db.session import SessionLocal
from app.models.ad_ecpm import AdEcpmRecord
from app.models.comparison import ComparisonRecord
from app.models.coupon_state import CouponSession
def _ecpm(trace_id: str, ecpm_raw: str, session_id: str, scene: str) -> AdEcpmRecord:
return AdEcpmRecord(
user_id=1, ad_type="draw", feed_scene=scene, ad_session_id=session_id,
ecpm_raw=ecpm_raw, trace_id=trace_id, report_date="2020-01-02",
created_at=datetime(2020, 1, 2, tzinfo=UTC),
)
def test_coupon_data_report_includes_ad_revenue() -> None:
"""领券看板明细行带本次广告收益;200+300 分 → 0.005 元。"""
db = SessionLocal()
try:
db.add(CouponSession(
trace_id="rev-cp-1", device_id="d1", status="completed", app_env="prod",
platforms=["meituan-waimai"], platform_success=["meituan-waimai"],
started_at=datetime(2020, 1, 2, tzinfo=UTC), started_date=date(2020, 1, 2),
))
db.add_all([
_ecpm("rev-cp-1", "200", "cp-sess-a", "coupon"),
_ecpm("rev-cp-1", "300", "cp-sess-b", "coupon"),
])
db.flush()
res = coupon_data_report(db, date_from="2020-01-02", date_to="2020-01-02", app_env="prod")
row = next(r for r in res["items"] if r["trace_id"] == "rev-cp-1")
assert row["ad_revenue_yuan"] == 0.005
finally:
db.rollback()
db.close()
def test_comparison_list_includes_ad_revenue() -> None:
"""比价记录列表项带本次广告收益;200 分 → 0.002 元。
用独有 user_id 过滤,确保本行必落在第一页(避免共享测试库里同 user 记录多分页把它挤掉)
"""
db = SessionLocal()
try:
db.add(ComparisonRecord(
trace_id="rev-cmp-1", user_id=987654, status="success", business_type="food",
created_at=datetime(2020, 1, 2, tzinfo=UTC),
))
db.add(_ecpm("rev-cmp-1", "200", "cmp-sess-a", "comparison"))
db.commit()
items, _next, _total = queries.list_comparison_records(db, user_id=987654)
row = next(it for it in items if it.trace_id == "rev-cmp-1")
assert row.ad_revenue_yuan == 0.002
finally:
db.execute(delete(AdEcpmRecord).where(AdEcpmRecord.trace_id.in_(["rev-cmp-1"])))
db.execute(delete(ComparisonRecord).where(ComparisonRecord.trace_id == "rev-cmp-1"))
db.commit()
db.close()
def test_coupon_user_records_includes_ad_revenue() -> None:
"""手机号抽屉的用户领券记录也带本次广告收益(400 分 → 0.004 元)。"""
db = SessionLocal()
try:
db.add(CouponSession(
trace_id="rev-drawer-1", device_id="d2", status="completed", app_env="prod",
user_id=987655, platforms=["meituan-waimai"], platform_success=["meituan-waimai"],
started_at=datetime(2020, 1, 2, tzinfo=UTC), started_date=date(2020, 1, 2),
))
db.add(_ecpm("rev-drawer-1", "400", "drawer-sess-a", "coupon"))
db.commit()
res = coupon_user_records(db, user_id=987655)
row = next(r for r in res["items"] if r["trace_id"] == "rev-drawer-1")
assert row["ad_revenue_yuan"] == 0.004
finally:
db.execute(delete(AdEcpmRecord).where(AdEcpmRecord.trace_id == "rev-drawer-1"))
db.execute(delete(CouponSession).where(CouponSession.trace_id == "rev-drawer-1"))
db.commit()
db.close()
+184
View File
@@ -0,0 +1,184 @@
"""LLM 调用成本计算 compute_llm_cost:按模型分桶累加 token × 单价;error/无 usage 跳过。"""
from __future__ import annotations
from app.services.llm_cost import compute_llm_cost
_PRICE = {
"per_model": {"qwen3.5-flash": {"input_per_1m": 0.8, "output_per_1m": 2.0}},
"default": {"input_per_1m": 3.0, "output_per_1m": 15.0},
}
def test_sums_per_model_single_model():
# 真实样本:4 次 qwen3.5-flash;Σprompt=6888、Σcompletion=337
calls = [
{"model": "qwen3.5-flash", "error": None, "usage": {"prompt_tokens": 1512, "completion_tokens": 22}},
{"model": "qwen3.5-flash", "error": None, "usage": {"prompt_tokens": 2111, "completion_tokens": 160}},
{"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}},
]
cost, snapshot = compute_llm_cost(calls, _PRICE)
# 6888/1e6*0.8 + 337/1e6*2.0 = 0.0055104 + 0.000674 = 0.0061844 → round(6)
assert cost == 0.006184
assert snapshot == {
"mode": "per_model",
"prices": {
"qwen3.5-flash": {"input_per_1m": 0.8, "output_per_1m": 2.0, "_source": "per_model"},
},
}
def test_multi_model_prices_each_bucket_separately():
calls = [
{"model": "qwen3.5-flash", "error": None, "usage": {"prompt_tokens": 1_000_000, "completion_tokens": 0}},
{"model": "gpt-x", "error": None, "usage": {"prompt_tokens": 0, "completion_tokens": 1_000_000}},
]
price = {
"per_model": {
"qwen3.5-flash": {"input_per_1m": 0.8, "output_per_1m": 2.0},
"gpt-x": {"input_per_1m": 10.0, "output_per_1m": 30.0},
},
"default": {"input_per_1m": 3.0, "output_per_1m": 15.0},
}
cost, snap = compute_llm_cost(calls, price)
assert cost == 30.8 # qwen 1M入×0.8=0.8 + gpt-x 1M出×30=30.0
assert set(snap["prices"]) == {"qwen3.5-flash", "gpt-x"}
def test_unknown_model_falls_back_to_default():
calls = [{"model": "mystery", "error": None, "usage": {"prompt_tokens": 1_000_000, "completion_tokens": 0}}]
price = {"per_model": {}, "default": {"input_per_1m": 3.0, "output_per_1m": 15.0}}
cost, snap = compute_llm_cost(calls, price)
assert cost == 3.0
assert snap["prices"]["mystery"]["_source"] == "default"
def test_unpriced_model_marked_and_zero_cost():
calls = [{"model": "mystery", "error": None, "usage": {"prompt_tokens": 1_000_000, "completion_tokens": 999}}]
cost, snap = compute_llm_cost(calls, {"per_model": {}}) # 无 default
assert cost == 0.0
assert snap["prices"]["mystery"]["unpriced"] is True
def test_error_and_missing_usage_calls_skipped():
calls = [
{"model": "qwen3.5-flash", "error": "boom", "usage": {"prompt_tokens": 9_999_999, "completion_tokens": 9_999_999}},
{"model": "qwen3.5-flash", "error": None, "usage": None}, # 无 usage
{"model": "qwen3.5-flash", "error": None, "usage": {"prompt_tokens": 1_000_000, "completion_tokens": 0}},
]
cost, _ = compute_llm_cost(calls, _PRICE)
assert cost == 0.8 # 只有第 3 条计入
def test_empty_or_all_error_returns_none():
assert compute_llm_cost([], _PRICE) == (None, None)
assert compute_llm_cost(None, _PRICE) == (None, None)
all_error = [{"model": "x", "error": "boom", "usage": {"prompt_tokens": 100, "completion_tokens": 100}}]
assert compute_llm_cost(all_error, _PRICE) == (None, None)
def test_malformed_price_entry_is_treated_as_unpriced_not_raised():
# 手改配置页可能存出残缺/非法单价(缺 output_per_1m、非 dict);不能抛异常连累 token 回填。
calls = [
{"model": "bad-a", "error": None, "usage": {"prompt_tokens": 1_000_000, "completion_tokens": 5}},
{"model": "bad-b", "error": None, "usage": {"prompt_tokens": 1_000_000, "completion_tokens": 5}},
{"model": "ok", "error": None, "usage": {"prompt_tokens": 1_000_000, "completion_tokens": 0}},
]
price = {
"per_model": {
"bad-a": {"input_per_1m": 0.8}, # 缺 output_per_1m
"bad-b": 5, # 非 dict
"ok": {"input_per_1m": 3.0, "output_per_1m": 15.0},
},
}
cost, snap = compute_llm_cost(calls, price) # 不得抛异常
assert cost == 3.0 # 只有 ok(1M 入 × 3.0)计入;两个残缺项按 unpriced
assert snap["prices"]["bad-a"].get("unpriced") is True
assert snap["prices"]["bad-b"].get("unpriced") is True
def test_get_llm_prices_falls_back_to_default_then_uses_override():
from app.db.session import SessionLocal
from app.models.app_config import AppConfig
from app.repositories import app_config
from app.services.llm_cost import get_llm_prices
db = SessionLocal()
try:
# 无 override → CONFIG_DEFS 默认(含 per_model / default)
prices = get_llm_prices(db)
assert "per_model" in prices and "default" in prices
# 有 override → 用 DB 值
app_config.set_value(
db, "llm_token_price",
{"per_model": {"m": {"input_per_1m": 1.0, "output_per_1m": 2.0}},
"default": {"input_per_1m": 0.0, "output_per_1m": 0.0}},
admin_id=1,
)
assert get_llm_prices(db)["per_model"]["m"]["input_per_1m"] == 1.0
finally:
row = db.get(AppConfig, "llm_token_price")
if row is not None:
db.delete(row)
db.commit()
db.close()
def test_backfill_llm_calls_stores_cost_and_snapshot(monkeypatch):
from datetime import UTC, datetime
from app.api.v1 import compare_record
from app.db.session import SessionLocal
from app.models.app_config import AppConfig
from app.models.comparison import ComparisonRecord
from app.repositories import app_config
sample = [
{"model": "qwen3.5-flash", "error": None, "usage": {"prompt_tokens": 1512, "completion_tokens": 22}},
{"model": "qwen3.5-flash", "error": None, "usage": {"prompt_tokens": 2111, "completion_tokens": 160}},
{"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)
db = SessionLocal()
try:
app_config.set_value(
db, "llm_token_price",
{"per_model": {"qwen3.5-flash": {"input_per_1m": 0.8, "output_per_1m": 2.0}},
"default": {"input_per_1m": 3.0, "output_per_1m": 15.0}},
admin_id=1,
)
rec = ComparisonRecord(
trace_id="llmcost-bf-1", status="success",
created_at=datetime.now(UTC).replace(tzinfo=None),
)
db.add(rec)
db.commit()
rid = rec.id
finally:
db.close()
compare_record._backfill_llm_calls(rid, "llmcost-bf-1") # 独立 session 内回填
db = SessionLocal()
try:
rec = db.get(ComparisonRecord, rid)
assert rec.llm_cost_yuan == 0.006184
assert rec.llm_price_snapshot["prices"]["qwen3.5-flash"]["input_per_1m"] == 0.8
assert rec.input_tokens == 6888 # 现有 token 派生仍在
finally:
db.delete(db.get(ComparisonRecord, rid))
row = db.get(AppConfig, "llm_token_price")
if row is not None:
db.delete(row)
db.commit()
db.close()
def test_admin_detail_schema_exposes_llm_cost_fields():
from app.admin.schemas.comparison import AdminComparisonDetail
fields = AdminComparisonDetail.model_fields
assert "llm_cost_yuan" in fields
assert "llm_price_snapshot" in fields
+57
View File
@@ -0,0 +1,57 @@
"""ratelimit 内存桶过期清理(GC)测试。
回归重点:_buckets **全局共享**混着不同窗口(60s 广告 / 3600s 登录 / 86400s 日闸) key
GC 必须按每个 key 自己存的 window_sec判过期,而不是当前调用方的窗口 否则高频的 60s 端点
触发 GC 时会把本该存活更久的 3600s/86400s 计数(如短信日闸)一并删掉,使其被反复清零限流失效
monkeypatch _GC_THRESHOLD 0 强制每次都扫,免造上万条(仿 test_auth 里对 sms._GC_THRESHOLD 的做法)
"""
from __future__ import annotations
from app.core import ratelimit
def test_purge_expired_respects_each_key_own_window(monkeypatch) -> None:
"""短窗口(60s)触发的 GC 只删真正过期的 key,不得删掉仍在自身窗口内的长窗口 key。"""
monkeypatch.setattr(ratelimit, "_GC_THRESHOLD", 0) # 强制每次都扫
ratelimit._buckets.clear()
now = 1_000_000.0
# 日闸:100s 前开窗、window=86400 → 远未过期,必须保留
ratelimit._buckets["sms-send-device-daily:D:IP"] = (now - 100, 7, 86400.0)
# 登录:1800s、window=3600 → 未过期,保留
ratelimit._buckets["sms-login-device:D:IP"] = (now - 1800, 2, 3600.0)
# 广告:120s、window=60 → 已过期,应删
ratelimit._buckets["ad-watch-report:IP2"] = (now - 120, 3, 60.0)
ratelimit._purge_expired(now)
assert "sms-send-device-daily:D:IP" in ratelimit._buckets
assert "sms-login-device:D:IP" in ratelimit._buckets
assert "ad-watch-report:IP2" not in ratelimit._buckets
def test_purge_expired_keeps_long_window_key_older_than_short_window(monkeypatch) -> None:
"""反证旧 bug:日闸 key 已老于 3600s,旧代码在 60s/3600s 端点触发 GC 时会误删它;
现在按自身 86400s 窗口判 未过期 必须保留"""
monkeypatch.setattr(ratelimit, "_GC_THRESHOLD", 0)
ratelimit._buckets.clear()
now = 2_000_000.0
# 3700s 前开窗(> 1 小时),但 window=86400 → 未过期
ratelimit._buckets["sms-send-device-daily:D:IP"] = (now - 3700, 20, 86400.0)
ratelimit._purge_expired(now)
assert "sms-send-device-daily:D:IP" in ratelimit._buckets
def test_purge_expired_noop_below_threshold(monkeypatch) -> None:
"""未超阈值时不扫(即便有过期 key 也不动),避免每次请求都 O(n) 扫全表。"""
monkeypatch.setattr(ratelimit, "_GC_THRESHOLD", 10)
ratelimit._buckets.clear()
now = 3_000_000.0
ratelimit._buckets["stale:IP"] = (now - 999, 1, 60.0) # 早过期,但没超阈值
ratelimit._purge_expired(now)
assert "stale:IP" in ratelimit._buckets # 桶数没超阈值 → 不清理