教你怎麼用Python監控愉客行車程
一、愉客行車程監控並通知
大概思路：用戶填寫指定信息在config.json文件中，通過定時訪問網頁，獲取指定信息，從而達到對指定車程的監控
1.分析網頁
按下F12，打開開發者工具，再刷新一下網頁
找到我們需要的信息
然後再分析一下它的請求方式
很直觀的就看到瞭幾條主要的信息
第一條和第三條是null不重要
第二條是起始站
第四條是終點站
第五條是個數字，經過反復嘗試，發現是固定參數
第六條乍一看應該是時間戳，經過驗證，的確是車票指定日期零點的時間戳
2.請求頭偽裝、帶參訪問指定網頁，獲取信息：
def get_html(startStation, endStation, timeStamp):
# 模擬請求
headers = {
'Accept': 'application/json, text/javascript, */*; q=0.01',
'Accept-Encoding': 'gzip, deflate, br',
'Accept-Language': 'zh-CN,zh;q=0.9,en-US;q=0.8,en;q=0.7,zh-HK;q=0.6',
'Connection': 'keep-alive',
'Content-Length': '124',
'Content-Type': 'application/json; charset=UTF-8',
'sec-ch-ua': '" Not A;Brand";v="99", "Chromium";v="90", "Google Chrome";v="90"',
'sec-ch-ua-mobile': '?0',
'Sec-Fetch-Dest': 'empty',
'Sec-Fetch-Mode': 'cors',
'Sec-Fetch-Site': 'cross-site',
'Host': 'busserver.cqyukexing.com',
'Origin': 'https://www.96096kp.com',
'Referer': 'https://www.96096kp.com/',
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/90.0.4430.72 Safari/537.36',
}
data = {
'departureName': startStation,
'destinationId': 'null',
'destinationName': endStation,
'opSource': '7',
# 指定日期時間戳
'queryDate': timeStamp,
}
data = json.dumps(data)
url = 'https://busserver.cqyukexing.com/busticket/schedule_list_310?channel=7'
response = requests.post(url, headers=headers, data=data, timeout=5)
if response.status_code == 200:
html = response.text
# print(html)
return html
3.將返回的數據解析
因為請求獲得的數據是json格式的，所以用jsonpath做數據解析
def parse_html(html):
# 解析獲取的數據
items = []
html = json.loads(html)
for i in range(len(jsonpath.jsonpath(html, '$..scheduleInfo'))):
item = {}
timeStamp = jsonpath.jsonpath(html, '$..scheduleInfo..departureTime')[i]
item["發車日期"] = time.strftime("%Y-%m-%d", time.localtime(timeStamp))
# 檢測是否過期
out_data(item["發車日期"])
item["發車時間"] = jsonpath.jsonpath(html, '$..scheduleInfo..departureTimeDesc')[i]
item["起始站"] = jsonpath.jsonpath(html, '$..departureStation..name')[i]
# item["地址"] = jsonpath.jsonpath(html, '$..departureStation..addr')[i]
item["終點站"] = jsonpath.jsonpath(html, '$..destinationStation..name')[i]
item["餘票"] = jsonpath.jsonpath(html, '$..scheduleInfo..remainSeatCnt')[i]
item["票價"] = jsonpath.jsonpath(html, '$..scheduleInfo..fullTicketPrice')[i]
item["車型"] = jsonpath.jsonpath(html, '$..scheduleInfo..busType')[i]
item["車牌號"] = jsonpath.jsonpath(html, '$..scheduleInfo..scheduleCode')[i]
item["路線"] = jsonpath.jsonpath(html, '$..scheduleInfo..lineName')[i][3:]
item["狀態"] = '\033[32m' if item["餘票"] > 0 else '\033[31m'
# item["途徑"] = jsonpath.jsonpath(html, '$..scheduleInfo..stopStation')[i]
items.append(item)
return items
4.篩選出有票的車次
這裡是將已經獲取過的車次保存到文件中，一旦檢測到新的車次，就準備通知，如果檢測到沒有新車次，不做通知
def watch_ticks(bus_list):
# 檢查目前還有票的車次
format_info(bus_list)
has_ticks = []
filename = 'tick_log of ' + bus_list[0]["起始站"] + '-' + bus_list[0]["終點站"] + '.txt'
# 如果log文件不存在，則新建一個空的文件
if not os.path.exists('./logs/' + filename):
f = open('./logs/' + filename, 'w')
f.close()
with open('./logs/' + filename, 'r+', encoding='utf-8') as file:
alreald_send = file.read()
for bus in bus_list:
if bus["餘票"] != 0 and bus["發車時間"] not in alreald_send or not len(alreald_send):
has_ticks.append(bus)
with open('./logs/tick_log of ' + bus["起始站"] + '-' + bus["終點站"] + '.txt', 'a+', encoding='utf-8') as file:
file.write(bus["發車時間"] + '\n')
# print(has_ticks)
return has_ticks
5.格式化終端輸出信息
輸出車程信息，這裡改瞭終端車次顯示的顏色，有票的是綠色、沒票的是紅色，很快就能識別出自己想要的
def format_info(bus_list):
print(bus_list[0]["發車日期"] + '\t' + bus_list[0]["起始站"] + '-' + bus_list[0]["終點站"])
print('-' * 120)
# print("\t發車時間"
# "\t\t\t起始站"
# "\t\t\t終點站"
# "\t\t餘票"
# "\t\t票價"
# "\t\t路線"
# "\t\t車型"
# "\t\t車牌號")
for bus in bus_list:
print(bus["狀態"] + "\t" + bus["發車時間"],
"\t\t" + bus["起始站"],
"\t\t" + bus["終點站"],
"\t\t" + str(bus["餘票"]),
"\t\t\t" + str(bus["票價"]),
"\t\t" + bus["路線"],
"\t\t" + bus["車型"],
"\t\t" + bus["車牌號"] + '\033[0m')
print('-' * 120)
6.設定郵件通知
這裡代碼是以前的，我直接拿來改瞭一下
def send_email(sendUser, mail_user, mail_pass, receivers, start, end, tick_date, message):
"""發送郵件"""
# 第三方 SMTP 服務
mail_host = 'smtp.qq.com' # 設置服務器
sender = mail_user
# 創建一個帶附件的案例
mail = MIMEMultipart()
mail['From'] = Header(sendUser, 'utf-8')
mail['To'] = ";".join(receivers)
subject = '愉客行有新的票務情況：' + tick_date + '-' + start + '-' + end # 郵件標題
mail['Subject'] = Header(subject, 'utf-8')
# 郵件正文內容
mail.attach(MIMEText(message, 'plain', 'utf-8'))
try:
smtpObj = smtplib.SMTP()
smtpObj.connect(mail_host, 25) # 25為端口號
smtpObj.login(mail_user, mail_pass)
smtpObj.sendmail(sender, receivers, mail.as_string())
print(receivers + "\t發送成功") # 郵件發送成功
except Exception as e:
pass
finally:
smtpObj.quit()
7.設定主函數
這裡把用戶輸入的信息轉換一下，將日期轉為時間戳，並且可支持多車程的監控，配置文件應一一對應。
將獲取到的車程信息保存
如果有變化，立刻發送郵件通知
設定瞭定時執行，這裡是每隔30分鐘執行一次
def main():
global timer_times
timer_times = timer_times + 1
for i in range(len(startStation)):
html = get_html(startStation[i], endStation[i], timeStamp[i])
bus_list = parse_html(html)
# pprint.pprint(bus_list)
has_ticks = watch_ticks(bus_list)
json.dump(bus_list,
open('./data/bus_list of ' + startStation[i] + '-' + endStation[i] + '.json', 'a+', encoding='utf-8'),
ensure_ascii=False)
if len(has_ticks):
json.dump(has_ticks, open('./data/has_ticks of ' + startStation[i] + '-' + endStation[i] + '.json', 'w+',
encoding='utf-8'), ensure_ascii=False)
message = '\n'.join([str(tick).replace(',', '\n') for tick in has_ticks])
send_email(sendUser[i], mail_user[i], mail_pass[i], receivers[i], startStation[i], endStation[i],
ticksDate[i], message)
# 定時延遲
now = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime())
log_message = ("\n定時任務已觸發至：第%s輪\n當前時間：%s\n" % (timer_times, now))
with open("./logs/log.txt", 'a+', encoding="utf-8") as file:
file.write(log_message)
print(log_message)
time.sleep(1800)
timer = threading.Timer(1800, main())
timer.start()
8.程序入口
獲取config.json文件的信息，執行main函數，開始定時任務
if __name__ == '__main__':
with open('config.json', 'r', encoding='utf-8') as file:
config = json.load(file)
startStation = config["起始站"]
endStation = config["終點站"]
ticksDate = config["車票日期"]
timeArray = [time.strptime(tick_date + ' 00:00:00', "%Y-%m-%d %H:%M:%S") for tick_date in config["車票日期"]]
timeStamp = [int(time.mktime(times)) for times in timeArray]
sendUser = config["發送人"]
mail_user = config["用戶名"]
mail_pass = config["第三方客戶端授權碼"]
receivers = config["接收方"]
# 定時延遲
timer_times = 0
timer = threading.Timer(1800, main())
timer.start()
本來是想掛到服務器上，就做瞭一個檢測日期的函數，如果車程日期在當前日期之前，就直接退出程序，最後還是在本地上運行的，就沒用的上
def out_data(date):
# 檢查車票跟蹤是否過時
# 是否過期一天
tomorrow = datetime.date.today() - datetime.timedelta(days=1)
if date == tomorrow:
print("車票跟蹤已過時！")
os.exit(0)
9.結果圖
二、目錄結構
三、完整代碼
import datetime
import os
import smtplib
import threading
import time
from email.header import Header
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
import requests
import json
import jsonpath
def get_html(startStation, endStation, timeStamp):
# 模擬請求
headers = {
'Accept': 'application/json, text/javascript, */*; q=0.01',
'Accept-Encoding': 'gzip, deflate, br',
'Accept-Language': 'zh-CN,zh;q=0.9,en-US;q=0.8,en;q=0.7,zh-HK;q=0.6',
'Connection': 'keep-alive',
'Content-Length': '124',
'Content-Type': 'application/json; charset=UTF-8',
'sec-ch-ua': '" Not A;Brand";v="99", "Chromium";v="90", "Google Chrome";v="90"',
'sec-ch-ua-mobile': '?0',
'Sec-Fetch-Dest': 'empty',
'Sec-Fetch-Mode': 'cors',
'Sec-Fetch-Site': 'cross-site',
'Host': 'busserver.cqyukexing.com',
'Origin': 'https://www.96096kp.com',
'Referer': 'https://www.96096kp.com/',
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/90.0.4430.72 Safari/537.36',
}
data = {
'departureName': startStation,
'destinationId': 'null',
'destinationName': endStation,
'opSource': '7',
# 指定日期時間戳
'queryDate': timeStamp,
}
data = json.dumps(data)
url = 'https://busserver.cqyukexing.com/busticket/schedule_list_310?channel=7'
response = requests.post(url, headers=headers, data=data, timeout=5)
if response.status_code == 200:
html = response.text
# print(html)
return html
def parse_html(html):
# 解析獲取的數據
items = []
html = json.loads(html)
for i in range(len(jsonpath.jsonpath(html, '$..scheduleInfo'))):
item = {}
timeStamp = jsonpath.jsonpath(html, '$..scheduleInfo..departureTime')[i]
item["發車日期"] = time.strftime("%Y-%m-%d", time.localtime(timeStamp))
# 檢測是否過期
out_data(item["發車日期"])
item["發車時間"] = jsonpath.jsonpath(html, '$..scheduleInfo..departureTimeDesc')[i]
item["起始站"] = jsonpath.jsonpath(html, '$..departureStation..name')[i]
# item["地址"] = jsonpath.jsonpath(html, '$..departureStation..addr')[i]
item["終點站"] = jsonpath.jsonpath(html, '$..destinationStation..name')[i]
item["餘票"] = jsonpath.jsonpath(html, '$..scheduleInfo..remainSeatCnt')[i]
item["票價"] = jsonpath.jsonpath(html, '$..scheduleInfo..fullTicketPrice')[i]
item["車型"] = jsonpath.jsonpath(html, '$..scheduleInfo..busType')[i]
item["車牌號"] = jsonpath.jsonpath(html, '$..scheduleInfo..scheduleCode')[i]
item["路線"] = jsonpath.jsonpath(html, '$..scheduleInfo..lineName')[i][3:]
item["狀態"] = '\033[32m' if item["餘票"] > 0 else '\033[31m'
# item["途徑"] = jsonpath.jsonpath(html, '$..scheduleInfo..stopStation')[i]
items.append(item)
return items
def watch_ticks(bus_list):
# 檢查目前還有票的車次
format_info(bus_list)
has_ticks = []
filename = 'tick_log of ' + bus_list[0]["起始站"] + '-' + bus_list[0]["終點站"] + '.txt'
# 如果log文件不存在，則新建一個空的文件
if not os.path.exists('./logs/' + filename):
f = open('./logs/' + filename, 'w')
f.close()
with open('./logs/' + filename, 'r+', encoding='utf-8') as file:
alreald_send = file.read()
for bus in bus_list:
if bus["餘票"] != 0 and bus["發車時間"] not in alreald_send or not len(alreald_send):
has_ticks.append(bus)
with open('./logs/tick_log of ' + bus["起始站"] + '-' + bus["終點站"] + '.txt', 'a+', encoding='utf-8') as file:
file.write(bus["發車時間"] + '\n')
# print(has_ticks)
return has_ticks
def out_data(date):
# 檢查車票跟蹤是否過時
# 是否過期一天
tomorrow = datetime.date.today() - datetime.timedelta(days=1)
if date == tomorrow:
print("車票跟蹤已過時！")
os.exit(0)
def format_info(bus_list):
print(bus_list[0]["發車日期"] + '\t' + bus_list[0]["起始站"] + '-' + bus_list[0]["終點站"])
print('-' * 120)
# print("\t發車時間"
# "\t\t\t起始站"
# "\t\t\t終點站"
# "\t\t餘票"
# "\t\t票價"
# "\t\t路線"
# "\t\t車型"
# "\t\t車牌號")
for bus in bus_list:
print(bus["狀態"] + "\t" + bus["發車時間"],
"\t\t" + bus["起始站"],
"\t\t" + bus["終點站"],
"\t\t" + str(bus["餘票"]),
"\t\t\t" + str(bus["票價"]),
"\t\t" + bus["路線"],
"\t\t" + bus["車型"],
"\t\t" + bus["車牌號"] + '\033[0m')
print('-' * 120)
def send_email(sendUser, mail_user, mail_pass, receivers, start, end, tick_date, message):
"""發送郵件"""
# 第三方 SMTP 服務
mail_host = 'smtp.qq.com' # 設置服務器
sender = mail_user
# 創建一個帶附件的案例
mail = MIMEMultipart()
mail['From'] = Header(sendUser, 'utf-8')
mail['To'] = ";".join(receivers)
subject = '愉客行有新的票務情況：' + tick_date + '-' + start + '-' + end # 郵件標題
mail['Subject'] = Header(subject, 'utf-8')
# 郵件正文內容
mail.attach(MIMEText(message, 'plain', 'utf-8'))
try:
smtpObj = smtplib.SMTP()
smtpObj.connect(mail_host, 25) # 25為端口號
smtpObj.login(mail_user, mail_pass)
smtpObj.sendmail(sender, receivers, mail.as_string())
print(receivers + "\t發送成功") # 郵件發送成功
except Exception as e:
pass
finally:
smtpObj.quit()
def main():
global timer_times
timer_times = timer_times + 1
for i in range(len(startStation)):
html = get_html(startStation[i], endStation[i], timeStamp[i])
bus_list = parse_html(html)
# pprint.pprint(bus_list)
has_ticks = watch_ticks(bus_list)
json.dump(bus_list,
open('./data/bus_list of ' + startStation[i] + '-' + endStation[i] + '.json', 'a+', encoding='utf-8'),
ensure_ascii=False)
if len(has_ticks):
json.dump(has_ticks, open('./data/has_ticks of ' + startStation[i] + '-' + endStation[i] + '.json', 'w+',
encoding='utf-8'), ensure_ascii=False)
message = '\n'.join([str(tick).replace(',', '\n') for tick in has_ticks])
send_email(sendUser[i], mail_user[i], mail_pass[i], receivers[i], startStation[i], endStation[i],
ticksDate[i], message)
# 定時延遲
now = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime())
log_message = ("\n定時任務已觸發至：第%s輪\n當前時間：%s\n" % (timer_times, now))
with open("./logs/log.txt", 'a+', encoding="utf-8") as file:
file.write(log_message)
print(log_message)
time.sleep(1800)
timer = threading.Timer(1800, main())
timer.start()
if __name__ == '__main__':
with open('config.json', 'r', encoding='utf-8') as file:
config = json.load(file)
startStation = config["起始站"]
endStation = config["終點站"]
ticksDate = config["車票日期"]
timeArray = [time.strptime(tick_date + ' 00:00:00', "%Y-%m-%d %H:%M:%S") for tick_date in config["車票日期"]]
timeStamp = [int(time.mktime(times)) for times in timeArray]
sendUser = config["發送人"]
mail_user = config["用戶名"]
mail_pass = config["第三方客戶端授權碼"]
receivers = config["接收方"]
# 定時延遲
timer_times = 0
timer = threading.Timer(1800, main())
timer.start()
四、config.json文件
{
"車票日期": [
"2021-4-30",
"2021-5-5"
],
"起始站": [
"萬州",
"彭水縣"
],
"終點站": [
"涪陵",
"萬州"
],
"發送人": [
"愉客行",
"愉客行"
],
"用戶名": [
"1*******[email protected]",
"1*******[email protected]"
],
"第三方客戶端授權碼": [
"oxms********iicj",
"oxms********iicj"
],
"接收方": [
"265******[email protected]",
"265******[email protected]"
]
}
到此這篇關於教你怎麼用Python監控愉客行車程的文章就介紹到這瞭,更多相關Python監控愉客行車程內容請搜索WalkonNet以前的文章或繼續瀏覽下面的相關文章希望大傢以後多多支持WalkonNet！
推薦閱讀：
- 解析python中的jsonpath 提取器
- 詳解Python如何實現發送帶附件的電子郵件
- Python常用配置文件ini、json、yaml讀寫總結
- python-json校驗-jsonpath解析
- Python實現yaml與json文件批量互轉