chenshuobo
900b64d4f9
把 main 的实时 feed 换成离线库版本 + status 降级兜底: - rec 智能推荐:改走离线库 meituan_coupon 筛佣金≥3%(SQL 去重+排序+分页),不再实时撞限流 - distance 距离最近:改实时搜索翻页(sortField=6),按用户真实位置由近及远排 - top_sales 销量最高:改 SQL DISTINCT ON 分页,修原全表拉取+全量解析的翻页卡顿 - 四路加 status(ok/empty/degraded):空库/上游失败均返 200 不再 5xx;/coupons 502 软化为降级 - FeedResponse/CouponListResponse 加 status 字段(默认 ok,向后兼容) 注意:替换 main #18 的实时 feed 实现,需 review 行为变更;依赖的离线库由 ETL 灌(见 meituan-etl PR)
298 lines
14 KiB
Python
298 lines
14 KiB
Python
"""美团 CPS 券列表 + 换链。
|
|
|
|
不需要登录，客户端传经纬度即可。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from concurrent.futures import ThreadPoolExecutor, as_completed
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException
|
|
from sqlalchemy import nullslast, select
|
|
from sqlalchemy.orm import Session, aliased
|
|
|
|
from app.core.config import settings
|
|
from app.db.session import get_db
|
|
from app.integrations.meituan import MeituanCpsError, _call, get_referral_link, query_coupon
|
|
from app.models.meituan_coupon import MeituanCoupon
|
|
from app.schemas.meituan import (
|
|
CouponCard,
|
|
CouponListRequest,
|
|
CouponListResponse,
|
|
FeedRequest,
|
|
FeedResponse,
|
|
ReferralLinkRequest,
|
|
ReferralLinkResponse,
|
|
TopSalesRequest,
|
|
)
|
|
|
|
logger = logging.getLogger("shagua.meituan")
|
|
|
|
router = APIRouter(prefix="/api/v1/meituan", tags=["meituan-cps"])
|
|
|
|
|
|
@router.post("/coupons", response_model=CouponListResponse, summary="券列表(为您推荐)")
|
|
def list_coupons(req: CouponListRequest) -> CouponListResponse:
|
|
if not settings.mt_cps_configured:
|
|
return CouponListResponse(items=[], has_next=False, search_id=None, status="degraded")
|
|
logger.info("[coupons] lon=%.6f lat=%.6f topic=%s", req.longitude, req.latitude, req.list_topic_id)
|
|
try:
|
|
raw = query_coupon(
|
|
longitude=req.longitude,
|
|
latitude=req.latitude,
|
|
platform=req.platform,
|
|
biz_line=req.biz_line,
|
|
list_topic_id=req.list_topic_id,
|
|
search_text=req.keyword,
|
|
search_id=req.search_id,
|
|
sort_field=req.sort_field,
|
|
page_no=req.page,
|
|
page_size=req.page_size,
|
|
)
|
|
except MeituanCpsError as e:
|
|
# 软降级:不再抛 502(避免前端弹报错 toast),返空 + degraded,前端显示「服务繁忙」。
|
|
logger.warning("[coupons] query_coupon 失败,降级返空: %s", e)
|
|
return CouponListResponse(items=[], has_next=False, search_id=None, status="degraded")
|
|
|
|
items: list[CouponCard] = []
|
|
for it in (raw.get("data") or []):
|
|
try:
|
|
items.append(CouponCard.from_raw(it))
|
|
except Exception: # noqa: BLE001
|
|
continue
|
|
return CouponListResponse(
|
|
items=items,
|
|
has_next=raw.get("hasNext", False),
|
|
search_id=raw.get("searchId"),
|
|
status="ok" if items else "empty",
|
|
)
|
|
|
|
|
|
_TOPIC_ROUNDS = [
|
|
(3, 3), # 爆款筛选
|
|
(2, 2), # 今日必推
|
|
(1, 5), # 精选 + 限时筛选
|
|
]
|
|
|
|
def _interleave(waimai: list[dict], daodian: list[dict]) -> list[CouponCard]:
|
|
items: list[CouponCard] = []
|
|
seen: set[str] = set()
|
|
i = j = 0
|
|
while i < len(waimai) or j < len(daodian):
|
|
for _ in range(2):
|
|
if i < len(waimai):
|
|
card = CouponCard.from_raw(waimai[i]); i += 1
|
|
if card.product_view_sign not in seen:
|
|
seen.add(card.product_view_sign); items.append(card)
|
|
if j < len(daodian):
|
|
card = CouponCard.from_raw(daodian[j]); j += 1
|
|
if card.product_view_sign not in seen:
|
|
seen.add(card.product_view_sign); items.append(card)
|
|
return items
|
|
|
|
|
|
def _commission_pct(card: CouponCard) -> float:
|
|
"""'1.4%' → 1.4;解析失败按 0(会被智能推荐过滤掉)。"""
|
|
try:
|
|
return float(card.commission_rate.rstrip("%"))
|
|
except (ValueError, AttributeError):
|
|
return 0.0
|
|
|
|
|
|
@router.post("/feed", response_model=FeedResponse, summary="混合feed(外卖+到店交叉);tab=rec智能推荐/distance距离最近")
|
|
def feed(req: FeedRequest, db: Session = Depends(get_db)) -> FeedResponse:
|
|
lon, lat = req.longitude, req.latitude
|
|
tab = (req.tab or "").strip()
|
|
logger.info("[feed] tab=%s page=%s lon=%.6f lat=%.6f", tab or "(default)", req.page, lon, lat)
|
|
|
|
# rec 是纯离线库查询,不依赖 MT 凭证、不实时打美团;其余 tab(distance / 默认)要实时打美团,
|
|
# 未配凭证直接降级返空(degraded),让前端区分「服务暂不可用」而非「暂无」。
|
|
if tab != "rec" and not settings.mt_cps_configured:
|
|
return FeedResponse(items=[], has_next=False, page=req.page, status="degraded")
|
|
|
|
def _fetch_topic(platform: int, biz_line: int | None, topic: int) -> tuple[list[dict], bool]:
|
|
"""返回(items, 是否调用失败)。失败标志用于把默认 feed 整页标 degraded。"""
|
|
try:
|
|
data = (query_coupon(
|
|
longitude=lon, latitude=lat,
|
|
platform=platform, biz_line=biz_line,
|
|
list_topic_id=topic, page_size=20,
|
|
).get("data") or [])
|
|
return data, False
|
|
except MeituanCpsError:
|
|
return [], True
|
|
|
|
# 距离最近:搜索召回(外卖搜"外卖" + 到店搜"美食",都 sortField=6 离我最近)一页页拉。
|
|
# 搜索翻页必须用 searchId(pageNo 翻不动),所以每个 feed 页顺序翻到第 N 页;两路并行、page 1 最快。
|
|
# 无状态、不改 APP(传页码即可);按你位置实时算距离(库里没存 POI 经纬度,只能实时)。
|
|
if tab == "distance":
|
|
lon_i, lat_i = int(lon * 1_000_000), int(lat * 1_000_000)
|
|
|
|
def _search_page_n(platform: int, biz_line: int | None, keyword: str, n: int) -> tuple[list[dict], bool, bool]:
|
|
"""顺序翻到第 n 页(搜索须 searchId 续页),返回(第 n 页 items, 是否还有下一页, 是否调用失败)。"""
|
|
sid: str | None = None
|
|
data: list[dict] = []
|
|
has_next = False
|
|
for pg in range(1, n + 1):
|
|
body: dict = {
|
|
"platform": platform, "searchText": keyword, "sortField": 6,
|
|
"longitude": lon_i, "latitude": lat_i, "pageSize": 20,
|
|
}
|
|
if biz_line is not None:
|
|
body["bizLine"] = biz_line
|
|
if sid:
|
|
body["searchId"] = sid
|
|
else:
|
|
body["pageNo"] = 1
|
|
try:
|
|
r = _call("/cps_open/common/api/v1/query_coupon", body)
|
|
except MeituanCpsError:
|
|
return [], False, True # 调用失败(用于标 degraded)
|
|
data = r.get("data") or []
|
|
sid = r.get("searchId")
|
|
has_next = bool(r.get("hasNext")) and bool(data)
|
|
if not data or (not has_next and pg < n):
|
|
return [], False, False # 没那么多页了(非错误)
|
|
return data, has_next, False
|
|
|
|
with ThreadPoolExecutor(max_workers=2) as pool:
|
|
f_wm = pool.submit(_search_page_n, 1, None, "外卖", req.page)
|
|
f_dd = pool.submit(_search_page_n, 2, 1, "美食", req.page)
|
|
(wm_data, wm_hn, wm_fail), (dd_data, dd_hn, dd_fail) = f_wm.result(), f_dd.result()
|
|
|
|
seen: set[str] = set()
|
|
cards: list[CouponCard] = []
|
|
for it in wm_data + dd_data:
|
|
try:
|
|
card = CouponCard.from_raw(it)
|
|
except Exception: # noqa: BLE001
|
|
continue
|
|
if card.product_view_sign and card.product_view_sign not in seen:
|
|
seen.add(card.product_view_sign)
|
|
cards.append(card)
|
|
cards.sort(key=lambda c: c.distance_meters if c.distance_meters is not None else float("inf"))
|
|
# 两路都失败且无数据 → degraded;否则有数据 ok / 无数据 empty
|
|
status = "degraded" if (not cards and wm_fail and dd_fail) else ("ok" if cards else "empty")
|
|
return FeedResponse(items=cards, has_next=wm_hn or dd_hn, page=req.page, status=status)
|
|
|
|
# 智能推荐(rec):走【离线库】筛佣金率 ≥ 3%,分页返回(SQL 侧去重+排序+分页,秒级、不打美团)。
|
|
# 实测库里佣金≥3% 去重后仅 ~578 条(几乎全是外卖;到店团购佣金普遍 <3%):实时按"同城热销榜单"
|
|
# 拉既撞限流、又填不满(该榜单中位佣金 ~0.8%,筛完每页剩 0-1 条),故从库出。佣金阈值逻辑不变。
|
|
if tab == "rec":
|
|
PAGE = 20
|
|
try:
|
|
base = select(MeituanCoupon).where(MeituanCoupon.commission_percent >= 3.0)
|
|
deduped = base.distinct(MeituanCoupon.dedup_key).order_by(
|
|
MeituanCoupon.dedup_key,
|
|
MeituanCoupon.commission_percent.desc(),
|
|
).subquery()
|
|
m = aliased(MeituanCoupon, deduped)
|
|
start = (req.page - 1) * PAGE
|
|
rows = db.execute(
|
|
select(m)
|
|
# 销量高的优先(无销量档排后),同档佣金高优先,id 兜底稳定分页
|
|
.order_by(nullslast(m.sale_volume_num.desc()), m.commission_percent.desc(), m.id)
|
|
.offset(start)
|
|
.limit(PAGE + 1)
|
|
).scalars().all()
|
|
except Exception: # noqa: BLE001
|
|
logger.exception("[feed] rec 库查询失败,降级返空")
|
|
return FeedResponse(items=[], has_next=False, page=req.page, status="degraded")
|
|
has_next = len(rows) > PAGE
|
|
cards: list[CouponCard] = []
|
|
for row in rows[:PAGE]:
|
|
try:
|
|
card = CouponCard.from_raw(row.raw or {})
|
|
except Exception: # noqa: BLE001
|
|
continue
|
|
if card.product_view_sign:
|
|
# 智能推荐不显示距离:库里的距离是相对城市默认点的(对用户无意义、且误导)。
|
|
# 置空后前端"距离 店名"那行只剩店名、自动顶到最左(店名移到原距离的位置)。
|
|
card.distance_text = None
|
|
card.distance_meters = None
|
|
cards.append(card)
|
|
return FeedResponse(items=cards, has_next=has_next, page=req.page,
|
|
status="ok" if cards else "empty")
|
|
|
|
# 默认(老客户端不传 tab):沿用逐轮分页的混合 feed,不筛。
|
|
page_idx = req.page - 1
|
|
if page_idx >= len(_TOPIC_ROUNDS):
|
|
return FeedResponse(items=[], has_next=False, page=req.page, status="empty")
|
|
|
|
wm_topic, dd_topic = _TOPIC_ROUNDS[page_idx]
|
|
with ThreadPoolExecutor(max_workers=2) as pool:
|
|
f_wm = pool.submit(_fetch_topic, 1, None, wm_topic)
|
|
f_dd = pool.submit(_fetch_topic, 2, 1, dd_topic)
|
|
(waimai, wm_fail), (daodian, dd_fail) = f_wm.result(), f_dd.result()
|
|
|
|
items = _interleave(waimai, daodian)
|
|
has_next = page_idx + 1 < len(_TOPIC_ROUNDS)
|
|
# 两路都失败且无数据 → degraded;否则有数据 ok / 无数据 empty
|
|
status = "degraded" if (not items and wm_fail and dd_fail) else ("ok" if items else "empty")
|
|
return FeedResponse(items=items, has_next=has_next, page=req.page, status=status)
|
|
|
|
|
|
@router.post("/referral-link", response_model=ReferralLinkResponse, summary="换取推广链接(点抢时调)")
|
|
def referral_link(req: ReferralLinkRequest) -> ReferralLinkResponse:
|
|
if not settings.mt_cps_configured:
|
|
return ReferralLinkResponse(link="", link_map={})
|
|
try:
|
|
raw = get_referral_link(
|
|
product_view_sign=req.product_view_sign,
|
|
platform=req.platform,
|
|
biz_line=req.biz_line,
|
|
sid=req.sid,
|
|
link_type_list=req.link_type_list,
|
|
)
|
|
except MeituanCpsError as e:
|
|
logger.error("get_referral_link failed: %s", e)
|
|
raise HTTPException(status_code=502, detail=f"meituan: {e}") from e
|
|
|
|
link_map = raw.get("referralLinkMap") or {}
|
|
link = raw.get("data") or link_map.get("1") or link_map.get("3") or next(iter(link_map.values()), "")
|
|
return ReferralLinkResponse(link=link, link_map=link_map)
|
|
|
|
|
|
@router.post("/top-sales", response_model=CouponListResponse,
|
|
summary="销量最高(从离线库 meituan_coupon 按销量降序 + 跨源去重,不实时打美团)")
|
|
def top_sales(req: TopSalesRequest, db: Session = Depends(get_db)) -> CouponListResponse:
|
|
# 去重 + 排序 + 分页全在 SQL 做,每页只取并解析当前页 ~20 条。
|
|
# (之前实现每翻一页都全表拉取 + 全量 from_raw 解析,翻页慢 → 客户端滑动卡顿/翻不动。)
|
|
# 库为空(prod 刚部署 / ETL 未跑完)时返空 + status=empty,不崩;库查询异常降级 degraded。
|
|
try:
|
|
# 1) DISTINCT ON (dedup_key):每个去重键(品牌|名|价)只留销量最高那条(同销量再按佣金)
|
|
base = select(MeituanCoupon).where(MeituanCoupon.sale_volume_num.isnot(None))
|
|
if req.platform is not None:
|
|
base = base.where(MeituanCoupon.platform == req.platform)
|
|
deduped = base.distinct(MeituanCoupon.dedup_key).order_by(
|
|
MeituanCoupon.dedup_key,
|
|
MeituanCoupon.sale_volume_num.desc(),
|
|
MeituanCoupon.commission_percent.desc(),
|
|
).subquery()
|
|
|
|
# 2) 对去重结果按销量降序分页;多取 1 条判断 has_next,只对本页做 from_raw
|
|
m = aliased(MeituanCoupon, deduped)
|
|
start = (req.page - 1) * req.page_size
|
|
rows = db.execute(
|
|
select(m)
|
|
# 加 id 作稳定 tiebreaker:同销量同佣金的并列项排序确定,避免跨页重复/漏项
|
|
.order_by(m.sale_volume_num.desc(), m.commission_percent.desc(), m.id)
|
|
.offset(start)
|
|
.limit(req.page_size + 1)
|
|
).scalars().all()
|
|
except Exception: # noqa: BLE001
|
|
logger.exception("[top-sales] 库查询失败,降级返空")
|
|
return CouponListResponse(items=[], has_next=False, search_id=None, status="degraded")
|
|
|
|
has_next = len(rows) > req.page_size
|
|
cards: list[CouponCard] = []
|
|
for row in rows[:req.page_size]:
|
|
try:
|
|
card = CouponCard.from_raw(row.raw or {})
|
|
except Exception: # noqa: BLE001
|
|
continue
|
|
if card.product_view_sign:
|
|
cards.append(card)
|
|
return CouponListResponse(items=cards, has_next=has_next, search_id=None,
|
|
status="ok" if cards else "empty")
|