#!/usr/bin/env python3 """ 铱云易订货 OpenAPI 2.0 Python Client 用于查询客户、商品、订单、库存、收付款等业务数据 """ import os import json import time import requests from typing import Optional, Dict, Any, List # 默认配置 DEFAULT_BASE_URL = "https://openapi.77ircloud.com" DEFAULT_TIMEOUT = 30 class IrunClient: """铱云易订货 API 客户端""" def __init__(self, client_id: str = None, client_secret: str = None, access_token: str = None, base_url: str = DEFAULT_BASE_URL): """ 初始化客户端 Args: client_id: 开发者ID client_secret: 密钥 access_token: 访问令牌（可直接传入，跳过授权步骤） base_url: API基础地址 """ self.client_id = client_id or os.getenv("IRUN_CLIENT_ID") self.client_secret = client_secret or os.getenv("IRUN_CLIENT_SECRET") self.access_token = access_token or os.getenv("IRUN_ACCESS_TOKEN") self.base_url = base_url self._token_expires_at = 0 def _get_headers(self) -> Dict[str, str]: """获取请求头""" return { "Content-Type": "application/json; charset=UTF-8", "access_token": self.access_token, "x-client-id": self.client_id } def _request(self, method: str, endpoint: str, params: Dict = None, data: Dict = None) -> Dict[str, Any]: """ 发送API请求 Args: method: 请求方法 (GET, POST, PUT, DELETE) endpoint: API端点 params: URL参数 data: 请求体数据 Returns: API响应数据 """ url = f"{self.base_url}{endpoint}" headers = self._get_headers() try: response = requests.request( method=method, url=url, headers=headers, params=params, json=data, timeout=DEFAULT_TIMEOUT ) response.raise_for_status() return response.json() except requests.exceptions.RequestException as e: return {"code": 500, "message": str(e), "data": None} def get_token(self) -> Dict[str, Any]: """ 获取访问令牌 Returns: 包含access_token的响应 """ if not self.client_id or not self.client_secret: return {"code": 400, "message": "缺少 client_id 或 client_secret"} url = f"{self.base_url}/oauth/token" data = { "grant_type": "client_credentials", "client_id": self.client_id, "client_secret": self.client_secret } try: response = requests.post(url, json=data, timeout=DEFAULT_TIMEOUT) response.raise_for_status() result = response.json() if result.get("code") == 200: self.access_token = result["data"]["access_token"] self._token_expires_at = time.time() + result["data"].get("expires_in", 7200) return result except requests.exceptions.RequestException as e: return {"code": 500, "message": str(e), "data": None} # ==================== 客户管理 ==================== def get_customer_list(self, page_num: int = 1, page_size: int = 10, name: str = None, category_id: str = None) -> Dict[str, Any]: """获取客户列表""" params = {"pageNum": page_num, "pageSize": page_size} if name: params["name"] = name if category_id: params["categoryId"] = category_id return self._request("GET", "/openapi/customer/list", params=params) def get_customer_detail(self, customer_id: str) -> Dict[str, Any]: """获取客户详情""" return self._request("GET", f"/openapi/customer/{customer_id}") def get_customer_categories(self) -> Dict[str, Any]: """获取客户分类列表""" return self._request("GET", "/openapi/customer/category/list") # ==================== 商品管理 ==================== def get_product_list(self, page_num: int = 1, page_size: int = 10, name: str = None, category_id: str = None) -> Dict[str, Any]: """获取商品列表""" params = {"pageNum": page_num, "pageSize": page_size} if name: params["name"] = name if category_id: params["categoryId"] = category_id return self._request("GET", "/openapi/product/list", params=params) def get_product_detail(self, product_id: str) -> Dict[str, Any]: """获取商品详情""" return self._request("GET", f"/openapi/product/{product_id}") def get_category_list(self) -> Dict[str, Any]: """获取商品分类列表""" return self._request("GET", "/openapi/category/list") # ==================== 订单管理 ==================== def get_order_list(self, page_num: int = 1, page_size: int = 10, order_no: str = None, customer_id: str = None, start_time: int = None, end_time: int = None, status: str = None) -> Dict[str, Any]: """获取订单列表""" params = {"pageNum": page_num, "pageSize": page_size} if order_no: params["orderNo"] = order_no if customer_id: params["customerId"] = customer_id if start_time: params["startTime"] = start_time if end_time: params["endTime"] = end_time if status: params["status"] = status return self._request("GET", "/openapi/order/list", params=params) def get_order_detail(self, order_id: str) -> Dict[str, Any]: """获取订单详情""" return self._request("GET", f"/openapi/order/{order_id}") def get_refund_list(self, page_num: int = 1, page_size: int = 10, order_no: str = None, customer_id: str = None, start_time: int = None, end_time: int = None) -> Dict[str, Any]: """获取退单列表""" params = {"pageNum": page_num, "pageSize": page_size} if order_no: params["orderNo"] = order_no if customer_id: params["customerId"] = customer_id if start_time: params["startTime"] = start_time if end_time: params["endTime"] = end_time return self._request("GET", "/openapi/refund/list", params=params) def get_refund_detail(self, refund_id: str) -> Dict[str, Any]: """获取退单详情""" return self._request("GET", f"/openapi/refund/{refund_id}") # ==================== 进销存 ==================== def get_inventory_list(self, warehouse_id: str = None, product_id: str = None) -> Dict[str, Any]: """查询库存列表""" params = {} if warehouse_id: params["warehouseId"] = warehouse_id if product_id: params["productId"] = product_id return self._request("GET", "/openapi/inventory/list", params=params) def get_warehouse_list(self) -> Dict[str, Any]: """获取仓库列表""" return self._request("GET", "/openapi/warehouse/list") def get_storage_list(self, storage_type: str = None, start_time: int = None, end_time: int = None) -> Dict[str, Any]: """获取出入库记录""" params = {} if storage_type: params["type"] = storage_type if start_time: params["startTime"] = start_time if end_time: params["endTime"] = end_time return self._request("GET", "/openapi/storage/list", params=params) # ==================== 资金管理 ==================== def get_payment_list(self, page_num: int = 1, page_size: int = 10, payment_type: str = None, start_time: int = None, end_time: int = None) -> Dict[str, Any]: """获取收付款记录""" params = {"pageNum": page_num, "pageSize": page_size} if payment_type: params["type"] = payment_type if start_time: params["startTime"] = start_time if end_time: params["endTime"] = end_time return self._request("GET", "/openapi/payment/list", params=params) # ==================== 基础数据 ==================== def get_employee_list(self) -> Dict[str, Any]: """获取员工列表""" return self._request("GET", "/openapi/employee/list") def get_area_list(self) -> Dict[str, Any]: """获取省市数据""" return self._request("GET", "/openapi/area/province") def get_logistics_company_list(self) -> Dict[str, Any]: """获取物流公司列表""" return self._request("GET", "/openapi/logistics/company") # ==================== 便捷函数 ==================== def create_client(client_id: str = None, client_secret: str = None, access_token: str = None) -> IrunClient: """创建API客户端""" return IrunClient(client_id, client_secret, access_token) def get_timestamp(days_ago: int = 0) -> int: """获取时间戳""" import datetime return int((datetime.datetime.now() - datetime.timedelta(days=days_ago)).timestamp() * 1000) if __name__ == "__main__": # 示例用法 client = IrunClient() # 获取Token # result = client.get_token() # print(result) # 查询客户 # result = client.get_customer_list(page_num=1, page_size=10) # print(result) # 查询订单 # result = client.get_order_list(start_time=get_timestamp(30)) # print(result) print("铱云易订货 API 客户端示例")