surugaya/app/models/scrape.py

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


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 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]] = []  # 面包屑


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


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 = ""


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