465 lines
19 KiB
Python
465 lines
19 KiB
Python
"""比价记录 CRUD:上报 upsert(按 user_id+trace_id 幂等) + 派生字段 + 明细分页。
|
||
|
||
派生逻辑:best_* / saved_amount_cents / is_source_best 全部从 comparison_results 算出
|
||
(协议保证已按 price 升序、rank=1 最便宜),客户端不用自己算、也不可信它算。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
from datetime import datetime
|
||
|
||
from sqlalchemy import func, select
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.core.rewards import CN_TZ
|
||
from app.models.ad_feed_reward import AdFeedRewardRecord
|
||
from app.models.comparison import ComparisonRecord
|
||
from app.models.savings import SavingsRecord
|
||
from app.schemas.compare_record import ComparisonRecordIn
|
||
|
||
|
||
def _yuan_to_cents(yuan: float | None) -> int | None:
|
||
"""元(float)→ 分(int)。None 透传。"""
|
||
if yuan is None:
|
||
return None
|
||
return round(yuan * 100)
|
||
|
||
|
||
def _derive(payload: ComparisonRecordIn) -> dict:
|
||
"""从上报 payload 派生结构化列(best/saved/is_source_best/status)。"""
|
||
results = payload.comparison_results
|
||
|
||
# 最优 = rank 最小的一条;协议已升序,但不信顺序,显式按 rank/price 兜底取最小价。
|
||
best = None
|
||
priced = [r for r in results if r.price is not None]
|
||
if priced:
|
||
best = min(
|
||
priced,
|
||
key=lambda r: (r.rank if r.rank is not None else 10**9, r.price),
|
||
)
|
||
|
||
source_price_cents = _yuan_to_cents(payload.source_price)
|
||
if source_price_cents is None:
|
||
# 源价没单独给,从 comparison_results 里的 is_source 行兜底
|
||
src_row = next((r for r in results if r.is_source and r.price is not None), None)
|
||
if src_row is not None:
|
||
source_price_cents = _yuan_to_cents(src_row.price)
|
||
|
||
best_price_cents = _yuan_to_cents(best.price) if best else None
|
||
|
||
saved_amount_cents = None
|
||
if source_price_cents is not None and best_price_cents is not None:
|
||
saved_amount_cents = source_price_cents - best_price_cents
|
||
|
||
is_source_best = best.is_source if best is not None else None
|
||
|
||
# status:客户端显式给了就用;否则有"非源且有价"的结果=success,否则 failed
|
||
status = payload.status
|
||
if status is None:
|
||
has_valid_target = any(
|
||
(not r.is_source) and r.price is not None for r in results
|
||
)
|
||
status = "success" if has_valid_target else "failed"
|
||
|
||
return {
|
||
"source_price_cents": source_price_cents,
|
||
"best_platform_id": best.platform_id if best else None,
|
||
"best_platform_name": best.platform_name if best else None,
|
||
"best_price_cents": best_price_cents,
|
||
"saved_amount_cents": saved_amount_cents,
|
||
"is_source_best": is_source_best,
|
||
"status": status,
|
||
}
|
||
|
||
|
||
def upsert_record(
|
||
db: Session, *, user_id: int, payload: ComparisonRecordIn
|
||
) -> ComparisonRecord:
|
||
"""按 **trace_id** 幂等写入(唯一键已从 user_id+trace_id 改为 trace_id):已存在则合并
|
||
(回填 null user_id + 覆盖字段,但**不降级 success**),否则新建。
|
||
|
||
灰度期老客户端 POST /compare/record 走这条,与后端 harvest 按 trace_id reconcile;
|
||
新客户端不再 POST(改由 compare.py 透传壳 harvest 落库)。
|
||
"""
|
||
derived = _derive(payload)
|
||
fields = dict(
|
||
device_id=payload.device_id,
|
||
business_type=payload.business_type,
|
||
store_name=payload.store_name,
|
||
source_platform_id=payload.source_platform_id,
|
||
source_platform_name=payload.source_platform_name,
|
||
source_package=payload.source_package,
|
||
information=payload.information,
|
||
best_deeplink=payload.best_deeplink,
|
||
trace_url=payload.trace_url,
|
||
total_dish_count=payload.total_dish_count,
|
||
skipped_dish_count=payload.skipped_dish_count,
|
||
items=[it.model_dump(exclude_none=True) for it in payload.items],
|
||
comparison_results=[r.model_dump() for r in payload.comparison_results],
|
||
skipped_dish_names=list(payload.skipped_dish_names),
|
||
# 客户端环境 / 性能(debug,客户端上报;旧客户端为 None)
|
||
device_model=payload.device_model,
|
||
device_manufacturer=payload.device_manufacturer,
|
||
rom_vendor=payload.rom_vendor,
|
||
rom_name=payload.rom_name,
|
||
rom_version=payload.rom_version,
|
||
android_version=payload.android_version,
|
||
android_sdk=payload.android_sdk,
|
||
app_version=payload.app_version,
|
||
app_version_code=payload.app_version_code,
|
||
source_app_version=payload.source_app_version,
|
||
longitude=payload.longitude,
|
||
latitude=payload.latitude,
|
||
total_ms=payload.total_ms,
|
||
step_count=payload.step_count,
|
||
raw_payload=payload.model_dump(),
|
||
**derived,
|
||
)
|
||
|
||
# 按 trace_id 定位(唯一键已改 trace_id)：后端 harvest 可能已建行,这里的客户端上报
|
||
# (灰度期老客户端 / 新客户端不再走这条)与之 reconcile。
|
||
existing = db.execute(
|
||
select(ComparisonRecord).where(
|
||
ComparisonRecord.trace_id == payload.trace_id,
|
||
)
|
||
).scalar_one_or_none()
|
||
|
||
if existing is not None:
|
||
# 不降级:harvest 或更早上报已落成 success,收尾期 fromFailure 的 cancelled/failed
|
||
# 不许把它盖回去(老 comparisonReported bug 的服务端兜底)——只补 user_id / trace_url。
|
||
if existing.status == "success" and fields.get("status") != "success":
|
||
if existing.user_id is None and user_id is not None:
|
||
existing.user_id = user_id
|
||
if fields.get("trace_url"):
|
||
existing.trace_url = fields["trace_url"]
|
||
db.commit()
|
||
db.refresh(existing)
|
||
return existing
|
||
# 只**填**空缺 user_id、不 reassign:trace_id 全局唯一,一条 trace 只属一个用户;
|
||
# 绝不把已有归属的记录改判给另一个上报者(仅 harvest 建的 null-user 行在此绑上)。
|
||
if existing.user_id is None:
|
||
existing.user_id = user_id
|
||
for k, v in fields.items():
|
||
setattr(existing, k, v)
|
||
db.commit()
|
||
db.refresh(existing)
|
||
return existing
|
||
|
||
# created_at 显式存 naive 北京 wall-clock(同 savings_record):客户端「比价记录」页
|
||
# formatRecordTime 原样切片 created_at 字符串、不转时区,而默认 server_default=func.now()
|
||
# 在 SQLite 下返回 UTC → 直接慢 8h。覆盖分支不动 created_at(保留首次创建时间)。
|
||
rec = ComparisonRecord(
|
||
user_id=user_id,
|
||
trace_id=payload.trace_id,
|
||
created_at=datetime.now(CN_TZ).replace(tzinfo=None),
|
||
**fields,
|
||
)
|
||
db.add(rec)
|
||
db.commit()
|
||
db.refresh(rec)
|
||
return rec
|
||
|
||
|
||
# ============================================================
|
||
# 后端 harvest：app-server 从 pricebot 透传响应里直接落库(不靠客户端上报)。
|
||
# 帧0 建行(running) → 最终 done 更新(success/failed) → finalize 更新(aborted)。
|
||
# 全按 trace_id upsert；success 行永不被后到的 failed/aborted 降级。
|
||
# ============================================================
|
||
|
||
|
||
def _derive_from_results(results: list[dict]) -> dict:
|
||
"""从 done 帧 comparison_results(pricebot 原始 dict 列表)派生结构化列。
|
||
等价 _derive,但吃原始字段(is_source/price/rank/platform_id/store_name...)而非 pydantic 对象。"""
|
||
priced = [r for r in results if r.get("price") is not None]
|
||
best = None
|
||
if priced:
|
||
best = min(
|
||
priced,
|
||
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)
|
||
|
||
source_price_cents = None
|
||
if src_row is not None and src_row.get("price") is not None:
|
||
source_price_cents = _yuan_to_cents(src_row["price"])
|
||
best_price_cents = _yuan_to_cents(best["price"]) if best else None
|
||
saved_amount_cents = None
|
||
if source_price_cents is not None and best_price_cents is not None:
|
||
saved_amount_cents = source_price_cents - best_price_cents
|
||
|
||
has_valid_target = any(
|
||
(not r.get("is_source")) and r.get("price") is not None for r in results
|
||
)
|
||
return {
|
||
"source_platform_id": (src_row or {}).get("platform_id"),
|
||
"source_platform_name": (src_row or {}).get("platform_name"),
|
||
"source_package": (src_row or {}).get("package"),
|
||
"source_price_cents": source_price_cents,
|
||
"best_platform_id": best.get("platform_id") if best else None,
|
||
"best_platform_name": best.get("platform_name") if best else None,
|
||
"best_price_cents": best_price_cents,
|
||
"saved_amount_cents": saved_amount_cents,
|
||
"is_source_best": best.get("is_source") if best else None,
|
||
"store_name": (src_row or {}).get("store_name") or None,
|
||
"status": "success" if has_valid_target else "failed",
|
||
}
|
||
|
||
|
||
def _device_cols_from_info(device_info: dict | None) -> dict:
|
||
"""step 帧 device_info({locale,brand,model,android_version,rom_version})→ 表列。
|
||
step 帧只带这几项;rom_name / android_sdk / app_version / 耗时步数 harvest 拿不到 → 留 None
|
||
(灰度期老客户端 fromComparison 会补齐;要新客户端也全带需扩 collectDeviceInfo,后续)。"""
|
||
d = device_info or {}
|
||
rv = str(d.get("rom_version") or "")
|
||
return {
|
||
"device_model": d.get("model") or None,
|
||
"device_manufacturer": d.get("brand") or None,
|
||
"android_version": d.get("android_version") or None,
|
||
"rom_version": int(rv) if rv.isdigit() else None,
|
||
}
|
||
|
||
|
||
def _get_by_trace(db: Session, trace_id: str) -> ComparisonRecord | None:
|
||
return db.execute(
|
||
select(ComparisonRecord).where(ComparisonRecord.trace_id == trace_id)
|
||
).scalar_one_or_none()
|
||
|
||
|
||
def harvest_running(
|
||
db: Session,
|
||
*,
|
||
trace_id: str,
|
||
user_id: int | None,
|
||
business_type: str = "food",
|
||
device_id: str | None = None,
|
||
device_info: dict | None = None,
|
||
trace_url: str | None = None,
|
||
) -> ComparisonRecord:
|
||
"""帧0(或任一尚未建行的帧)建/补 running 行。幂等:已存在只补空缺(user_id/trace_url/
|
||
device_id/机型),绝不动已落定的 status / 结果。"""
|
||
rec = _get_by_trace(db, trace_id)
|
||
if rec is None:
|
||
rec = ComparisonRecord(
|
||
trace_id=trace_id,
|
||
user_id=user_id,
|
||
business_type=business_type or "food",
|
||
device_id=device_id,
|
||
status="running",
|
||
trace_url=trace_url,
|
||
created_at=datetime.now(CN_TZ).replace(tzinfo=None),
|
||
**_device_cols_from_info(device_info),
|
||
)
|
||
db.add(rec)
|
||
db.commit()
|
||
db.refresh(rec)
|
||
return rec
|
||
changed = False
|
||
if rec.user_id is None and user_id is not None:
|
||
rec.user_id = user_id
|
||
changed = True
|
||
if not rec.trace_url and trace_url:
|
||
rec.trace_url = trace_url
|
||
changed = True
|
||
if rec.device_id is None and device_id:
|
||
rec.device_id = device_id
|
||
changed = True
|
||
for k, v in _device_cols_from_info(device_info).items():
|
||
if getattr(rec, k) is None and v is not None:
|
||
setattr(rec, k, v)
|
||
changed = True
|
||
if changed:
|
||
db.commit()
|
||
db.refresh(rec)
|
||
return rec
|
||
|
||
|
||
def harvest_done(
|
||
db: Session,
|
||
*,
|
||
trace_id: str,
|
||
user_id: int | None,
|
||
done_params: dict,
|
||
business_type: str = "food",
|
||
device_id: str | None = None,
|
||
device_info: dict | None = None,
|
||
trace_url: str | None = None,
|
||
) -> tuple[ComparisonRecord, bool]:
|
||
"""最终 done 帧:running 行 → 终态(success/failed)+结果+trace_url。
|
||
返回 (记录, 是否本次**新**落成 success)——供调用方据此幂等发一次邀请奖。
|
||
行不存在(理论上帧0已建;防御)则新建。"""
|
||
results = done_params.get("comparison_results") or []
|
||
derived = _derive_from_results(results)
|
||
fields = dict(
|
||
business_type=business_type or "food",
|
||
information=done_params.get("information") or None,
|
||
# best_deeplink 来自客户端剪贴板采集,harvest 拿不到 → 留空(灰度期 fromComparison 会补;
|
||
# 纯 harvest 行「再次比价」退化为按 package 拉起 App。要精确深链需客户端另传,后续)。
|
||
trace_url=trace_url or done_params.get("trace_url"),
|
||
total_dish_count=done_params.get("total_dish_count"),
|
||
skipped_dish_count=done_params.get("skipped_dish_count"),
|
||
skipped_dish_names=list(done_params.get("skipped_dish_names") or []),
|
||
comparison_results=results,
|
||
# 菜品:pricebot 已把源单菜品塞进 comparison_results[源行].items
|
||
items=next((r.get("items") or [] for r in results if r.get("is_source")), []),
|
||
raw_payload=done_params,
|
||
**derived,
|
||
**_device_cols_from_info(device_info),
|
||
)
|
||
|
||
rec = _get_by_trace(db, trace_id)
|
||
was_success = rec is not None and rec.status == "success"
|
||
if rec is None:
|
||
rec = ComparisonRecord(
|
||
trace_id=trace_id,
|
||
user_id=user_id,
|
||
created_at=datetime.now(CN_TZ).replace(tzinfo=None),
|
||
device_id=device_id,
|
||
**fields,
|
||
)
|
||
db.add(rec)
|
||
else:
|
||
if user_id is not None and rec.user_id is None:
|
||
rec.user_id = user_id
|
||
if device_id and rec.device_id is None:
|
||
rec.device_id = device_id
|
||
for k, v in fields.items():
|
||
setattr(rec, k, v)
|
||
db.commit()
|
||
db.refresh(rec)
|
||
newly_success = (rec.status == "success") and not was_success
|
||
return rec, newly_success
|
||
|
||
|
||
def harvest_abort(
|
||
db: Session,
|
||
*,
|
||
trace_id: str,
|
||
status: str,
|
||
reason: str | None,
|
||
trace_url: str | None = None,
|
||
) -> ComparisonRecord | None:
|
||
"""finalize(用户终止/超时/异常,无 done 帧):running 行 → aborted/failed + trace_url。
|
||
**不降级 success**:行已 success(收尾取消那种 finalize 后到)只 refresh trace_url。
|
||
行不存在(极少:帧0没建成)→ 返回 None,不凭空造。"""
|
||
rec = _get_by_trace(db, trace_id)
|
||
if rec is None:
|
||
return None
|
||
if trace_url and not rec.trace_url:
|
||
rec.trace_url = trace_url
|
||
if rec.status != "success":
|
||
rec.status = status or "cancelled"
|
||
if reason:
|
||
rec.information = reason
|
||
db.commit()
|
||
db.refresh(rec)
|
||
return rec
|
||
|
||
|
||
def _ordered_shop_names(db: Session, user_id: int) -> set[str]:
|
||
"""该用户「真实下单」(source='compare')覆盖到的店名集合,用来给比价记录打「已下单」。
|
||
|
||
只认 compare(归因命中后真实上报),demo 演示数据不算。下单上报不带 trace_id,
|
||
只能按店名对齐——两边店名同源(都来自比价意图识别阶段的门店名 query),精确相等即视为同店。
|
||
语义=店级:同一家店比价过多次,这些记录会一并标「已下单」。
|
||
"""
|
||
rows = db.execute(
|
||
select(SavingsRecord.shop_name).where(
|
||
SavingsRecord.user_id == user_id,
|
||
SavingsRecord.source == "compare",
|
||
SavingsRecord.shop_name.is_not(None),
|
||
)
|
||
).scalars().all()
|
||
return {s for s in rows if s}
|
||
|
||
|
||
def _ad_coins_by_trace(db: Session, user_id: int, trace_ids: list[str]) -> dict[str, int]:
|
||
"""各 trace_id「看广告赚的金币」:按 trace_id 聚合该用户已发放(granted)的信息流广告金币。
|
||
|
||
广告金币来自比价等候期信息流(biz_type=feed_ad_reward);客户端结算时把本场 trace_id 带上,
|
||
落到 ad_feed_reward_record.trace_id。此处按 trace_id 求和 = 这次比价看广告实发的金币。
|
||
trace_id 仅比价场景客户端带(领券/福利/旧客户端为 NULL),故按 trace_id 过滤天然只算比价广告。
|
||
空集合直接返回(避免 IN () 非法)。
|
||
"""
|
||
if not trace_ids:
|
||
return {}
|
||
rows = db.execute(
|
||
select(
|
||
AdFeedRewardRecord.trace_id,
|
||
func.coalesce(func.sum(AdFeedRewardRecord.coin), 0),
|
||
).where(
|
||
AdFeedRewardRecord.user_id == user_id,
|
||
AdFeedRewardRecord.status == "granted",
|
||
AdFeedRewardRecord.trace_id.in_(trace_ids),
|
||
).group_by(AdFeedRewardRecord.trace_id)
|
||
).all()
|
||
return {tid: int(coin) for tid, coin in rows if tid}
|
||
|
||
|
||
def list_records(
|
||
db: Session,
|
||
user_id: int,
|
||
*,
|
||
limit: int = 20,
|
||
cursor: int | None = None,
|
||
) -> tuple[list[ComparisonRecord], int | None]:
|
||
"""比价记录分页(按创建时间倒序、id 兜底,游标式)。附「已下单」店级标记 + 「看广告赚的金币」(瞬态,不写库)。"""
|
||
stmt = select(ComparisonRecord).where(ComparisonRecord.user_id == user_id)
|
||
if cursor is not None:
|
||
stmt = stmt.where(ComparisonRecord.id < cursor)
|
||
stmt = stmt.order_by(ComparisonRecord.created_at.desc(), ComparisonRecord.id.desc()).limit(limit)
|
||
|
||
items = list(db.execute(stmt).scalars().all())
|
||
next_cursor = items[-1].id if len(items) == limit else None
|
||
|
||
# 「已下单」标记:本页记录的 store_name 若落在该用户真实下单的店名集合里即 True。
|
||
# ordered / ad_coins_earned 均非 ORM 列,仅挂实例上供 ComparisonRecordOut(from_attributes) 读出,不持久化。
|
||
ordered_shops = _ordered_shop_names(db, user_id)
|
||
# 「本次比价看广告赚的金币」:按本页 trace_id 一次性聚合(同 ordered 范式)。
|
||
ad_coins = _ad_coins_by_trace(db, user_id, [it.trace_id for it in items])
|
||
for it in items:
|
||
it.ordered = bool(it.store_name and it.store_name in ordered_shops)
|
||
it.ad_coins_earned = ad_coins.get(it.trace_id, 0)
|
||
|
||
return items, next_cursor
|
||
|
||
|
||
def count_success(db: Session, user_id: int) -> int:
|
||
"""该用户成功比价(status='success')的条数。比价战绩里程碑的解锁进度源。"""
|
||
return db.execute(
|
||
select(func.count(ComparisonRecord.id)).where(
|
||
ComparisonRecord.user_id == user_id,
|
||
ComparisonRecord.status == "success",
|
||
)
|
||
).scalar_one()
|
||
|
||
|
||
def get_stats(db: Session, user_id: int) -> tuple[int, int]:
|
||
"""「我的」页省钱战绩卡的**比价口径**聚合:(成功比价数, 累计发现可省额/分)。
|
||
|
||
- 完成比价 = status='success' 的比价记录数(同 count_success)。
|
||
- 累计发现可省 = 各成功比价 saved_amount_cents(源价−最低价)之和;null(没省/未识别)不计。
|
||
与「下单成交记账」savings_record 解耦:比过价就计,不要求下单(见 savings.py)。
|
||
"""
|
||
row = db.execute(
|
||
select(
|
||
func.count(ComparisonRecord.id),
|
||
func.coalesce(func.sum(ComparisonRecord.saved_amount_cents), 0),
|
||
).where(
|
||
ComparisonRecord.user_id == user_id,
|
||
ComparisonRecord.status == "success",
|
||
)
|
||
).one()
|
||
return int(row[0]), int(row[1])
|
||
|
||
|
||
def get_record(db: Session, user_id: int, record_id: int) -> ComparisonRecord | None:
|
||
"""取单条(限本人,避免越权读他人记录)。附「看广告赚的金币」瞬态属性(同 list_records)。"""
|
||
rec = db.execute(
|
||
select(ComparisonRecord).where(
|
||
ComparisonRecord.id == record_id,
|
||
ComparisonRecord.user_id == user_id,
|
||
)
|
||
).scalar_one_or_none()
|
||
if rec is not None:
|
||
rec.ad_coins_earned = _ad_coins_by_trace(db, user_id, [rec.trace_id]).get(rec.trace_id, 0)
|
||
return rec
|