duobibi-app-mock/duobibi(mock)backend/duobibi-server/app/mock_extractor.py

"""Mock 模式的"提取/归簇/估常见价"实现。

用于占坑期 iOS 端开发与审核演示:不调真实 LLM,直接按图片字节哈希在
预置商品池中轮询返回,稳定可重放,且能自然制造"归簇命中"和"新建簇"两种路径。

接真模型时:
- 把 routes 里调用 mock_extract_image / mock_extract_text 的位置切换到调用真实 extractor
- 或在外层加 USE_MOCK 环境变量分流(本占坑期方案不引入开关,接模型时直接换实现)
"""
from __future__ import annotations

import asyncio
import hashlib
import logging
import random
from typing import Optional

from app.schemas import ClusterDtoStr

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

# 10 件预置商品。挑选原则:覆盖 7 个白名单平台 + 3C/食品/服饰/小家电 多品类,
# 价格分布从 28 元(快餐)到 9999 元(手机)跨度大,便于测试统计页/趋势图视觉效果。
MOCK_PRODUCTS: list[dict] = [
    {"title": "iPhone 15 Pro Max 256GB 暮光紫", "price": 9999.0, "source_app": "淘宝", "typical_price": 13999.0},
    {"title": "海底捞外卖经典套餐 2-3 人餐", "price": 158.0, "source_app": "美团", "typical_price": 199.0},
    {"title": "元气森林白桃味气泡水 480ml*15", "price": 49.9, "source_app": "京东", "typical_price": 65.0},
    {"title": "Nike Air Force 1 三色刺绣男款 41", "price": 599.0, "source_app": "拼多多", "typical_price": 799.0},
    {"title": "海尔不锈钢真空保温杯 500ml", "price": 89.0, "source_app": "京东", "typical_price": 129.0},
    {"title": "麦当劳麦辣鸡腿堡套餐(大份)", "price": 28.0, "source_app": "美团", "typical_price": 35.0},
    {"title": "飞利浦多功能旋转剃须刀 S5586", "price": 469.0, "source_app": "淘宝", "typical_price": 599.0},
    {"title": "Lululemon Align 高腰瑜伽裤 25", "price": 750.0, "source_app": "抖音", "typical_price": 980.0},
    {"title": "蒙牛特仑苏纯牛奶 250ml*16 礼盒", "price": 65.0, "source_app": "京东", "typical_price": 88.0},
    {"title": "戴森 V12 Detect Slim 无线吸尘器", "price": 3699.0, "source_app": "淘宝", "typical_price": 4490.0},
]


# 让"同一张图触发同一个商品"可复现,但同一个 mock 商品在不同 record 里,
# 价格上下浮动 ±5%(模拟用户在不同时间记到的不同到手价),便于看趋势图效果。
def _picked_with_jitter(base: dict, jitter_seed: int) -> dict:
    rnd = random.Random(jitter_seed)
    delta = rnd.uniform(-0.05, 0.05)
    price = round(base["price"] * (1 + delta), 2)
    return {
        "title": base["title"],
        "price": price,
        "source_app": base["source_app"],
        "typical_price": base["typical_price"],
    }


def _pick_by_bytes(blob: bytes) -> dict:
    """字节稳定哈希到预置商品。同样字节 → 同样商品,便于真机重放。"""
    h = hashlib.sha256(blob).digest()
    idx = h[0] % len(MOCK_PRODUCTS)
    # 用第二个字节作为价格抖动种子,让"同一商品"在多次记账时价格略有差异
    jitter_seed = h[1]
    return _picked_with_jitter(MOCK_PRODUCTS[idx], jitter_seed)


def _pick_by_text(title: str, price: float) -> dict:
    """文本场景: 客户端已知 title/price,后端只决定 typical_price。
    根据 title 哈希挑一个 mock 商品的 typical_price 作为参考价。"""
    h = hashlib.sha256(title.encode("utf-8")).digest()
    idx = h[0] % len(MOCK_PRODUCTS)
    return MOCK_PRODUCTS[idx]


def _match_cluster(picked_title: str, clusters: list[ClusterDtoStr]) -> Optional[str]:
    """简易归簇: 取 picked title 前 4 个字符,在 clusters 里找包含该子串的;
    找到就返回该 cluster.id (字符串)。

    这不是真实 LLM 的语义归簇能力,只是为了让客户端在测试中能稳定触发"归簇命中"路径:
    用户第一次记某商品 → 新建簇;再次记同 hash 的图片 → 命中已有簇。
    """
    if not clusters:
        return None
    needle = picked_title[:4] if len(picked_title) >= 4 else picked_title
    for c in clusters:
        if needle and needle in c.title:
            return c.id
    return None


async def _simulate_latency() -> None:
    """模拟真实 LLM 延迟 1.5-2.5 秒,让 iOS 端 loading UI 测起来真实。

    **必须用 asyncio.sleep 而非 time.sleep**:本模块被 async 路由 (parse_image) 调用,
    同步 sleep 会阻塞整个 event loop —— 任何一个请求 sleep 期间,server 无法处理
    任何其他请求(健康检查、其他识别请求都会卡)。占坑期 QPS 低也会因为并发审核
    被串行化,严重影响体验。
    """
    await asyncio.sleep(random.uniform(1.5, 2.5))


async def mock_extract_image(image_bytes: bytes, clusters: list[ClusterDtoStr]) -> dict:
    """图片版 mock 提取。返回与真实 extractor 同形状的 dict。

    入参:
      - image_bytes: 客户端上传的截图原始字节(已被客户端降采样)
      - clusters: 客户端发来的已有商品簇列表

    返回(success):
      {success, title, price, source_app, cluster_id, typical_price}
    """
    await _simulate_latency()
    picked = _pick_by_bytes(image_bytes)
    cluster_id = _match_cluster(picked["title"], clusters)

    logger.info(
        "mock image: bytes=%d picked=%r cluster_id=%s",
        len(image_bytes),
        picked["title"],
        cluster_id,
    )
    return {
        "success": True,
        "title": picked["title"],
        "price": picked["price"],
        "source_app": picked["source_app"],
        "cluster_id": cluster_id,
        "typical_price": picked["typical_price"],
    }


async def mock_extract_text(
    title: str,
    price: float,
    source_app: str,
    clusters: list[ClusterDtoStr],
) -> dict:
    """文本版 mock: 客户端已知 title/price/source_app,后端只决定归簇 + 估常见价。"""
    await _simulate_latency()
    picked = _pick_by_text(title, price)
    typical_price = picked["typical_price"]
    cluster_id = _match_cluster(title, clusters)

    logger.info(
        "mock text: title=%r price=%.2f source=%s cluster_id=%s typical=%.2f",
        title,
        price,
        source_app,
        cluster_id,
        typical_price,
    )
    return {
        "success": True,
        "title": title,
        "price": price,
        "source_app": source_app,
        "cluster_id": cluster_id,
        "typical_price": typical_price,
    }


async def mock_extract_ocr(ocr_text: str, clusters: list[ClusterDtoStr]) -> dict:
    """OCR 文本版 mock:按整段 OCR 文本的哈希挑预置商品,稳定可重放。

    与 image / text 版同形状返回,客户端复用同一套结果处理逻辑。
    (此前 main.py 引用了 parse_ocr router 但实现缺失会导致启动 ImportError,这里补齐对应 mock。)
    """
    await _simulate_latency()
    h = hashlib.sha256(ocr_text.encode("utf-8")).digest()
    picked = _picked_with_jitter(MOCK_PRODUCTS[h[0] % len(MOCK_PRODUCTS)], h[1])
    cluster_id = _match_cluster(picked["title"], clusters)

    logger.info(
        "mock ocr: chars=%d picked=%r cluster_id=%s",
        len(ocr_text),
        picked["title"],
        cluster_id,
    )
    return {
        "success": True,
        "title": picked["title"],
        "price": picked["price"],
        "source_app": picked["source_app"],
        "cluster_id": cluster_id,
        "typical_price": picked["typical_price"],
    }