教你怎么用Python監(jiān)控愉客行車程
更新時間：2021年04月29日 15:50:03 作者：昱禹
這篇文章主要介紹了教你怎么用Python監(jiān)控愉客行車程,文中有非常詳細的代碼示例,對正在學習python的小伙伴們有非常好的幫助,需要的朋友可以參考下
一、愉客行車程監(jiān)控并通知
大概思路：用戶填寫指定信息在config.json文件中，通過定時訪問網頁，獲取指定信息，從而達到對指定車程的監(jiān)控
1.分析網頁
按下F12，打開開發(fā)者工具，再刷新一下網頁
找到我們需要的信息
然后再分析一下它的請求方式
很直觀的就看到了幾條主要的信息
第一條和第三條是null不重要
第二條是起始站
第四條是終點站
第五條是個數(shù)字，經過反復嘗試，發(fā)現(xiàn)是固定參數(shù)
第六條乍一看應該是時間戳，經過驗證，的確是車票指定日期零點的時間戳
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.將返回的數(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)控網站瀏覽記錄實現(xiàn)過程詳解
- python實現(xiàn)批量監(jiān)控網站
- 利用Python自動監(jiān)控網站并發(fā)送郵件告警的方法
- python監(jiān)控網站運行異常并發(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)控你的朋友都在瀏覽哪些網站?
相關文章
在本篇文章里小編給大家整理了一篇關于django中顯示字符串的實例方法，有興趣的朋友們可以跟著學習參考下。2021-03-03
這篇文章主要為大家詳細介紹了python使用minimax算法實現(xiàn)五子棋，具有一定的參考價值，感興趣的小伙伴們可以參考一下2019-07-07
這篇文章主要為大家介紹了Python人工智能構建簡單聊天機器人示例詳解，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進步，早日升職加薪2023-03-03
這篇文章主要為大家詳細介紹了python實現(xiàn)登錄與注冊系統(tǒng)，文中示例代碼介紹的非常詳細，具有一定的參考價值，感興趣的小伙伴們可以參考一下2020-11-11
今天小編就為大家分享一篇pip安裝時ReadTimeoutError的解決方法，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧2018-06-06
這篇文章主要介紹了Python中的異步for循環(huán),我們將討論 Python 庫 asyncio 和運行異步代碼所需的函數(shù)，需要的朋友可以參考下2023-05-05
這篇文章主要介紹了Python根據(jù)指定文件生成XML的方法，文中代碼非常詳細，幫助大家更好的理解和學習，感興趣的朋友可以了解下2020-06-06
ansible是一個python package，是個完全的unpack and play軟件，對客戶端唯一的要求是有ssh有python，并且裝了python-simplejson包，部署上簡單到發(fā)指。下面這篇文章就給大家主要介紹了ansible作為python模塊庫使用的方法實例,需要的朋友可以參考借鑒。2017-01-01
這篇文章主要介紹了python 提高開發(fā)效率的5個小技巧，幫助大家更好的進行python開發(fā)，感興趣的朋友可以了解下2020-10-10
大家好，本篇文章主要講的是Python實現(xiàn)簡易的限流器介紹，感興趣的同學趕快來看一看吧，對你有幫助的話記得收藏一下2022-01-01