物流爬虫实战：某丰快递信息实时追踪技术全解析

import requests
import hashlib
import time

def get_sf_track(tracking_number):
    app_key = "YOUR_APP_KEY"
    app_secret = "YOUR_APP_SECRET"
    timestamp = str(int(time.time()))
    
    # 生成签名
    sign_str = f"{app_key}{app_secret}{tracking_number}{timestamp}"
    sign = hashlib.md5(sign_str.encode()).hexdigest().upper()
    
    url = "https://bsp-ois.***.com/bsp-ois/express/service/queryTrack"
    params = {
        "appKey": app_key,
        "trackNumber": tracking_number,
        "timestamp": timestamp,
        "sign": sign
    }
    
    response = requests.get(url, params=params)
    return response.json()

from bs4 import BeautifulSoup
import requests

def scrape_sf_page(tracking_number):
    url = f"https://www.***.com/cn/sc/dynamic_function/waybill/#query/{tracking_number}"
    headers = {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/130.0.0.0 Safari/537.36"
    }
    
    response = requests.get(url, headers=headers)
    soup = BeautifulSsoup(response.text, 'html.parser')
    
    # 解析物流节点（实际需根据页面结构调整选择器）
    steps = soup.select('.track-list li')
    return [step.get_text(strip=True) for step in steps]

from selenium import webdriver
from selenium.webdriver.chrome.options import Options
import time

def selenium_scrape(tracking_number):
    options = Options()
    options.add_argument("--headless")  # 无头模式
    driver = webdriver.Chrome(options=options)
    
    driver.get(f"https://www.***.com/cn/sc/dynamic_function/waybill/#query/{tracking_number}")
    time.sleep(3)  # 等待数据加载
    
    # 执行JavaScript获取动态内容
    steps = driver.execute_script("return Array.from(document.querySelectorAll('.track-list li')).map(el => el.innerText)")
    driver.quit()
    return steps

import requests

def kuaidi100_query(tracking_number, company="shunfeng"):
    url = "https://poll.***.com/poll/query.do"
    params = {
        "customer": "YOUR_CUSTOMER_ID",
        "sign": "YOUR_ENCRYPTED_SIGN",
        "com": company,
        "num": tracking_number
    }
    
    response = requests.get(url, params=params)
    return response.json()

import random
from proxy_pool import ProxyPool  # 假设的代理池类

def get_random_proxy():
    pool = ProxyPool()
    return random.choice(pool.get_proxies())

def request_with_proxy(url):
    proxy = get_random_proxy()
    proxies = {
        "http": f"http://{proxy}",
        "https": f"https://{proxy}"
    }
    return requests.get(url, proxies=proxies, timeout=10)

from pymongo import MongoClient

client = MongoClient("mongodb://localhost:27017/")
db = client["logistics_db"]
collection = db["sf_tracks"]

def save_track_data(tracking_number, steps):
    data = {
        "tracking_number": tracking_number,
        "steps": steps,
        "update_time": datetime.now(),
        "status": steps[-1].split("]")[0].strip() if steps else "未知"
    }
    collection.insert_one(data)

def detect_anomalies(steps):
    time_diffs = []
    for i in range(1, len(steps)):
        # 提取时间部分（需根据实际格式调整）
        prev_time = parse_time(steps[i-1].split("]")[0])
        curr_time = parse_time(steps[i].split("]")[0])
        time_diffs.append((curr_time - prev_time).total_seconds()/3600)  # 转换为小时
    
    # 超过6小时未更新视为异常
    return any(diff > 6 for diff in time_diffs)

[用户输入] → [运单号校验] → [路由选择]
       ↓                   ↓
[API查询] ←→ [代理IP池] ←→ [网页爬取]
       ↓
[数据清洗] → [MongoDB存储] → [异常预警]
       ↓
[可视化看板] ← [定时任务调度]