"""SDS POD 平台 REST API 客户端(移植自 @inkreach/sds-mng)。 Base URL: https://pod-api.sdspod.com 认证: username + password + 公司名(印美达) -> access-token 头,401 自动重登。 """ from __future__ import annotations import time from typing import Any from api._common import TTLCache, chunk, run_concurrent, split_lines from api._common.http import TokenHttpClient from api.sds.models import ( FiltratePanelResponse, IssuingBayAreaResponse, LogisticsItem, LogisticsPageRequest, LogisticsPageResponse, LoginResult, OrderPageRequest, OrderPageResponse, OrderQueryParams, OrderStatistics, OrderStatus, ProductListRequest, ProductListResponse, ) BASE_URL = "https://pod-api.sdspod.com" COMPANY_NO = "印美达" KEYWORDS_BATCH_SIZE = 200 FILTERS_FIELDS = ( "factoryList,merchantList,logisticsList,earlyWarningList," "issuingBayAreaList,logisticsSourceList,extendStatusList" ) EXTRA_INFO = ( "eyJpcCI6IjM4LjcxLjEyNy4xODkiLCJsb2NhdGlvbiI6eyJsYXQiOjM4Ljg4MzMsImxuZyI6LTc3fSwiYWRfaW5mbyI6" "eyJuYXRpb24iOiLnrKw0OSIsInByb3ZpbmNlIjoiIiwiY2l0eSI6IiIsImRpc3RyaWN0IjoiIiwiYWRjb2RlIjotMSwi" "bmF0aW9uX2NvZGUiOjg0MH0sImZpbmdlcnByaW50SWQiOiI3YzdjOWUwNTFhZTY0OTM5MTMwZjk5YjQ2OGJjYWFhNSIs" "InVzZXJBZ2VudCI6Ik1vemlsbGEvNS4wIChXaW5kb3dzIE5UIDEwLjA7IFdpbjY0OyB4NjQpIEFwcGxlV2ViS2l0LzUz" "Ny4zNiAoS0hUTUwsIGxpa2UgR2Vja28pIENocm9tZS8xMzguMC4wLjAgU2FmYXJpLzUzNy4zNiIsImhyZWYiOiJodHRw" "czovL21hbmFnZS5zZHNwb2QuY29tL3VzZXIvbG9naW4ifQ==" ) class SdsMngClient(TokenHttpClient): """SDS POD 系统客户端:认证 / 订单 / 产品 / 物流。""" def __init__( self, username: str, password: str, *, cache_ttl_seconds: int | float = 300, concurrency: int = 10, ) -> None: super().__init__( BASE_URL, token_header="access-token", fixed_headers=None, login_markers=("/pod/auth/login",), ) self._username = username self._password = password self._cache: TTLCache[Any] = TTLCache(cache_ttl_seconds) self._concurrency = concurrency # ----- 认证 ----- def _login(self) -> str: data = self.post( f"/pod/auth/login?t={int(time.time() * 1000)}", body={ "no": COMPANY_NO, "password": self._password, "username": self._username, "extraInfo": EXTRA_INFO, }, ) result = LoginResult.model_validate(data["data"]) self._token = result.access_token return self._token # ----- 订单 ----- @staticmethod def _build_order_params(params: OrderPageRequest) -> dict[str, Any]: query: dict[str, Any] = {"productionType": 2, "t": int(time.time() * 1000)} if params.size is not None: query["size"] = params.size if params.page is not None: query["page"] = params.page if params.keywords is not None: query["keywords"] = params.keywords if params.created_date_begin is not None: query["createdDateBegin"] = params.created_date_begin if params.created_date_end is not None: query["createdDateEnd"] = params.created_date_end if params.pay_date_begin is not None: query["payDateBegin"] = params.pay_date_begin if params.pay_date_end is not None: query["payDateEnd"] = params.pay_date_end if params.confirm_date_begin is not None: query["confirmDateBegin"] = params.confirm_date_begin if params.confirm_date_end is not None: query["confirmDateEnd"] = params.confirm_date_end if params.finish_date_begin is not None: query["finishDateBegin"] = params.finish_date_begin if params.finish_date_end is not None: query["finishDateEnd"] = params.finish_date_end if params.order_status_code is not None: query["orderStatusCode"] = params.order_status_code if params.product_keywords is not None: query["productKeywords"] = params.product_keywords if params.order_time_range_type is not None: query["orderTimeRangeType"] = params.order_time_range_type if params.factory_ids is not None: query["factoryIds"] = params.factory_ids if params.merchant_ids is not None: query["merchantIds"] = params.merchant_ids if params.logistics_ids is not None: query["logisticsIds"] = params.logistics_ids if params.issuing_bay_area_id is not None: query["issuingBayAreaId"] = params.issuing_bay_area_id if params.extend_status is not None: query["extendStatus"] = params.extend_status if params.early_warning_type is not None: query["earlyWarningType"] = params.early_warning_type if params.item_num_type is not None: query["itemNumType"] = params.item_num_type return query @staticmethod def _build_query_params(params: OrderQueryParams | None) -> dict[str, Any]: params = params or OrderQueryParams() query: dict[str, Any] = {"t": int(time.time() * 1000)} for py_name, api_name in [ ("created_date_begin", "createdDateBegin"), ("created_date_end", "createdDateEnd"), ("pay_date_begin", "payDateBegin"), ("pay_date_end", "payDateEnd"), ("confirm_date_begin", "confirmDateBegin"), ("confirm_date_end", "confirmDateEnd"), ("finish_date_begin", "finishDateBegin"), ("finish_date_end", "finishDateEnd"), ("order_time_range_type", "orderTimeRangeType"), ("product_keywords", "productKeywords"), ("factory_ids", "factoryIds"), ("merchant_ids", "merchantIds"), ("logistics_ids", "logisticsIds"), ("issuing_bay_area_id", "issuingBayAreaId"), ("extend_status", "extendStatus"), ("early_warning_type", "earlyWarningType"), ("item_num_type", "itemNumType"), ]: value = getattr(params, py_name, None) if value: query[api_name] = value return query def _get_single_page(self, params: OrderPageRequest) -> OrderPageResponse: query = self._build_order_params( params.model_copy(update={"size": params.size or 50, "page": params.page or 1}) ) data = self.get("/pod/orders/page", params=query) return OrderPageResponse.model_validate(data) def _get_all_pages(self, params: OrderPageRequest) -> OrderPageResponse: page_size = params.size or 200 first = self._get_single_page(params.model_copy(update={"size": page_size, "page": 1})) total_pages = -(-first.total_count // page_size) if total_pages <= 1: return first def fetch(i: int) -> OrderPageResponse: return self._get_single_page(params.model_copy(update={"size": page_size, "page": i})) remaining = run_concurrent( [lambda i=i: fetch(i) for i in range(2, total_pages + 1)], limit=self._concurrency ) merged = list(first.records) for r in remaining: merged.extend(r.records) return OrderPageResponse(total_count=first.total_count, records=merged) def get_orders(self, params: OrderPageRequest | None = None) -> OrderPageResponse: """获取订单。keywords>200 自动分批;limit 取首页 N 条;all 全量分页。""" params = params or OrderPageRequest() if params.keywords: return self._get_orders_with_keywords(params) if params.limit: return self._get_single_page(params.model_copy(update={"size": params.limit, "page": 1})) if params.all: return self._get_all_pages(params) return self._get_single_page(params) def _get_orders_with_keywords(self, params: OrderPageRequest) -> OrderPageResponse: batches = chunk(split_lines(params.keywords or ""), KEYWORDS_BATCH_SIZE) or [[]] def fetch(batch: list[str]) -> OrderPageResponse: batch_params = params.model_copy(update={"keywords": "\n".join(batch)}) if params.limit: return self._get_single_page( batch_params.model_copy(update={"size": params.limit, "page": 1}) ) return self._get_all_pages(batch_params) results = run_concurrent([lambda b=b: fetch(b) for b in batches], limit=self._concurrency) merged: list[Any] = [] total = 0 for r in results: merged.extend(r.records) total += r.total_count if params.limit: merged = merged[: params.limit] return OrderPageResponse(total_count=total, list=merged) def get_filters(self, params: OrderQueryParams | None = None) -> FiltratePanelResponse: cache_key = f"filters:{(params.model_dump_json(exclude_none=True)) if params else '{}'}" cached = self._cache.get(cache_key) if cached is not None: return cached query = {"fields": FILTERS_FIELDS, **self._build_query_params(params)} data = self.get("/pod/orders/filtratePanel", params=query) result = FiltratePanelResponse.model_validate(data) self._cache.set(cache_key, result) return result def get_status(self, params: OrderQueryParams | None = None) -> list[OrderStatus]: cache_key = f"status:{(params.model_dump_json(exclude_none=True)) if params else '{}'}" cached = self._cache.get(cache_key) if cached is not None: return cached query = {"productionType": 2, **self._build_query_params(params)} data = self.get("/pod/orders/statistics", params=query) result = [OrderStatus.model_validate(item) for item in data] self._cache.set(cache_key, result) return result def get_statistics( self, params: OrderQueryParams | None = None, production_type: int = 2 ) -> list[OrderStatistics]: params = params or OrderQueryParams() cache_key = f"statistics:{production_type}:{params.model_dump_json(exclude_none=True)}" cached = self._cache.get(cache_key) if cached is not None: return cached query = { "fields": FILTERS_FIELDS, "productionType": production_type, **self._build_query_params(params), } data = self.get("/pod/orders/statistics", params=query) result = [OrderStatistics.model_validate(item) for item in data] self._cache.set(cache_key, result) return result # ----- 产品 ----- def get_products(self, params: ProductListRequest | None = None) -> ProductListResponse: params = params or ProductListRequest() if params.limit: return self._get_products_page(params.model_copy(update={"size": params.limit, "page": 1})) if params.all: return self._get_all_products(params) return self._get_products_page(params) def _get_products_page(self, params: ProductListRequest) -> ProductListResponse: query = { "keyword": params.keyword or "", "page": params.page or 1, "size": params.size or 50, "publicStatus": params.public_status if params.public_status is not None else 1, "t": int(time.time() * 1000), } data = self.get("/pod/product/self", params=query) return ProductListResponse.model_validate(data) def _get_all_products(self, params: ProductListRequest) -> ProductListResponse: page_size = params.size or 200 first = self._get_products_page(params.model_copy(update={"size": page_size, "page": 1})) total_pages = -(-first.total_count // page_size) if total_pages <= 1: return first remaining = run_concurrent( [ lambda i=i: self._get_products_page(params.model_copy(update={"size": page_size, "page": i})) for i in range(2, total_pages + 1) ], limit=self._concurrency, ) merged = list(first.records) for r in remaining: merged.extend(r.records) return ProductListResponse(total_count=first.total_count, records=merged) # ----- 物流 ----- def get_logistics_areas(self) -> IssuingBayAreaResponse: cache_key = "logistics:areas" cached = self._cache.get(cache_key) if cached is not None: return cached data = self.get("/pod/issuing_bay_areas", params={"t": int(time.time() * 1000)}) result = IssuingBayAreaResponse.model_validate(data) self._cache.set(cache_key, result) return result def get_logistics_page(self, params: LogisticsPageRequest) -> LogisticsPageResponse: if params.limit: return self._get_logistics_single( params.model_copy(update={"size": params.limit, "page": 1}) ) if params.all: return self._get_logistics_all(params) return self._get_logistics_single(params) def _get_logistics_single(self, params: LogisticsPageRequest) -> LogisticsPageResponse: query = { "keyword": params.keyword or "", "deliveryMethod": params.delivery_method or "FBM", "issuingBayAreaId": params.issuing_bay_area_id, "page": params.page or 1, "size": params.size or 50, "t": int(time.time() * 1000), } data = self.get("/pod/logistics/page", params=query) return LogisticsPageResponse.model_validate(data) def _get_logistics_all(self, params: LogisticsPageRequest) -> LogisticsPageResponse: page_size = params.size or 200 first = self._get_logistics_single( params.model_copy(update={"size": page_size, "page": 1}) ) total_count = first.total_count or first.total_count_alt or 0 total_pages = -(-total_count // page_size) if total_pages <= 1: return first remaining = run_concurrent( [ lambda i=i: self._get_logistics_single(params.model_copy(update={"size": page_size, "page": i})) for i in range(2, total_pages + 1) ], limit=self._concurrency, ) merged: list[LogisticsItem] = list(first.items) for r in remaining: merged.extend(r.items) return LogisticsPageResponse( items=merged, page=1, size=len(merged), total_count=total_count )