279 lines
9.9 KiB
Python
279 lines
9.9 KiB
Python
#!/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 客户端示例")
|