q792602257
118667d01f
/api/order/get_shipping_fee 与购物车加入逻辑此前只认官方自营详情页，加盟店运费分桶 仅靠自由文本 supplier 判断，未与其他店铺列表的 tenpo_cd 打通。补充 OrderItem/ CargoItemRequest 的 tenpo_cd/branch_number 字段，运费计算优先按 tenpo_cd 分桶 （无则回退 supplier 保持兼容），cargo_service 加购 URL 同步带上店铺参数。 Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
258 lines
8.1 KiB
Python
258 lines
8.1 KiB
Python
"""API 数据模型：请求体和响应体定义"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from typing import Any, Generic, TypeVar
|
|
|
|
from pydantic import BaseModel, ConfigDict, Field, HttpUrl, field_validator
|
|
|
|
T = TypeVar("T")
|
|
|
|
|
|
class ApiResponse(BaseModel, Generic[T]):
|
|
"""统一 API 响应格式"""
|
|
success: bool
|
|
msg: str
|
|
data: T | None = None
|
|
code: int
|
|
|
|
|
|
class CategoryData(BaseModel):
|
|
"""商品分类信息"""
|
|
id: str
|
|
name: str
|
|
pid: str = ""
|
|
icon: str | None = None
|
|
href: str = ""
|
|
children: list["CategoryData"] = Field(default_factory=list)
|
|
|
|
|
|
class SearchRequest(BaseModel):
|
|
"""搜索请求参数"""
|
|
model_config = ConfigDict(extra="allow")
|
|
|
|
search_word: str = ""
|
|
page: int = Field(default=1, ge=1, le=100)
|
|
search_url: HttpUrl | None = None
|
|
|
|
|
|
class DetailRequest(BaseModel):
|
|
"""商品详情请求参数，支持 id 或 product_url"""
|
|
id: str | None = Field(default=None, min_length=3, max_length=40)
|
|
product_url: HttpUrl | None = None
|
|
tenpo_cd: str | None = None # 指定店铺代码；跳转查看某个加盟店/市场店铺的报价时传入
|
|
branch_number: str | None = None # 指定分支编号，与 tenpo_cd 搭配使用
|
|
include_other_shops: bool = False # 为 True 时，随详情额外抓取并返回 other_shop_list（多付出一次页面请求）
|
|
|
|
|
|
class ProductSummary(BaseModel):
|
|
"""搜索结果中的商品摘要信息"""
|
|
goods_id: str
|
|
goods_name: str
|
|
goods_image: str = ""
|
|
goods_price: int = 0 # 当前售价（日元）
|
|
original_price: int = 0 # 原价（日元）
|
|
third_shop_price: int = 0 # 第三方市场价（日元）
|
|
goods_link: str
|
|
on_sold: int = 0 # 1: 在售, 0: 售罄
|
|
created_at: str = ""
|
|
updated_at: str = ""
|
|
|
|
|
|
class SearchResultData(BaseModel):
|
|
"""搜索结果数据"""
|
|
query: str
|
|
page: int
|
|
page_size: int = 0
|
|
total_count: int = 0
|
|
has_more: int = 0
|
|
session_id: str
|
|
items: list[ProductSummary]
|
|
|
|
|
|
class OtherShopItem(BaseModel):
|
|
"""商品「他のショップ」列表中的单个店铺报价
|
|
|
|
来自 /product/other/{goods_id} 页面，含官方自营与加盟店/市场店铺。
|
|
"""
|
|
tenpo_cd: str = "" # 店铺代码；空串 = 官方自营
|
|
shop_name: str = "" # 店铺名称；官方自营固定为"駿河屋"
|
|
shop_url: str = "" # 店铺主页链接
|
|
is_official: bool = False # 是否为官方自营（tenpo_cd 为空）
|
|
condition: str = "" # 成色/规格描述，如"中古"
|
|
price: int = 0 # 该店铺售价（日元）
|
|
branch_number: str = "" # 分支编号
|
|
rating_score: str = "" # 店铺评分，如"5.0"
|
|
rating_count: int = 0 # 评价数量
|
|
shipping_note: str = "" # 发货时效描述，如"1日〜3日以内に発送します"
|
|
detail_url: str = "" # 该报价对应的商品详情页链接（含 tenpo_cd/branch_number）
|
|
|
|
|
|
class ProductDetailData(BaseModel):
|
|
"""商品详情数据"""
|
|
goods_id: str
|
|
goods_name: str
|
|
goods_image: str = ""
|
|
goods_price: int = 0 # 当前售价（日元）
|
|
on_sold: int = 0 # 1: 在售, 0: 售罄
|
|
stock: int | None = None
|
|
description: str | None = None
|
|
session_id: str | None = None
|
|
supplier: str | None = None # 供应商名称
|
|
price_note: str | None = None # 价格备注
|
|
shipping_comission_fee_text: str | None = None # 运费描述
|
|
shipping_comission_fee: int = 0 # 运费
|
|
sku_list: list[dict[str, Any]] # 规格列表
|
|
sku: str = "" # 规格
|
|
breadcrumb_list: list[dict[str, Any]] = [] # 面包屑
|
|
other_shop_list: list[OtherShopItem] = [] # 仅当请求 include_other_shops=True 时才会填充，否则为空列表
|
|
|
|
|
|
class OtherShopListData(BaseModel):
|
|
"""商品「他のショップ」全部店铺（含官方自营+加盟店/市场店铺）报价列表"""
|
|
goods_id: str
|
|
other_shop_list: list[OtherShopItem] = []
|
|
|
|
|
|
class HealthData(BaseModel):
|
|
"""健康检查响应数据"""
|
|
status: str
|
|
browser_enabled: bool
|
|
browser_ready: bool
|
|
browser_startup_error: str | None = None
|
|
redis_enabled: bool
|
|
active_session_id: str | None = None
|
|
|
|
|
|
class CargoItemRequest(BaseModel):
|
|
id: str
|
|
number: int = 1
|
|
price_limit: int | None = None
|
|
tenpo_cd: str | None = None # 指定店铺代码;跳转加入指定加盟店/市场店铺的商品时传入
|
|
branch_number: str | None = None # 指定分支编号,与 tenpo_cd 搭配使用
|
|
|
|
|
|
class CargoPayload(BaseModel):
|
|
item_list: list[CargoItemRequest]
|
|
|
|
|
|
class CargoRequest(BaseModel):
|
|
record_video: str = "0"
|
|
payload: CargoPayload
|
|
|
|
|
|
class CargoItemData(BaseModel):
|
|
"""购物车商品信息"""
|
|
goods_id: str
|
|
goods_name: str
|
|
goods_price: int
|
|
goods_image: str = ""
|
|
quantity: int = 1
|
|
|
|
|
|
class CargoInfoData(BaseModel):
|
|
"""购物车整体信息"""
|
|
items: list[CargoItemData]
|
|
total_goods_amount: int = 0
|
|
total_shipping_fee: int = 0
|
|
total_commission_fee: int = 0 # 邮购手续费
|
|
|
|
|
|
class OrderItem(BaseModel):
|
|
goods_id: str = Field(min_length=1, max_length=40)
|
|
goods_number: int = Field(ge=1, le=9999)
|
|
goods_price: int = Field(ge=0)
|
|
# 店铺标识: 空串 = 官方店铺;非空 = 加盟店标识(同名视为同一家加盟店)
|
|
supplier: str = ""
|
|
# 店铺代码,来自 other_shop_list 中的 tenpo_cd;非空 = 加盟店/市场店铺。
|
|
# 优先于 supplier 用于判定"同一家店铺"(同 tenpo_cd 视为同一家);未提供时回退使用 supplier。
|
|
tenpo_cd: str = ""
|
|
branch_number: str = "" # 分支编号,可选,与 tenpo_cd 搭配使用
|
|
|
|
|
|
class OrderShippingFeeRequest(BaseModel):
|
|
item_list: list[OrderItem] = Field(min_length=1)
|
|
|
|
|
|
class OrderShippingFeeData(BaseModel):
|
|
goods_amount: int
|
|
shipping_fee: int
|
|
order_amount: int
|
|
handle_amount: int
|
|
|
|
|
|
class PurchaseTaskRequest(BaseModel):
|
|
"""添加采购任务请求参数"""
|
|
model_config = ConfigDict(extra="allow")
|
|
|
|
task_id: str
|
|
topic: str
|
|
purchase_account: str
|
|
site_id: str
|
|
record_video: str
|
|
is_dev: str
|
|
payload: dict[str, Any]
|
|
|
|
class TradeMonitorRequest(BaseModel):
|
|
"""添加采购监控请求参数"""
|
|
model_config = ConfigDict(extra="allow")
|
|
|
|
node_id: str
|
|
payload: dict[str, Any]
|
|
is_dev: str
|
|
|
|
|
|
class CheckTaskCallbackUrlRequest(BaseModel):
|
|
"""测试回调地址请求参数"""
|
|
app_id: str = Field(min_length=1, max_length=100)
|
|
app_secret: str = Field(min_length=1, max_length=200)
|
|
callback_url: HttpUrl
|
|
data: dict[str, Any]
|
|
|
|
@field_validator("app_id", "app_secret", mode="before")
|
|
@classmethod
|
|
def strip_text_fields(cls, value: Any) -> Any:
|
|
if isinstance(value, str):
|
|
return value.strip()
|
|
return value
|
|
|
|
@field_validator("callback_url", mode="before")
|
|
@classmethod
|
|
def normalize_callback_url(cls, value: Any) -> Any:
|
|
if isinstance(value, str):
|
|
return value.strip().strip("`").strip()
|
|
return value
|
|
|
|
@field_validator("data")
|
|
@classmethod
|
|
def validate_data(cls, value: dict[str, Any]) -> dict[str, Any]:
|
|
if not value:
|
|
raise ValueError("data must not be empty")
|
|
|
|
try:
|
|
json.dumps(value, ensure_ascii=False)
|
|
except (TypeError, ValueError) as exc:
|
|
raise ValueError("data must be a valid JSON object") from exc
|
|
|
|
return value
|
|
|
|
|
|
class CheckTaskCallbackUrlData(BaseModel):
|
|
"""测试回调地址响应数据"""
|
|
response_status: int
|
|
response_body: Any
|
|
|
|
|
|
class NewsItemData(BaseModel):
|
|
"""单条「お知らせ」通知，由 worker 解析 action_news 页面后写入 Redis。"""
|
|
model_config = ConfigDict(extra="allow")
|
|
|
|
data_id: str # 站点侧通知唯一标识（单调递增）
|
|
date: str = "" # 通知日期，如 2026/06/12
|
|
message: str = "" # 通知文案，如 【販売】(取引番号：S2606115133)発送準備中です。
|
|
data_type: str = "" # 站点 data-type
|
|
data_url: str | None = None # 站点 data-url
|
|
trade_code: str | None = None # 从 data-url 解析的交易单号（非交易通知为 null）
|
|
account_id: str = "" # 该通知所属抓取账号
|
|
fetched_at: int = 0 # worker 抓取时间（epoch 秒）
|