目錄
- 一、愉客行車程監(jiān)控并通知
- 二、目錄結構
- 三、完整代碼
- 四、config.json文件
一、愉客行車程監(jiān)控并通知
大概思路：用戶填寫指定信息在config.json文件中，通過定時訪問網(wǎng)頁，獲取指定信息，從而達到對指定車程的監(jiān)控
1.分析網(wǎng)頁
按下F12，打開開發(fā)者工具，再刷新一下網(wǎng)頁
找到我們需要的信息
然后再分析一下它的請求方式
很直觀的就看到了幾條主要的信息
第一條和第三條是null不重要
第二條是起始站
第四條是終點站
第五條是個數(shù)字，經過反復嘗試，發(fā)現(xiàn)是固定參數(shù)
第六條乍一看應該是時間戳，經過驗證，的確是車票指定日期零點的時間戳
2.請求頭偽裝、帶參訪問指定網(wǎng)頁，獲取信息：
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.將返回的數(shù)據(jù)解析
因為請求獲得的數(shù)據(jù)是json格式的，所以用jsonpath做數(shù)據(jù)解析
def parse_html(html):
# 解析獲取的數(shù)據(jù)
items = []
html = json.loads(html)
for i in range(len(jsonpath.jsonpath(html, '$..scheduleInfo'))):
item = {}
timeStamp = jsonpath.jsonpath(html, '$..scheduleInfo..departureTime')[i]
item["發(fā)車日期"] = time.strftime("%Y-%m-%d", time.localtime(timeStamp))
# 檢測是否過期
out_data(item["發(fā)車日期"])
item["發(fā)車時間"] = 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["狀態(tài)"] = '\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["發(fā)車時間"] 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["發(fā)車時間"] + '\n')
# print(has_ticks)
return has_ticks
5.格式化終端輸出信息
輸出車程信息，這里改了終端車次顯示的顏色，有票的是綠色、沒票的是紅色，很快就能識別出自己想要的
def format_info(bus_list):
print(bus_list[0]["發(fā)車日期"] + '\t' + bus_list[0]["起始站"] + '-' + bus_list[0]["終點站"])
print('-' * 120)
# print("\t發(fā)車時間"
# "\t\t\t起始站"
# "\t\t\t終點站"
# "\t\t余票"
# "\t\t票價"
# "\t\t路線"
# "\t\t車型"
# "\t\t車牌號")
for bus in bus_list:
print(bus["狀態(tài)"] + "\t" + bus["發(fā)車時間"],
"\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):
"""發(fā)送郵件"""
# 第三方 SMTP 服務
mail_host = 'smtp.qq.com' # 設置服務器
sender = mail_user
# 創(chuàng)建一個帶附件的案例
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發(fā)送成功") # 郵件發(fā)送成功
except Exception as e:
pass
finally:
smtpObj.quit()
7.設定主函數(shù)
這里把用戶輸入的信息轉換一下，將日期轉為時間戳，并且可支持多車程的監(jiān)控，配置文件應一一對應。
將獲取到的車程信息保存
如果有變化，立刻發(fā)送郵件通知
設定了定時執(zhí)行，這里是每隔30分鐘執(zhí)行一次
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定時任務已觸發(fā)至：第%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文件的信息，執(zhí)行main函數(shù)，開始定時任務
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["發(fā)送人"]
mail_user = config["用戶名"]
mail_pass = config["第三方客戶端授權碼"]
receivers = config["接收方"]
# 定時延遲
timer_times = 0
timer = threading.Timer(1800, main())
timer.start()
本來是想掛到服務器上，就做了一個檢測日期的函數(shù)，如果車程日期在當前日期之前，就直接退出程序，最后還是在本地上運行的，就沒用的上
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):
# 解析獲取的數(shù)據(jù)
items = []
html = json.loads(html)
for i in range(len(jsonpath.jsonpath(html, '$..scheduleInfo'))):
item = {}
timeStamp = jsonpath.jsonpath(html, '$..scheduleInfo..departureTime')[i]
item["發(fā)車日期"] = time.strftime("%Y-%m-%d", time.localtime(timeStamp))
# 檢測是否過期
out_data(item["發(fā)車日期"])
item["發(fā)車時間"] = 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["狀態(tài)"] = '\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["發(fā)車時間"] 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["發(fā)車時間"] + '\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]["發(fā)車日期"] + '\t' + bus_list[0]["起始站"] + '-' + bus_list[0]["終點站"])
print('-' * 120)
# print("\t發(fā)車時間"
# "\t\t\t起始站"
# "\t\t\t終點站"
# "\t\t余票"
# "\t\t票價"
# "\t\t路線"
# "\t\t車型"
# "\t\t車牌號")
for bus in bus_list:
print(bus["狀態(tài)"] + "\t" + bus["發(fā)車時間"],
"\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):
"""發(fā)送郵件"""
# 第三方 SMTP 服務
mail_host = 'smtp.qq.com' # 設置服務器
sender = mail_user
# 創(chuàng)建一個帶附件的案例
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發(fā)送成功") # 郵件發(fā)送成功
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定時任務已觸發(fā)至：第%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["發(fā)送人"]
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"
],
"起始站": [
"萬州",
"彭水縣"
],
"終點站": [
"涪陵",
"萬州"
],
"發(fā)送人": [
"愉客行",
"愉客行"
],
"用戶名": [
"1*******27@qq.com",
"1*******27@qq.com"
],
"第三方客戶端授權碼": [
"oxms********iicj",
"oxms********iicj"
],
"接收方": [
"265******8@qq.com",
"265******8@qq.com"
]
}
到此這篇關于教你怎么用Python監(jiān)控愉客行車程的文章就介紹到這了,更多相關Python監(jiān)控愉客行車程內容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家！
您可能感興趣的文章:- Python實時監(jiān)控網(wǎng)站瀏覽記錄實現(xiàn)過程詳解
- python實現(xiàn)批量監(jiān)控網(wǎng)站
- 利用Python自動監(jiān)控網(wǎng)站并發(fā)送郵件告警的方法
- python監(jiān)控網(wǎng)站運行異常并發(fā)送郵件的方法
- 用python實現(xiàn)監(jiān)控視頻人數(shù)統(tǒng)計
- 用Python監(jiān)控NASA TV直播畫面的實現(xiàn)步驟
- Python實戰(zhàn)之能監(jiān)控文件變化的神器—看門狗
- Python實現(xiàn)用手機監(jiān)控遠程控制電腦的方法
- python實現(xiàn)的web監(jiān)控系統(tǒng)
- 用Python監(jiān)控你的朋友都在瀏覽哪些網(wǎng)站?
