python如何爬取快递100能查询的物流信息

发布时间：2021-10-26 09:42:46 作者：柒染
来源：亿速云阅读：447

# Python如何爬取快递100能查询的物流信息

## 前言

在电商和物流行业蓬勃发展的今天，物流信息的实时查询成为刚需。快递100作为国内知名的物流信息聚合平台，提供了丰富的API接口供开发者调用。本文将详细介绍如何使用Python爬取快递100的物流信息，涵盖从基础API调用到高级反反爬策略的完整解决方案。

---

## 一、准备工作

### 1.1 注册快递100开发者账号
访问[快递100开放平台](https://api.kuaidi100.com/)完成注册，获取以下关键信息：
- CustomerID（企业ID）
- API Key（接口密钥）

### 1.2 安装必备Python库
```bash
pip install requests pandas hashlib json time random

1.3 接口文档概览

快递100提供两种主要接口： - 实时查询API（免费版有QPS限制） - 订阅推送API（适合大规模查询）

二、基础API调用实现

2.1 实时查询接口

import requests
import hashlib
import json

def query_express(customer, key, com, num):
    url = "https://poll.kuaidi100.com/poll/query.do"
    
    # 构造请求参数
    param = {
        "com": com,        # 快递公司代码
        "num": num,       # 快递单号
        "phone": ""        # 收/寄件人手机号（部分快递需要）
    }
    
    # 生成签名
    sign_str = json.dumps(param) + key + customer
    sign = hashlib.md5(sign_str.encode()).hexdigest().upper()
    
    payload = {
        "customer": customer,
        "sign": sign,
        "param": json.dumps(param)
    }
    
    headers = {'Content-Type': "application/x-www-form-urlencoded"}
    response = requests.post(url, data=payload, headers=headers)
    return response.json()

# 使用示例
result = query_express(
    customer="YOUR_CUSTOMER_ID",
    key="YOUR_API_KEY",
    com="yuantong",  # 圆通快递代码
    num="YT123456789"
)
print(result)

2.2 返回数据结构解析

典型响应示例：

{
    "message": "ok",
    "state": "3",
    "status": "200",
    "data": [
        {
            "time": "2023-05-01 08:00:00",
            "context": "已签收，感谢使用圆通速递",
            "location": ""
        },
        {
            "time": "2023-04-30 15:30:00",
            "context": "[杭州市] 浙江杭州滨江公司派件员：张三（138****1234）正在为您派件",
            "location": "杭州"
        }
    ]
}

关键字段说明： - state: 物流状态（0-在途，1-揽收，2-疑难，3-签收，4-退签） - data: 物流轨迹数组（按时间倒序排列）

三、高级应用技巧

3.1 自动识别快递公司

当不确定快递公司时，使用智能识别接口：

def auto_detect_company(key, num):
    url = f"https://www.kuaidi100.com/autonumber/auto?num={num}&key={key}"
    response = requests.get(url)
    return response.json()  # 返回可能的快递公司列表

3.2 多单号批量查询

import pandas as pd

def batch_query(orders):
    results = []
    for order in orders:
        try:
            res = query_express(**order)
            res['订单号'] = order['num']
            results.append(res)
        except Exception as e:
            print(f"查询失败 {order['num']}: {str(e)}")
    return pd.DataFrame(results)

3.3 数据持久化存储

import sqlite3

def save_to_db(data):
    conn = sqlite3.connect('express.db')
    c = conn.cursor()
    
    # 创建表（如果不存在）
    c.execute('''CREATE TABLE IF NOT EXISTS logistics
                 (id INTEGER PRIMARY KEY AUTOINCREMENT,
                  order_no TEXT,
                  status TEXT,
                  update_time TEXT,
                  context TEXT)''')
    
    # 插入数据
    for item in data['data']:
        c.execute("INSERT INTO logistics VALUES (NULL,?,?,?,?)",
                 (data['num'], data['state'], item['time'], item['context']))
    
    conn.commit()
    conn.close()

四、反反爬策略实战

4.1 请求头伪装

headers = {
    'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
    'Referer': 'https://www.kuaidi100.com/',
    'Origin': 'https://www.kuaidi100.com'
}

4.2 请求频率控制

import time
import random

def safe_query(params):
    delay = random.uniform(1.2, 3.5)  # 随机延迟
    time.sleep(delay)
    return query_express(**params)

4.3 代理IP池实现

proxy_list = [
    "http://112.85.164.220:9999",
    "http://117.69.201.148:9999"
]

def rotate_proxy_request(url, params):
    proxy = {"http": random.choice(proxy_list)}
    return requests.post(url, proxies=proxy, **params)

五、异常处理与日志记录

5.1 常见异常类型

429错误：请求过于频繁
601错误：单号不存在
600错误：参数错误

5.2 健壮性增强代码

import logging

logging.basicConfig(filename='express.log', level=logging.INFO)

def robust_query(params):
    try:
        result = query_express(**params)
        if result.get('status') != '200':
            raise Exception(result.get('message', 'Unknown error'))
        return result
    except requests.exceptions.RequestException as e:
        logging.error(f"Request failed: {str(e)}")
    except Exception as e:
        logging.error(f"API error: {str(e)}")
    return None

六、完整项目示例

6.1 项目结构

/express-tracker
│── config.py       # 存储API密钥
│── tracker.py      # 核心查询逻辑
│── utils.py        # 工具函数
│── main.py         # 主程序入口
└── requirements.txt

6.2 主程序实现

# main.py
from tracker import batch_query
from utils import load_orders, save_results

if __name__ == "__main__":
    orders = load_orders("orders.csv")  # 从CSV加载待查询单号
    results = batch_query(orders)
    save_results(results, "output.xlsx")
    print(f"成功查询 {len(results)} 条物流信息")

七、法律与道德注意事项

遵守平台规则：严格遵循快递100的API使用条款
数据使用限制：不得将数据用于商业售卖等违规用途
查询频率控制：免费版限制为每分钟5次请求
用户隐私保护：妥善处理包含个人信息的物流数据

结语

通过本文介绍的方法，您可以构建一个功能完善的物流信息查询系统。建议在实际应用中： 1. 使用异步IO提高查询效率（如aiohttp） 2. 考虑使用官方推荐的订阅推送模式 3. 重要业务场景建议购买企业版API服务

完整代码示例已上传GitHub：[示例仓库链接]（此处替换为实际地址） “`

（注：实际字数约2180字，可根据需要调整细节部分）