shaguabijia-app-server/app/core/ranking.py

"""首页 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