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

"""价格观测落库:一次比价的整批观测幂等写入。

幂等键 = (trace_id, platform, scope)。pricebot 重试 / replay 会重复上报同一 trace,
这里先查该 trace 已存在的 (platform, scope),只插新的——跨方言(PG / SQLite dev)都安全,
不依赖 ON CONFLICT。极端并发下若仍撞唯一约束,IntegrityError 回滚后忽略(本批本就幂等)。
"""
from __future__ import annotations

import logging

from sqlalchemy import select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session

from app.models.price_observation import PriceObservation
from app.schemas.price_observation import PriceObservationBatchIn

logger = logging.getLogger("shagua.price_obs")


def insert_batch(db: Session, payload: PriceObservationBatchIn) -> tuple[int, int]:
    """写入一次比价的整批观测,返回 (inserted, skipped)。

    skipped = 因 (trace_id, platform, scope) 已存在而幂等跳过的条数。
    """
    if not payload.observations:
        return 0, 0

    # 该 trace 已有的 (platform, scope),用于幂等过滤
    existing: set[tuple[str, str]] = set(
        db.execute(
            select(PriceObservation.platform, PriceObservation.scope).where(
                PriceObservation.trace_id == payload.trace_id
            )
        ).all()
    )

    inserted = 0
    skipped = 0
    for obs in payload.observations:
        key = (obs.platform, obs.scope)
        if key in existing:
            skipped += 1
            continue
        existing.add(key)  # 同批内去重(防 pricebot 同帧给重复平台)
        db.add(
            PriceObservation(
                trace_id=payload.trace_id,
                business_type=payload.business_type,
                platform=obs.platform,
                platform_store_id=obs.platform_store_id,
                store_name=payload.store_name,
                city=payload.city,
                geohash=payload.geohash,
                lng=payload.lng,
                lat=payload.lat,
                is_source=obs.is_source,
                scope=obs.scope,
                price_cents=obs.price_cents,
                coupon_saved_cents=obs.coupon_saved_cents,
                coupon_name=obs.coupon_name,
                store_closed=obs.store_closed,
                rank=obs.rank,
                dishes=payload.dishes,
                attrs=obs.attrs,
                source_device_id=payload.source_device_id,
                source_user_id=payload.source_user_id,
            )
        )
        inserted += 1

    if inserted == 0:
        return 0, skipped

    try:
        db.commit()
    except IntegrityError:
        # 并发下另一个请求刚插了同 (trace,platform,scope) → 唯一约束撞了。
        # 本批本就幂等,回滚后当作全跳过(不重试,失败就失败,见 P0 约定)。
        db.rollback()
        logger.warning(
            "price_observation 并发幂等冲突 trace=%s,本批回滚跳过", payload.trace_id
        )
        return 0, skipped + inserted

    return inserted, skipped