shaguabijia-app-server/app/api/v1/meituan.py

"""美团 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,
)
from app.utils.meituan_city import get_meituan_city

logger = logging.getLogger("shagua.meituan")


def _resolve_city_id(latitude: float, longitude: float) -> str:
    """经纬度 → 美团城市 ID;解析失败返 ""（调用方应降级返空）。"""
    try:
        return get_meituan_city(latitude, longitude).get("city_id", "")
    except Exception:
        logger.exception("get_meituan_city 失败")
        return ""

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":
        city_id = _resolve_city_id(lat, lon)
        if not city_id:
            return FeedResponse(items=[], has_next=False, page=req.page, status="degraded")
        PAGE = 20
        try:
            base = select(MeituanCoupon).where(
                MeituanCoupon.commission_percent >= 3.0,
                MeituanCoupon.city_id == city_id,
            )
            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)
        if not cards and req.page == 1:
            # 命中城市却 0 券:该城确无 ≥3% 券,或 ETL 灌的 city_id 与 city_dict 口径不一致。
            logger.info("[feed] rec city_id=%s 命中 0 券(该城确无券?或 ETL/city_dict 的 city_id 口径不一致)", city_id)
        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:
    # 按设备经纬度定位城市,只查同城券;老客户端不带坐标 → 降级返空(不 422、不误返全城)。
    if req.latitude is None or req.longitude is None:
        return CouponListResponse(items=[], has_next=False, search_id=None, status="degraded")
    city_id = _resolve_city_id(req.latitude, req.longitude)
    if not city_id:
        return CouponListResponse(items=[], has_next=False, search_id=None, status="degraded")

    # 去重 + 排序 + 分页全在 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),
            MeituanCoupon.city_id == city_id,
        )
        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:
            # 不显示距离:库里的距离是相对城市默认点的(对用户无意义、且误导)。
            # 置空后前端"距离 店名"那行只剩店名、自动顶到最左(店名移到原距离的位置)。
            # 逻辑与推荐流保持一致
            card.distance_text = None
            card.distance_meters = None
            cards.append(card)
    if not cards and req.page == 1:
        # 命中城市却 0 券:可能该城确无券,也可能 ETL 灌的 city_id 与 city_dict 口径不一致(静默降级的隐患)。
        logger.info("[top-sales] city_id=%s 命中 0 券(该城确无券?或 ETL/city_dict 的 city_id 口径不一致)", city_id)
    return CouponListResponse(items=cards, has_next=has_next, search_id=None,
                              status="ok" if cards else "empty")