surugaya/app/api/routes/scrape.py

from __future__ import annotations
import asyncio
import json
import time
from typing import Any
from urllib import error, request

from fastapi import APIRouter, Depends
from pydantic import ValidationError

from app.api.dependencies import get_container, require_bearer_token
from app.core.container import ServiceContainer
from app.models.scrape import (
    ApiResponse,
    CategoryData,
    CheckTaskCallbackUrlData,
    CheckTaskCallbackUrlRequest,
    DetailRequest,
    OrderShippingFeeData,
    OrderShippingFeeRequest,
    ProductDetailData,
    PurchaseTaskRequest,
    SearchRequest,
    SearchResultData, TradeMonitorRequest,
)
from surugaya_common.http_utils import generate_signed_headers
from app.utils.parse_util import (
    get_franchise_shipping_fee,
    get_handle_fee,
    get_shipping_fee,
)
from surugaya_common import redis_keys
from surugaya_common.models import SurugayaOrderDetail

router = APIRouter(prefix="/api", tags=["scrape"])


def _parse_callback_response_body(content_type: str, body_bytes: bytes) -> Any:
    if not body_bytes:
        return None

    body_text = body_bytes.decode("utf-8", errors="replace")
    if "application/json" not in content_type.lower():
        return body_text

    try:
        return json.loads(body_text)
    except json.JSONDecodeError:
        return body_text


def _post_callback_request(
        callback_url: str,
        request_body: dict[str, Any],
        headers: dict[str, str],
) -> tuple[int, dict[str, str], Any]:
    body_bytes = json.dumps(request_body, ensure_ascii=False).encode("utf-8")
    http_request = request.Request(
        url=callback_url,
        data=body_bytes,
        headers=headers,
        method="POST",
    )

    try:
        with request.urlopen(http_request, timeout=15) as response:
            response_body = response.read()
            response_headers = dict(response.headers.items())
            content_type = response_headers.get("Content-Type", "")
            return response.status, response_headers, _parse_callback_response_body(content_type, response_body)
    except error.HTTPError as exc:
        response_body = exc.read()
        response_headers = dict(exc.headers.items()) if exc.headers is not None else {}
        content_type = response_headers.get("Content-Type", "")
        return exc.code, response_headers, _parse_callback_response_body(content_type, response_body)


@router.post(
    "/search",
    response_model=ApiResponse[SearchResultData],
    dependencies=[Depends(require_bearer_token)],
)
async def search(
        payload: SearchRequest,
        container: ServiceContainer = Depends(get_container),
) -> ApiResponse[SearchResultData]:
    """搜索商品列表 Authorization	Bearer NilsO8Il8yoT1wazJFs2eeOmlPRfWuSx

    根据关键词和页码抓取骏河屋搜索结果，返回商品列表、总数和分页信息。
    """
    data = await container.surugaya_client.search(payload)
    return ApiResponse[SearchResultData](
        success=True,
        msg="success",
        data=data,
        code=0,
    )


@router.post(
    "/item_detail",
    response_model=ApiResponse[ProductDetailData],
    dependencies=[Depends(require_bearer_token)],
)
async def item_detail(
        payload: DetailRequest,
        container: ServiceContainer = Depends(get_container),
) -> ApiResponse[ProductDetailData]:
    """获取商品详情

    根据商品 ID 或 URL 抓取骏河屋商品详情页，返回名称、价格、图片等信息。
    """
    data = await container.surugaya_client.fetch_detail(payload)
    return ApiResponse[ProductDetailData](
        success=True,
        msg="success",
        data=data,
        code=0,
    )


@router.get(
    "/categories",
    response_model=ApiResponse[list[CategoryData]],
    dependencies=[Depends(require_bearer_token)],
)
async def categories(
        container: ServiceContainer = Depends(get_container),
) -> ApiResponse[list[CategoryData]]:
    """获取商品分类

    获取骏河屋商品分类列表，包含分类名称和分类 ID。数据带有缓存。
    """
    data = await container.surugaya_client.fetch_categories()
    return ApiResponse[list[CategoryData]](
        success=True,
        msg="success",
        data=data,
        code=0,
    )


@router.post(
    "/order/get_shipping_fee",
    response_model=ApiResponse[OrderShippingFeeData],
    dependencies=[Depends(require_bearer_token)],
)
async def order_get_shipping_fee(payload: OrderShippingFeeRequest) -> ApiResponse[OrderShippingFeeData]:
    """计算订单运费与手续费。

    规则:
      - 官方店铺(supplier 为空)运费: 按官方部分商品总额套阶梯价(440/385/0)
        * 订单中无任何官方商品时,官方运费 0
      - 加盟店运费(supplier 非空,同名视为同一家): 每家统一 800 日元
      - 通贩手续费: 按全部商品(含官方+加盟店)总额套阶梯价(240/0)
    """
    official_amount = 0
    franchise_stores: set[str] = set()
    goods_amount = 0
    for item in payload.item_list:
        subtotal = item.goods_number * item.goods_price
        goods_amount += subtotal
        if item.supplier == "":
            official_amount += subtotal
        else:
            franchise_stores.add(item.supplier)

    official_shipping_fee = get_shipping_fee(official_amount) if official_amount > 0 else 0
    franchise_shipping_fee = get_franchise_shipping_fee(len(franchise_stores))
    shipping_fee = official_shipping_fee + franchise_shipping_fee
    handle_amount = get_handle_fee(goods_amount)
    order_amount = goods_amount + shipping_fee + handle_amount

    return ApiResponse[OrderShippingFeeData](
        success=True,
        msg="success",
        data=OrderShippingFeeData(
            goods_amount=goods_amount,
            shipping_fee=shipping_fee,
            order_amount=order_amount,
            handle_amount=handle_amount,
        ),
        code=0,
    )


@router.post(
    "/order/add_purchase_task",
    response_model=ApiResponse[None],
    dependencies=[Depends(require_bearer_token)],
)
async def order_add_purchase_task(
        payload: PurchaseTaskRequest,
        container: ServiceContainer = Depends(get_container),
) -> ApiResponse[None]:
    """添加采购任务
    
    将采购任务加入 Redis Stream 队列。
    """
    if container.session_store._redis is None:
        return ApiResponse[None](
            success=False,
            msg="Redis 未配置",
            data=None,
            code=500,
        )

    stream_name = redis_keys.purchase_task_stream(payload.is_dev)
    params_str = payload.model_dump_json(by_alias=True)

    await container.session_store._redis.xadd(
        name=stream_name,
        fields={"payload": params_str},
        maxlen=100,
        approximate=True
    )

    return ApiResponse[None](
        success=True,
        msg="success",
        data=None,
        code=0,
    )


@router.post(
    "/order/add_trade_monitor",
    response_model=ApiResponse[None],
    dependencies=[Depends(require_bearer_token)],
)
async def order_add_trade_monitor(
        payload: TradeMonitorRequest,
        container: ServiceContainer = Depends(get_container),
) -> ApiResponse[None]:
    """添加采购监控

    将采购监控加入 Redis Stream 队列。
    """
    if container.session_store._redis is None:
        return ApiResponse[None](
            success=False,
            msg="Redis 未配置",
            data=None,
            code=500,
        )

    node_id = payload.node_id
    if node_id is None:
        return ApiResponse[None](
            success=False,
            msg="node_id 不能为空",
            data=None,
            code=400,
        )

    trade_code = payload.payload.get("trade_code")
    if not trade_code or trade_code == "":
        return ApiResponse[None](
            success=False,
            msg="trade_code 不能为空",
            data=None,
            code=400,
        )

    queue_name = redis_keys.trade_delay_zset(node_id, payload.is_dev)

    score = int(time.time()) + 3
    member = json.dumps(payload.payload)
    await container.session_store._redis.zadd(
        name=queue_name,
        mapping={member: score}
    )

    return ApiResponse[None](
        success=True,
        msg="success",
        data=None,
        code=0,
    )


@router.get(
    "/trade",
    response_model=ApiResponse[SurugayaOrderDetail],
    dependencies=[Depends(require_bearer_token)],
)
async def trade(
        trade_code: str,
        container: ServiceContainer = Depends(get_container),
) -> ApiResponse[SurugayaOrderDetail]:
    """
    获取订单详情
    """
    if not trade_code or trade_code.strip() == "":
        return ApiResponse[SurugayaOrderDetail](
            success=False,
            msg="trade_code 不能为空",
            data=None,
            code=400,
        )

    if container.session_store._redis is None:
        return ApiResponse[SurugayaOrderDetail](
            success=False,
            msg="Redis 未配置",
            data=None,
            code=500,
        )

    trade_data = await container.session_store._redis.hget(
        name=redis_keys.trade_hash(),
        key=trade_code,
    )

    if not trade_data:
        # 去重：同一 trade_code 在 3600s 内不重复下发
        dedup_key = redis_keys.trade_pending_key(trade_code)
        dedup_exists = await container.session_store._redis.get(dedup_key)
        if not dedup_exists:
            await container.session_store._redis.setex(dedup_key, 3600, "1")
            await container.session_store._redis.xadd(
                name=redis_keys.trade_fetch_stream(),
                fields={"payload": json.dumps({"trade_code": trade_code})},
                maxlen=100,
                approximate=True,
            )

        return ApiResponse[SurugayaOrderDetail](
            success=False,
            msg="未找到该交易",
            data=None,
            code=404,
        )

    # Redis 中存储的是 worker 写入的 SurugayaOrderDetail 序列化结果，
    # 用共享契约校验并类型化；历史脏数据校验失败时降级为结构化错误而非 500。
    try:
        order_detail = SurugayaOrderDetail.model_validate_json(trade_data)
    except ValidationError as exc:
        return ApiResponse[SurugayaOrderDetail](
            success=False,
            msg=f"订单详情数据结构异常: {exc.error_count()} 处校验失败",
            data=None,
            code=500,
        )

    return ApiResponse[SurugayaOrderDetail](
        success=True,
        msg="success",
        data=order_detail,
        code=0,
    )



@router.post(
    "/test/check_task_callback_url",
    response_model=ApiResponse[CheckTaskCallbackUrlData],
    dependencies=[Depends(require_bearer_token)],
)
async def check_task_callback_url(
        payload: CheckTaskCallbackUrlRequest,
) -> ApiResponse[CheckTaskCallbackUrlData]:
    """测试回调地址可用性并返回回调响应。"""
    request_body = payload.data
    headers = generate_signed_headers(
        app_id=payload.app_id,
        app_secret=payload.app_secret,
        data=request_body,
        method="POST",
    )

    try:
        response_status, response_headers, response_body = await asyncio.to_thread(
            _post_callback_request,
            str(payload.callback_url),
            request_body,
            headers,
        )
    except error.URLError as exc:
        return ApiResponse[CheckTaskCallbackUrlData](
            success=False,
            msg=f"回调请求失败: {exc.reason}",
            data=None,
            code=500,
        )
    except Exception as exc:
        return ApiResponse[CheckTaskCallbackUrlData](
            success=False,
            msg=f"回调请求异常: {exc}",
            data=None,
            code=500,
        )

    success = 200 <= response_status < 300
    return ApiResponse[CheckTaskCallbackUrlData](
        success=success,
        msg="调用正常" if success else "回调请求有异常",
        data=CheckTaskCallbackUrlData(
            response_status=response_status,
            response_body=response_body,
        ),
        code=0 if success else response_status,
    )