feat: 首页推荐 feed 实现 MVP 排序策略 + 标签格式化

排序策略 (app/core/ranking.py 新建):
- parse_sale_volume: "热销8.5万+" → 85000 整数销量
- get_distance_km: 距离统一(到店米 ÷1000), >50km 视为脏数据
- filter_items: 售价≤0 / 距离>8km / 脏数据剔除
- sort_by_sales: 按销量降序
- split_pages + shuffle_pages: 切页 + 页内 Fisher-Yates 打乱
- inject_billboard: 榜单商品 4 条/页注入前 5 页, 随机位置
- merge_category_pages: 外卖第K页 + 到店第K页合并后 shuffle

Feed 接口重写 (app/api/v1/meituan.py):
- 旧逻辑: 三页固定 listTopiId 主题轮, 美团给什么序就什么序
- 新逻辑: searchText="外卖"/"到店餐饮" + sortField=6 翻页 LBS 召回,
  4 任务并发(外卖召回 / 到店召回 / 外卖榜单 / 到店榜单),
  走完 ranking pipeline 后做坐标级内存缓存(5min TTL)
- 翻页上限 5 页/品类(性价比最高: 3.4s 拿到 238 条/品类)
- 直接调 _call 而非 query_coupon, 因为 searchId 翻页时不能传 pageNo

标签格式化 (app/schemas/meituan.py):
- price_label: "比日常团购省3.5元" → "比团购省 3.5 元"
- rank_label: "2小时北京外卖销量榜第1名" → "外卖榜第 1"
- rating_label: "4.6分" → "点评 4.6 分"

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

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 feed(req: FeedRequest) -> FeedResponse:
-    page_idx = req.page - 1
-    lon, lat = req.longitude, req.latitude
-    logger.info("[feed] page=%s lon=%.6f lat=%.6f", req.page, lon, lat)
+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
 
-    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 [])
+            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 []
 
-    if page_idx >= len(_TOPIC_ROUNDS):
+
+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:
+    lon, lat = req.longitude, req.latitude
+    page_idx = req.page - 1
+    logger.info("[feed] page=%s lon=%.6f lat=%.6f", req.page, lon, lat)
+
+    pages = _get_or_build_feed(lon, lat)
+
+    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)
 
 

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

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")),
         )