鼎熠智控-物联网管理系统文档

搜索文档

搜索文档内容

鼎熠智控-物联网管理系统

SDK 示例

本文档提供多种编程语言的 SDK 客户端调用示例。所有示例都遵循同一套签名规则：按参数名排序、排除 signature、使用 timestamp 和 nonce 参与 HMAC-SHA256 签名。

可用 SDK

Python 客户端

适用于后端服务、脚本和自动化任务。

Go 客户端

适用于服务端集成和命令行工具。

Node.js 客户端

适用于 Node.js 服务端和构建脚本。

PHP 客户端

适用于 PHP Web 应用后端。

Java 客户端

适用于 Java 服务端集成。

快速开始

1. 初始化客户端

# Python
client = WLWOpenAPIClient(
    base_url="https://api.example.com",
    access_key="ak_xxxxxxxxxx",
    secret_key="sk_xxxxxxxxxx"
)
// Go
client := NewWLWOpenAPIClient(
    "https://api.example.com",
    "ak_xxxxxxxxxx",
    "sk_xxxxxxxxxx"
)
// Node.js
const client = new WLWOpenAPIClient(
    'https://api.example.com',
    'ak_xxxxxxxxxx',
    'sk_xxxxxxxxxx'
);
// PHP
$client = new WLWOpenAPIClient(
    'https://api.example.com',
    'ak_xxxxxxxxxx',
    'sk_xxxxxxxxxx'
);
// Java
WLWOpenAPIClient client = new WLWOpenAPIClient(
    "https://api.example.com",
    "ak_xxxxxxxxxx",
    "sk_xxxxxxxxxx"
);

2. 调用接口

# Python
try:
    card_info = client.get_card_info("CARD123456789")
    print(f"余额: {'{'}card_info['balance']{'}'}")
except APIError as e:
    print(f"错误: {'{'}e.message{'}'}")
// Go
cardInfo, err := client.GetCardInfo("CARD123456789")
if err != nil {'{'}
    log.Printf("错误: %v", err)
{'}'} else {'{'}
    fmt.Printf("余额: %.2f\n", cardInfo.Balance)
{'}'}
// Node.js
try {'{'}
    const cardInfo = await client.getCardInfo('CARD123456789');
    console.log(`余额: ${'{'}cardInfo.balance{'}'}`);
{'}'} catch (error) {'{'}
    console.error(`错误: ${'{'}error.message{'}'}`);
{'}'}
// PHP
try {'{'}
    $cardInfo = $client->getCardInfo('CARD123456789');
    echo "余额: " . $cardInfo['balance'];
{'}'} catch (APIError $e) {'{'}
    echo "错误: " . $e->getMessage();
{'}'}
// Java
try {'{'}
    CardInfo cardInfo = client.getCardInfo("CARD123456789");
    System.out.println("余额: " + cardInfo.getBalance());
{'}'} catch (APIError e) {'{'}
    System.err.println("错误: " + e.getMessage());
{'}'}

主要功能

接入 SDK 时建议封装以下核心能力：

基础功能

  • ✅ HMAC-SHA256 签名生成
  • ✅ 自动参数排序和签名
  • ✅ 请求/响应处理
  • ✅ 错误处理和异常封装

接口方法

  • heartbeat() - 心跳检测
  • getAgentBalance() - 查询当前代理余额
  • listConnectablePackages(packageCode) - 获取可对接套餐列表
  • getCardInfo(cardNo) - 查询卡板信息
  • getCardRealNameUrl(cardNo) - 获取卡板实名链接
  • refreshCardFlow(cardNo) - 刷新卡板流量
  • restartCard(cardNo) - 卡板复机
  • stopCard(cardNo) - 卡板停机
  • orderCardPackage(cardNo, packageCode, strategy) - 卡板套餐订购
  • getMobileToken(channelKey) - 通过渠道密钥获取移动 Token（无需签名）
  • getDeviceInfo(deviceNo) - 查询设备信息
  • getDeviceRealNameUrls(deviceNo) - 获取设备实名链接
  • orderDevicePackage(deviceNo, packageCode, strategy) - 设备套餐订购
  • sendDeviceCommand(deviceNo, command, params) - 执行设备操作指令

错误处理

所有 SDK 都实现了统一的错误处理机制：

错误类型 错误码 说明
认证失败 InvalidSignature、InvalidTimestamp、InvalidAccessKey 检查签名、时间戳和密钥
权限不足 PermissionDenied、Forbidden 检查池中池余额、账单提取配置的对象绑定和禁止接口动作，或联系管理员
频率限制 RateLimited 降低请求频率或实现重试
业务错误 其他错误码 查看错误消息详情

最佳实践

  1. 密钥管理

    • 使用环境变量存储密钥
    • 不要在代码中硬编码密钥
    • 定期轮换密钥

  2. 错误处理

    • 捕获并处理所有异常
    • 实现重试逻辑（指数退避）
    • 记录详细的错误日志

  3. 性能优化

    • 复用客户端实例
    • 设置合理的超时时间
    • 实现连接池（适用于高并发场景）

  4. 监控告警

    • 监控 API 调用成功率
    • 监控响应时间
    • 设置异常告警

更多资源

完整 SDK 源码

来自后端 demo 的 SDK 示例文件。

Python SDKpython.txt
import hmac
import hashlib
import time
import uuid
import json
import requests
from urllib.parse import urlencode

class WLWClient:
    def __init__(self, base_url, access_key, secret_key):
        self.base_url = base_url
        self.access_key = access_key
        self.secret_key = secret_key
        self.session = requests.Session()
        self.session.timeout = 30

    def _generate_signature(self, method, path, params):
        # 排序参数
        sorted_params = sorted(params.items())
        if 'signature' in dict(sorted_params):
            sorted_params = [(k, v) for k, v in sorted_params if k != 'signature']

        # 构造参数字符串
        param_str = '&'.join([f"{k}={v}" for k, v in sorted_params])

        # 构造签名字符串
        sign_str = f"{method}{path}{params['timestamp']}{params['nonce']}"
        if param_str:
            sign_str += f"{param_str}"

        # HMAC-SHA256 签名
        return hmac.new(
            self.secret_key.encode(),
            sign_str.encode(),
            hashlib.sha256
        ).hexdigest()

    def heartbeat(self):
        params = {
            'access_key': self.access_key,
            'timestamp': str(int(time.time())),
            'nonce': str(uuid.uuid4())
        }

        params['signature'] = self._generate_signature('GET', '/api/v1/heartbeat', params)

        url = f"{self.base_url}/api/v1/heartbeat"
        response = self.session.get(url, params=params)

        return response.json()

    def get_card_info(self, card_no):
        params = {
            'access_key': self.access_key,
            'card_no': card_no,
            'timestamp': str(int(time.time())),
            'nonce': str(uuid.uuid4())
        }

        params['signature'] = self._generate_signature('GET', '/api/v1/info', params)

        url = f"{self.base_url}/api/v1/info"
        response = self.session.get(url, params=params)

        return response.json()

    def restart_card(self, card_no):
        params = {
            'access_key': self.access_key,
            'card_no': card_no,
            'timestamp': str(int(time.time())),
            'nonce': str(uuid.uuid4())
        }

        params['signature'] = self._generate_signature('POST', '/api/v1/start', params)

        url = f"{self.base_url}/api/v1/start"
        response = self.session.post(url, data=params)

        return response.json()

    def stop_card(self, card_no):
        params = {
            'access_key': self.access_key,
            'card_no': card_no,
            'timestamp': str(int(time.time())),
            'nonce': str(uuid.uuid4())
        }

        params['signature'] = self._generate_signature('POST', '/api/v1/stop', params)

        url = f"{self.base_url}/api/v1/stop"
        response = self.session.post(url, data=params)

        return response.json()

    def order_package(self, card_no, package_code, strategy):
        params = {
            'access_key': self.access_key,
            'card_no': card_no,
            'package_code': package_code,
            'strategy': str(strategy),
            'timestamp': str(int(time.time())),
            'nonce': str(uuid.uuid4())
        }

        params['signature'] = self._generate_signature('POST', '/api/v1/orders', params)

        url = f"{self.base_url}/api/v1/orders"
        response = self.session.post(url, data=params)

        return response.json()

# 使用示例
if __name__ == "__main__":
    client = WLWClient(
        base_url="http://localhost:8080",
        access_key="your_access_key",
        secret_key="your_secret_key"
    )

    try:
        # 心跳检测
        result = client.heartbeat()
        print("心跳检测成功:", result)
    except Exception as e:
        print("心跳检测失败:", e)

    try:
        # 查询卡板信息
        result = client.get_card_info("CARD123456789")
        print("卡板信息:", result)
    except Exception as e:
        print("查询卡板信息失败:", e)

    try:
        # 卡板复机
        result = client.restart_card("CARD123456789")
        print("卡板复机:", result)
    except Exception as e:
        print("卡板复机失败:", e)

    try:
        # 卡板停机
        result = client.stop_card("CARD123456789")
        print("卡板停机:", result)
    except Exception as e:
        print("卡板停机失败:", e)

    try:
        # 订购套餐
        result = client.order_package("CARD123456789", "pkg_d4k9...", 1)
        print("订购套餐:", result)
    except Exception as e:
        print("订购套餐失败:", e)