shaguabijia-app-server/app/schemas/meituan.py

"""美团 CPS 券列表 / 换链相关 schemas。"""
from __future__ import annotations

from typing import Any, Literal

from pydantic import BaseModel, Field

# feed / 列表类接口的结果状态,供前端区分占位文案:
#   ok       = 正常有数据
#   empty    = 后端正常、查询成功但确实没有数据(显示「暂无优惠券」)
#   degraded = 上游(美团)或库查询失败,已降级返空(显示「服务繁忙,下拉重试」)
# 老客户端不读该字段,默认 ok,向后兼容。
FeedStatus = Literal["ok", "empty", "degraded"]


# ───────────────── 头图传输优化（统一缩放 + 转 WebP） ─────────────────
# 美团图片 CDN 支持在 URL 后拼 @<w>w_<h>h_1e_1c[.webp] 让服务端缩放/转码。feed 卡片只占
# 124dp(≈372px@3x),而库里 head_url 是原图(实测中位 137KB、长尾到 17MB、GIF 均 4.8MB),
# 是首页 feed 图片加载偏慢的来源。按产品要求,头图**统一**缩到 124dp + 转 WebP 再下发
# (全部压缩、不设大小阈值),由美团 CDN 服务端缩放/转码,实测省 76–99% 体积。
FEED_THUMB_PARAM = "@375w_375h_1e_1c.webp"          # 124dp@3x 缩略 + 转 WebP


def feed_image_url(head_url: str) -> str:
    """统一给头图 URL 拼缩放参数(缩到 124dp + 转 WebP),让美团 CDN 出小图;空 URL 原样返回。"""
    if not head_url:
        return head_url
    return head_url.split("@")[0] + FEED_THUMB_PARAM


# ───────────────── 券卡片（归一化后给客户端） ─────────────────

class CouponCard(BaseModel):
    product_view_sign: str
    platform: int = Field(..., description="1=外卖/到家, 2=到店")
    biz_line: int | None = Field(None, description="到店子类: 1到餐 2到综 3酒店 4门票")

    name: str
    head_image_url: str
    brand_name: str | None = None
    brand_logo_url: str | None = None

    sell_price: str
    original_price: str
    discount_amount: str = Field(..., description="优惠金额 = 原价 - 现价")

    commission_rate: str = Field(..., description="佣金比例, 如 1.4%")
    commission_amount: str | None = Field(None, description="预估佣金元")

    sale_volume: str | None = None
    poi_name: str | None = Field(None, description="最近门店名")
    distance_text: str | None = Field(None, description="格式化距离, 如 600m / 1.5km")
    distance_meters: float | None = Field(None, description="原始距离(米)")
    available_poi_num: int | None = Field(None, description="可用门店数")

    coupon_num: int | None = Field(None, description="券张数(外卖兑换券)")
    valid_days: int | None = Field(None, description="券有效天数")
    price_label: str | None = Field(None, description="如 15天低价")
    rank_label: str | None = Field(None, description="如 2小时北京外卖销量榜第1名")
    rating_label: str | None = Field(None, description="如 4.6分")

    @classmethod
    def from_raw(cls, item: dict[str, Any]) -> CouponCard:
        cpd = item.get("couponPackDetail") or {}
        brand = item.get("brandInfo") or {}
        comm = item.get("commissionInfo") or {}
        poi = item.get("deliverablePoiInfo") or {}
        avail = item.get("availablePoiInfo") or {}
        label = cpd.get("productLabel") or {}
        price_lbl = label.get("pricePowerLabel") or {}
        valid_info = item.get("couponValidTimeInfo") or {}

        platform = cpd.get("platform") or 1
        biz_line = cpd.get("bizLine")

        head_url = cpd.get("headUrl") or ""
        if "@" in head_url:
            head_url = head_url.split("@")[0]

        sell = cpd.get("sellPrice") or "0"
        orig = cpd.get("originalPrice") or "0"
        try:
            discount = f"{float(orig) - float(sell):.2f}"
        except (ValueError, TypeError):
            discount = "0"

        cp = comm.get("commissionPercent") or "0"
        try:
            rate = f"{float(cp) / 100:.1f}%"
        except (ValueError, TypeError):
            rate = "0%"

        raw_dist = poi.get("deliveryDistance")
        dist_meters: float | None = None
        dist_text: str | None = None
        if raw_dist not in (None, "null", ""):
            try:
                d = float(raw_dist)
                dist_meters = d * 1000 if platform == 1 else d
                if dist_meters < 1000:
                    dist_text = f"{int(dist_meters)}m"
                else:
                    dist_text = f"{dist_meters / 1000:.1f}km"
            except (ValueError, TypeError):
                pass

        return cls(
            product_view_sign=cpd.get("productViewSign") or cpd.get("skuViewId") or "",
            platform=platform,
            biz_line=biz_line,
            name=cpd.get("name") or "",
            head_image_url=feed_image_url(head_url),
            brand_name=brand.get("brandName"),
            brand_logo_url=brand.get("brandLogoUrl"),
            sell_price=sell,
            original_price=orig,
            discount_amount=discount,
            commission_rate=rate,
            commission_amount=comm.get("commission"),
            sale_volume=cpd.get("saleVolume"),
            poi_name=poi.get("poiName"),
            distance_text=dist_text,
            distance_meters=dist_meters,
            available_poi_num=avail.get("availablePoiNum"),
            coupon_num=cpd.get("couponNum"),
            valid_days=valid_info.get("couponValidDay"),
            price_label=price_lbl.get("historyPriceLabel") or price_lbl.get("beatMTLabel"),
            rank_label=label.get("productRankLabel"),
            rating_label=label.get("dianPingRankLabel"),
        )


# ───────────────── 券列表 请求 / 响应 ─────────────────

class CouponListRequest(BaseModel):
    longitude: float = Field(..., description="经度, 如 116.404")
    latitude: float = Field(..., description="纬度, 如 39.928")
    platform: int = Field(1, description="1=外卖/到家, 2=到店")
    biz_line: int | None = Field(None, description="到店子类: 1到餐 2到综 3酒店 4门票; 外卖不填")
    list_topic_id: int | None = Field(3, description="榜单: 1精选 2今日必推 3爆款筛选 5限时筛选")
    keyword: str | None = Field(None, description="搜索关键词(填了会忽略 list_topic_id)")
    sort_field: int | None = Field(None, description="1价格 2销量 6离我最近; 搜索默认6")
    search_id: str | None = Field(None, description="翻页token, 首页不填")
    page: int = Field(1, ge=1)
    page_size: int = Field(20, ge=1, le=20)


class CouponListResponse(BaseModel):
    items: list[CouponCard]
    has_next: bool = False
    search_id: str | None = None
    status: FeedStatus = Field("ok", description="ok 有数据 / empty 暂无 / degraded 上游失败已降级")


class FeedRequest(BaseModel):
    longitude: float = Field(..., description="经度")
    latitude: float = Field(..., description="纬度")
    page: int = Field(1, ge=1)
    page_size: int = Field(20, ge=1, le=20)
    # 筛选/排序口径,后端据此处理后返回(前端不再自己筛/排):
    #  rec      = 智能推荐:榜单混合 feed 去掉佣金率 < 3%(分页)
    #  distance = 距离最近:拉齐全部轮次后全局按距离由近及远(一次性返回, has_next=False)
    #  留空/其它 = 原混合 feed 不筛(老客户端兼容,新 app 会显式传 tab)
    # (销量最高 sales 走 /coupons 同城热销,不在本接口)
    tab: str = Field("", description="rec 智能推荐 / distance 距离最近 / 空=不筛(兼容)")


class FeedResponse(BaseModel):
    items: list[CouponCard]
    has_next: bool = False
    page: int = 1
    status: FeedStatus = Field("ok", description="ok 有数据 / empty 暂无 / degraded 上游失败已降级")


class TopSalesRequest(BaseModel):
    """销量最高 tab:从离线库 meituan_coupon 按销量降序取(不实时打美团)。"""
    # 可选:老客户端(本次改动前发版)不带经纬度。缺省时后端降级返空(status=degraded),
    # 不做 422 硬拒,也不误返"全城"结果。新客户端会传坐标 → 按城市过滤。
    longitude: float | None = Field(None, description="经度(用于定位城市;缺省=老客户端,降级返空)")
    latitude: float | None = Field(None, description="纬度(用于定位城市;缺省=老客户端,降级返空)")
    page: int = Field(1, ge=1)
    page_size: int = Field(20, ge=1, le=50)
    platform: int | None = Field(None, description="可选: 1只外卖 / 2只到店; 不填=全部(同城销量)")


# ───────────────── 换链 请求 / 响应 ─────────────────

class ReferralLinkRequest(BaseModel):
    product_view_sign: str = Field(..., min_length=1)
    platform: int = Field(1, description="1=外卖/到家, 2=到店")
    biz_line: int | None = None
    sid: str | None = Field(None, description="渠道追踪标识, 不填用默认")
    link_type_list: list[int] | None = Field(None, description="1=H5长链 2=H5短链 3=deeplink; 不填默认[1,3]")


class ReferralLinkResponse(BaseModel):
    link: str = Field(..., description="默认推广链接")
    link_map: dict[str, str] = Field(default_factory=dict, description="各类型链接 {linkType: url}")