Python(二十)实现各大跨境船公司物流查询新增[ARKAS Line]船司
一、前言
本章主要在之前的基础上，新增一个 【ARKAS LINE】船司 物流信息查询，插入excel数据方法
- 基于已经实现了常用 跨境物流船司 基础信息查询功能 及新增优化功能 的基础上；
- 再加一个船司信息查询
https://blog.csdn.net/Makasa/article/details/145600527?spm=1001.2014.3001.5501
二、代码实现
- 我这里就不拆开了，还是直接复制整个代码
具体优化：
1、新增方法
- deal_ARKAS_eta_time_format(date) ：日期转换
- get_ARKAS_tracking_info_by_bill_number(tracking_number)
- get_ARKAS_tracking_info(tracking_number)
- get_ARKAS_related_time(tracking_number)
- get_ARKAS_shipment_detail(bill_number)
2、添加船司提单号判断
- insert_ship_detail_to_excel(tracking_number)
- assign_shipper(tracking_number)
# -*- codeing = utf-8 -*-
# @time: 2025/2/12 14：26
# @Author : Mikasa
import json
import time
from datetime import datetime
import pandas as pd
import requests
from bs4 import BeautifulSoup
from pathlib import Path
def deal_ARKAS_eta_time_format(date):
"""
处理ARKAS Line 船司预期时间格式 : 2025-02-19T09:00:00 - 2025/2/19 9:00:00
:param date:
:return:
"""
dt = datetime.strptime(date, "%Y-%m-%dT%H:%M:%S")
year = dt.year
month = dt.month if dt.month >= 10 else dt.month
day = dt.day if dt.day >= 10 else dt.day
hour = dt.hour if dt.hour >= 10 else dt.hour
minute = f"{dt.minute:02}"
second = f"{dt.second:02}"
format_time = f"{year}/{month}/{day} {hour}:{minute}:{second}"
return format_time
def get_ARKAS_tracking_info_by_bill_number(tracking_number):
"""
查询ARKAS LINE 船司物流信息
:param tracking_number: 提单号
:return:
"""
params = "{'key':'" + tracking_number + "'}"
url = "https://webtracking.arkasline.com.tr/api/request/Get?controllerMethod=webtracking/api/shipmenttracking/GetDocumentationAsync&prms=" + params
headers = {
"user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/132.0.0.0 Safari/537.36"
}
try:
response = requests.get(url, headers=headers, timeout=10)
result_json = response.json()
print("===========================================")
print("ARKAS LINE船信息：", result_json)
return result_json
except requests.exceptions.RequestException as e:
print(f"请求失败: {e}")
except json.JSONDecodeError:
print("响应不是有效的 JSON 格式")
def get_ARKAS_tracking_info(tracking_number):
"""
获取ARKAS柜号物流信息
:param tracking_number:
:return:
"""
response = get_ARKAS_tracking_info_by_bill_number(tracking_number)
# 默认拿到第一个柜子的document_id
documentation_id = response['data']['data'][0]['documentationId']
# print("documentationId:", documentation_id)
params = "{'documentationId':'" + str(documentation_id) + "'}"
url = "https://webtracking.arkasline.com.tr/api/request/Get?controllerMethod=webtracking/api/shipmenttracking/GetDocumentationHistoryAsync&prms=" + params
headers = {
"user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/132.0.0.0 Safari/537.36"
}
try:
response = requests.get(url, headers=headers, timeout=10)
result_json = response.json()
print("===========================================")
print("ARKAS LINE详细物流信息：", result_json)
return result_json
except requests.exceptions.RequestException as e:
print(f"请求失败: {e}")
except json.JSONDecodeError:
print("响应不是有效的 JSON 格式")
def get_ARKAS_related_time(tracking_number):
"""
获取ARKAS船期信息
:param tracking_number:
:return:
"""
response = get_ARKAS_tracking_info(tracking_number)
container_list = response['data']
for container_info in container_list:
if container_info['status'] == "Discharge":
eventDate = container_info['eventDate']
new_eta_format = deal_ARKAS_eta_time_format(eventDate)
print("------------------------------------------")
print(f"提单号【{tracking_number}】船期信息如下：")
print(f"预计到港时间：{new_eta_format}")
return new_eta_format
def get_ARKAS_shipment_detail(bill_number):
"""
获取ARKAS LINE物流详细信息
:return:
"""
response = get_ARKAS_tracking_info_by_bill_number(bill_number)
response2 = get_ARKAS_tracking_info(bill_number)['data'][0]
container_list = response['data']['data']
date = get_local_time()
shipping = "ARKAS LINE"
index = 1
bl_number = bill_number
origin = response2['port']
destination = container_list[0]['finalDestination']
eta = get_ARKAS_related_time(bill_number)
telex_status = "N"
data_list = []
for container_info in container_list:
container_number = container_info['containerNo']
container_size = str(container_info['size'])
str1 = "20"
str2 = "40"
container_size_status = 1
if str1 in container_size:
data_total = [index, date, "", shipping, "", bl_number, "", container_number, date, eta,
telex_status, "", "", "", "", container_size_status, origin, destination]
elif str2 in container_size:
data_total = [index, date, "", shipping, "", bl_number, "", container_number, date, eta,
telex_status, "", "", "", container_size_status, "", origin, destination]
else:
print("当前集装箱尺寸不存在，请详细查明尺寸参数")
data_list.append(data_total)
index = index + 1
print(f"ARKAS LINE船司：共插入【{len(data_list)}】条数据")
print("ARKAS LINE船司新插入数据list：", data_list)
return data_list
def get_MSK_tracking_info(tracking_number):
"""
查询MSK船司物流信息
:param tracking_number:提单号
:return:
"""
url = "https://api.maersk.com.cn/synergy/tracking/" + tracking_number
params = {"operator": "MAEU"}
headers = {
"consumer-key": "UtMm6JCDcGTnMGErNGvS2B98kt1Wl25H", # 关键认证头 必传
"user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/132.0.0.0 Safari/537.36"
}
try:
response = requests.get(url, params=params, headers=headers, timeout=10)
result_json = response.json()
print("===========================================")
print("MSK船信息：", result_json)
return result_json
except requests.exceptions.RequestException as e:
print(f"请求失败: {e}")
except json.JSONDecodeError:
print("响应不是有效的 JSON 格式")
def deal_MSK_eta_time_format(date):
"""
处理MSK船司预期时间格式
:param date:
:return:
"""
dt = datetime.strptime(date, "%Y-%m-%dT%H:%M:%S.%f")
formatted_str = dt.strftime("%Y/%#m/%#d %H:%M:%S")
return formatted_str
def get_MSK_related_time(tracking_number):
"""
获取MSK船期时间
:param tracking_number:
:return:
"""
result = get_MSK_tracking_info(tracking_number)
location_list = result['containers'][0]['locations']
target_terminal = "APAPA PORT"
index = 0
for i, location in enumerate(location_list):
if location["terminal"] == target_terminal:
index = i # 拿到目的地为apapa PORT的数据的index
break
ship_info = location_list[index]['events'][0]
event_time_type = ship_info['event_time_type']
if event_time_type == "EXPECTED":
# 提单还未到港
ETA = ship_info['expected_time']
elif event_time_type == "ACTUAL":
# 提单已到港，拿真实时间
ETA = ship_info['actual_time']
new_ETA_fomat = deal_MSK_eta_time_format(ETA)
print("------------------------------------------")
print(f"提单号【{tracking_number}】船期信息如下：")
print(f"预计到港时间：{new_ETA_fomat}")
return new_ETA_fomat
def get_ONE_tracking_info_by_bill_number(bill_number):
"""
根据提单号查询ONE物流信息
:param bill_number:
:return:
"""
str = "ONEY"
if str in bill_number:
bill_number = bill_number.replace("ONEY", "")
headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/132.0.0.0 Safari/537.36",
"Content-Type": "application/x-www-form-urlencoded; charset=UTF-8",
"X-Requested-With": "XMLHttpRequest"
}
url = "https://ecomm.one-line.com/ecom/CUP_HOM_3301GS.do?" \
"sord=asc" \
"&f_cmd=121" \
"&search_type=B" \
f"&search_name={bill_number}"
response = requests.get(url, headers=headers)
result = response.json()
print("BL_response:", result)
return result
def get_ONE_tracking_info(bill_number):
"""
查询ONE船司物流信息
:param tracking_number:
:return:
"""
result = get_ONE_tracking_info_by_bill_number(bill_number)
container_list = result['list'][0] # 默认取第一个集装箱
container_number = container_list['cntrNo']
cop_no = container_list['copNo']
bkg_no = container_list['bkgNo']
url = "https://ecomm.one-line.com/ecom/CUP_HOM_3301GS.do"
headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/132.0.0.0 Safari/537.36",
"Content-Type": "application/x-www-form-urlencoded; charset=UTF-8",
"X-Requested-With": "XMLHttpRequest"
}
payload = {
"f_cmd": "125",
"cntr_no": container_number,
"bkg_no": bkg_no,
"cop_no": cop_no
}
response = requests.post(url, headers=headers, data=payload)
if response.status_code == 200:
data = response.json()
print("===========================================")
print("ONE船信息:", response.json())
return data
def get_ONE_related_time(bill_number):
"""
查询ONE船期时间
:param bill_number:
:return:
"""
data = get_ONE_tracking_info(bill_number)
track_info_list = data['list']
eta = "Arrival at Port of Discharging"
actual_pod = "POD Berthing Destination"
discharge_cargo = "Port of Discharging"
print("------------------------------------------")
print(f"提单号【{bill_number}】船期信息如下：")
for track_info in track_info_list:
statusNm = track_info['statusNm']
if eta in statusNm:
print(f"预期到港时间：", track_info['eventDt'])
elif actual_pod in statusNm:
print(f"实际抵达卸货港时间：", track_info['eventDt'])
elif discharge_cargo in statusNm:
print(f"实际卸货时间：", track_info['eventDt'])
return track_info['eventDt'] # 直接返回的最新一套信息的时间
def get_PIL_cookie():
"""
获取PIL的cookie
:return:
"""
# 生成北京当前时间戳 单位为ms
timestamp_ms = int(time.time() * 1000)
url = "https://www.pilship.com/wp-content/themes/hello-theme-child-master/pil-api/common/get-n.php?timestamp=" + str(
timestamp_ms)
headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/132.0.0.0 Safari/537.36",
"Referer": "https://www.pilship.com/digital-solutions/?tab=customer&id=track-trace&label=containerTandT&module=TrackTraceBL&refNo=NGRI50097000"
}
response = requests.get(url, headers=headers)
return response
def get_PIL_tracking_info_by_bill_number(bill_number):
"""
根据提单号查询PIL物流信息
:param bill_number:
:return:
"""
response = get_PIL_cookie() # 拿到n
n = response.json()['n']
timestamp = str(int(time.time() * 1000))
referer = "https://www.pilship.com/digital-solutions/?tab=customer&id=track-trace&label=containerTandT&module=TrackTraceBL&refNo=" + bill_number
headers = {
'user-agent': "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/132.0.0.0 Safari/537.36",
'referer': referer,
'X-Requested-With': "XMLHttpRequest",
'cookie': "CookieConsent={stamp:%27LUQ2eOuWASANePCHNSXS2yeSAiOmJoQhXm7pEbeVpMEr7kRRRhczfw==%27%2Cnecessary:true%2Cpreferences:true%2Cstatistics:true%2Cmarketing:true%2Cmethod:%27explicit%27%2Cver:1%2Cutc:1734771385816%2Cregion:%27ng%27}; _ga=GA1.1.1137581632.1734770854; pil_session_cookie=!WFtN5njOPpQS1hf8xTD27kmUREcR4A5xQFGbSdeZWh6xne7uf6cynXyQzICRX7DAVtXpZHsVQFbXaA==; wp-wpml_current_language=en; TSPD_101=08c4718e7dab2800c57f82cee5e846814246783a4ea9308f585af084e53e57c851f4d7af7507b6ba01b43d68d07a6fc9084f292334051800b0bdc60b7433c56f989c7ca0dcd64b787093688cd95badc6; TS00000000076=08c4718e7dab28007707032a2a3c087717436b105e0ea27086f21be845e6238e426c659eed97e9faac24d9498ea8e2f90825bede2809d000f4ff876b60b5d60819d6e9be3e6f3d6f2d0ad9cc544d6ba809938284a1f852c824efa411a4094ef683ea56f2d383a67379539aa5b28a9a52823dde83af710cc1065a8cb738359c2102062f1f41799313cc887fe2f7f84015edf73a4443f90343838461044c8dbccae6660ca3322e787bb229d3ae25557ce32af1669c3ddeb451eb967fe99ac32cc0ef550af1567da52efbed7d17b811159ef5b40416eb7178abca86b4f273868b80aee1cd0ab7a38515f98e7b91baf133b22894982a220077e68b7c314d8ca5c0f89a4bbc55a09b538e; TSPD_101_DID=08c4718e7dab28007707032a2a3c087717436b105e0ea27086f21be845e6238e426c659eed97e9faac24d9498ea8e2f90825bede28063800dd525af8f96ac9cff1388ccc906e36023328029cfae4893759356223de54c62e94a0c306a33859e46108198cac5061a89c8fa7c14267e4fc; TS7ff733e9077=08c4718e7dab280085f60f04ba264a2e0ac0875b36b59c53defab055f5fea0c7c7ce7b40ed9afcf756781ecc450e1b470807626b7517200050d650fcfa6773405f5a17954cff4628d908f416f4234200dbaa3cc3ec46b52a; _ga_M39J8YZNDE=GS1.1.1738786704.56.0.1738786704.0.0.0; TS015cd57e=011698f8e78ca3bc42cad4dd0e721fdcb048e98d12c3529eecbcc961ce0855aa8b6eef2788cf1b3cc8bbe190c65fcf3d0666de4efd377ed9d22d0f951ed126bf67a554e95c; TS7ff733e9029=08c4718e7dab2800e6a5d0f6faeca1dc19685dfff567b4dd1369f1b4b696f84893905e24cb58bc50ffb1a716a0e029fa; TS97a2723b027=08c4718e7dab2000fd4b11416d6e7e1c3da097165f9bd7b084e973e022fa2ec9d671050f9f2f2966081a278b7c1130004fb6fbf91b1d6cb4cd98e6bcb7bd8410d5bed0a0f4b4e1e764e4f4d75f4d5666b6a1bb124f33758104b457ee4acaf2c2"
}
# 否则根据提单号查询
url = f"https://www.pilship.com/wp-content/themes/hello-theme-child-master/pil-api/trackntrace-containertnt.php?" \
f"module=TrackTraceBL" \
f"&refNo={bill_number}" \
f"&n={n}" \
f"×tamp={timestamp}"
response = requests.get(url, headers=headers)
print("url:", response.url) # 查看最终请求的 URL
if response.status_code == 200:
response_json = json.loads(response.text)
result_json = analysis_PIL_response_by_bill_number(response_json) # 解析
return result_json
else:
print("通过【提单号】查询PIL船司信息失败")
def get_PIL_tracking_info(bill_number):
"""
查询PIL船公司物流信息（完整工作流）
- 包含动态Token获取和Cookie管理
- 必传校验：n | 经过多次测试，发现cookie只穿
"""
response = get_PIL_cookie() # 拿到n
n = response.json()['n']
timestamp = str(int(time.time() * 1000))
referer = "https://www.pilship.com/digital-solutions/?tab=customer&id=track-trace&label=containerTandT&module=TrackTraceBL&refNo=" + bill_number
headers = {
'user-agent': "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/132.0.0.0 Safari/537.36",
'referer': referer,
'X-Requested-With': "XMLHttpRequest",
'cookie': "CookieConsent={stamp:%27LUQ2eOuWASANePCHNSXS2yeSAiOmJoQhXm7pEbeVpMEr7kRRRhczfw==%27%2Cnecessary:true%2Cpreferences:true%2Cstatistics:true%2Cmarketing:true%2Cmethod:%27explicit%27%2Cver:1%2Cutc:1734771385816%2Cregion:%27ng%27}; _ga=GA1.1.1137581632.1734770854; pil_session_cookie=!WFtN5njOPpQS1hf8xTD27kmUREcR4A5xQFGbSdeZWh6xne7uf6cynXyQzICRX7DAVtXpZHsVQFbXaA==; wp-wpml_current_language=en; TSPD_101=08c4718e7dab2800c57f82cee5e846814246783a4ea9308f585af084e53e57c851f4d7af7507b6ba01b43d68d07a6fc9084f292334051800b0bdc60b7433c56f989c7ca0dcd64b787093688cd95badc6; TS00000000076=08c4718e7dab28007707032a2a3c087717436b105e0ea27086f21be845e6238e426c659eed97e9faac24d9498ea8e2f90825bede2809d000f4ff876b60b5d60819d6e9be3e6f3d6f2d0ad9cc544d6ba809938284a1f852c824efa411a4094ef683ea56f2d383a67379539aa5b28a9a52823dde83af710cc1065a8cb738359c2102062f1f41799313cc887fe2f7f84015edf73a4443f90343838461044c8dbccae6660ca3322e787bb229d3ae25557ce32af1669c3ddeb451eb967fe99ac32cc0ef550af1567da52efbed7d17b811159ef5b40416eb7178abca86b4f273868b80aee1cd0ab7a38515f98e7b91baf133b22894982a220077e68b7c314d8ca5c0f89a4bbc55a09b538e; TSPD_101_DID=08c4718e7dab28007707032a2a3c087717436b105e0ea27086f21be845e6238e426c659eed97e9faac24d9498ea8e2f90825bede28063800dd525af8f96ac9cff1388ccc906e36023328029cfae4893759356223de54c62e94a0c306a33859e46108198cac5061a89c8fa7c14267e4fc; TS7ff733e9077=08c4718e7dab280085f60f04ba264a2e0ac0875b36b59c53defab055f5fea0c7c7ce7b40ed9afcf756781ecc450e1b470807626b7517200050d650fcfa6773405f5a17954cff4628d908f416f4234200dbaa3cc3ec46b52a; _ga_M39J8YZNDE=GS1.1.1738786704.56.0.1738786704.0.0.0; TS015cd57e=011698f8e78ca3bc42cad4dd0e721fdcb048e98d12c3529eecbcc961ce0855aa8b6eef2788cf1b3cc8bbe190c65fcf3d0666de4efd377ed9d22d0f951ed126bf67a554e95c; TS7ff733e9029=08c4718e7dab2800e6a5d0f6faeca1dc19685dfff567b4dd1369f1b4b696f84893905e24cb58bc50ffb1a716a0e029fa; TS97a2723b027=08c4718e7dab2000fd4b11416d6e7e1c3da097165f9bd7b084e973e022fa2ec9d671050f9f2f2966081a278b7c1130004fb6fbf91b1d6cb4cd98e6bcb7bd8410d5bed0a0f4b4e1e764e4f4d75f4d5666b6a1bb124f33758104b457ee4acaf2c2"
}
result_json = get_PIL_tracking_info_by_bill_number(bill_number)
container_number = result_json['data']['table2'][0]['Container #']
container_number = container_number[:11] # 截取固定bl固定长度11位
print("container_number:", container_number)
if container_number != '':
# 箱号存在，就根据箱号查询
url = f"https://www.pilship.com/wp-content/themes/hello-theme-child-master/pil-api/trackntrace-containertnt-trace.php?" \
f"module=TrackTraceBL" \
f"&reference_no={bill_number}" \
f"&cntr_no={container_number}" \
f"&n={n}" \
f"×tamp={timestamp}"
response = requests.get(url, headers=headers)
# print("url:", response.url) # 查看最终请求的 URL
if response.status_code == 200:
response_json = json.loads(response.text)
resule_json = analysis_PIL_response_by_container_number(response_json) # 解析
return resule_json
else:
print("通过【箱号】查询PIL船司信息失败")
def analysis_PIL_response_by_container_number(response_json):
"""
解析通过【提单号+箱号】查询PIL船信息response的前端属性
:param bill_number:
:param container_number:
:return:
"""
html_data = response_json["data"] # 提取data里的html
soup = BeautifulSoup(html_data, "html.parser") # 解析
# 提取表头
header_row = soup.find("tr", class_="bg-lightblue text-fc-black text-fw-bold")
headers = [th.text.strip() for th in header_row.find_all("td")]
# 提取数据行
data_rows = soup.find_all("tr", class_="bg-lightblue text-fc-black")
table_data = []
for row in data_rows:
cells = row.find_all("td")
row_data = {headers[i]: cells[i].text.strip() for i in range(len(headers))}
table_data.append(row_data)
# 构建最终的 JSON 结构
result = {
"success": response_json["success"],
"data": table_data
}
return result
def analysis_PIL_response_by_bill_number(response_json):
"""
解析通过【提单号】查询PIL船信息response的前端属性
- 因为通过【提单号】/【箱号】返回的dom树结构不一样，所以解析方法不一致
- 注意：解析不兼容，提单号里面没有箱号树的情况 例如：TAJM40625400
:param bill_number:
:return:
"""
html_data = response_json["data"] # 提取html
soup = BeautifulSoup(html_data, "html.parser") # 解析 HTML
bl_number = soup.find("b").text
table1 = soup.find("table")
table1_data = []
headers = [th.text for th in table1.find("tr", class_="table-header").find_all("th")]
rows = table1.find_all("tr", class_="resultrow")
for row in rows:
cells = row.find_all("td")
row_data = {headers[i]: cells[i].get_text(separator=" ").strip() for i in range(len(headers))}
table1_data.append(row_data)
# 提取第二个表格的数据
table2 = soup.find("table", class_="table")
table2_data = []
headers = [td.text for td in table2.find("thead").find_all("td")]
rows = table2.find("tbody").find_all("tr")
for row in rows:
# todo: 解析还存在点问题 没有适配所有情况 例如： TAJM40625400 就报错
cells = row.find_all("td")
row_data = {headers[i]: cells[i].get_text(separator=" ").strip() for i in range(len(headers))}
table2_data.append(row_data)
# 构建最终的 JSON 结构
result_json = {
"success": response_json["success"],
"data": {
"bl_number": bl_number,
"table1": table1_data,
"table2": table2_data
}
}
return result_json
def get_PIL_related_time(bill_number):
"""
获取PIL船期时间
- 传container_number 原因是为了得到dom树里面更具体的时间;
- 只根据提单查询的话，container_number传空字符串''
:param bill_number:
:param container_number:
:return:
"""
result_json = get_PIL_tracking_info(bill_number)
print("result_json:", result_json)
dom_list = result_json['data']
for dom in dom_list:
# 根据箱号查询的dom树json
if dom['Event Name'] == "Vessel Discharge" and dom['Event Location'] == "LAGOS (APAPA SEAPORT)":
new_eta_format = deal_PIL_eta_date_format(dom['Event Date'])
print("------------------------------------------")
print(f"提单号【{bill_number}】，船期信息如下：")
print("预期到港时间：", new_eta_format)
return new_eta_format
def deal_PIL_eta_date_format(date_str):
"""
处理PIL日期格式: 将 18-Jan-2025 18:46:00 转化为 2025/1/18 18:46:00
- 这里需要加个判断：因为根据箱号查询返回的日期前面会有个前缀，【* 】，我们需去除后再进行转换
:param date_str:
:return:
"""
try:
if "* " in date_str:
date_str = date_str.lstrip("* ").strip()
dt = datetime.strptime(date_str, "%d-%b-%Y %H:%M:%S")
import platform
if platform.system() == "Windows":
formatted_date = dt.strftime("%Y/%#m/%#d %H:%M:%S")
else:
formatted_date = dt.strftime("%Y/%-m/%-d %H:%M:%S")
return formatted_date
except ValueError as e:
print("日期时间格式错误:", e)
def get_ZIM_access_token():
session = requests.session()
api_url = "https://www.goldstarline.com/#/track_ship"
session.get(api_url)
cookies_dict = session.cookies.get_dict()
headers = {
'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/132.0.0.0 Safari/537.36',
'cookie': "; ".join([f"{k}={v}" for k, v in cookies_dict.items()]),
}
payload = {
'country_code': 'HK'
}
url = "https://www.goldstarline.com/react-rest//access_token"
response = session.post(url, data=payload, headers=headers)
return response
def get_ZIM_tracking_info(tracking_number):
"""
获取ZIM船司物流信息
:param tracking_number:提单号
权限校验：必传cookie & authorization
- 格式：Bearer 52851984
- ZIM接口经常不稳定，当一次调用不成功，可重复再调用即可
:return:
"""
# todo: GOSUNGB20416758 bug调试
access_response = get_ZIM_access_token()
time.sleep(1)
token = access_response.json()['token']
cookie = access_response.cookies.get_dict()['ci_session']
cookie = "ci_session=" + cookie + ";"
authorization = "Bearer " + token
# print("cookie:", cookie)
# print("authorization:", authorization)
session = requests.Session()
api_url = "https://www.goldstarline.com/#/track_shipment"
session.get(api_url)
url = "https://www.goldstarline.com/react-rest//get_track_shipment_val"
payload = {
"containerid": tracking_number
}
# cookie 通过测试，只需要拿到ci_session即可
headers = {
'content-type': "multipart/form-data; boundary=----WebKitFormBoundary7MA4YWxkTrZu0gW",
'Content-Type': "application/x-www-form-urlencoded",
'cookie': cookie,
'user-agent': "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/132.0.0.0 Safari/537.36",
'authorization': authorization,
'accept': "application/json, text/plain, */*",
'Cache-Control': "no-cache",
}
try:
response = session.request("POST", url, data=payload, headers=headers)
if response.status_code == 200:
data = response.json()
print("===========================================")
print("ZIM船信息：", data)
if data['status'] == '0':
print("请检查Cookie或authorization是否有效")
return
return data
else:
print("请求失败，请重新再试")
except requests.exceptions.RequestException as e:
return f"请求错误: {str(e)}"
def deal_ZIM_eta_time_format(date):
"""
处理ZIM船司ETA时间格式
- 将 04-Mar-2025 转换为 2025/3/4
:param date:
:return:
"""
dt = datetime.strptime(date, "%d-%b-%Y")
formatted_date = f"{dt.year}/{dt.month}/{dt.day}"
return formatted_date
def get_ZIM_related_time(tracking_number):
"""
获取ZIM相关船期时间
通过观察ZIM查询逻辑： 如果POD里面获取不到值就去取arrivalDateTz - dateTime
我们加一层判断
:param tracking_number:
:return:
"""
data = get_ZIM_tracking_info(tracking_number)
try:
if data['status'] == '0':
print("ZIM船司-提单号查询返回数据为空，请检查提单号是否有误，或者船司还未同步更新")
else:
estimated_time_of_arrival = data['message']['pod']['estimated_time_of_arrival']
if estimated_time_of_arrival == '':
arrivalDateTz = data['message']['response']['blRouteLegs'][0]['arrivalDateTz']['dateTime']
print("------------------------------------------")
print(f"提单号【{tracking_number}】的船期信息如下：")
print("预计到港时间：", arrivalDateTz)
return arrivalDateTz
else:
new_eta_format = deal_ZIM_eta_time_format(estimated_time_of_arrival)
print("------------------------------------------")
print(f"提单号【{tracking_number}】的船期信息如下：")
print("预计到港时间：", new_eta_format)
return new_eta_format
except requests.exceptions.RequestException as e:
return f"请求数据失败: {str(e)}"
def get_MSC_tracking_info(tracking_number):
"""
查询MSC船公司信息
:param tracking_number: 提单号/预约号
:param tracking_number: 为0根据提单号查询；为1根据预约号查询
这里后端直接加个判断：如果传进来的tracking_number的前面4位数不以[MEDU]开头
则为置为1；否则默认为0
:return: 具体船期时间
"""
session = requests.Session()
api_url = "https://www.msc.com/api/feature/tools/TrackingInfo"
headers = {
"Content-Type": "application/json",
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/132.0.0.0 Safari/537.36",
"X-Requested-With": "XMLHttpRequest",
}
if tracking_number[:4] == "MEDU":
tracking_mode = "0" # 提单号查询
else:
tracking_mode = "1" # 预约号查询
payload = {
"trackingNumber": tracking_number,
"trackingMode": tracking_mode
}
try:
response = session.post(api_url, json=payload, headers=headers)
if response.status_code == 200:
data = response.json()
print("===========================================")
print("MSC船信息:", data)
return data
except requests.exceptions.RequestException as e:
return f"请求错误: {str(e)}"
def deal_MSC_ETA_date_format(date):
"""
:param Pod_eta_date: MSC返回日期为15/02/2025 转为2025/2/15
:return:
"""
date_obj = datetime.strptime(date, "%d/%m/%Y")
new_date_str = date_obj.strftime("%Y/%m/%d")
return new_date_str
def get_MSC_related_time(tracking_number):
"""
根据单号获取MSC相关时间
:param tracking_number: 提单号/预约号
:return:
"""
data = get_MSC_tracking_info(tracking_number)
if data is not None:
# time.sleep(5)
actualShipment_info = data['Data']['BillOfLadings'][0]['ContainersInfo']
print("actualShipment_info:", actualShipment_info)
PodEtaDate = actualShipment_info[0]['PodEtaDate']
if PodEtaDate == '':
# 若ETA为空，则输出最近一条物流消息
print("===========================================")
print(f"该提单【{tracking_number}】目前暂无ETA信息")
latest_event = actualShipment_info[0]['Events'][0] # MSC是倒序的,所以取第一条
date = latest_event['Date']
PodEtaDate = deal_MSC_ETA_date_format(date) # 日期格式转换
EquipmentHandling = latest_event['EquipmentHandling']['Name']
print(f"其最新一条物流消息如下：\n日期：{PodEtaDate}，位于 {EquipmentHandling};")
else:
PodEtaDate = deal_MSC_ETA_date_format(PodEtaDate) # 日期格式转换
print("------------------------------------------")
print(f"提单号【{tracking_number}】的船期信息如下：")
print("预计到港时间：", PodEtaDate)
return PodEtaDate
else:
print("请求存在延迟，返回数据为None，需要再次调用")
def get_COSCO_tracking_info(tracking_number):
"""
根据提单号查询COSCO船司物流信息
:param bill_of_lading: 目前只能根据提单号查询；预约号无法查询到
:return: 返回
"""
tracking_number = deal_COSCO_bill_number(tracking_number) # 处理前缀COSU
session = requests.Session()
headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36",
"Accept": "application/json",
"X-Requested-With": "XMLHttpRequest"
}
try:
# 假设发现的API端点（需要实际分析）
api_url = f"https://elines.coscoshipping.com/ebtracking/public/bill/{tracking_number}"
print("api_url:", api_url)
response = session.get(api_url, headers=headers)
if response.status_code == 200:
data = response.json()
print("===========================================")
print("COSCO船信息：", data)
return data
else:
print("COSCO船信息查询失败")
print("错误返回：", response)
return
except Exception as e:
return f"请求失败: {str(e)}"
def deal_COSCO_bill_number(bill_number):
"""
处理COSCO提单号：COSU6407335642 过滤掉单号前4位前缀
因为COSCO只能根据数值查询，添加了前缀无法查询
:param bill_number: 提单号
:return:
"""
remove_substring = "COSU"
if remove_substring in bill_number:
new_bill_number = bill_number.replace(remove_substring, "")
else:
new_bill_number = bill_number
return new_bill_number
def get_COSCO_related_time(tracking_number):
"""
获取COSCO单号相关时间
:param tracking_number: 提单号
:return:
"""
data = get_COSCO_tracking_info(tracking_number)
# COSCO查询逻辑
actualShipment_list = data['data']['content']['actualShipment']
print("actualShipment_list:", actualShipment_list)
len_actualShipment_list = len(actualShipment_list)
if len_actualShipment_list == 0:
# 若里面只有一条数据，就取这里面的第一条
actualShipment_info = actualShipment_list[0]
print("actualShipment_info:", actualShipment_info)
estimatedDateOfArrival = actualShipment_info['estimatedDateOfArrival']
actualArrivalDate = actualShipment_info['actualArrivalDate']
actualDischargeDate = actualShipment_info['actualDischargeDate']
else:
# 否则取list中最后条数据
actualShipment_info = actualShipment_list[len_actualShipment_list - 1]
print("actualShipment_info:", actualShipment_info)
estimatedDateOfArrival = actualShipment_info['estimatedDateOfArrival']
actualArrivalDate = actualShipment_info['actualArrivalDate']
actualDischargeDate = actualShipment_info['actualDischargeDate']
print("------------------------------------------")
print(f"提单号【{tracking_number}】的船期信息如下：")
print("预计到港时间：", estimatedDateOfArrival)
print("实际到港时间：", actualArrivalDate)
print("实际卸货时间：", actualDischargeDate)
return estimatedDateOfArrival
def assign_shipper(tracking_number):
"""
分配对应公司查询 COSCO | ZIM | MSK | MSC | ONE | PIL
- 使用前缀判断
:param tracking_number:
:return:
"""
if type(tracking_number) != "str":
tracking_number = str(tracking_number)
if "COSU" in tracking_number or tracking_number[0] == '6' or tracking_number[0] == '9':
print("===========================================")
print("【COSCO船司物流信息查询】")
eta_time = get_COSCO_related_time(tracking_number)
elif tracking_number[0] == '2':
print("===========================================")
print("【MSK船司物流信息查询】")
eta_time = get_MSK_related_time(tracking_number)
elif "GOSU" in tracking_number:
print("===========================================")
print("【ZIM船司物流信息查询】")
eta_time = get_ZIM_related_time(tracking_number)
elif "ONEY" in tracking_number:
print("===========================================")
print("【ONE船司物流信息查询】")
eta_time = get_ONE_related_time(tracking_number)
elif "MEDU" in tracking_number or tracking_number[0] == '1':
print("===========================================")
print("【MSC船司物流信息查询】")
eta_time = get_MSC_related_time(tracking_number)
elif "000" in tracking_number:
eta_time = get_ARKAS_related_time(tracking_number)
else:
print("===========================================")
print("【PIL船司物流信息查询】")
eta_time = get_PIL_related_time(tracking_number)
return eta_time
def search_tracking_info_by_list(bl_list):
"""
遍历list查询物流信息
:param bl_list: 提单列表
:return: eta_time_List 预计到港时间列表
"""
eta_time_List = []
for tracking_number in bl_list:
eta_time = assign_shipper(tracking_number)
eta_time_List.append(eta_time)
return eta_time_List
def get_latest_eta_time_and_insert_excel():
"""
1、读取excel里指定船公司的bill
2、将通过bill查询返回的各大船司[预期到港时间]组成一个list
3、在原excel里新增【new_ETA】列，并将最新查询到的【预期到港时间】写入
- ZIM查询暂不稳定，所以先不写进去
:return:
"""
sheet_name = "Sheet1"
file_path = "./update_eta_time_client_list/update_eta_time_list.xlsx"
df = pd.read_excel(file_path, sheet_name=sheet_name)
bill_list = df["B/L"].to_list()
print("bill_list", bill_list)
eta_time_list = search_tracking_info_by_list(bill_list)
print("eta_time_list:", eta_time_list)
new_column_name = "new_ETA" # 新增新ETA列
new_loc = df.columns.get_loc('ETA') + 1
df.insert(loc=new_loc, column=new_column_name, value=eta_time_list) # 插入
print("df:", df)
df.to_excel(file_path, index=False)
def get_local_time():
"""
获取本地日期 - 格式：2025/2/10
:return:
"""
local_time = datetime.now()
formatted_date = local_time.strftime('%Y/%#m/%#d')
return formatted_date
def get_MSK_shipment_detail(bill_number):
"""
获取MSK物流详情
:param bill_number:
:return: 返回符合excel表头的数据list
"""
response = get_MSK_tracking_info(bill_number)
date = get_local_time()
shipping = "MSK"
index = 1
bl_number = bill_number
origin = response['origin']['city']
destination = response['destination']['city']
eta = get_MSK_related_time(bill_number)
container_list = response['containers']
telex_status = "N"
data_list = []
for container in container_list:
print("container:", container)
container_number = container['container_num']
container_size = container['container_size']
container_size_status = 1
str1 = "20"
str2 = "40"
if str1 in container_size:
data_total = [index, date, "", shipping, "", bl_number, "", container_number, date, eta,
telex_status, "", "", "", "", container_size_status, origin, destination]
elif str2 in container_size:
data_total = [index, date, "", shipping, "", bl_number, "", container_number, date, eta,
telex_status, "", "", "", container_size_status, "", origin, destination]
data_list.append(data_total)
index = index + 1
print(f"MSK船司：共插入【{len(data_list)}】条数据")
print("MSK船司新插入数据list：", data_list)
return data_list
def get_COSCO_shipment_detail(bill_number):
"""
获取COSCO物流跟踪详情
:param bill_number:
:return: list | 尺寸未填写
"""
response = get_COSCO_tracking_info(bill_number)
date = get_local_time()
content = response['data']['content']
shipping = "COSCO"
index = 1
bl_number = bill_number
origin = content['actualShipment'][0]['portOfLoading']
destination = content['actualShipment'][0]['portOfDischarge']
eta = get_COSCO_related_time(bill_number)
telex_status = "N"
data_list = []
container_list = content['cargoTrackingContainer']
for container_info in container_list:
container_number = container_info['cntrNum']
# print("container_info:", container_info)
# print("ContainerNumber:", container_number)
data_total = [index, date, "", shipping, "", bl_number, "", container_number, date, eta,
telex_status, "", "", "", "", "", origin, destination]
data_list.append(data_total)
index = index + 1
print(f"COSCO船司：共插入【{len(data_list)}】条数据")
print("COSCO船司新插入数据list：", data_list)
return data_list
def get_MSC_shipment_detail(bill_number):
"""
获取MSC船司物流详细信息
:param bill_number:
:return: list
"""
response = get_MSC_tracking_info(bill_number)
bill_of_ladings = response['Data']['BillOfLadings'][0]
print(bill_of_ladings)
date = get_local_time()
shipping = "MSC"
index = 1
bl_number = bill_number
origin = bill_of_ladings['GeneralTrackingInfo']['ShippedFrom']
destination = bill_of_ladings['GeneralTrackingInfo']['ShippedTo']
eta = get_MSC_related_time(bill_number)
container_list = bill_of_ladings['ContainersInfo']
telex_status = "N"
data_list = []
for container_info in container_list:
container_number = container_info['ContainerNumber']
container_size = container_info['ContainerType']
container_size_status = 1
str1 = "20"
str2 = "40"
print("container_info:", container_info)
print("ContainerNumber:", container_number)
if str1 in container_size:
data_total = [index, date, "", shipping, "", bl_number, "", container_number, date, eta,
telex_status, "", "", "", "", container_size_status, origin, destination]
elif str2 in container_size:
data_total = [index, date, "", shipping, "", bl_number, "", container_number, date, eta,
telex_status, "", "", "", container_size_status, "", origin, destination]
data_list.append(data_total)
index = index + 1
print(f"MSC船司：共插入【{len(data_list)}】条数据")
print("MSC船司新插入数据list：", data_list)
return data_list
def get_ONE_shipment_detail(bill_number):
"""
获取ONE船司物流详情
:param bill_number:
:return:
"""
response = get_ONE_tracking_info_by_bill_number(bill_number)
response2 = get_ONE_tracking_info(bill_number)
bill_of_ladings = response['list']
bill_of_ladings2 = response2['list']
date = get_local_time()
shipping = "ONE"
index = 1
bl_number = bill_number
origin_connect = bill_of_ladings2[0]['placeNm'] # 'QINGDAO, SHANDONG, CHINA'
origin = origin_connect.split(',')[0] # QINGDAO
des_str = "Arrival at Port of Discharging"
for content in bill_of_ladings2:
if des_str in content['statusNm']:
destination = content['placeNm'].split(',')[0]
eta = get_ONE_related_time(bill_number)
telex_status = "N"
data_list = []
data_total = []
for container_info in bill_of_ladings:
container_number = container_info['cntrNo']
container_size = container_info['cntrTpszNm']
container_size_status = 1
str = "20'DRY" # 20'DRY HC. / 20'DRY ST.
str2 = "40'DRY"
if str in container_size:
data_total = [index, date, "", shipping, "", bl_number, "", container_number, date, eta,
telex_status, "", "", "", "", container_size_status, origin, destination]
print(data_total)
elif str2 in container_size:
data_total = [index, date, "", shipping, "", bl_number, "", container_number, date, eta,
telex_status, "", "", "", container_size_status, "", origin, destination]
print(data_total)
else:
print("集装箱尺寸不存在，请详细检查返回参数：cntrTpszNm")
data_list.append(data_total)
index = index + 1
print(f"ONE船司：共插入【{len(data_list)}】条数据")
print("ONE船司新插入数据list：", data_list)
return data_list
def get_ZIM_shipment_detail(bill_number):
"""
获取ZIM船司物流详情
:param tracking_number:
:return: list
"""
response = get_ZIM_tracking_info(bill_number)
blRouteLegs = response['message']['response']['blRouteLegs']
container_list = response['message']['result']
print("blRouteLegs:", blRouteLegs)
date = get_local_time()
shipping = "ZIM"
index = 1
bl_number = bill_number
origin = blRouteLegs[0]['portNameFrom']
destination = blRouteLegs[0]['portNameTo']
eta = get_ZIM_related_time(bill_number)
telex_status = "N"
data_list = []
for container_info in container_list:
str = container_info['container']
parts = str.split() # 自动合并多个空格/制表符等空白字符
part1, part2 = parts[0], parts[1]
container_number = part1
container_size = part2
str1 = "20"
str2 = "40"
# print("container_number:", container_number)
# print("container_size:", container_size)
container_size_status = 1
if str1 in container_size:
data_total = [index, date, "", shipping, "", bl_number, "", container_number, date, eta,
telex_status, "", "", "", "", container_size_status, origin, destination]
elif str2 in container_size:
data_total = [index, date, "", shipping, "", bl_number, "", container_number, date, eta,
telex_status, "", "", "", container_size_status, "", origin, destination]
else:
print("当前集装箱尺寸不存在，请详细查明尺寸参数")
data_list.append(data_total)
index = index + 1
print(f"ZIM船司：共插入【{len(data_list)}】条数据")
print("ZIM船司新插入数据list：", data_list)
return data_list
def get_PIL_shipment_detail(bill_number):
"""
获取PIL船司物流详情
# todo: 解析方法需优化，适应各种情况；目前解析只默认拿取第一条柜子数据
:param tracking_number:
:return: list
"""
response = get_PIL_tracking_info_by_bill_number(bill_number)
print("PIL船信息:", response)
data = response['data']
date = get_local_time()
shipping = "PIL"
index = 1
bl_number = bill_number
container_list = data['table2']
origin = container_list[0]['Place']
str = data['table1'][0]['Next Location']
parts = str.split() # 自动合并多个空格/制表符等空白字符
part1, part2 = parts[0], parts[1]
destination = part1
eta = get_PIL_related_time(bill_number)
telex_status = "N"
data_list = []
for container_info in container_list:
str = container_info['Container #']
parts = str.split() # 自动合并多个空格/制表符等空白字符
part1, part2 = parts[0], parts[1]
container_number = part1
container_size = container_info['Size/Type']
print("container_number:", container_number)
print("container_size:", container_size)
str1 = "20"
str2 = "40"
container_size_status = 1
if str1 in container_size:
data_total = [index, date, "", shipping, "", bl_number, "", container_number, date, eta,
telex_status, "", "", "", "", container_size_status, origin, destination]
elif str2 in container_size:
data_total = [index, date, "", shipping, "", bl_number, "", container_number, date, eta,
telex_status, "", "", "", container_size_status, "", origin, destination]
else:
print("当前集装箱尺寸不存在")
data_list.append(data_total)
index = index + 1
print(f"PIL船司：共插入【{len(data_list)}】条数据")
print("PIL船司新插入数据list：", data_list)
return data_list
def insert_ship_detail_to_excel(tracking_number):
"""
将提单物流详情自动插入到excel中
:param data_total: 数据list
:param headers: 表头 - 这里我按需求写死就行
:return:
"""
if type(tracking_number) != "str":
tracking_number = str(tracking_number)
if "COSU" in tracking_number or tracking_number[0] == '6' or tracking_number[0] == '9':
print("===========================================")
data_total = get_COSCO_shipment_detail(tracking_number)
elif tracking_number[0] == '2':
print("===========================================")
data_total = get_MSK_shipment_detail(tracking_number)
elif "GOSU" in tracking_number:
print("===========================================")
data_total = get_ZIM_shipment_detail(tracking_number)
elif "ONEY" in tracking_number:
print("===========================================")
data_total = get_ONE_shipment_detail(tracking_number)
elif "MEDU" in tracking_number or tracking_number[0] == '1':
print("===========================================")
data_total = get_MSC_shipment_detail(tracking_number)
elif "000" in tracking_number:
data_total = get_ARKAS_shipment_detail(tracking_number)
else:
print("===========================================")
data_total = get_PIL_shipment_detail(tracking_number)
headers = ["Index", "Date", "Client", "Shipping",
"Goods", "B/L", "提单确认", "Container No.", "COPY件",
"ETA", "电放", "电放日期", "B/L2", "正本接收时间",
"40Q", "20Q", "Discharge", "Destination"]
file_path = "./update_eta_time_client_list/auto_insert_new_bill_info.xlsx"
if not Path(file_path).exists():
pd.DataFrame(columns=headers) # 不存在文件, 创建表头
else:
df = pd.read_excel(file_path) # 读取现有数据
new_df = pd.DataFrame(data_total, columns=headers)
combined_df = pd.concat([df, new_df], ignore_index=True) # 合并新旧数据
combined_df.to_excel(file_path, index=False) # 不保留索引列
print(f"数据已追加至 {file_path}")
def insert_ship_detail_list_to_excel(bl_list):
"""
批量插入物流信息到excel
:param bl_list:
:return:
"""
for bill_number in bl_list:
insert_ship_detail_to_excel(bill_number)
# assign_shipper("GOSUNGB20516257") # 单个查询eta时间 - 自行判断哪个船司
# assign_shipper("285233296")
# assign_shipper("HLCUTA12411DAEC0")
# get_latest_eta_time_and_insert_excel() # 获取最新eta时间并写入excel
bl_list = [
"BCN000007279",
# "GOSUNGB20515308",
# "COSU6405220124"
]
# insert_ship_detail_list_to_excel(bl_list) # 批量插入数excel数据
# insert_ship_detail_to_excel("COSU6405220124") # 自动将船司查询到的信息按规定表头写入excel表格
# insert_ship_detail_to_excel("COSU6405220124") # 自动将船司查询到的信息按规定表头写入excel表格
bl_list = [
"BCN000007279",
# "247719498",
# "MEDUKT297182"
]
search_tracking_info_by_list(bl_list) # 批量查询eta时间，传list
# OOCL / CMA /HPL 加密,暂时无法破解；