383 lines
14 KiB
Python
383 lines
14 KiB
Python
import time
|
||
import json
|
||
import re
|
||
from DrissionPage import ChromiumPage, ChromiumOptions
|
||
|
||
# 设置浏览器路径
|
||
CHROME_PATH = r'C:\Program Files\Google\Chrome\Application\chrome.exe'
|
||
|
||
# 全局浏览器实例
|
||
global_page = None
|
||
|
||
def get_global_browser():
|
||
"""获取全局浏览器实例"""
|
||
global global_page
|
||
if global_page is None:
|
||
print("正在初始化浏览器...")
|
||
print(f"浏览器路径: {CHROME_PATH}")
|
||
|
||
# 导入 os 检查文件是否存在
|
||
import os
|
||
if not os.path.exists(CHROME_PATH):
|
||
raise FileNotFoundError(f"找不到 Chrome 浏览器，路径: {CHROME_PATH}")
|
||
|
||
options = ChromiumOptions()
|
||
options.set_browser_path(CHROME_PATH)
|
||
|
||
# DrissionPage 默认应该是有界面的浏览器
|
||
# 参考 jd.py 和 tb.py 的实现，直接创建即可
|
||
# 如果需要最大化窗口，可以尝试添加参数（可选）
|
||
try:
|
||
options.set_argument('--start-maximized')
|
||
except:
|
||
pass # 如果设置失败就忽略，不影响浏览器启动
|
||
|
||
print("正在启动浏览器，请稍候...")
|
||
print("如果浏览器没有自动打开，请检查 Chrome 是否正确安装")
|
||
|
||
try:
|
||
global_page = ChromiumPage(options)
|
||
print("✅ 浏览器已成功启动！")
|
||
print(f"当前页面 URL: {global_page.url}")
|
||
# 等待浏览器完全启动
|
||
time.sleep(2)
|
||
except Exception as e:
|
||
print(f"❌ 浏览器启动失败: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
raise
|
||
else:
|
||
print("使用已存在的浏览器实例")
|
||
return global_page
|
||
|
||
|
||
def extract_logistics_info(tracking_url):
|
||
"""
|
||
从京东物流追踪页面提取运单号、承运人等信息
|
||
|
||
Args:
|
||
tracking_url: 物流追踪页面 URL，例如 https://3.cn/2t-Iibig
|
||
|
||
Returns:
|
||
dict: 包含运单号、承运人、承运人电话、物流跟踪信息等的字典
|
||
"""
|
||
page = get_global_browser()
|
||
|
||
try:
|
||
print(f"\n正在打开物流追踪页面: {tracking_url}")
|
||
page.get(tracking_url)
|
||
print("页面加载中，请稍候...")
|
||
time.sleep(5) # 等待页面加载
|
||
|
||
# 检查页面是否成功加载
|
||
current_url = page.url
|
||
print(f"当前页面 URL: {current_url}")
|
||
|
||
# 检查页面标题
|
||
try:
|
||
title = page.title
|
||
print(f"页面标题: {title}")
|
||
except:
|
||
print("无法获取页面标题")
|
||
|
||
# 检查页面是否有内容
|
||
try:
|
||
html_length = len(page.html)
|
||
print(f"页面 HTML 长度: {html_length} 字符")
|
||
if html_length < 100:
|
||
print("⚠️ 警告: 页面内容可能未完全加载")
|
||
except Exception as e:
|
||
print(f"⚠️ 无法获取页面 HTML: {e}")
|
||
|
||
result = {
|
||
"waybill_no": None, # 运单号
|
||
"carrier": None, # 国内承运人
|
||
"carrier_phone": None, # 国内承运人电话
|
||
"tracking_info": [], # 物流跟踪信息列表
|
||
"raw_html": None # 原始 HTML（用于调试）
|
||
}
|
||
|
||
# 方法1: 监听网络请求，查找物流数据 API
|
||
print("方法1: 监听网络请求...")
|
||
page.listen.start()
|
||
|
||
# 滚动页面触发可能的请求
|
||
page.scroll.down(300)
|
||
time.sleep(2)
|
||
page.scroll.to_bottom()
|
||
time.sleep(3)
|
||
|
||
# 检查监听到的请求
|
||
responses = page.listen.get()
|
||
print(f"监听到 {len(responses)} 个请求")
|
||
|
||
# 查找可能的物流数据接口
|
||
possible_urls = [
|
||
'track', 'logistics', 'waybill', 'express',
|
||
'delivery', '3.cn', 'jd.com/logistics',
|
||
'api.m.jd.com', 'mapi.jd.com'
|
||
]
|
||
|
||
for resp in responses:
|
||
url = resp.url if hasattr(resp, 'url') else ''
|
||
url_lower = url.lower()
|
||
|
||
# 检查是否可能是物流相关的 API
|
||
if any(keyword in url_lower for keyword in possible_urls):
|
||
print(f"发现可能的物流 API: {url[:100]}")
|
||
try:
|
||
if hasattr(resp, 'response') and hasattr(resp.response, 'body'):
|
||
body = resp.response.body
|
||
|
||
# 处理 JSON 响应
|
||
if isinstance(body, dict):
|
||
json_data = body
|
||
elif isinstance(body, str):
|
||
try:
|
||
json_data = json.loads(body)
|
||
except:
|
||
continue
|
||
else:
|
||
continue
|
||
|
||
# 尝试从 JSON 中提取运单号等信息
|
||
extracted = extract_from_json(json_data)
|
||
if extracted:
|
||
result.update(extracted)
|
||
print("成功从 API 响应中提取数据")
|
||
return result
|
||
except Exception as e:
|
||
print(f"解析 API 响应时出错: {e}")
|
||
|
||
# 方法2: 从页面 HTML/DOM 中提取
|
||
print("\n方法2: 从页面 DOM 提取数据...")
|
||
|
||
html = page.html
|
||
result['raw_html'] = html[:5000] # 保存部分 HTML 用于调试
|
||
|
||
# 从 HTML 文本中提取运单号
|
||
waybill_patterns = [
|
||
r'运单号[：:\s]*(\d+)',
|
||
r'waybill[_\s]*no["\']?\s*[:：]\s*["\']?(\d+)',
|
||
r'tracking[_\s]*number["\']?\s*[:：]\s*["\']?(\d+)',
|
||
r'"waybillNo"\s*[:：]\s*["\']?(\d+)',
|
||
r'"trackingNumber"\s*[:：]\s*["\']?(\d+)',
|
||
]
|
||
|
||
for pattern in waybill_patterns:
|
||
matches = re.findall(pattern, html, re.IGNORECASE)
|
||
if matches:
|
||
result['waybill_no'] = matches[0]
|
||
print(f"找到运单号: {result['waybill_no']}")
|
||
break
|
||
|
||
# 提取承运人
|
||
carrier_patterns = [
|
||
r'国内承运人[：:\s]*([^\s<，,]+)',
|
||
r'carrier[：:\s]*([^\s<，,]+)',
|
||
r'"carrier"\s*[:：]\s*["\']?([^"\']+)',
|
||
]
|
||
|
||
for pattern in carrier_patterns:
|
||
matches = re.findall(pattern, html, re.IGNORECASE)
|
||
if matches:
|
||
result['carrier'] = matches[0].strip()
|
||
print(f"找到承运人: {result['carrier']}")
|
||
break
|
||
|
||
# 提取承运人电话
|
||
phone_patterns = [
|
||
r'国内承运人电话[：:\s]*(\d+)',
|
||
r'carrier[_\s]*phone[：:\s]*(\d+)',
|
||
r'"carrierPhone"\s*[:：]\s*["\']?(\d+)',
|
||
]
|
||
|
||
for pattern in phone_patterns:
|
||
matches = re.findall(pattern, html, re.IGNORECASE)
|
||
if matches:
|
||
result['carrier_phone'] = matches[0]
|
||
print(f"找到承运人电话: {result['carrier_phone']}")
|
||
break
|
||
|
||
# 方法3: 从 DOM 元素中提取
|
||
print("\n方法3: 从 DOM 元素提取数据...")
|
||
|
||
# 尝试查找运单号元素
|
||
waybill_elements = page.eles('xpath=//*[contains(text(), "运单号") or contains(text(), "运单")]')
|
||
for elem in waybill_elements:
|
||
text = elem.text
|
||
parent_text = elem.parent().text if elem.parent() else ""
|
||
full_text = text + " " + parent_text
|
||
|
||
# 从文本中提取数字作为运单号
|
||
numbers = re.findall(r'\d{8,}', full_text)
|
||
if numbers and not result['waybill_no']:
|
||
result['waybill_no'] = numbers[0]
|
||
print(f"从元素文本中找到运单号: {result['waybill_no']}")
|
||
|
||
# 提取承运人
|
||
if '承运人' in text and not result['carrier']:
|
||
carrier_match = re.search(r'承运人[：:\s]*([^\s<，,]+)', full_text)
|
||
if carrier_match:
|
||
result['carrier'] = carrier_match.group(1).strip()
|
||
print(f"从元素文本中找到承运人: {result['carrier']}")
|
||
|
||
# 提取电话
|
||
if '电话' in text and not result['carrier_phone']:
|
||
phone_match = re.search(r'电话[：:\s]*(\d+)', full_text)
|
||
if phone_match:
|
||
result['carrier_phone'] = phone_match.group(1)
|
||
print(f"从元素文本中找到电话: {result['carrier_phone']}")
|
||
|
||
# 提取物流跟踪信息（时间线）
|
||
print("\n提取物流跟踪信息...")
|
||
tracking_elements = page.eles('xpath=//*[contains(@class, "track") or contains(@class, "logistics") or contains(@class, "timeline")]')
|
||
|
||
if not tracking_elements:
|
||
# 尝试查找包含时间戳的元素
|
||
tracking_elements = page.eles('xpath=//*[contains(text(), "2025") or contains(text(), "货物") or contains(text(), "到达")]')
|
||
|
||
tracking_info = []
|
||
for elem in tracking_elements[:20]: # 限制数量
|
||
text = elem.text
|
||
if text and len(text) > 5:
|
||
# 尝试提取时间戳
|
||
time_match = re.search(r'(\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2})', text)
|
||
if time_match or any(keyword in text for keyword in ['货物', '到达', '揽收', '运输', '配送', '签收']):
|
||
tracking_info.append({
|
||
'text': text.strip(),
|
||
'time': time_match.group(1) if time_match else None
|
||
})
|
||
|
||
result['tracking_info'] = tracking_info[:10] # 最多保存10条
|
||
|
||
return result
|
||
|
||
except Exception as e:
|
||
print(f"提取物流信息时出错: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
return None
|
||
|
||
|
||
def extract_from_json(json_data):
|
||
"""
|
||
从 JSON 数据中提取物流信息
|
||
|
||
Args:
|
||
json_data: JSON 字典
|
||
|
||
Returns:
|
||
dict: 提取到的物流信息
|
||
"""
|
||
result = {}
|
||
|
||
def search_dict(d, key_patterns):
|
||
"""递归搜索字典中的值"""
|
||
if isinstance(d, dict):
|
||
for k, v in d.items():
|
||
# 检查键名
|
||
for pattern in key_patterns:
|
||
if re.search(pattern, k, re.IGNORECASE):
|
||
return v
|
||
# 递归搜索值
|
||
if isinstance(v, (dict, list)):
|
||
found = search_dict(v, key_patterns)
|
||
if found:
|
||
return found
|
||
elif isinstance(d, list):
|
||
for item in d:
|
||
found = search_dict(item, key_patterns)
|
||
if found:
|
||
return found
|
||
return None
|
||
|
||
# 搜索运单号
|
||
waybill = search_dict(json_data, [r'waybill', r'tracking.*number', r'运单号', r'waybillNo'])
|
||
if waybill:
|
||
result['waybill_no'] = str(waybill)
|
||
|
||
# 搜索承运人
|
||
carrier = search_dict(json_data, [r'carrier', r'承运人', r'carrierName'])
|
||
if carrier:
|
||
result['carrier'] = str(carrier)
|
||
|
||
# 搜索承运人电话
|
||
phone = search_dict(json_data, [r'carrier.*phone', r'承运人电话', r'carrierPhone', r'phone'])
|
||
if phone:
|
||
result['carrier_phone'] = str(phone)
|
||
|
||
# 搜索物流跟踪信息
|
||
tracking = search_dict(json_data, [r'track', r'logistics', r'物流', r'轨迹', r'history'])
|
||
if tracking:
|
||
if isinstance(tracking, list):
|
||
result['tracking_info'] = tracking
|
||
elif isinstance(tracking, dict):
|
||
result['tracking_info'] = [tracking]
|
||
|
||
return result if result else None
|
||
|
||
|
||
def print_result(result):
|
||
"""打印提取结果"""
|
||
if not result:
|
||
print("未能提取到物流信息")
|
||
return
|
||
|
||
print("\n" + "="*50)
|
||
print("物流信息提取结果:")
|
||
print("="*50)
|
||
print(f"运单号: {result.get('waybill_no', '未找到')}")
|
||
print(f"国内承运人: {result.get('carrier', '未找到')}")
|
||
print(f"国内承运人电话: {result.get('carrier_phone', '未找到')}")
|
||
|
||
if result.get('tracking_info'):
|
||
print(f"\n物流跟踪信息 (共 {len(result['tracking_info'])} 条):")
|
||
for idx, info in enumerate(result['tracking_info'], 1):
|
||
if isinstance(info, dict):
|
||
text = info.get('text', str(info))
|
||
time_str = info.get('time', '')
|
||
print(f" {idx}. {text}")
|
||
if time_str:
|
||
print(f" 时间: {time_str}")
|
||
else:
|
||
print(f" {idx}. {info}")
|
||
else:
|
||
print("\n物流跟踪信息: 未找到")
|
||
|
||
print("="*50)
|
||
|
||
|
||
# 主程序
|
||
if __name__ == '__main__':
|
||
# 测试 URL
|
||
tracking_url = "https://3.cn/2t-Iibig"
|
||
|
||
print("="*60)
|
||
print("京东物流信息提取工具")
|
||
print("="*60)
|
||
print(f"目标 URL: {tracking_url}")
|
||
print("开始提取物流信息...\n")
|
||
|
||
try:
|
||
result = extract_logistics_info(tracking_url)
|
||
except Exception as e:
|
||
print(f"\n❌ 执行过程中出错: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
result = None
|
||
|
||
if result:
|
||
print_result(result)
|
||
|
||
# 保存结果到文件
|
||
output_file = "logistics_result.json"
|
||
with open(output_file, 'w', encoding='utf-8') as f:
|
||
json.dump(result, f, ensure_ascii=False, indent=2)
|
||
print(f"\n结果已保存到: {output_file}")
|
||
else:
|
||
print("提取失败")
|
||
|
||
print("\n脚本执行完成，浏览器保持打开状态用于调试")
|
||
|