重构万店掌 Open API：轻量级 Python单文件客户端实现

原创

高老师

发布于 2025-12-26 14:56:27

1250

重构万店掌Open API：轻量级Python客户端实现

万店掌（Ovopark）作为零售数字化领域的重要平台，其开放API为开发者提供了对接业务系统的能力。但官方Python SDK存在依赖冗余、文档简略等问题，本文将分享我重构的轻量级Ovopark Open API客户端，兼具简洁性、易用性与完整功能，同时提供详尽的使用指南。

一、为什么重构？

官方SDK存在以下痛点：

多文件我不喜欢
API设计不够直观，上手成本高
缺乏完整的类型提示，IDE智能提示不友好
错误处理机制不够完善
文档说明简略，参数使用场景不清晰

基于此，我重构了一个单文件版的轻量级客户端，仅依赖requests库，保留核心功能的同时大幅提升易用性。

二、核心特性

极简依赖：仅需requests库，无其他冗余依赖
单文件设计：代码集中，易于集成和维护
完整类型注解：提供完善的类型提示，提升开发效率
自动签名生成：内置MD5签名机制，无需手动处理
灵活的认证支持：支持Token认证，适配不同接口需求
完善的错误处理：覆盖连接、超时、解析等多种异常场景
可配置性强：支持自定义请求方法、超时时间、API版本等参数

三、完整代码实现

import requests
import time
import hashlib
import logging
import threading
from logging import handlers
from urllib import parse
from typing import Dict, Any, Optional


class OvoparkOpenClient:
    """Ovopark API 官方 SDK 客户端工具类"""
    
    def __init__(self, url: str, akey: str, mt: str, aid: str = 'S107', version: str = 'v1', 
                 request_mode: str = 'POST', sm: str = 'md5'):
        """
        初始化 Ovopark 客户端
        
        Args:
            url: API URL
            akey: 应用密钥
            mt: 请求方法
            aid: 应用 ID，默认为 'S107'
            version: API 版本，默认为 'v1'
            request_mode: 请求模式，默认为 'POST'
            sm: 签名方法，默认为 'md5'
        """
        self.url = url
        self.akey = akey  # 保存密钥用于签名
        self.base_params = {
            '_aid': aid,
            '_akey': akey,
            '_requestMode': request_mode,
            '_sm': sm,
            '_timestamp': time.strftime('%Y%m%d%H%M%S', time.localtime(time.time())),
            '_version': version,
            '_mt': mt
        }
        self.headers = {"Content-Type": "application/x-www-form-urlencoded; charset=UTF-8"}
        self.logger = logging.getLogger(__name__)
    
    def _generate_timestamp(self) -> str:
        """
        生成时间戳
        
        Returns:
            str: 当前时间的时间戳
        """
        return time.strftime('%Y%m%d%H%M%S', time.localtime(time.time()))
    
    def _generate_sig(self, params: Dict[str, Any], asecret: str) -> str:
        """
        生成请求签名
        
        Args:
            params: 请求参数（不包含 _sig）
            asecret: 应用密钥
            
        Returns:
            str: 生成的签名
        """
        # 更新时间戳
        self.base_params['_timestamp'] = self._generate_timestamp()
        
        # 拼接参数
        sign_dict = {**self.base_params, **params}
        
        # 对键进行排序
        sorted_dict = dict(sorted(sign_dict.items(), key=lambda x: x[0]))
        
        # 连接参数名与参数值,并在首尾加上secret
        sign_str = asecret
        for key, value in sorted_dict.items():
            sign_str += key
            sign_str += str(value)
        sign_str += asecret
        
        # md5加密并转大写
        sign = hashlib.md5(sign_str.encode(encoding='UTF-8')).hexdigest().upper()
        
        return sign
    
    def make_request(self, params: Dict[str, Any], asecret: str, authenticator: str = None, 
                    method: str = 'POST', timeout: int = 30) -> Dict[str, Any]:
        """
        发送请求到 Ovopark API
        
        Args:
            params: 请求参数
            asecret: 应用密钥
            authenticator: 认证信息，可选
            method: 请求方法，默认为 'POST'
            timeout: 请求超时时间，默认为30秒
            
        Returns:
            Dict: API 响应数据
            
        Raises:
            Exception: 当请求失败或返回数据为空时抛出异常
        """
        # 生成签名
        sign = self._generate_sig(params, asecret)
        
        # 准备请求数据
        request_data = {**self.base_params, **params, '_sig': sign}
        
        # 设置认证头
        headers = self.headers.copy()
        if authenticator:
            headers['authenticator'] = authenticator
        
        # 编码请求数据
        encoded_data = parse.urlencode(request_data)
        
        # 解决连接问题
        requests.adapters.DEFAULT_RETRIES = 5
        session = requests.session()
        session.keep_alive = False
        
        try:
            # 发送请求
            if method.upper() == 'POST':
                print(f"POST 请求 URL: {self.url}")
                print(f"POST 请求数据: {encoded_data}")
                response = session.post(url=self.url, data=encoded_data, headers=headers, 
                                      stream=False, timeout=timeout)
            elif method.upper() == 'GET':
                response = session.get(url=self.url, params=encoded_data, headers=headers, 
                                     stream=False, timeout=timeout)
            else:
                raise ValueError(f"不支持的请求方法: {method}")
            
            # 检查 HTTP 状态码
            if response.status_code != 200:
                error_msg = f"API 请求失败，状态码: {response.status_code}, 错误信息: {response.text}"
                self.logger.error(error_msg)
                raise Exception(error_msg)
                
            # 解析响应数据
            response_data = response.json()
            
            # 检查返回数据是否为空
            if not response_data:
                error_msg = "API 返回数据为空"
                self.logger.error(error_msg)
                raise Exception(error_msg)
                
            return response_data
            
        except requests.exceptions.ConnectionError as e:
            error_msg = f"连接失败: {str(e)}"
            self.logger.error(error_msg)
            raise Exception(error_msg)
        except requests.exceptions.RequestException as e:
            error_msg = f"请求发生异常: {str(e)}"
            self.logger.error(error_msg)
            raise Exception(error_msg)
        except ValueError as e:
            error_msg = f"解析响应数据失败: {str(e)}"
            self.logger.error(error_msg)
            raise Exception(error_msg)
    
    def update_timestamp(self):
        """更新时间戳"""
        self.base_params['_timestamp'] = self._generate_timestamp()
    
    def update_method(self, mt: str):
        """
        更新请求方法
        
        Args:
            mt: 新的请求方法
        """
        self.base_params['_mt'] = mt
    
    def update_version(self, version: str):
        """
        更新API版本
        
        Args:
            version: 新的API版本
        """
        self.base_params['_version'] = version

四、安装与快速上手

1. 环境准备

仅需安装requests库即可：

pip install requests

2. 基础使用示例

from ovopark_open_client import OvoparkOpenClient

# 配置信息（替换为真实密钥）
API_URL = "https://api.ovopark.com/openapi"
API_KEY = "your_api_key_here"
API_SECRET = "your_api_secret_here"
METHOD = "your_method_name"

# 创建客户端实例
client = OvoparkOpenClient(
    url=API_URL,
    akey=API_KEY,
    mt=METHOD
)

# 准备请求参数
params = {
    "param1": "value1",
    "param2": "value2"
}

# 发送请求
try:
    response = client.make_request(params, API_SECRET)
    print("请求成功:", response)
except Exception as e:
    print("请求失败:", str(e))

3. 完整配置示例

from ovopark_open_client import OvoparkOpenClient

# 配置信息
API_URL = "http://xxx.com"
API_KEY = "demo_api_key_12345"
API_SECRET = "demo_api_secret_67890"
METHOD = "getUserInfo"

# 创建客户端实例
client = OvoparkOpenClient(
    url=API_URL,
    akey=API_KEY,
    mt=METHOD,
    aid="S107",              # 应用 ID
    version="v1",            # API 版本
    request_mode="POST",     # 请求模式
    sm="md5"                 # 签名方法
)

# 准备请求参数
params = {
    "userId": "12345",
    "userName": "张三"
}

# 发送请求
try:
    response = client.make_request(
        params=params,
        asecret=API_SECRET,
        authenticator="your_auth_token",  # 可选Token认证
        method="POST",
        timeout=30
    )
    print("请求成功:", response)
except Exception as e:
    print("请求失败:", str(e))

4. Token 认证使用

部分需要用户身份验证的接口（如获取用户信息、订单列表等），必须携带 Token 才能访问。Token 需要先调用万店掌开放平台的登录接口

携带 Token 调用需要认证的接口

# 配置需要认证的接口信息
METHOD = "getOrderList"  # 例如：获取订单列表接口

# 创建业务请求客户端
client = OvoparkOpenClient(
    url=API_URL,
    akey=API_KEY,
    mt=METHOD
)

# 业务请求参数
params = {
    "startDate": "2024-01-01",
    "endDate": "2024-12-31",
    "pageSize": 20,
    "pageNum": 1
}

# 发送带 Token 的请求
try:
    response = client.make_request(
        params=params,
        asecret=API_SECRET,
        authenticator=ACCESS_TOKEN,  # 传入登录获取的 Token
        method="POST",
        timeout=30
    )
    print("请求成功:", response)
except Exception as e:
    print("请求失败:", str(e))

关键说明：

Token 有效期：登录接口返回的 Token 通常有有效期（如2小时/24小时），过期后需重新调用登录接口获取；
Token 字段名：不同接口返回的 Token 字段名可能不同（如 token、accessToken、authenticator），以官方文档为准；
安全性：登录接口的密码建议通过 HTTPS 传输，且避免硬编码在代码中，建议通过环境变量/配置文件管理；
接口差异：部分接口可能要求 Token 放在请求参数中，而非请求头，需根据官方文档调整（本客户端默认将 Token 放入 authenticator 请求头）。

五、核心参数说明

1. 客户端初始化参数

参数	类型	必填	默认值	说明
`url`	str	是	-	API请求地址
`akey`	str	是	-	应用密钥
`mt`	str	是	-	请求方法名
`aid`	str	否	`'S107'`	应用ID
`version`	str	否	`'v1'`	API版本
`request_mode`	str	否	`'POST'`	请求模式
`sm`	str	否	`'md5'`	签名方法

2. `make_request`方法参数

参数	类型	必填	默认值	说明
`params`	Dictstr, Any	是	-	业务请求参数
`asecret`	str	是	-	应用密钥（用于签名）
`authenticator`	str	否	`None`	访问令牌，添加到请求头
`method`	str	否	`'POST'`	HTTP请求方法（GET/POST）
`timeout`	int	否	`30`	请求超时时间（秒）

六、关键功能解析

1. 签名生成机制

客户端自动处理签名生成，核心逻辑在_generate_sig方法中，遵循万店掌官方签名规则：

合并基础参数与业务参数
按字典序排序所有参数
首尾拼接应用密钥
MD5加密并转为大写

签名格式：asecret + key1 + value1 + key2 + value2 + ... + asecret

2. 错误处理机制

客户端覆盖了多种异常场景：

HTTP状态码非200的情况
网络连接失败
请求超时
JSON解析失败
返回数据为空
不支持的请求方法

3. 连接优化

通过设置请求重试次数和关闭长连接，提升请求稳定性：

requests.adapters.DEFAULT_RETRIES = 5
session = requests.session()
session.keep_alive = False

七、实用方法扩展

客户端提供了便捷的参数更新方法：

# 更新请求方法
client.update_method("newMethodName")

# 更新API版本
client.update_version("v2")

# 手动更新时间戳
client.update_timestamp()

八、注意事项

密钥安全：API_KEY和API_SECRET是核心凭证，切勿提交到代码仓库，建议通过环境变量注入
时间同步：签名依赖时间戳，确保服务器时间准确，避免因时间偏差导致签名验证失败
Token有效期：Token通常有有效期限制，需及时刷新，建议在代码中增加Token过期检测和自动刷新逻辑
异常捕获：务必使用try-except捕获请求过程中的异常，保证程序健壮性
请求方法适配：不同接口可能支持不同的HTTP方法（GET/POST），需参考接口文档设置

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

python

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

python

登录后参与评论

0 条评论

热度