"""RIIN 分销平台 REST API 客户端(移植自 @inkreach/riin-dis)。 Base URL: https://tshirt.riin.com 认证: username + MD5(password) -> Bearer token(Authorization 头),401 自动重登。 """ from __future__ import annotations import hashlib from typing import Any from pydantic import BaseModel, ConfigDict from pydantic.alias_generators import to_camel from api._common import TTLCache, chunk, parse_id_text, run_concurrent from api._common.http import TokenHttpClient from api.riin.models import ( Article, Color, CustomerListResponse, CustomerShopListResponse, ExpressInfo, FactoryInfoListResponse, LoginResult, OrderGoodsResponse, OrderListCountResponse, OrderLogResponse, OrderPageRequest, OrderPageResponse, Size, ThirdInterfaceCustomer, ) BASE_URL = "https://tshirt.riin.com" BATCH_SIZE = 500 class _ListResponse(BaseModel): """通用列表响应壳(辅助主数据用)。""" model_config = ConfigDict(extra="allow", populate_by_name=True, alias_generator=to_camel) successful: bool | None = None message: str | None = None error_code: str | None = None data: list[Any] = [] class _PageResponse(BaseModel): model_config = ConfigDict(extra="allow", populate_by_name=True, alias_generator=to_camel) successful: bool | None = None message: str | None = None error_code: str | None = None data: dict[str, Any] = {} class RiinDisClient(TokenHttpClient): """RIIN 分销系统客户端:认证 / 订单 / 店铺 / 辅助主数据。""" def __init__( self, username: str, password: str, *, cache_ttl_seconds: int | float = 300, concurrency: int = 10, ) -> None: super().__init__( BASE_URL, token_header="Authorization", fixed_headers={"X-Time-Zone": "UTC+8", "X-Requested-With": "XMLHttpRequest"}, login_markers=("auth/api/auth/login",), ) self._username = username self._password = password self._cache: TTLCache[Any] = TTLCache(cache_ttl_seconds) self._concurrency = concurrency # ----- 认证 ----- def _login(self) -> str: md5_password = hashlib.md5(self._password.encode()).hexdigest() data = self.post("auth/api/auth/login", body={"username": self._username, "password": md5_password}) result = LoginResult.model_validate(data["data"]) self._token = result.token return self._token # ----- 订单 ----- def _build_order_page_body(self, params: OrderPageRequest) -> dict[str, Any]: has_oid_or_courier = bool(params.platform_oid_list or params.courier_number_list) body: dict[str, Any] = { "pageIndex": params.page_index if params.page_index is not None else 1, "pageSize": params.page_size if params.page_size is not None else 20, "shopCodeList": params.shop_code_list or [], "orderStatus": params.order_status if params.order_status is not None else None, "dayRangeGroup": params.day_range_group if params.day_range_group is not None else None, "hasRefund": params.has_refund if params.has_refund is not None else False, } if not has_oid_or_courier: body["timeType"] = params.time_type if params.time_type is not None else 3 if params.start_time: body["startTime"] = params.start_time if params.end_time: body["endTime"] = params.end_time elif params.time_type is not None: body["timeType"] = params.time_type if params.platform_oid_list: body["platformOidList"] = parse_id_text(params.platform_oid_list) if params.courier_number_list: body["courierNumberList"] = parse_id_text(params.courier_number_list) if params.sub_id_list: body["subIdList"] = params.sub_id_list if params.goods_title: body["goodsTitle"] = params.goods_title if params.mapping_sku_code: body["mappingSkuCode"] = params.mapping_sku_code if params.platform_type is not None: body["platformType"] = params.platform_type if params.order_source is not None: body["orderSource"] = params.order_source if params.factory_ids: body["factoryIds"] = params.factory_ids if params.express_source is not None: body["expressSource"] = params.express_source if params.delivery_courier: body["deliveryCourier"] = params.delivery_courier if params.receiver_country: body["receiverCountry"] = params.receiver_country if params.region: body["region"] = params.region if params.receiver_province: body["receiverProvince"] = params.receiver_province if params.craft_type is not None: body["craftType"] = params.craft_type if params.style_code_list: body["styleCodeList"] = params.style_code_list if params.color_code_list: body["colorCodeList"] = params.color_code_list if params.size_code_list: body["sizeCodeList"] = params.size_code_list if params.seller_remark: body["sellerRemark"] = params.seller_remark if params.urgent_order is not None: body["urgentOrder"] = params.urgent_order if params.third_interface_customer_id: body["thirdInterfaceCustomerId"] = params.third_interface_customer_id if params.platform_status_list: body["platformStatusList"] = params.platform_status_list if params.is_third_flag is not None: body["isThirdFlag"] = params.is_third_flag if params.push_failed is not None: body["pushFailed"] = params.push_failed if params.assigned_fail_reason: body["assignedFailReason"] = params.assigned_fail_reason if params.push_fail_reason: body["pushFailReason"] = params.push_fail_reason if params.factory_remark_param: body["factoryRemarkParam"] = params.factory_remark_param if params.distributor_remark_param: body["distributorRemarkParam"] = params.distributor_remark_param return body def _get_single_page(self, params: OrderPageRequest) -> OrderPageResponse: body = self._build_order_page_body( params.model_copy(update={"page_index": params.page_index or 1, "page_size": params.page_size or 20}) ) data = self.post("distribution/order/page", body=body) return OrderPageResponse.model_validate(data) def _get_all_pages(self, params: OrderPageRequest) -> OrderPageResponse: page_size = params.page_size or 200 first = self._get_single_page(params.model_copy(update={"page_index": 1, "page_size": page_size})) total_count = first.data.total total_pages = -(-total_count // page_size) # ceil if total_pages <= 1: return first remaining_indices = list(range(2, total_pages + 1)) def fetch(i: int) -> OrderPageResponse: return self._get_single_page(params.model_copy(update={"page_index": i, "page_size": page_size})) remaining = run_concurrent([lambda i=i: fetch(i) for i in remaining_indices], limit=self._concurrency) records = list(first.data.records) for r in remaining: records.extend(r.data.records) return OrderPageResponse( successful=True, data={"records": records, "total": total_count}, ) def get_orders(self, params: OrderPageRequest | None = None) -> OrderPageResponse: """获取订单。支持三种模式:limit(取首页 N 条)/ all(全量分页)/ 单页。""" params = params or OrderPageRequest() needs_batching = ( (params.platform_oid_list and len(parse_id_text(params.platform_oid_list)) > BATCH_SIZE) or (params.courier_number_list and len(parse_id_text(params.courier_number_list)) > BATCH_SIZE) ) if needs_batching: return self._get_orders_batched(params) if params.limit: return self._get_single_page(params.model_copy(update={"page_index": 1, "page_size": params.limit})) if params.all: return self._get_all_pages(params) return self._get_single_page(params) def _get_orders_batched(self, params: OrderPageRequest) -> OrderPageResponse: oid_batches = chunk(parse_id_text(params.platform_oid_list), BATCH_SIZE) if params.platform_oid_list else [None] courier_batches = ( chunk(parse_id_text(params.courier_number_list), BATCH_SIZE) if params.courier_number_list else [None] ) def fetch(oid_batch: list[str] | None, courier_batch: list[str] | None) -> OrderPageResponse: update: dict[str, Any] = {} if oid_batch is not None: update["platform_oid_list"] = ",".join(oid_batch) if courier_batch is not None: update["courier_number_list"] = ",".join(courier_batch) batch_params = params.model_copy(update=update) if params.limit: return self._get_single_page( batch_params.model_copy(update={"page_index": 1, "page_size": params.limit}) ) return self._get_all_pages(batch_params) calls = [ (lambda o=o, c=c: fetch(o, c)) for o in oid_batches for c in courier_batches ] results = run_concurrent(calls, limit=self._concurrency) records: list[Any] = [] total = 0 for r in results: records.extend(r.data.records) total += r.data.total if params.limit: records = records[: params.limit] return OrderPageResponse(successful=True, data={"records": records, "total": total}) def get_order_goods(self, order_id: str) -> OrderGoodsResponse: data = self.post("distribution/order/goods", body={"orderId": order_id}) return OrderGoodsResponse.model_validate(data) def get_order_goods_batch(self, ids: list[str]) -> OrderGoodsResponse: """批量获取订单商品(distribution/order/goods/list)。 上游要求:单批 ≤200 个 id 且串行调用;高频/并发会被软限流(成功但空 data)。 """ data = self.post("distribution/order/goods/list", body={"ids": ids}) return OrderGoodsResponse.model_validate(data) def get_order_count( self, params: OrderPageRequest | None = None ) -> OrderListCountResponse: params = params or OrderPageRequest() cache_key = f"orderCount:{params.model_dump_json(exclude_none=True)}" cached = self._cache.get(cache_key) if cached is not None: return cached body = self._build_order_page_body(params) body.pop("pageIndex", None) body.pop("pageSize", None) data = self.post("distribution/order/list/count", body=body) result = OrderListCountResponse.model_validate(data) self._cache.set(cache_key, result) return result def get_order_logs(self, order_id: str, platform_oid: str) -> OrderLogResponse: data = self.post("distribution/order/logs", body={"orderId": order_id, "platformOid": platform_oid}) return OrderLogResponse.model_validate(data) # ----- 店铺 / 工厂 / 客户 ----- def get_customer_shop_list(self) -> CustomerShopListResponse: cache_key = "shops:customerShopList" cached = self._cache.get(cache_key) if cached is not None: return cached data = self.post("distribution/order/customer/shop/list", body={}) result = CustomerShopListResponse.model_validate(data) self._cache.set(cache_key, result) return result def get_factory_list(self) -> FactoryInfoListResponse: cache_key = "shops:factoryList" cached = self._cache.get(cache_key) if cached is not None: return cached data = self.get("distribution/tenant/getFactoryInfoList") result = FactoryInfoListResponse.model_validate(data) self._cache.set(cache_key, result) return result def get_customer_list(self) -> CustomerListResponse: cache_key = "shops:customerList" cached = self._cache.get(cache_key) if cached is not None: return cached data = self.post("distribution/customer/listTenantNoDisable", body={}) result = CustomerListResponse.model_validate(data) self._cache.set(cache_key, result) return result # ----- 辅助主数据 ----- def get_articles(self, params: dict[str, Any] | None = None) -> list[Article]: data = self.post("distribution/articleManagement/list", body=params or {}) resp = _ListResponse.model_validate(data) return [Article.model_validate(item) for item in resp.data] def get_sizes(self, params: dict[str, Any] | None = None) -> list[Size]: data = self.post("distribution/sizeManagement/size/list", body=params or {}) resp = _ListResponse.model_validate(data) return [Size.model_validate(item) for item in resp.data] def get_colors(self, params: dict[str, Any] | None = None) -> list[Color]: data = self.post("distribution/colorManagement/list", body=params or {}) resp = _ListResponse.model_validate(data) return [Color.model_validate(item) for item in resp.data] def get_third_party_customers( self, params: dict[str, Any] | None = None ) -> list[ThirdInterfaceCustomer]: data = self.post("distribution/thirdInterfaceCustomer/page", body=params or {}) resp = _PageResponse.model_validate(data) records = resp.data.get("records", []) if isinstance(resp.data, dict) else [] return [ThirdInterfaceCustomer.model_validate(item) for item in records] def get_express_info(self) -> list[ExpressInfo]: data = self.post("distribution/order/tenant/delivery/express", body={}) resp = _ListResponse.model_validate(data) return [ExpressInfo.model_validate(item) for item in resp.data]