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 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 app.utils.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 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[dict[str, Any]], dependencies=[Depends(require_bearer_token)], ) async def trade( trade_code: str, container: ServiceContainer = Depends(get_container), ) -> ApiResponse[dict[str, Any]]: """ 获取订单详情 """ if not trade_code or trade_code.strip() == "": return ApiResponse[dict[str, Any]]( success=False, msg="trade_code 不能为空", data=None, code=400, ) if container.session_store._redis is None: return ApiResponse[dict[str, Any]]( 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[dict[str, Any]]( success=False, msg="未找到该交易", data=None, code=404, ) return ApiResponse[dict[str, Any]]( success=True, msg="success", data=json.loads(trade_data), 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, )