阿里菜鸟物流云API调用--python

工作需要写一个程序调用菜鸟物流云五级地址查询api，此处决定用python实现。

 

调用实例如图：

接口文档：

先申请获得appkey与resourceid：

需要查询的数据集：

读取文件存入一个list：

def get_address():
    try:
        with open(os.getcwd() + r'address.txt', 'r') as f:
            address = f.readlines()
        return address
    except:
        print('打开文件失败')
        return ''

构造请求参数：

def get_raw_input(address):
    raw_input = {
        "address": address,
        "limit": "20"
    }
    return raw_input

构造公共请求参数：

def get_param(sign, content):
    param = {
        'msg_type': 'CNDZK_ADDRESS_QUERY',
        'data_digest': sign,
        'logistic_provider_id': 'd0119848ab......e1df5d8d6dc149',
        'logistics_interface': content
    }
    return param

计算请求签名（MD5值算法与base64编码）：

def get_data_digest(inputs, keys):
    m1 = hashlib.md5()
    m1.update((inputs + keys).encode('utf-8'))

    # base64.b64encode(m1.hexdigest()) 得到错误值!!!
    return base64.b64encode(m1.digest())

此处踩到几个坑卡了很久，主要是m1.hexdigest()与m1.digest()的不同，导致计算出的结果与java，js算出的不同

一开始以为是str转bytes的问题，注意python的str.encode('utf-8')与java的getBytes()结果在显示上有所不同（前者返回bytes类型，后者返回byte[]类型）

请求调用api并写入文件：

address_list = get_address()

for x in address_list:
    # unicode形态转中文，去掉空格（否则查不出）
    inputs = json.dumps(get_raw_input(x)).encode('utf-8').decode('unicode_escape').replace(' ', '')

    result = requests.post(url, data=get_param(get_data_digest(inputs, keys), inputs), headers=headers)
    result = result.content.decode(encoding='utf-8')
    # print(result)
    results = results + result + '
'


try:
    with open(os.getcwd()+r'
esult.txt', 'w') as f:
        f.write(results)
        print("写入成功")
except:
    print("写文件失败")

 此处又踩到一个小坑-。-，由于python3.x的str类型中文数据默认采用unicode编码显示，所以会得到中文变成u...的数据，导致查询失败

json.dumps()将dict转化为json字符串返回，先utf-8编码为bytes类型

再用decode('unicode_escape')解码让中文字符正常显示

同时请求参数中有空格也会导致查询失败（阿里api文档太简单什么都没说，参数只能自己慢慢调，差评）

所以此处用replace(' ','')除去空格

得到结果集：

完整代码：

# coding:utf-8
# by NeilShi 11/29/2017
import requests
import hashlib
import base64
import json
import os


# 'appkey': '154430',
# 'AppSecret': 'S7ib51Kp5......Qd10Lt4490QTwpr',
# 资源：d0119848ab5......df5d8d6dc149

url = 'http://link.cainiao.com/gateway/link.do'
keys = 'S7ib51Kp5O......d10Lt4490QTwpr'

headers = {
    'content-type': 'application/x-www-form-urlencoded; charset=UTF-8'
}

results = ''   # 存放结果集


def get_address():
    try:
        with open(os.getcwd() + r'address.txt', 'r') as f:
            address = f.readlines()
        return address
    except:
        print('打开文件失败')
        return ''


def get_raw_input(address):
    raw_input = {
        "address": address,
        "limit": "20"
    }
    return raw_input


def get_param(sign, content):
    param = {
        'msg_type': 'CNDZK_ADDRESS_QUERY',
        'data_digest': sign,
        'logistic_provider_id': 'd011984......d7ae1df5d8d6dc149',
        'logistics_interface': content
    }
    return param


def get_data_digest(inputs, keys):
    m1 = hashlib.md5()
    m1.update((inputs + keys).encode('utf-8'))

    # base64.b64encode(m1.hexdigest()) 得到错误值!!!原因未知
    return base64.b64encode(m1.digest())


address_list = get_address()

for x in address_list:
    # unicode形态转中文，去掉空格（否则查不出）
    inputs = json.dumps(get_raw_input(x)).encode('utf-8').decode('unicode_escape').replace(' ', '')

    result = requests.post(url, data=get_param(get_data_digest(inputs, keys), inputs), headers=headers)
    result = result.content.decode(encoding='utf-8')
    # print(result)
    results = results + result + '
'


try:
    with open(os.getcwd()+r'
esult.txt', 'w') as f:
        f.write(results)
        print("写入成功")
except:
    print("写文件失败")