shaguabijia-app-server/app/repositories/comparison.py

"""比价记录 CRUD:上报 upsert(按 user_id+trace_id 幂等) + 派生字段 + 明细分页。

派生逻辑:best_* / saved_amount_cents / is_source_best 全部从 comparison_results 算出
(协议保证已按 price 升序、rank=1 最便宜),客户端不用自己算、也不可信它算。
"""
from __future__ import annotations

from sqlalchemy import func, select
from sqlalchemy.orm import Session

from app.models.comparison import ComparisonRecord
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:
    """按 (user_id, trace_id) 幂等写入:已存在则覆盖(更完整的重试上报胜出),否则新建。"""
    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,
        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),
        raw_payload=payload.model_dump(),
        **derived,
    )

    existing = db.execute(
        select(ComparisonRecord).where(
            ComparisonRecord.user_id == user_id,
            ComparisonRecord.trace_id == payload.trace_id,
        )
    ).scalar_one_or_none()

    if existing is not None:
        for k, v in fields.items():
            setattr(existing, k, v)
        db.commit()
        db.refresh(existing)
        return existing

    rec = ComparisonRecord(user_id=user_id, trace_id=payload.trace_id, **fields)
    db.add(rec)
    db.commit()
    db.refresh(rec)
    return rec


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.id.desc()).limit(limit)

    items = list(db.execute(stmt).scalars().all())
    next_cursor = items[-1].id if len(items) == limit else None
    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_record(db: Session, user_id: int, record_id: int) -> ComparisonRecord | None:
    """取单条(限本人,避免越权读他人记录)。"""
    return db.execute(
        select(ComparisonRecord).where(
            ComparisonRecord.id == record_id,
            ComparisonRecord.user_id == user_id,
        )
    ).scalar_one_or_none()