如何编写物流信息查询脚本

wen 实用脚本 2026-06-28 27

本文目录导读：

如何编写物流信息查询脚本

使用快递鸟API（推荐）
使用快递100API
使用爬虫方式（仅用于学习）
批量查询脚本（带缓存）
命令行工具版本
注意事项

使用快递鸟API（推荐）

import requests
import json
class LogisticsQuery:
    def __init__(self, app_id, app_key):
        self.app_id = app_id
        self.app_key = app_key
        self.url = "https://api.kdniao.com/Ebusiness/EbusinessOrderHandle.aspx"
    def query(self, exp_code, exp_no):
        """查询物流信息
        exp_code: 快递公司编码 (如 SFEXPRESS, YTO)
        exp_no: 运单号
        """
        request_data = {
            "OrderCode": "",
            "ShipperCode": exp_code,
            "LogisticCode": exp_no
        }
        # 生成签名
        data_str = json.dumps(request_data, separators=(',',':'))
        sign = self._build_sign(data_str)
        params = {
            "RequestType": "1002",
            "EBusinessID": self.app_id,
            "RequestData": data_str,
            "DataSign": sign,
            "DataType": "2"
        }
        response = requests.post(self.url, data=params)
        return response.json()
    def _build_sign(self, data_str):
        import hashlib
        sign_str = data_str + self.app_key
        sign = hashlib.md5(sign_str.encode()).hexdigest()
        return sign
# 使用示例
if __name__ == "__main__":
    # 需要在快递鸟官网注册获取
    app_id = "你的APP_ID"
    app_key = "你的APP_KEY"
    query = LogisticsQuery(app_id, app_key)
    result = query.query("SFEXPRESS", "SF1234567890")
    print(json.dumps(result, indent=2, ensure_ascii=False))

使用快递100API

import requests
import hashlib
import json
class Kuaidi100Query:
    def __init__(self, key, customer):
        self.key = key
        self.customer = customer
        self.url = "https://poll.kuaidi100.com/poll/query.do"
    def query(self, com, num):
        """查询物流信息"""
        param = {
            "com": com,  # 快递公司编码
            "num": num   # 运单号
        }
        param_str = json.dumps(param)
        sign = hashlib.md5((param_str + self.key + self.customer).encode()).hexdigest().upper()
        data = {
            "customer": self.customer,
            "param": param_str,
            "sign": sign
        }
        response = requests.post(self.url, data=data)
        return response.json()
# 使用示例
query = Kuaidi100Query("你的KEY", "你的CUSTOMER")
result = query.query("yuantong", "YT1234567890")

使用爬虫方式（仅用于学习）

import requests
from bs4 import BeautifulSoup
import re
class SimpleLogisticsQuery:
    def __init__(self):
        self.headers = {
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
        }
    def query_baidu(self, exp_no):
        """使用百度查询物流信息"""
        url = f"https://www.baidu.com/s?wd={exp_no}"
        response = requests.get(url, headers=self.headers)
        # 解析HTML提取物流信息
        soup = BeautifulSoup(response.text, 'html.parser')
        # 这里需要根据实际页面结构解析
    def query_17track(self, exp_no):
        """使用17track查询"""
        url = f"https://www.17track.net/en/track?nums={exp_no}"
        response = requests.get(url, headers=self.headers)
        # 解析结果
# 注意：爬虫方式不稳定，且可能违反网站服务条款

批量查询脚本（带缓存）

import json
import time
import sqlite3
from datetime import datetime
class BatchLogisticsQuery:
    def __init__(self, api_instance):
        self.api = api_instance
        self.cache = {}
        self.init_db()
    def init_db(self):
        """初始化SQLite缓存数据库"""
        self.conn = sqlite3.connect('logistics_cache.db')
        self.cursor = self.conn.cursor()
        self.cursor.execute('''
            CREATE TABLE IF NOT EXISTS cache (
                exp_no TEXT PRIMARY KEY,
                result TEXT,
                update_time TIMESTAMP
            )
        ''')
        self.conn.commit()
    def get_from_cache(self, exp_no):
        """从缓存获取数据"""
        self.cursor.execute(
            "SELECT result, update_time FROM cache WHERE exp_no=?",
            (exp_no,)
        )
        result = self.cursor.fetchone()
        if result:
            # 180分钟内的缓存有效
            update_time = datetime.strptime(result[1], '%Y-%m-%d %H:%M:%S')
            if (datetime.now() - update_time).seconds < 10800:
                return json.loads(result[0])
        return None
    def save_to_cache(self, exp_no, result):
        """保存到缓存"""
        self.cursor.execute(
            "REPLACE INTO cache (exp_no, result, update_time) VALUES (?, ?, ?)",
            (exp_no, json.dumps(result), datetime.now().strftime('%Y-%m-%d %H:%M:%S'))
        )
        self.conn.commit()
    def batch_query(self, exp_list, delay=1):
        """批量查询（带延迟避免被封）"""
        results = {}
        for exp_no in exp_list:
            # 先查缓存
            cached = self.get_from_cache(exp_no)
            if cached:
                results[exp_no] = cached
                print(f"{exp_no}: 使用缓存数据")
            else:
                # 实时查询
                try:
                    result = self.api.query(exp_no)
                    results[exp_no] = result
                    self.save_to_cache(exp_no, result)
                    print(f"{exp_no}: 查询成功")
                    time.sleep(delay)  # 避免请求过快
                except Exception as e:
                    print(f"{exp_no}: 查询失败 - {str(e)}")
        return results

命令行工具版本

import argparse
import sys
def main():
    parser = argparse.ArgumentParser(description='物流信息查询工具')
    parser.add_argument('exp_no', help='快递单号')
    parser.add_argument('--company', '-c', help='快递公司编码（可选）')
    parser.add_argument('--output', '-o', help='输出文件路径')
    parser.add_argument('--format', '-f', choices=['json', 'text'], default='text', help='输出格式')
    args = parser.parse_args()
    # 查询逻辑
    result = query_logistics(args.exp_no, args.company)
    # 输出
    if args.format == 'json':
        output = json.dumps(result, indent=2, ensure_ascii=False)
    else:
        output = format_text(result)
    if args.output:
        with open(args.output, 'w', encoding='utf-8') as f:
            f.write(output)
    else:
        print(output)
if __name__ == "__main__":
    main()

注意事项

API使用：推荐使用正规API服务，需要注册获取密钥
爬虫限制：爬虫方式不稳定且可能违法，仅用于学习
频率限制：批量查询时要控制请求频率
数据缓存：避免重复请求同一运单
错误处理：做好异常捕获和重试机制

需要我详细讲解某个具体实现吗？