"""美团 CPS 券列表 / 换链相关 schemas。""" from __future__ import annotations from typing import Any from pydantic import BaseModel, Field # ───────────────── 券卡片(归一化后给客户端) ───────────────── 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=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 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) class FeedResponse(BaseModel): items: list[CouponCard] has_next: bool = False page: int = 1 # ───────────────── 换链 请求 / 响应 ───────────────── 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}")