Compare commits
4 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| bf02b846a7 | |||
| b2a528eba1 | |||
| cdd49c6421 | |||
| 775a503d6f |
@@ -0,0 +1,31 @@
|
||||
"""guide video play count is independent for coupon and comparison
|
||||
|
||||
Revision ID: guide_video_scene_unique
|
||||
Revises: d8dd2106e438, risk_monitor_generic
|
||||
"""
|
||||
from alembic import op
|
||||
|
||||
revision = "guide_video_scene_unique"
|
||||
down_revision = ("d8dd2106e438", "risk_monitor_generic")
|
||||
branch_labels = None
|
||||
depends_on = None
|
||||
|
||||
|
||||
def upgrade() -> None:
|
||||
op.drop_index("uq_guide_video_play_user_seq", table_name="guide_video_play")
|
||||
op.create_index(
|
||||
"uq_guide_video_play_user_scene_seq",
|
||||
"guide_video_play",
|
||||
["user_id", "scene", "seq"],
|
||||
unique=True,
|
||||
)
|
||||
|
||||
|
||||
def downgrade() -> None:
|
||||
op.drop_index("uq_guide_video_play_user_scene_seq", table_name="guide_video_play")
|
||||
op.create_index(
|
||||
"uq_guide_video_play_user_seq",
|
||||
"guide_video_play",
|
||||
["user_id", "seq"],
|
||||
unique=True,
|
||||
)
|
||||
@@ -12,10 +12,11 @@
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from sqlalchemy import func, select
|
||||
from sqlalchemy import func, or_, select
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from app.core import rewards
|
||||
from app.models.ad_ecpm import AdEcpmRecord
|
||||
from app.models.ad_feed_reward import AdFeedRewardRecord
|
||||
from app.models.ad_reward import AdRewardRecord
|
||||
from app.repositories.ad_feed_reward import FEED_REWARD_UNIT_SECONDS
|
||||
@@ -55,10 +56,21 @@ def _reward_video_rows(
|
||||
if user_id is not None:
|
||||
stmt = stmt.where(AdRewardRecord.user_id == user_id)
|
||||
|
||||
records = list(db.execute(stmt).scalars())
|
||||
# S2S 发奖回调不携带实际填充 ADN;用相同用户和 ad_session_id 的展示记录回填。
|
||||
session_ids = {record.ad_session_id for record in records if record.ad_session_id}
|
||||
impression_by_session = {
|
||||
(record.user_id, record.ad_session_id): record
|
||||
for record in db.execute(
|
||||
select(AdEcpmRecord).where(AdEcpmRecord.ad_session_id.in_(session_ids))
|
||||
).scalars()
|
||||
} if session_ids else {}
|
||||
|
||||
# 用本日之前的累计份数做起点,当日 granted 在其上继续递增 → 与 _granted_cumulative+1 对齐
|
||||
granted_n: dict[int, int] = _prior_granted_counts(db, date=date, user_id=user_id)
|
||||
rows: list[dict] = []
|
||||
for rec in db.execute(stmt).scalars():
|
||||
for rec in records:
|
||||
impression = impression_by_session.get((rec.user_id, rec.ad_session_id))
|
||||
if rec.status == "granted":
|
||||
nth = granted_n.get(rec.user_id, 0) + 1
|
||||
granted_n[rec.user_id] = nth
|
||||
@@ -68,6 +80,8 @@ def _reward_video_rows(
|
||||
"record_id": rec.id,
|
||||
"user_id": rec.user_id,
|
||||
"ad_session_id": rec.ad_session_id,
|
||||
"adn": impression.adn if impression is not None else None,
|
||||
"slot_id": impression.slot_id if impression is not None else None,
|
||||
"app_env": rec.app_env,
|
||||
"our_code_id": rec.our_code_id,
|
||||
"created_at": rec.created_at,
|
||||
@@ -90,6 +104,8 @@ def _reward_video_rows(
|
||||
"record_id": rec.id,
|
||||
"user_id": rec.user_id,
|
||||
"ad_session_id": rec.ad_session_id,
|
||||
"adn": impression.adn if impression is not None else None,
|
||||
"slot_id": impression.slot_id if impression is not None else None,
|
||||
"app_env": rec.app_env,
|
||||
"our_code_id": rec.our_code_id,
|
||||
"created_at": rec.created_at,
|
||||
@@ -149,6 +165,78 @@ def _feed_scene_matches(rec: AdFeedRewardRecord, scene: str | None) -> bool:
|
||||
return True
|
||||
|
||||
|
||||
def _nonblank(value: str | None) -> str | None:
|
||||
value = value.strip() if value else None
|
||||
return value or None
|
||||
|
||||
|
||||
def _unique_ad_source(records: list[AdEcpmRecord]) -> tuple[str | None, str | None]:
|
||||
"""仅在候选展示记录指向唯一 ADN 时回填来源,避免错误归因。"""
|
||||
adns = {_nonblank(record.adn) for record in records}
|
||||
adns.discard(None)
|
||||
if len(adns) != 1:
|
||||
return None, None
|
||||
|
||||
slots = {_nonblank(record.slot_id) for record in records}
|
||||
slots.discard(None)
|
||||
return next(iter(adns)), next(iter(slots)) if len(slots) == 1 else None
|
||||
|
||||
|
||||
def _feed_source_fallbacks(
|
||||
db: Session, records: list[AdFeedRewardRecord]
|
||||
) -> tuple[
|
||||
dict[tuple[int, str], tuple[str | None, str | None]],
|
||||
dict[tuple[int, str, str], tuple[str | None, str | None]],
|
||||
]:
|
||||
"""为旧信息流发奖记录构建安全来源索引。"""
|
||||
session_ids = {record.ad_session_id for record in records if record.ad_session_id}
|
||||
trace_ids = {record.trace_id for record in records if record.trace_id}
|
||||
if not session_ids and not trace_ids:
|
||||
return {}, {}
|
||||
|
||||
filters = []
|
||||
if session_ids:
|
||||
filters.append(AdEcpmRecord.ad_session_id.in_(session_ids))
|
||||
if trace_ids:
|
||||
filters.append(AdEcpmRecord.trace_id.in_(trace_ids))
|
||||
impressions = list(db.execute(select(AdEcpmRecord).where(or_(*filters))).scalars())
|
||||
|
||||
by_session: dict[tuple[int, str], list[AdEcpmRecord]] = {}
|
||||
by_trace_ecpm: dict[tuple[int, str, str], list[AdEcpmRecord]] = {}
|
||||
for impression in impressions:
|
||||
if impression.ad_session_id:
|
||||
by_session.setdefault((impression.user_id, impression.ad_session_id), []).append(impression)
|
||||
if impression.trace_id:
|
||||
by_trace_ecpm.setdefault(
|
||||
(impression.user_id, impression.trace_id, impression.ecpm_raw), []
|
||||
).append(impression)
|
||||
|
||||
return (
|
||||
{key: _unique_ad_source(value) for key, value in by_session.items()},
|
||||
{key: _unique_ad_source(value) for key, value in by_trace_ecpm.items()},
|
||||
)
|
||||
|
||||
|
||||
def _feed_source(
|
||||
record: AdFeedRewardRecord,
|
||||
*,
|
||||
by_session: dict[tuple[int, str], tuple[str | None, str | None]],
|
||||
by_trace_ecpm: dict[tuple[int, str, str], tuple[str | None, str | None]],
|
||||
) -> tuple[str | None, str | None]:
|
||||
"""返回本条发奖广告的来源;无唯一证据时保留原始空值。"""
|
||||
adn, slot_id = _nonblank(record.adn), _nonblank(record.slot_id)
|
||||
if adn and slot_id:
|
||||
return adn, slot_id
|
||||
|
||||
candidate = by_session.get((record.user_id, record.ad_session_id or ""))
|
||||
if candidate is None and record.trace_id:
|
||||
candidate = by_trace_ecpm.get((record.user_id, record.trace_id, record.ecpm_raw))
|
||||
if candidate is None:
|
||||
return adn, slot_id
|
||||
candidate_adn, candidate_slot_id = candidate
|
||||
return adn or candidate_adn, slot_id or candidate_slot_id
|
||||
|
||||
|
||||
def _feed_rows(
|
||||
db: Session, *, date: str, user_id: int | None, scene: str | None = None
|
||||
) -> list[dict]:
|
||||
@@ -167,11 +255,17 @@ def _feed_rows(
|
||||
if user_id is not None:
|
||||
stmt = stmt.where(AdFeedRewardRecord.user_id == user_id)
|
||||
|
||||
records = list(db.execute(stmt).scalars())
|
||||
by_session, by_trace_ecpm = _feed_source_fallbacks(db, records)
|
||||
|
||||
# 本日之前的累计**条数**做起点,与发奖侧 granted_unit_total(COUNT granted)对齐
|
||||
granted_count: dict[int, int] = _feed_prior_granted_count(db, date=date, user_id=user_id)
|
||||
rows: list[dict] = []
|
||||
for rec in db.execute(stmt).scalars():
|
||||
for rec in records:
|
||||
keep = _feed_scene_matches(rec, scene) # 累计照常推进,这里只决定是否展示本行
|
||||
adn, slot_id = _feed_source(
|
||||
rec, by_session=by_session, by_trace_ecpm=by_trace_ecpm
|
||||
)
|
||||
if rec.status == "granted":
|
||||
# 一条广告 = 1 份(与 grant_feed_reward 同口径:看满一份即发该条满额,不按 unit_count 累加)。
|
||||
# nth = 账号累计第几**条**(含本日之前),与发奖侧 granted_unit_total+1 对齐;累计照常推进
|
||||
@@ -188,6 +282,8 @@ def _feed_rows(
|
||||
"record_id": rec.id,
|
||||
"user_id": rec.user_id,
|
||||
"ad_session_id": rec.ad_session_id,
|
||||
"adn": adn,
|
||||
"slot_id": slot_id,
|
||||
"trace_id": rec.trace_id,
|
||||
"app_env": rec.app_env,
|
||||
"our_code_id": rec.our_code_id,
|
||||
@@ -214,6 +310,8 @@ def _feed_rows(
|
||||
"record_id": rec.id,
|
||||
"user_id": rec.user_id,
|
||||
"ad_session_id": rec.ad_session_id,
|
||||
"adn": adn,
|
||||
"slot_id": slot_id,
|
||||
"trace_id": rec.trace_id,
|
||||
"app_env": rec.app_env,
|
||||
"our_code_id": rec.our_code_id,
|
||||
|
||||
@@ -96,7 +96,11 @@ _REWARD_DETAIL_KEYS = (
|
||||
|
||||
def _reward_detail(row: dict) -> dict:
|
||||
"""从 audit 行抽出发奖复算明细(给前端展开行渲染因子1/因子2/份数/LT/应发实发)。"""
|
||||
return {k: row[k] for k in _REWARD_DETAIL_KEYS}
|
||||
detail = {key: row[key] for key in _REWARD_DETAIL_KEYS}
|
||||
# 聚合父行可能包含多个 ADN,来源必须保留在每一条发奖明细上。
|
||||
detail["adn"] = row.get("adn")
|
||||
detail["slot_id"] = row.get("slot_id")
|
||||
return detail
|
||||
|
||||
|
||||
def ad_revenue_report(
|
||||
@@ -244,8 +248,8 @@ def ad_revenue_report(
|
||||
"impressions": 0,
|
||||
"ecpm": row["ecpm"],
|
||||
"revenue_yuan": 0.0,
|
||||
"adn": None,
|
||||
"slot_id": None,
|
||||
"adn": row.get("adn"),
|
||||
"slot_id": row.get("slot_id"),
|
||||
"has_reward": True,
|
||||
"status": row["status"],
|
||||
"expected_coin": int(row["expected_coin"]),
|
||||
|
||||
@@ -9,7 +9,7 @@ from datetime import date, datetime, time, timedelta, timezone
|
||||
from decimal import ROUND_HALF_UP, Decimal
|
||||
from zoneinfo import ZoneInfo
|
||||
|
||||
from sqlalchemy import Select, and_, asc, case, desc, func, or_, select
|
||||
from sqlalchemy import Select, asc, case, desc, func, or_, select
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from app.core import rewards
|
||||
@@ -45,72 +45,6 @@ _FEED_SCENE_LABEL = {
|
||||
}
|
||||
|
||||
|
||||
_DEVICE_MARKETING_NAMES = {
|
||||
"23078RKD5C": "Redmi K60 至尊版",
|
||||
"M2012K11AC": "Redmi K40",
|
||||
"PJA110": "一加 Ace 2 Pro",
|
||||
"PPG-AN00": "荣耀 GT Pro",
|
||||
"V2166BA": "vivo Y77e",
|
||||
"V2309A": "vivo X100",
|
||||
}
|
||||
|
||||
|
||||
def _device_marketing_name(model: str | None) -> str | None:
|
||||
"""把线上已知 Build.MODEL 编码转成用户可识别的商品名。"""
|
||||
if not model:
|
||||
return None
|
||||
return _DEVICE_MARKETING_NAMES.get(model.strip().upper())
|
||||
|
||||
|
||||
def _attach_feedback_device_details(db: Session, feedbacks: list[Feedback]) -> None:
|
||||
"""按同一用户、同一设备编码及提交时间补齐厂商和 ROM 大版本。"""
|
||||
candidates = [
|
||||
item for item in feedbacks if item.device_model and item.device_model.strip()
|
||||
]
|
||||
for item in candidates:
|
||||
item.device_model_name = _device_marketing_name(item.device_model)
|
||||
item.device_manufacturer = None
|
||||
item.rom_version = None
|
||||
if not candidates:
|
||||
return
|
||||
|
||||
ranked = (
|
||||
select(
|
||||
Feedback.id.label("feedback_id"),
|
||||
ComparisonRecord.device_manufacturer.label("device_manufacturer"),
|
||||
ComparisonRecord.rom_version.label("rom_version"),
|
||||
func.row_number()
|
||||
.over(
|
||||
partition_by=Feedback.id,
|
||||
order_by=(
|
||||
ComparisonRecord.created_at.desc(),
|
||||
ComparisonRecord.id.desc(),
|
||||
),
|
||||
)
|
||||
.label("row_num"),
|
||||
)
|
||||
.join(
|
||||
ComparisonRecord,
|
||||
and_(
|
||||
ComparisonRecord.user_id == Feedback.user_id,
|
||||
ComparisonRecord.device_model == Feedback.device_model,
|
||||
ComparisonRecord.created_at <= Feedback.created_at,
|
||||
),
|
||||
)
|
||||
.where(Feedback.id.in_([item.id for item in candidates]))
|
||||
.subquery()
|
||||
)
|
||||
details = {
|
||||
row.feedback_id: row
|
||||
for row in db.execute(select(ranked).where(ranked.c.row_num == 1)).all()
|
||||
}
|
||||
|
||||
for item in candidates:
|
||||
detail = details.get(item.id)
|
||||
item.device_manufacturer = detail.device_manufacturer if detail else None
|
||||
item.rom_version = detail.rom_version if detail else None
|
||||
|
||||
|
||||
def cursor_paginate(
|
||||
db: Session, stmt: Select, id_col, *, limit: int, cursor: int | None
|
||||
) -> tuple[list, int | None]:
|
||||
@@ -883,7 +817,6 @@ def list_feedbacks(
|
||||
db, stmt, (order_fn(sort_col), id_order), limit=limit, cursor=cursor
|
||||
)
|
||||
_attach_user_info(db, items) # 列表展示完整手机号(点手机号查该用户全部反馈)
|
||||
_attach_feedback_device_details(db, items)
|
||||
return items, next_cursor, total
|
||||
|
||||
|
||||
@@ -1295,6 +1228,11 @@ def _cn_wall_to_utc(dt: datetime) -> datetime:
|
||||
return dt.replace(tzinfo=rewards.CN_TZ).astimezone(timezone.utc).replace(tzinfo=None)
|
||||
|
||||
|
||||
def _coin_record_sort_key(row: dict) -> datetime:
|
||||
"""金币明细跨数据源排序键:兼容 SQLite naive 与 PostgreSQL aware 时间。"""
|
||||
return _as_utc(row["created_at"])
|
||||
|
||||
|
||||
def user_coin_records(
|
||||
db: Session,
|
||||
user_id: int,
|
||||
@@ -1386,7 +1324,10 @@ def user_coin_records(
|
||||
"coin": rec.amount,
|
||||
})
|
||||
|
||||
rows.sort(key=lambda r: r["created_at"], reverse=True)
|
||||
# SQLite 常返回 naive datetime,PostgreSQL timestamptz 返回 aware datetime;
|
||||
# 统一成 aware UTC 排序,避免线上合并广告记录与签到记录时抛
|
||||
# “can't compare offset-naive and offset-aware datetimes”。
|
||||
rows.sort(key=_coin_record_sort_key, reverse=True)
|
||||
has_more = len(rows) > offset + limit
|
||||
|
||||
# 总数 = 三源在窗口内 granted 计数之和(供前端页码分页渲染页码/共 N 条)
|
||||
|
||||
@@ -9,7 +9,7 @@ client_max_body_size,见 shaguabijia-admin-web/deploy/nginx/admin.shaguabijia.co
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Annotated
|
||||
from typing import Annotated, Literal
|
||||
|
||||
from fastapi import APIRouter, Depends, File, HTTPException, Request, UploadFile
|
||||
|
||||
@@ -27,14 +27,21 @@ router = APIRouter(
|
||||
)
|
||||
|
||||
|
||||
def _out(db: AdminDb) -> GuideVideoConfigOut:
|
||||
GuideScene = Literal["coupon", "comparison"]
|
||||
|
||||
|
||||
def _out(db: AdminDb, scene: GuideScene) -> GuideVideoConfigOut:
|
||||
"""配置 + 播放统计合成响应(四个写接口都以最新状态返回,前端一次同步到位)。"""
|
||||
return GuideVideoConfigOut(**guide_video.get_config(db), **guide_video.play_stats(db))
|
||||
return GuideVideoConfigOut(
|
||||
scene=scene,
|
||||
**guide_video.get_config(db, scene),
|
||||
**guide_video.play_stats(db, scene),
|
||||
)
|
||||
|
||||
|
||||
@router.get("", response_model=GuideVideoConfigOut, summary="新手引导视频配置(领券浮层)")
|
||||
def get_config(db: AdminDb) -> GuideVideoConfigOut:
|
||||
return _out(db)
|
||||
def get_config(db: AdminDb, scene: GuideScene = "coupon") -> GuideVideoConfigOut:
|
||||
return _out(db, scene)
|
||||
|
||||
|
||||
@router.patch("", response_model=GuideVideoConfigOut, summary="改开关/次数/金币(带审计)")
|
||||
@@ -43,21 +50,23 @@ def update_config(
|
||||
request: Request,
|
||||
admin: Annotated[AdminUser, Depends(require_role("operator"))],
|
||||
db: AdminDb,
|
||||
scene: GuideScene = "coupon",
|
||||
) -> GuideVideoConfigOut:
|
||||
before, after = guide_video.update_config(
|
||||
db,
|
||||
enabled=body.enabled,
|
||||
max_plays=body.max_plays,
|
||||
reward_coin=body.reward_coin,
|
||||
scene=scene,
|
||||
admin_id=admin.id,
|
||||
commit=False,
|
||||
)
|
||||
write_audit(
|
||||
db, admin, action="guide_video.update", target_type="guide_video", target_id=None,
|
||||
detail={"before": before, "after": after}, ip=get_client_ip(request), commit=False,
|
||||
detail={"scene": scene, "before": before, "after": after}, ip=get_client_ip(request), commit=False,
|
||||
)
|
||||
db.commit()
|
||||
return _out(db)
|
||||
return _out(db, scene)
|
||||
|
||||
|
||||
@router.post("/video", response_model=GuideVideoConfigOut, summary="上传新手引导视频(MP4,带审计)")
|
||||
@@ -65,23 +74,26 @@ async def upload_video(
|
||||
request: Request,
|
||||
admin: Annotated[AdminUser, Depends(require_role("operator"))],
|
||||
db: AdminDb,
|
||||
file: UploadFile = File(...),
|
||||
file: Annotated[UploadFile, File()],
|
||||
scene: GuideScene = "coupon",
|
||||
) -> GuideVideoConfigOut:
|
||||
data = await file.read()
|
||||
try:
|
||||
url = media.save_guide_video(data)
|
||||
except media.MediaError as e:
|
||||
raise HTTPException(status_code=400, detail=str(e)) from e
|
||||
before, after = guide_video.set_video(db, url, admin_id=admin.id, commit=False)
|
||||
before, after = guide_video.set_video(
|
||||
db, url, scene=scene, admin_id=admin.id, commit=False
|
||||
)
|
||||
write_audit(
|
||||
db, admin, action="guide_video.set_video", target_type="guide_video", target_id=None,
|
||||
detail={"before": before.get("video_url"), "after": url, "bytes": len(data)},
|
||||
detail={"scene": scene, "before": before.get("video_url"), "after": url, "bytes": len(data)},
|
||||
ip=get_client_ip(request), commit=False,
|
||||
)
|
||||
db.commit()
|
||||
# 提交成功后再删旧片,避免新片没落库就把旧片丢了
|
||||
media.delete_guide_video(before.get("video_url"))
|
||||
return _out(db)
|
||||
return _out(db, scene)
|
||||
|
||||
|
||||
@router.delete("/video", response_model=GuideVideoConfigOut, summary="移除新手引导视频(带审计)")
|
||||
@@ -89,13 +101,16 @@ def delete_video(
|
||||
request: Request,
|
||||
admin: Annotated[AdminUser, Depends(require_role("operator"))],
|
||||
db: AdminDb,
|
||||
scene: GuideScene = "coupon",
|
||||
) -> GuideVideoConfigOut:
|
||||
"""移除后 /guide-video/start 一律返回 should_play=false,领券浮层回到「只放广告」。"""
|
||||
before, after = guide_video.set_video(db, None, admin_id=admin.id, commit=False)
|
||||
before, after = guide_video.set_video(
|
||||
db, None, scene=scene, admin_id=admin.id, commit=False
|
||||
)
|
||||
write_audit(
|
||||
db, admin, action="guide_video.delete_video", target_type="guide_video", target_id=None,
|
||||
detail={"before": before.get("video_url")}, ip=get_client_ip(request), commit=False,
|
||||
detail={"scene": scene, "before": before.get("video_url")}, ip=get_client_ip(request), commit=False,
|
||||
)
|
||||
db.commit()
|
||||
media.delete_guide_video(before.get("video_url"))
|
||||
return _out(db)
|
||||
return _out(db, scene)
|
||||
|
||||
@@ -86,7 +86,7 @@ def get_user_reward_stats(
|
||||
date_to: Annotated[datetime | None, Query()] = None,
|
||||
) -> UserRewardStats:
|
||||
"""提现详情抽屉「用户统计区」。date_from/date_to 都不传 = 注册至今(全量)。"""
|
||||
if user_repo.get_user_by_id(db, user_id) is None:
|
||||
if not user_repo.user_exists(db, user_id):
|
||||
raise HTTPException(status_code=404, detail="用户不存在")
|
||||
return UserRewardStats(
|
||||
**queries.user_reward_stats(db, user_id, date_from=date_from, date_to=date_to)
|
||||
|
||||
@@ -40,6 +40,8 @@ class AdRevenueRecord(BaseModel):
|
||||
expected_coin: int = Field(..., description="按公式复算应发金币")
|
||||
actual_coin: int = Field(..., description="实际入账金币")
|
||||
matched: bool = Field(..., description="复算与实发是否一致")
|
||||
adn: str | None = Field(None, description="本条发奖对应的实际填充 ADN 子渠道")
|
||||
slot_id: str | None = Field(None, description="本条发奖对应的底层 mediation rit")
|
||||
|
||||
|
||||
class AdRevenueDaily(BaseModel):
|
||||
|
||||
@@ -32,10 +32,7 @@ class FeedbackOut(BaseModel):
|
||||
# 提交端环境快照(feedback 表列):提交版本号 / 机型OS版本;改版前的历史反馈为 None
|
||||
app_version: str | None = None
|
||||
device_model: str | None = None
|
||||
device_model_name: str | None = None
|
||||
device_manufacturer: str | None = None
|
||||
rom_name: str | None = None
|
||||
rom_version: int | None = None
|
||||
android_version: str | None = None
|
||||
# 联表瞬态字段(queries._attach_user_info 挂):列表展示完整手机号,点手机号查该用户全部反馈
|
||||
phone: str | None = None
|
||||
|
||||
@@ -7,6 +7,7 @@ from app.repositories.guide_video import MAX_PLAYS_LIMIT, REWARD_COIN_LIMIT
|
||||
|
||||
|
||||
class GuideVideoConfigOut(BaseModel):
|
||||
scene: str
|
||||
enabled: bool
|
||||
video_url: str | None = None # 相对地址 /media/guide_video/xxx.mp4;未配片 = None
|
||||
max_plays: int
|
||||
|
||||
@@ -38,7 +38,7 @@ class GuideVideoPlay(Base):
|
||||
# start_play 捕获 IntegrityError 降级成"这次不放视频"。
|
||||
# 用 unique Index 而非 UniqueConstraint:与迁移里的 create_index 对齐(SQLite 加约束
|
||||
# 要整表重建),autogenerate 才不会每次报一条假 diff。
|
||||
Index("uq_guide_video_play_user_seq", "user_id", "seq", unique=True),
|
||||
Index("uq_guide_video_play_user_scene_seq", "user_id", "scene", "seq", unique=True),
|
||||
)
|
||||
|
||||
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
|
||||
|
||||
@@ -55,13 +55,22 @@ def _product_names_from_items(items: list | None) -> str | None:
|
||||
def _derive(payload: ComparisonRecordIn) -> dict:
|
||||
"""从上报 payload 派生结构化列(best/saved/is_source_best/status)。"""
|
||||
results = payload.comparison_results
|
||||
_pr = payload.platform_results or {}
|
||||
|
||||
# 最优 = rank 最小的一条;协议已升序,但不信顺序,显式按 rank/price 兜底取最小价。
|
||||
best = None
|
||||
def _is_short(r) -> bool:
|
||||
# 缺菜(漏菜)店: 少买了菜总价虚低, 不参与最优评选。逐平台 skipped 在 platform_results, 行里没有。
|
||||
# platform_results 内层结构宽松(pricebot/老客户端透传, 可伪造), 值非 dict 时按"不缺菜"处理, 不崩。
|
||||
info = _pr.get(r.platform_id) if r.platform_id else None
|
||||
return isinstance(info, dict) and (info.get("skipped_dish_count") or 0) > 0
|
||||
|
||||
# 最优 = 非缺菜里 rank 最小(=最便宜)的一条;协议已升序,但不信顺序,显式按 rank/price 取。
|
||||
# 源平台永远全菜, 故全目标缺菜时回落到源(is_source_best、saved=0), 不把虚低价当最低。
|
||||
priced = [r for r in results if r.price is not None]
|
||||
if priced:
|
||||
clean = [r for r in priced if not _is_short(r)]
|
||||
best = None
|
||||
if clean:
|
||||
best = min(
|
||||
priced,
|
||||
clean,
|
||||
key=lambda r: (r.rank if r.rank is not None else 10**9, r.price),
|
||||
)
|
||||
|
||||
@@ -196,14 +205,29 @@ def upsert_record(
|
||||
# ============================================================
|
||||
|
||||
|
||||
def _derive_from_results(results: list[dict]) -> dict:
|
||||
def _derive_from_results(
|
||||
results: list[dict], platform_results: dict | None = None
|
||||
) -> dict:
|
||||
"""从 done 帧 comparison_results(pricebot 原始 dict 列表)派生结构化列。
|
||||
等价 _derive,但吃原始字段(is_source/price/rank/platform_id/store_name...)而非 pydantic 对象。"""
|
||||
等价 _derive,但吃原始字段(is_source/price/rank/platform_id/store_name...)而非 pydantic 对象。
|
||||
|
||||
platform_results(done.params.platform_results): 逐平台 skipped_dish_count 在这里(行里没有)。
|
||||
传入则派生 best 时排除缺菜(漏菜)店 —— 少买了菜总价虚低, 不能当记录级"最低价"/算虚假省额;
|
||||
源平台永远全菜, 故全目标缺菜时 best 回落到源(is_source_best、不虚报省)。不传→纯 rank/price, 行为不变。"""
|
||||
_pr = platform_results or {}
|
||||
|
||||
def _is_short(r: dict) -> bool:
|
||||
# platform_results 内层结构宽松(pricebot/客户端透传), 值非 dict 时按"不缺菜"处理, 不崩。
|
||||
pid = r.get("platform_id")
|
||||
info = _pr.get(pid) if pid else None
|
||||
return isinstance(info, dict) and (info.get("skipped_dish_count") or 0) > 0
|
||||
|
||||
priced = [r for r in results if r.get("price") is not None]
|
||||
clean = [r for r in priced if not _is_short(r)] # 缺菜店排除出最优评选
|
||||
best = None
|
||||
if priced:
|
||||
if clean:
|
||||
best = min(
|
||||
priced,
|
||||
clean,
|
||||
key=lambda r: (r.get("rank") if r.get("rank") is not None else 10**9, r["price"]),
|
||||
)
|
||||
src_row = next((r for r in results if r.get("is_source")), None)
|
||||
@@ -389,7 +413,7 @@ def harvest_done(
|
||||
返回 (记录, 是否本次**新**落成 success)——供调用方据此幂等发一次邀请奖。
|
||||
行不存在(理论上帧0已建;防御)则新建。"""
|
||||
results = done_params.get("comparison_results") or []
|
||||
derived = _derive_from_results(results)
|
||||
derived = _derive_from_results(results, done_params.get("platform_results"))
|
||||
# 菜品:pricebot 已把源单菜品塞进 comparison_results[源行].items
|
||||
items = next((r.get("items") or [] for r in results if r.get("is_source")), [])
|
||||
fields = dict(
|
||||
|
||||
@@ -30,7 +30,11 @@ from app.models.app_config import AppConfig
|
||||
from app.models.guide_video import GuideVideoPlay
|
||||
from app.repositories import wallet as crud_wallet
|
||||
|
||||
_KEY = "coupon_guide_video"
|
||||
SCENES = ("coupon", "comparison")
|
||||
_KEY_BY_SCENE = {
|
||||
"coupon": "coupon_guide_video",
|
||||
"comparison": "comparison_guide_video",
|
||||
}
|
||||
|
||||
#: 金币流水 biz_type。客户端收益明细按它显示「新手引导视频奖励」。
|
||||
BIZ_TYPE = "guide_video"
|
||||
@@ -41,7 +45,7 @@ _DEFAULTS: dict[str, Any] = {
|
||||
"enabled": True,
|
||||
"video_url": None, # None/空 = 未配片 → 不下发,浮层照旧放广告
|
||||
"max_plays": 3, # 每个账号前 N 次浮层放引导视频
|
||||
"reward_coin": 120, # 每次固定金币
|
||||
"reward_coin": 100, # 每次固定金币
|
||||
}
|
||||
|
||||
_FIELDS = tuple(_DEFAULTS.keys())
|
||||
@@ -65,19 +69,28 @@ def _merge(raw: Any) -> dict[str, Any]:
|
||||
return out
|
||||
|
||||
|
||||
def get_config(db: Session) -> dict[str, Any]:
|
||||
def _config_key(scene: str) -> str:
|
||||
if scene not in _KEY_BY_SCENE:
|
||||
raise ValueError(f"unsupported guide video scene: {scene}")
|
||||
return _KEY_BY_SCENE[scene]
|
||||
|
||||
|
||||
def get_config(db: Session, scene: str = "coupon") -> dict[str, Any]:
|
||||
"""完整配置 + updated_at(admin 读 / 业务读共用)。"""
|
||||
row = db.get(AppConfig, _KEY)
|
||||
row = db.get(AppConfig, _config_key(scene))
|
||||
cfg = _merge(row.value if row is not None else None)
|
||||
cfg["updated_at"] = row.updated_at.isoformat() if row is not None and row.updated_at else None
|
||||
return cfg
|
||||
|
||||
|
||||
def _write(db: Session, value: dict[str, Any], *, admin_id: int, commit: bool) -> dict[str, Any]:
|
||||
def _write(
|
||||
db: Session, value: dict[str, Any], *, scene: str, admin_id: int, commit: bool
|
||||
) -> dict[str, Any]:
|
||||
"""整体覆写该行(value 须为完整字段 dict),返回合并后的完整配置(含 updated_at)。"""
|
||||
row = db.get(AppConfig, _KEY)
|
||||
key = _config_key(scene)
|
||||
row = db.get(AppConfig, key)
|
||||
if row is None:
|
||||
row = AppConfig(key=_KEY, value=value, updated_by_admin_id=admin_id)
|
||||
row = AppConfig(key=key, value=value, updated_by_admin_id=admin_id)
|
||||
db.add(row)
|
||||
else:
|
||||
row.value = value # 整体重新赋值,SQLAlchemy 才侦测得到变更
|
||||
@@ -98,11 +111,12 @@ def update_config(
|
||||
enabled: bool | None = None,
|
||||
max_plays: int | None = None,
|
||||
reward_coin: int | None = None,
|
||||
scene: str = "coupon",
|
||||
admin_id: int,
|
||||
commit: bool = True,
|
||||
) -> tuple[dict[str, Any], dict[str, Any]]:
|
||||
"""改开关 / 次数 / 金币(只改传了的字段;视频走 set_video)。返回 (before, after) 供审计。"""
|
||||
row = db.get(AppConfig, _KEY)
|
||||
row = db.get(AppConfig, _config_key(scene))
|
||||
before = _merge(row.value if row is not None else None)
|
||||
new_value = {k: before[k] for k in _FIELDS}
|
||||
if enabled is not None:
|
||||
@@ -111,45 +125,52 @@ def update_config(
|
||||
new_value["max_plays"] = max(0, min(int(max_plays), MAX_PLAYS_LIMIT))
|
||||
if reward_coin is not None:
|
||||
new_value["reward_coin"] = max(0, min(int(reward_coin), REWARD_COIN_LIMIT))
|
||||
after = _write(db, new_value, admin_id=admin_id, commit=commit)
|
||||
after = _write(db, new_value, scene=scene, admin_id=admin_id, commit=commit)
|
||||
return before, after
|
||||
|
||||
|
||||
def set_video(
|
||||
db: Session, video_url: str | None, *, admin_id: int, commit: bool = True
|
||||
db: Session, video_url: str | None, *, scene: str = "coupon",
|
||||
admin_id: int, commit: bool = True
|
||||
) -> tuple[dict[str, Any], dict[str, Any]]:
|
||||
"""设置/清空引导视频地址。返回 (before, after);before['video_url'] 供调用方删旧文件。"""
|
||||
row = db.get(AppConfig, _KEY)
|
||||
row = db.get(AppConfig, _config_key(scene))
|
||||
before = _merge(row.value if row is not None else None)
|
||||
new_value = {k: before[k] for k in _FIELDS}
|
||||
new_value["video_url"] = video_url
|
||||
after = _write(db, new_value, admin_id=admin_id, commit=commit)
|
||||
after = _write(db, new_value, scene=scene, admin_id=admin_id, commit=commit)
|
||||
return before, after
|
||||
|
||||
|
||||
# ===== 播放计次 =====
|
||||
|
||||
|
||||
def used_plays(db: Session, user_id: int) -> int:
|
||||
def used_plays(db: Session, user_id: int, scene: str = "coupon") -> int:
|
||||
"""该账号已用掉的引导视频次数(开播即算,含未发币的)。"""
|
||||
return int(
|
||||
db.execute(
|
||||
select(func.count()).select_from(GuideVideoPlay).where(
|
||||
GuideVideoPlay.user_id == user_id
|
||||
GuideVideoPlay.user_id == user_id,
|
||||
GuideVideoPlay.scene == scene,
|
||||
)
|
||||
).scalar_one()
|
||||
)
|
||||
|
||||
|
||||
def play_stats(db: Session) -> dict[str, int]:
|
||||
def play_stats(db: Session, scene: str = "coupon") -> dict[str, int]:
|
||||
"""全站播放统计(admin 页展示):总播放次数 / 其中已发币次数。"""
|
||||
total = int(
|
||||
db.execute(select(func.count()).select_from(GuideVideoPlay)).scalar_one()
|
||||
db.execute(
|
||||
select(func.count()).select_from(GuideVideoPlay).where(
|
||||
GuideVideoPlay.scene == scene
|
||||
)
|
||||
).scalar_one()
|
||||
)
|
||||
granted = int(
|
||||
db.execute(
|
||||
select(func.count()).select_from(GuideVideoPlay).where(
|
||||
GuideVideoPlay.status == "granted"
|
||||
GuideVideoPlay.status == "granted",
|
||||
GuideVideoPlay.scene == scene,
|
||||
)
|
||||
).scalar_one()
|
||||
)
|
||||
@@ -168,11 +189,12 @@ def start_play(
|
||||
reward_coin 播完/中途关闭都发的固定金币
|
||||
seq / remaining 第几次 / 发完这次还剩几次(仅展示与排查用)
|
||||
"""
|
||||
cfg = get_config(db)
|
||||
_config_key(scene)
|
||||
cfg = get_config(db, scene)
|
||||
video_url = (cfg.get("video_url") or "").strip()
|
||||
max_plays = int(cfg.get("max_plays") or 0)
|
||||
reward_coin = int(cfg.get("reward_coin") or 0)
|
||||
used = used_plays(db, user_id)
|
||||
used = used_plays(db, user_id, scene)
|
||||
|
||||
def _miss(used_now: int) -> dict[str, Any]:
|
||||
return {
|
||||
@@ -194,7 +216,7 @@ def start_play(
|
||||
scene=scene,
|
||||
seq=seq,
|
||||
video_url=video_url,
|
||||
coin=0,
|
||||
coin=reward_coin,
|
||||
status="playing",
|
||||
completed=0,
|
||||
started_at=datetime.now(rewards.CN_TZ).replace(tzinfo=None),
|
||||
@@ -210,7 +232,7 @@ def start_play(
|
||||
db.flush()
|
||||
except IntegrityError:
|
||||
db.rollback()
|
||||
return _miss(used_plays(db, user_id))
|
||||
return _miss(used_plays(db, user_id, scene))
|
||||
return {
|
||||
"should_play": True,
|
||||
"video_url": video_url,
|
||||
@@ -241,7 +263,8 @@ def grant_play(
|
||||
"""
|
||||
token = (play_token or "").strip()
|
||||
# 金币额度以**服务端配置**为准,不信客户端(客户端只上报"播完/关闭")。
|
||||
coin = int(get_config(db).get("reward_coin") or 0)
|
||||
play = _find_play(db, user_id, token)
|
||||
coin = int(play.coin if play is not None else 0)
|
||||
|
||||
# 幂等核心:把 status 放进 WHERE 做条件更新(compare-and-set),而不是"先读再判再写"。
|
||||
# 「播完」与「✕ 关闭」抢跑、或客户端超时重试时,两个请求会都读到 status='playing',
|
||||
|
||||
@@ -93,6 +93,11 @@ def get_user_by_id(db: Session, user_id: int) -> User | None:
|
||||
return db.get(User, user_id)
|
||||
|
||||
|
||||
def user_exists(db: Session, user_id: int) -> bool:
|
||||
"""只查主键判断用户是否存在,避免只读统计接口依赖完整用户表结构。"""
|
||||
return db.scalar(select(User.id).where(User.id == user_id)) is not None
|
||||
|
||||
|
||||
def get_user_by_phone(db: Session, phone: str) -> User | None:
|
||||
stmt = select(User).where(User.phone == phone)
|
||||
return db.execute(stmt).scalar_one_or_none()
|
||||
|
||||
@@ -1,13 +1,15 @@
|
||||
"""新手引导视频(领券等候浮层前 N 次替代广告)的客户端请求/响应契约。"""
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Literal
|
||||
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
|
||||
class GuideVideoStartIn(BaseModel):
|
||||
"""开播询问。scene 目前只有 coupon(领券浮层);预留给日后比价等场景。"""
|
||||
|
||||
scene: str = Field(default="coupon", max_length=16)
|
||||
scene: Literal["coupon", "comparison"] = "coupon"
|
||||
|
||||
|
||||
class GuideVideoStartOut(BaseModel):
|
||||
|
||||
@@ -175,22 +175,22 @@ def seed(db) -> list[Feedback]:
|
||||
# 普通反馈 · 新端(带环境快照)· 无图
|
||||
fb("13255550001", "签到金币到账有时候会延迟一两分钟,能不能做成实时到账?",
|
||||
source="profile",
|
||||
app_version="2.3.1", device_model="PJA110", rom_name="ColorOS 14", android_version="14",
|
||||
app_version="2.3.1", device_model="PJA110", rom_name="ColorOS", android_version="14",
|
||||
created=ago(minutes=6)),
|
||||
# 比价反馈 · scene=优惠不对 · 新端 · 2 图
|
||||
fb("13255550002", "这家店京东外卖的到手价比你们算出来的最低价还低,截图为证,麻烦核实。",
|
||||
source="comparison", scene="优惠不对", images=[imgs[0], imgs[1]],
|
||||
app_version="2.3.1", device_model="M2012K11AC", rom_name="MIUI 14", android_version="13",
|
||||
app_version="2.3.1", device_model="M2012K11AC", rom_name="MIUI", android_version="13",
|
||||
created=ago(minutes=22)),
|
||||
# 比价反馈 · scene=找错商品 · 新端 · 无图
|
||||
fb("13255550002", "比价结果里的商品跟我搜的不是同一个规格,数量对不上。",
|
||||
source="comparison", scene="找错商品",
|
||||
app_version="2.3.0", device_model="V2309A", rom_name="OriginOS 4", android_version="14",
|
||||
app_version="2.3.0", device_model="V2309A", rom_name="OriginOS", android_version="14",
|
||||
created=ago(hours=1)),
|
||||
# 普通反馈 · 新端 · 1 图(表扬 + 小问题)
|
||||
fb("13255550003", "提现秒到账,好评!顺手反馈个小 bug:金币记录页偶尔白屏,要退出去重进。",
|
||||
source="profile", images=[imgs[2]],
|
||||
app_version="2.3.1", device_model="23078RKD5C", rom_name="MIUI 14", android_version="14",
|
||||
app_version="2.3.1", device_model="23078RKD5C", rom_name="MIUI", android_version="14",
|
||||
created=ago(hours=3)),
|
||||
# 比价反馈 · scene=比价太慢 · 历史数据(env 全 NULL、contact 有值)
|
||||
fb("13255550004", "比价转圈太久了,经常要等十几秒才出结果,体验不太好。",
|
||||
@@ -206,7 +206,7 @@ def seed(db) -> list[Feedback]:
|
||||
source="profile", images=[imgs[3]],
|
||||
status="adopted", reward_coins=2000,
|
||||
review_note="有效产品建议,已排期到 2.4.0", admin_reply="感谢反馈!该功能已在规划中,金币奖励已发放~",
|
||||
app_version="2.2.8", device_model="PJA110", rom_name="ColorOS 13", android_version="13",
|
||||
app_version="2.2.8", device_model="PJA110", rom_name="ColorOS", android_version="13",
|
||||
created=ago(days=2)),
|
||||
|
||||
# ===== 未采纳 rejected(带原因 + 回复)=====
|
||||
@@ -214,7 +214,7 @@ def seed(db) -> list[Feedback]:
|
||||
source="comparison", scene="价格不准",
|
||||
status="rejected", reject_reason="截图价格为限时活动价且已过期,不满足「长期可复现更低价」条件,暂不采纳。",
|
||||
admin_reply="感谢参与,本次未通过,欢迎继续上报有效更低价~",
|
||||
app_version="2.3.0", device_model="M2012K11AC", rom_name="MIUI 14", android_version="13",
|
||||
app_version="2.3.0", device_model="M2012K11AC", rom_name="MIUI", android_version="13",
|
||||
created=ago(days=3)),
|
||||
]
|
||||
db.add_all(feedbacks)
|
||||
|
||||
@@ -8,6 +8,7 @@ from sqlalchemy import delete
|
||||
from app.admin.repositories import ad_revenue
|
||||
from app.db.session import SessionLocal
|
||||
from app.models.ad_ecpm import AdEcpmRecord
|
||||
from app.models.ad_feed_reward import AdFeedRewardRecord
|
||||
from app.models.ad_pangle_revenue import AdPangleDailyRevenue
|
||||
from app.models.ad_reward import AdRewardRecord
|
||||
from app.models.user import User
|
||||
@@ -208,3 +209,112 @@ def test_reward_video_incomplete_playback_has_zero_revenue() -> None:
|
||||
db.execute(delete(User).where(User.phone == phone))
|
||||
db.commit()
|
||||
db.close()
|
||||
|
||||
|
||||
def test_feed_reward_details_keep_each_record_adn() -> None:
|
||||
db = SessionLocal()
|
||||
phone = "18800009994"
|
||||
detail_date = "2040-02-06"
|
||||
try:
|
||||
user = User(phone=phone, username="29999999994", register_channel="sms")
|
||||
db.add(user)
|
||||
db.flush()
|
||||
db.add_all([
|
||||
AdFeedRewardRecord(
|
||||
client_event_id="detail-adn-pangle", user_id=user.id,
|
||||
reward_date=detail_date, duration_seconds=20, unit_count=1,
|
||||
ecpm_raw="12000", adn="pangle", slot_id="rit-pangle",
|
||||
ad_type="draw", feed_scene="coupon", trace_id="detail-adn-trace",
|
||||
app_env="prod", our_code_id="104098712", coin=12, status="granted",
|
||||
created_at=datetime(2040, 2, 6, 1, tzinfo=UTC),
|
||||
),
|
||||
AdFeedRewardRecord(
|
||||
client_event_id="detail-adn-gdt", user_id=user.id,
|
||||
reward_date=detail_date, duration_seconds=20, unit_count=1,
|
||||
ecpm_raw="25000", adn="gdt", slot_id="rit-gdt",
|
||||
ad_type="draw", feed_scene="coupon", trace_id="detail-adn-trace",
|
||||
app_env="prod", our_code_id="104098712", coin=25, status="granted",
|
||||
created_at=datetime(2040, 2, 6, 2, tzinfo=UTC),
|
||||
),
|
||||
])
|
||||
db.commit()
|
||||
|
||||
result = ad_revenue.ad_revenue_report(
|
||||
db, date_from=detail_date, date_to=detail_date,
|
||||
user_id=user.id, app_env="prod", revenue_scope="all",
|
||||
)
|
||||
|
||||
item = next(row for row in result["items"] if row["event_key"].startswith("feedgrp-"))
|
||||
assert [detail["adn"] for detail in item["sub_rewards"]] == ["pangle", "gdt"]
|
||||
assert [detail["slot_id"] for detail in item["sub_rewards"]] == ["rit-pangle", "rit-gdt"]
|
||||
finally:
|
||||
db.rollback()
|
||||
db.execute(delete(AdFeedRewardRecord).where(AdFeedRewardRecord.reward_date == detail_date))
|
||||
db.execute(delete(User).where(User.phone == phone))
|
||||
db.commit()
|
||||
db.close()
|
||||
|
||||
|
||||
def test_feed_reward_source_fallback_requires_unique_trace_and_ecpm() -> None:
|
||||
db = SessionLocal()
|
||||
phone = "18800009995"
|
||||
fallback_date = "2040-02-07"
|
||||
try:
|
||||
user = User(phone=phone, username="29999999995", register_channel="sms")
|
||||
db.add(user)
|
||||
db.flush()
|
||||
db.add_all([
|
||||
AdFeedRewardRecord(
|
||||
client_event_id="source-fallback-unique", user_id=user.id,
|
||||
reward_date=fallback_date, duration_seconds=3, unit_count=0,
|
||||
ecpm_raw="4700", ad_session_id="flow-session", trace_id="source-trace",
|
||||
app_env="prod", our_code_id="104098712", coin=0, status="too_short",
|
||||
ad_type="draw", feed_scene="comparison",
|
||||
created_at=datetime(2040, 2, 7, 1, tzinfo=UTC),
|
||||
),
|
||||
AdFeedRewardRecord(
|
||||
client_event_id="source-fallback-ambiguous", user_id=user.id,
|
||||
reward_date=fallback_date, duration_seconds=3, unit_count=0,
|
||||
ecpm_raw="4800", ad_session_id="flow-session", trace_id="source-trace",
|
||||
app_env="prod", our_code_id="104098712", coin=0, status="too_short",
|
||||
ad_type="draw", feed_scene="comparison",
|
||||
created_at=datetime(2040, 2, 7, 2, tzinfo=UTC),
|
||||
),
|
||||
AdEcpmRecord(
|
||||
user_id=user.id, ad_type="draw", ad_session_id="impression-unique",
|
||||
trace_id="source-trace", ecpm_raw="4700", adn="baidu", slot_id="rit-baidu",
|
||||
app_env="prod", our_code_id="104098712", report_date=fallback_date,
|
||||
created_at=datetime(2040, 2, 7, 1, tzinfo=UTC),
|
||||
),
|
||||
AdEcpmRecord(
|
||||
user_id=user.id, ad_type="draw", ad_session_id="impression-ambiguous-a",
|
||||
trace_id="source-trace", ecpm_raw="4800", adn="baidu", slot_id="rit-baidu",
|
||||
app_env="prod", our_code_id="104098712", report_date=fallback_date,
|
||||
created_at=datetime(2040, 2, 7, 2, tzinfo=UTC),
|
||||
),
|
||||
AdEcpmRecord(
|
||||
user_id=user.id, ad_type="draw", ad_session_id="impression-ambiguous-b",
|
||||
trace_id="source-trace", ecpm_raw="4800", adn="ks", slot_id="rit-ks",
|
||||
app_env="prod", our_code_id="104098712", report_date=fallback_date,
|
||||
created_at=datetime(2040, 2, 7, 2, 1, tzinfo=UTC),
|
||||
),
|
||||
])
|
||||
db.commit()
|
||||
|
||||
result = ad_revenue.ad_revenue_report(
|
||||
db, date_from=fallback_date, date_to=fallback_date,
|
||||
user_id=user.id, app_env="prod", revenue_scope="all",
|
||||
)
|
||||
item = next(row for row in result["items"] if row["event_key"].startswith("feedgrp-"))
|
||||
details = {detail["ecpm"]: detail for detail in item["sub_rewards"]}
|
||||
assert details["4700"]["adn"] == "baidu"
|
||||
assert details["4700"]["slot_id"] == "rit-baidu"
|
||||
assert details["4800"]["adn"] is None
|
||||
assert details["4800"]["slot_id"] is None
|
||||
finally:
|
||||
db.rollback()
|
||||
db.execute(delete(AdFeedRewardRecord).where(AdFeedRewardRecord.reward_date == fallback_date))
|
||||
db.execute(delete(AdEcpmRecord).where(AdEcpmRecord.report_date == fallback_date))
|
||||
db.execute(delete(User).where(User.phone == phone))
|
||||
db.commit()
|
||||
db.close()
|
||||
|
||||
@@ -1,71 +0,0 @@
|
||||
from datetime import datetime, timedelta, timezone
|
||||
|
||||
from app.admin.repositories import queries
|
||||
from app.db.session import SessionLocal
|
||||
from app.models.comparison import ComparisonRecord
|
||||
from app.models.feedback import Feedback
|
||||
from app.repositories import user as user_repo
|
||||
|
||||
|
||||
def test_feedback_device_details_use_same_user_and_model() -> None:
|
||||
with SessionLocal() as db:
|
||||
submitted_at = datetime.now(timezone.utc)
|
||||
user = user_repo.upsert_user_for_login(
|
||||
db,
|
||||
phone="13800009876",
|
||||
register_channel="sms",
|
||||
)
|
||||
db.add_all(
|
||||
[
|
||||
ComparisonRecord(
|
||||
user_id=user.id,
|
||||
trace_id="feedback-device-v2166ba",
|
||||
device_model="V2166BA",
|
||||
device_manufacturer="vivo",
|
||||
rom_name="OriginOS",
|
||||
rom_version=13,
|
||||
created_at=submitted_at - timedelta(minutes=5),
|
||||
),
|
||||
ComparisonRecord(
|
||||
user_id=user.id,
|
||||
trace_id="feedback-device-other",
|
||||
device_model="OTHER-CODE",
|
||||
device_manufacturer="Other",
|
||||
rom_name="OtherOS",
|
||||
rom_version=99,
|
||||
created_at=submitted_at - timedelta(minutes=5),
|
||||
),
|
||||
ComparisonRecord(
|
||||
user_id=user.id,
|
||||
trace_id="feedback-device-future-upgrade",
|
||||
device_model="V2166BA",
|
||||
device_manufacturer="vivo-new",
|
||||
rom_name="OriginOS",
|
||||
rom_version=99,
|
||||
created_at=submitted_at + timedelta(minutes=5),
|
||||
),
|
||||
Feedback(
|
||||
user_id=user.id,
|
||||
content="设备信息补全测试",
|
||||
contact="",
|
||||
status="pending",
|
||||
device_model="V2166BA",
|
||||
rom_name="OriginOS",
|
||||
android_version="13",
|
||||
created_at=submitted_at,
|
||||
),
|
||||
]
|
||||
)
|
||||
db.commit()
|
||||
|
||||
items, _next_cursor, total = queries.list_feedbacks(
|
||||
db,
|
||||
user_id=user.id,
|
||||
limit=20,
|
||||
)
|
||||
|
||||
assert total == 1
|
||||
assert items[0].device_model == "V2166BA"
|
||||
assert items[0].device_model_name == "vivo Y77e"
|
||||
assert items[0].device_manufacturer == "vivo"
|
||||
assert items[0].rom_version == 13
|
||||
@@ -1,7 +1,7 @@
|
||||
"""Admin M2 读接口测试:大盘聚合 + 用户/流水/提现/反馈列表 + 鉴权拦截。"""
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime
|
||||
from datetime import UTC, datetime
|
||||
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
@@ -9,6 +9,7 @@ from sqlalchemy import event
|
||||
|
||||
from app.admin.main import admin_app
|
||||
from app.admin.repositories import admin_user as admin_repo
|
||||
from app.admin.repositories import queries
|
||||
from app.db.session import SessionLocal, engine
|
||||
from app.models.comparison import ComparisonRecord
|
||||
from app.models.feedback import Feedback
|
||||
@@ -144,6 +145,8 @@ def test_user_reward_detail_does_not_select_unrelated_new_ad_columns(
|
||||
) -> None:
|
||||
if "ad_reward_record.boost_round_id" in statement:
|
||||
raise AssertionError("提现详情不应查询未使用的 boost_round_id")
|
||||
if "FROM user" in statement and "user.phone" in statement:
|
||||
raise AssertionError("奖励统计的用户存在性检查不应展开完整 user 表")
|
||||
|
||||
event.listen(engine, "before_cursor_execute", reject_full_ad_reward_projection)
|
||||
try:
|
||||
@@ -162,6 +165,17 @@ def test_user_reward_detail_does_not_select_unrelated_new_ad_columns(
|
||||
assert records.status_code == 200, records.text
|
||||
|
||||
|
||||
def test_user_coin_record_sort_accepts_mixed_timezone_datetimes() -> None:
|
||||
"""线上 PostgreSQL 返回 aware,SQLite/历史转换可能返回 naive,二者必须可混排。"""
|
||||
naive = datetime(2038, 1, 1, 8, 0)
|
||||
aware = datetime(2038, 1, 1, 7, 0, tzinfo=UTC)
|
||||
|
||||
rows = [{"created_at": aware}, {"created_at": naive}]
|
||||
rows.sort(key=queries._coin_record_sort_key, reverse=True)
|
||||
|
||||
assert rows == [{"created_at": naive}, {"created_at": aware}]
|
||||
|
||||
|
||||
def test_user_filter_by_status(admin_client: TestClient, admin_token: str) -> None:
|
||||
_seed_user_with_data("13800000003")
|
||||
r = admin_client.get("/admin/api/users", params={"status": "active"}, headers=_auth(admin_token))
|
||||
|
||||
@@ -0,0 +1,102 @@
|
||||
"""_derive_from_results / _derive: 缺菜(漏菜)店总价虚低, 不当记录级"最低价"。
|
||||
|
||||
回归: pricebot comparison_results[].rank 是纯价格排序(含缺菜), server 派生 best 若照单全收,
|
||||
会把缺菜店的虚低价当 best_price → 记录页戴"最低"红框 + 虚假省额。
|
||||
修复: 派生 best 时按 platform_results[pid].skipped_dish_count 排除缺菜店(源平台永远全菜, 仍可当 best)。
|
||||
纯函数, 不碰 DB。
|
||||
"""
|
||||
from app.repositories.comparison import _derive, _derive_from_results
|
||||
from app.schemas.compare_record import ComparisonRecordIn, ComparisonResultIn
|
||||
|
||||
|
||||
def test_derive_from_results_excludes_short_ordered_from_best():
|
||||
# jd 缺 2 道菜 → 虚低 ¥25(rank=1); tb 全有 ¥38.5; 源美团 ¥42。best 应是 tb(干净最便宜), 不是 jd。
|
||||
results = [
|
||||
{"platform_id": "meituan", "platform_name": "美团", "price": 42.0, "is_source": True, "rank": 3},
|
||||
{"platform_id": "jd_waimai", "platform_name": "京东外卖", "price": 25.0, "is_source": False, "rank": 1},
|
||||
{"platform_id": "taobao_flash", "platform_name": "淘宝闪购", "price": 38.5, "is_source": False, "rank": 2},
|
||||
]
|
||||
platform_results = {
|
||||
"jd_waimai": {"skipped_dish_count": 2},
|
||||
"taobao_flash": {"skipped_dish_count": 0},
|
||||
"meituan": {"is_source": True},
|
||||
}
|
||||
d = _derive_from_results(results, platform_results)
|
||||
assert d["best_platform_id"] == "taobao_flash"
|
||||
assert d["best_price_cents"] == 3850
|
||||
assert d["saved_amount_cents"] == 4200 - 3850 # 350, 用干净店算省额, 不是缺菜虚低价 42-25=17元
|
||||
assert d["is_source_best"] is False
|
||||
|
||||
|
||||
def test_derive_from_results_all_targets_short_falls_back_to_source():
|
||||
# 唯一比源便宜的都是缺菜 → 不crown缺菜店; 源全菜 → best=源, is_source_best, 不虚报省额。
|
||||
results = [
|
||||
{"platform_id": "meituan", "price": 42.0, "is_source": True, "rank": 2},
|
||||
{"platform_id": "jd_waimai", "price": 25.0, "is_source": False, "rank": 1},
|
||||
]
|
||||
platform_results = {"jd_waimai": {"skipped_dish_count": 3}}
|
||||
d = _derive_from_results(results, platform_results)
|
||||
assert d["best_platform_id"] == "meituan"
|
||||
assert d["is_source_best"] is True
|
||||
assert d["saved_amount_cents"] == 0
|
||||
|
||||
|
||||
def test_derive_from_results_no_platform_results_keeps_old_behavior():
|
||||
# 不传 platform_results(老 harvest / 无缺菜信息)→ 行为不变: 纯 rank/price 选 best。
|
||||
results = [
|
||||
{"platform_id": "meituan", "price": 42.0, "is_source": True, "rank": 2},
|
||||
{"platform_id": "jd_waimai", "price": 25.0, "is_source": False, "rank": 1},
|
||||
]
|
||||
d = _derive_from_results(results)
|
||||
assert d["best_platform_id"] == "jd_waimai"
|
||||
assert d["best_price_cents"] == 2500
|
||||
|
||||
|
||||
def test_derive_from_results_malformed_platform_results_no_crash():
|
||||
# 内层值非 dict(异常/伪造上报)→ 不抛 AttributeError, 按"不缺菜"处理, 照常选最便宜。
|
||||
results = [
|
||||
{"platform_id": "meituan", "price": 42.0, "is_source": True, "rank": 2},
|
||||
{"platform_id": "jd_waimai", "price": 25.0, "is_source": False, "rank": 1},
|
||||
]
|
||||
d = _derive_from_results(results, {"jd_waimai": "oops"})
|
||||
assert d["best_platform_id"] == "jd_waimai"
|
||||
assert d["best_price_cents"] == 2500
|
||||
|
||||
|
||||
def test_derive_pydantic_excludes_short():
|
||||
# _derive(老客户端 POST 路径)同样排除缺菜店: jd 缺菜虚低 ¥25 不当 best, 取干净的淘宝 ¥38.5。
|
||||
payload = ComparisonRecordIn(
|
||||
trace_id="t-short-pyd",
|
||||
source_price=42.0,
|
||||
source_platform_id="meituan",
|
||||
comparison_results=[
|
||||
ComparisonResultIn(platform_id="meituan", platform_name="美团", price=42.0, is_source=True, rank=3),
|
||||
ComparisonResultIn(platform_id="jd_waimai", platform_name="京东外卖", price=25.0, is_source=False, rank=1),
|
||||
ComparisonResultIn(platform_id="taobao_flash", platform_name="淘宝闪购", price=38.5, is_source=False, rank=2),
|
||||
],
|
||||
platform_results={
|
||||
"jd_waimai": {"skipped_dish_count": 2},
|
||||
"taobao_flash": {"skipped_dish_count": 0},
|
||||
},
|
||||
)
|
||||
d = _derive(payload)
|
||||
assert d["best_platform_id"] == "taobao_flash"
|
||||
assert d["best_price_cents"] == 3850
|
||||
assert d["saved_amount_cents"] == 4200 - 3850
|
||||
assert d["is_source_best"] is False
|
||||
|
||||
|
||||
def test_derive_pydantic_malformed_platform_results_no_crash():
|
||||
# _derive 的 platform_results 来自老客户端透传(可伪造): 内层非 dict 不应打 500。
|
||||
payload = ComparisonRecordIn(
|
||||
trace_id="t-malformed-pyd",
|
||||
source_price=42.0,
|
||||
comparison_results=[
|
||||
ComparisonResultIn(platform_id="meituan", price=42.0, is_source=True, rank=2),
|
||||
ComparisonResultIn(platform_id="jd_waimai", price=25.0, is_source=False, rank=1),
|
||||
],
|
||||
platform_results={"jd_waimai": "oops"},
|
||||
)
|
||||
d = _derive(payload) # 不抛 AttributeError
|
||||
assert d["best_platform_id"] == "jd_waimai"
|
||||
assert d["best_price_cents"] == 2500
|
||||
@@ -201,7 +201,7 @@ def test_start_loses_seq_race_degrades_to_ad(client, guide_configured, monkeypat
|
||||
assert first["should_play"] is True and first["seq"] == 1
|
||||
|
||||
# 本次请求读到的是过期计数 → 仍会算出 seq=1
|
||||
monkeypatch.setattr(crud_guide, "used_plays", lambda db, user_id: 0)
|
||||
monkeypatch.setattr(crud_guide, "used_plays", lambda db, user_id, scene="coupon": 0)
|
||||
with SessionLocal() as db:
|
||||
result = crud_guide.start_play(db, uid)
|
||||
|
||||
|
||||
Reference in New Issue
Block a user