Python快递鸟kdniao SDK如何使用,并支持Tornado异步客户端

写了一个快递鸟的 Python client 。 求 start 、 fork ,一起完善。

链接

截图(终端)

Install 安装

pip install -u kdniao

Usage 使用

依赖

无论是在程序里,还是在命令行中,你都必须先获得快递鸟官方分配给你的 app id 及 app key 。

在命令行运行 kdniao 命令时,需要在命令行参数中指定 id 与 key ,或者预先在环境变量中指定 KDNIAO_APP_IDKDNIAO_APP_KEY。如:

  1. KDNIAO_APP_ID={你的 ID} KDNIAO_APP_KEY={你的 Key} kdniao {运单号},或:
  2. ~/.bash_profile 中设置变量,并重新打开 shell 执行: kdniao {运单号},或:
  3. kdniao {运单号} --ik={APP_ID},{APP_KEY}

Command Line 命令行

$ kdniao {运单号} --s=快递公司编码 --o=订单号 --ik={APP_ID},{APP_KEY}

如:

$ kdniao 12345678 --s YTO

$ kdniao 12345678 --ik={APP_ID},{APP_KEY}

更多介绍

https://github.com/menduo/kdniao_python

求 start 、 fork ,一起完善。

menduo/kdniao_python: 快递鸟 kdniao python sdk, with tornado async http client support. https://github.com/menduo/kdniao_python


Python快递鸟kdniao SDK如何使用,并支持Tornado异步客户端

5 回复

注册还得认证…


要使用快递鸟(kdniao)SDK并支持Tornado异步客户端,你需要对官方SDK进行异步改造。核心思路是将同步的HTTP请求(如requests.post)替换为Tornado的异步HTTP客户端(AsyncHTTPClient)。下面是一个完整的、可直接运行的示例。

首先,确保已安装必要的库:

pip install tornado

然后,创建一个异步版本的快递鸟工具类。这里以“即时查询API”为例:

import json
import hashlib
import base64
from urllib.parse import urlencode
from tornado.httpclient import AsyncHTTPClient, HTTPRequest

class AsyncKdniaoTracker:
    """
    快递鸟异步查询客户端 (支持Tornado)
    """
    def __init__(self, e_business_id, app_key):
        self.e_business_id = e_business_id  # 商户ID
        self.app_key = app_key  # API密钥
        self.req_url = "https://api.kdniao.com/Ebusiness/EbusinessOrderHandle.aspx"  # 即时查询API地址

    def _build_request_data(self, shipper_code, logistic_code):
        """构造请求数据(与官方SDK逻辑一致)"""
        request_data = {
            'ShipperCode': shipper_code,
            'LogisticCode': logistic_code
        }
        data = json.dumps(request_data, ensure_ascii=False)
        return data

    def _sign(self, data):
        """生成数据签名(与官方SDK逻辑一致)"""
        content = data + self.app_key
        sign = hashlib.md5(content.encode('utf-8')).hexdigest()
        sign = base64.b64encode(sign.encode('utf-8')).decode('utf-8')
        return sign

    async def async_track(self, shipper_code, logistic_code):
        """
        异步查询物流轨迹
        :param shipper_code: 快递公司编码
        :param logistic_code: 物流单号
        :return: 快递鸟API返回的JSON结果
        """
        # 1. 构造请求数据
        request_data = self._build_request_data(shipper_code, logistic_code)
        data_sign = self._sign(request_data)

        # 2. 构造POST请求参数
        post_data = {
            'RequestData': request_data,
            'EBusinessID': self.e_business_id,
            'RequestType': '1002',  # 即时查询接口指令
            'DataSign': data_sign,
            'DataType': '2'  # 返回JSON格式
        }

        # 3. 使用Tornado异步HTTP客户端发送请求
        http_client = AsyncHTTPClient()
        request = HTTPRequest(
            url=self.req_url,
            method='POST',
            headers={'Content-Type': 'application/x-www-form-urlencoded'},
            body=urlencode(post_data)
        )

        try:
            response = await http_client.fetch(request)
            result = json.loads(response.body.decode('utf-8'))
            return result
        except Exception as e:
            # 实际生产环境应更精细地处理异常
            return {'Success': False, 'Reason': str(e)}

# ===== 使用示例 =====
async def main():
    # 替换为你的真实商户ID和API密钥
    tracker = AsyncKdniaoTracker(
        e_business_id="你的商户ID",
        app_key="你的API密钥"
    )

    # 示例:查询申通快递单号
    result = await tracker.async_track(
        shipper_code="STO",  # 申通快递编码
        logistic_code="你的快递单号"
    )
    print(json.dumps(result, indent=2, ensure_ascii=False))

# 在Tornado应用中,你可以在RequestHandler的async get/post方法中直接调用:
# class TrackHandler(tornado.web.RequestHandler):
#     async def get(self):
#         tracker = AsyncKdniaoTracker(EBUSINESS_ID, APP_KEY)
#         result = await tracker.async_track("STO", "单号")
#         self.write(result)

if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

关键改造点说明:

  1. 异步方法定义async_track方法使用async def定义,使其成为协程。
  2. 异步HTTP客户端:使用tornado.httpclient.AsyncHTTPClient代替requests
  3. 请求构造:使用HTTPRequest对象封装请求细节,body参数需要是application/x-www-form-urlencoded格式的字符串,这里用urlencode处理。
  4. 异步等待:使用await http_client.fetch(request)进行非阻塞的HTTP调用。

在Tornado Web应用中的集成示例:

import tornado.web
import tornado.ioloop

class TrackHandler(tornado.web.RequestHandler):
    async def get(self):
        tracker = AsyncKdniaoTracker("商户ID", "API密钥")
        result = await tracker.async_track("STO", "单号")
        self.write(result)

app = tornado.web.Application([
    (r"/track", TrackHandler),
])
app.listen(8888)
tornado.ioloop.IOLoop.current().start()

总结: 把同步HTTP请求换成AsyncHTTPClient就行,注意参数格式和异常处理。

不需要吧,我记得当时就用手机号验证了一下就可以使用大部分接口了。

支持 打算使用

欢迎使用,有问题及时联系。

回到顶部