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

"""价格观测表(price_observation)—— 比价沉淀的价格资产层。

每完成一次比价,pricebot 在 done 帧把"在某平台某店,这一篮菜/这单到手价多少钱"
按平台逐条算出来,server→server 内部上报落这里。**与登录无关、不依赖客户端上报**——
比价透传链路当前不鉴权,只要比价跑到 done 就记,匿名用户也记(来源记 device_id,
user_id 客户端带上时一并记)。

与 comparison_record 的区别:
- comparison_record:用户视角的「我的比价记录」,客户端登录后主动上报,按 user_id 存。
- price_observation:**平台/门店视角的价格事实**,server 侧无条件沉淀,按 (平台,门店,菜,时间)
  组织,是未来"别人查过同店→直接秒回价格"大数据的源头。两表独立,互不影响。

「先存下来、用法后说」:结构化列给将来按 平台/门店/菜/时间/地理 查询聚合用;dishes / attrs
(JSONB)兜底存灵活明细,免得每多记一个字段就迁移 schema。raw trace(逐 step 全过程)
另存 pricebot 本地 jsonl.gz,本表只存提炼后的价格事实(本表可从 raw 重算)。
"""
from __future__ import annotations

from datetime import datetime

from sqlalchemy import (
    JSON,
    Boolean,
    DateTime,
    Float,
    Integer,
    String,
    UniqueConstraint,
    func,
)
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.orm import Mapped, mapped_column

from app.db.base import Base

# PG 上用 JSONB(可建 GIN 索引),SQLite(本地/测试)退化为通用 JSON(同 comparison_record)。
_JSON = JSON().with_variant(JSONB(), "postgresql")


class PriceObservation(Base):
    __tablename__ = "price_observation"
    __table_args__ = (
        # 一次比价(trace)里,同一平台、同一口径(scope)只记一条:pricebot 重试 / 客户端
        # replay 重复上报时幂等去重(insert-or-ignore)。一次外卖比价 = 源 + N 个目标平台
        # 各一条 scope='order'。
        UniqueConstraint(
            "trace_id", "platform", "scope",
            name="uq_price_obs_trace_platform_scope",
        ),
    )

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)

    # 观测时刻(≈ 比价 done 时;服务端落库时取 now)。按时间查"最近有效价"用,建索引。
    observed_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now(), index=True, nullable=False
    )
    # pricebot 侧 trace_id:回指 A 层原始 trace(能溯源 / 重算)+ 幂等去重键
    trace_id: Mapped[str] = mapped_column(String(64), index=True, nullable=False)
    # 业务类型:food(外卖,当前唯一接通)/ ecom(电商,二期)。预留扩展。
    business_type: Mapped[str] = mapped_column(
        String(16), nullable=False, default="food", index=True
    )

    # ===== 门店维度(先 denormalize 成文本,不建维度表;实体归一二期再做)=====
    platform: Mapped[str] = mapped_column(String(32), index=True, nullable=False)
    # 平台内门店 ID(抓得到就存,将来归一最稳的 key;现在多为空)
    platform_store_id: Mapped[str | None] = mapped_column(String(64), nullable=True)
    store_name: Mapped[str | None] = mapped_column(String(128), index=True, nullable=True)
    city: Mapped[str | None] = mapped_column(String(64), nullable=True)
    # 地理(现在拿不到 → 留空;二期把 device 经纬度透传进比价请求后回填)
    geohash: Mapped[str | None] = mapped_column(String(16), index=True, nullable=True)
    lng: Mapped[float | None] = mapped_column(Float, nullable=True)
    lat: Mapped[float | None] = mapped_column(Float, nullable=True)

    # ===== 价格事实 =====
    # 是否源平台(发起比价那家)。源平台价也是真实观测,照记。
    is_source: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
    # 口径:order(整单到手价,外卖当前唯一)/ dish(单菜单价,引擎暂不逐菜读,二期)
    scope: Mapped[str] = mapped_column(String(16), nullable=False, default="order")
    # 该口径的到手价(分);None = 该平台采集失败 / 门店打烊,无有效价
    price_cents: Mapped[int | None] = mapped_column(Integer, nullable=True)
    coupon_saved_cents: Mapped[int | None] = mapped_column(Integer, nullable=True)
    coupon_name: Mapped[str | None] = mapped_column(String(64), nullable=True)
    # 该平台目标店打烊原因(非空时 price_cents 为 None)
    store_closed: Mapped[str | None] = mapped_column(String(32), nullable=True)
    # 本次比价里的全局名次(1=最便宜;按到手价升序,源/目标统一排)
    rank: Mapped[int | None] = mapped_column(Integer, nullable=True)

    # ===== 明细(JSON)=====
    # 菜篮 [{name, qty, subtotal?}](外卖)。dish 级单价二期从这里 + 引擎增强推。
    dishes: Mapped[list | None] = mapped_column(_JSON, nullable=True)
    # 灵活字段兜底(配送费 / 起送 / 活动名 / 跳过菜数 等),免得加字段就迁移。
    attrs: Mapped[dict | None] = mapped_column(_JSON, nullable=True)

    # ===== 来源(溯源 / 去重 / 置信)=====
    source_device_id: Mapped[str | None] = mapped_column(String(64), index=True, nullable=True)
    # user_id 当前比价链路不鉴权拿不到,客户端带上时才有;先可空。
    source_user_id: Mapped[int | None] = mapped_column(Integer, index=True, nullable=True)
    confidence: Mapped[float] = mapped_column(Float, nullable=False, default=1.0)

    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now(), nullable=False
    )

    def __repr__(self) -> str:  # pragma: no cover
        return (
            f"<PriceObservation id={self.id} platform={self.platform} "
            f"store={self.store_name!r} price_cents={self.price_cents} "
            f"trace_id={self.trace_id}>"
        )