diff --git a/app/api/v1/meituan.py b/app/api/v1/meituan.py index 4d56b73..9380eeb 100644 --- a/app/api/v1/meituan.py +++ b/app/api/v1/meituan.py @@ -5,11 +5,22 @@ from __future__ import annotations import logging -from concurrent.futures import ThreadPoolExecutor, as_completed - +import time +from concurrent.futures import ThreadPoolExecutor +from threading import Lock from fastapi import APIRouter, HTTPException -from app.integrations.meituan import MeituanCpsError, get_referral_link, query_coupon +from app.core.ranking import ( + dedup, + filter_items, + get_distance_km, + inject_billboard, + merge_category_pages, + shuffle_pages, + sort_by_sales, + split_pages, +) +from app.integrations.meituan import MeituanCpsError, _call as mt_call, get_referral_link, query_coupon from app.schemas.meituan import ( CouponCard, CouponListRequest, @@ -24,8 +35,15 @@ logger = logging.getLogger("shagua.meituan") router = APIRouter(prefix="/api/v1/meituan", tags=["meituan-cps"]) +_MAX_RECALL_PAGES = 5 +_MAX_DISTANCE_KM = 8.0 +_FEED_CACHE_TTL = 300 -@router.post("/coupons", response_model=CouponListResponse, summary="券列表(为您推荐)") +_feed_cache: dict[str, tuple[float, list[list[dict]]]] = {} +_feed_lock = Lock() + + +@router.post("/coupons", response_model=CouponListResponse, summary="券列表(通用查询)") def list_coupons(req: CouponListRequest) -> CouponListResponse: logger.info("[coupons] lon=%.6f lat=%.6f topic=%s", req.longitude, req.latitude, req.list_topic_id) try: @@ -53,56 +71,146 @@ def list_coupons(req: CouponListRequest) -> CouponListResponse: ) -_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 +# ────────────────────── Feed 排序策略 (MVP) ────────────────────── -@router.post("/feed", response_model=FeedResponse, summary="混合feed(外卖+到店交叉, 无限流)") +def _lbs_recall( + keyword: str, + lon: float, + lat: float, + is_daodian: bool, +) -> list[dict]: + """searchText + sortField=6 翻页召回,收集 ≤8km 商品,遇到整页都 >8km 或翻满 10 页停止。""" + lon_i = int(lon * 1_000_000) + lat_i = int(lat * 1_000_000) + all_items: list[dict] = [] + search_id: str | None = None + + for _ in range(_MAX_RECALL_PAGES): + body: dict = { + "longitude": lon_i, + "latitude": lat_i, + "searchText": keyword, + "sortField": 6, + "pageSize": 20, + } + if search_id: + body["searchId"] = search_id + + try: + data = mt_call("/cps_open/common/api/v1/query_coupon", body) + except MeituanCpsError: + break + + items = data.get("data") or [] + search_id = data.get("searchId") + has_next = data.get("hasNext", False) + + if not items: + break + + page_has_valid = False + for item in items: + dist = get_distance_km(item, is_daodian) + if dist is not None and dist <= _MAX_DISTANCE_KM: + all_items.append(item) + page_has_valid = True + + if not page_has_valid or not has_next: + break + + return dedup(all_items) + + +def _fetch_billboard( + platform: int, + topic_id: int, + lon: float, + lat: float, +) -> list[dict]: + """listTopiId 拉榜单,固定 20 条,不做距离过滤。""" + try: + data = query_coupon( + longitude=lon, + latitude=lat, + platform=platform, + list_topic_id=topic_id, + ) + return data.get("data") or [] + except MeituanCpsError: + return [] + + +def _build_feed(lon: float, lat: float) -> list[list[dict]]: + """完整 pipeline:并发召回 → 过滤 → 销量排序 → 分页 → shuffle → 榜单加成 → 合并。""" + with ThreadPoolExecutor(max_workers=4) as pool: + f_wm = pool.submit(_lbs_recall, "外卖", lon, lat, False) + f_dd = pool.submit(_lbs_recall, "到店餐饮", lon, lat, True) + f_wm_bill = pool.submit(_fetch_billboard, 1, 1, lon, lat) + f_dd_bill = pool.submit(_fetch_billboard, 2, 3, lon, lat) + + wm_raw = f_wm.result() + dd_raw = f_dd.result() + wm_billboard = f_wm_bill.result() + dd_billboard = f_dd_bill.result() + + logger.info( + "[feed:build] waimai=%d daodian=%d wm_bill=%d dd_bill=%d", + len(wm_raw), len(dd_raw), len(wm_billboard), len(dd_billboard), + ) + + wm_filtered = filter_items(wm_raw, is_daodian=False, max_km=_MAX_DISTANCE_KM) + dd_filtered = filter_items(dd_raw, is_daodian=True, max_km=_MAX_DISTANCE_KM) + + wm_sorted = sort_by_sales(wm_filtered) + dd_sorted = sort_by_sales(dd_filtered) + + wm_pages = shuffle_pages(split_pages(wm_sorted)) + dd_pages = shuffle_pages(split_pages(dd_sorted)) + + wm_pages = inject_billboard(wm_pages, wm_billboard, per_page=4, max_inject_pages=5) + dd_pages = inject_billboard(dd_pages, dd_billboard, per_page=4, max_inject_pages=5) + + return merge_category_pages(wm_pages, dd_pages) + + +def _cache_key(lon: float, lat: float) -> str: + return f"{lon:.4f},{lat:.4f}" + + +def _get_or_build_feed(lon: float, lat: float) -> list[list[dict]]: + key = _cache_key(lon, lat) + now = time.time() + + with _feed_lock: + entry = _feed_cache.get(key) + if entry and now - entry[0] < _FEED_CACHE_TTL: + return entry[1] + + pages = _build_feed(lon, lat) + + with _feed_lock: + _feed_cache[key] = (time.time(), pages) + expired = [k for k, (t, _) in _feed_cache.items() if now - t > _FEED_CACHE_TTL] + for k in expired: + del _feed_cache[k] + + return pages + + +@router.post("/feed", response_model=FeedResponse, summary="首页推荐 feed (MVP 排序策略)") def feed(req: FeedRequest) -> FeedResponse: - page_idx = req.page - 1 lon, lat = req.longitude, req.latitude + page_idx = req.page - 1 logger.info("[feed] page=%s lon=%.6f lat=%.6f", req.page, lon, lat) - def _fetch_topic(platform: int, biz_line: int | None, topic: int) -> list[dict]: - try: - return (query_coupon( - longitude=lon, latitude=lat, - platform=platform, biz_line=biz_line, - list_topic_id=topic, page_size=20, - ).get("data") or []) - except MeituanCpsError: - return [] + pages = _get_or_build_feed(lon, lat) - if page_idx >= len(_TOPIC_ROUNDS): + if page_idx >= len(pages): return FeedResponse(items=[], has_next=False, page=req.page) - 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, daodian = f_wm.result(), f_dd.result() - - items = _interleave(waimai, daodian) - has_next = page_idx + 1 < len(_TOPIC_ROUNDS) + items = [CouponCard.from_raw(it) for it in pages[page_idx]] + has_next = page_idx + 1 < len(pages) return FeedResponse(items=items, has_next=has_next, page=req.page) diff --git a/app/core/ranking.py b/app/core/ranking.py new file mode 100644 index 0000000..e9ac58f --- /dev/null +++ b/app/core/ranking.py @@ -0,0 +1,143 @@ +"""首页 Feed 排序策略 (MVP 版) + +LBS 召回 → 距离过滤 → 销量重排 → 分页 shuffle → 榜单加成 → 双品类同页交错。 +""" +from __future__ import annotations + +import random +import re +from typing import Any + + +def parse_sale_volume(text: str | None) -> int: + """'热销8.5万+' → 85000, '热销1k+' → 1000, None → 0""" + if not text: + return 0 + m = re.search(r"([\d.]+)\s*(万|k)?", text, re.IGNORECASE) + if not m: + return 0 + num = float(m.group(1)) + unit = (m.group(2) or "").lower() + if unit == "万": + num *= 10000 + elif unit == "k": + num *= 1000 + return int(num) + + +def get_distance_km(item: dict[str, Any], is_daodian: bool) -> float | None: + """提取距离(km)。外卖单位千米,到店单位米需÷1000。>50km 视为脏数据返回 None。""" + raw = (item.get("deliverablePoiInfo") or {}).get("deliveryDistance") + if raw is None: + return None + try: + d = float(raw) + except (ValueError, TypeError): + return None + if is_daodian: + d /= 1000 + if d > 50: + return None + return d + + +def get_sell_price(item: dict[str, Any]) -> float | None: + raw = (item.get("couponPackDetail") or {}).get("sellPrice") + if raw is None: + return None + try: + p = float(raw) + return p if p > 0 else None + except (ValueError, TypeError): + return None + + +def get_product_sign(item: dict[str, Any]) -> str: + cpd = item.get("couponPackDetail") or {} + return cpd.get("productViewSign") or cpd.get("skuViewId") or "" + + +def filter_items( + items: list[dict], is_daodian: bool, max_km: float = 8.0, +) -> list[dict]: + """距离 ≤ max_km、售价 > 0、去脏数据。""" + result = [] + for item in items: + if get_sell_price(item) is None: + continue + dist = get_distance_km(item, is_daodian) + if dist is None or dist > max_km: + continue + result.append(item) + return result + + +def dedup(items: list[dict]) -> list[dict]: + seen: set[str] = set() + result = [] + for item in items: + sign = get_product_sign(item) + if sign and sign not in seen: + seen.add(sign) + result.append(item) + return result + + +def sort_by_sales(items: list[dict]) -> list[dict]: + def _key(item: dict) -> int: + vol = (item.get("couponPackDetail") or {}).get("saleVolume") + return parse_sale_volume(vol) + return sorted(items, key=_key, reverse=True) + + +def split_pages(items: list[dict], page_size: int = 20) -> list[list[dict]]: + if not items: + return [] + return [items[i : i + page_size] for i in range(0, len(items), page_size)] + + +def shuffle_pages(pages: list[list[dict]]) -> list[list[dict]]: + """每页内 Fisher-Yates shuffle,不跨页。""" + result = [] + for page in pages: + shuffled = page[:] + random.shuffle(shuffled) + result.append(shuffled) + return result + + +def inject_billboard( + pages: list[list[dict]], + billboard: list[dict], + per_page: int = 4, + max_inject_pages: int = 5, +) -> list[list[dict]]: + """将榜单商品分配到前 N 页,每页额外加 per_page 个,随机位置插入。""" + existing = {get_product_sign(it) for p in pages for it in p} + unique = [it for it in billboard if get_product_sign(it) not in existing] + + idx = 0 + for i in range(min(max_inject_pages, len(pages))): + batch = unique[idx : idx + per_page] + idx += per_page + for it in batch: + pages[i].insert(random.randint(0, len(pages[i])), it) + return pages + + +def merge_category_pages( + waimai_pages: list[list[dict]], + daodian_pages: list[list[dict]], +) -> list[list[dict]]: + """同页合并 + shuffle,页数取两者最大值。""" + n = max(len(waimai_pages), len(daodian_pages)) + result = [] + for k in range(n): + merged: list[dict] = [] + if k < len(waimai_pages): + merged.extend(waimai_pages[k]) + if k < len(daodian_pages): + merged.extend(daodian_pages[k]) + random.shuffle(merged) + result.append(merged) + return result diff --git a/app/schemas/meituan.py b/app/schemas/meituan.py index b8cf1ab..a406d4c 100644 --- a/app/schemas/meituan.py +++ b/app/schemas/meituan.py @@ -1,11 +1,46 @@ """美团 CPS 券列表 / 换链相关 schemas。""" from __future__ import annotations +import re from typing import Any from pydantic import BaseModel, Field +def _format_price_label(price_lbl: dict[str, Any]) -> str | None: + history = price_lbl.get("historyPriceLabel") + if history: + return history + beat = price_lbl.get("beatMTLabel") + if not beat: + return None + m = re.match(r"比日常团购省([\d.]+)元", beat) + if m: + return f"比团购省 {m.group(1)} 元" + return beat + + +def _format_rank_label(raw: str | None) -> str | None: + if not raw: + return None + m = re.search(r"(外卖|美食|饮品|轻食|奶茶|咖啡|火锅|烧烤|甜品|快餐).*?第(\d+)名", raw) + if m: + return f"{m.group(1)}榜第 {m.group(2)}" + m2 = re.search(r"第(\d+)名", raw) + if m2: + return f"销量榜第 {m2.group(1)}" + return raw + + +def _format_rating_label(raw: str | None) -> str | None: + if not raw: + return None + m = re.search(r"([\d.]+)\s*分", raw) + if m: + return f"点评 {m.group(1)} 分" + return raw + + # ───────────────── 券卡片(归一化后给客户端) ───────────────── class CouponCard(BaseModel): @@ -102,9 +137,9 @@ class CouponCard(BaseModel): 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"), + price_label=_format_price_label(price_lbl), + rank_label=_format_rank_label(label.get("productRankLabel")), + rating_label=_format_rating_label(label.get("dianPingRankLabel")), )