教你怎么用Python監(jiān)控愉客行車程
一、愉客行車程監(jiān)控并通知
大概思路:用戶填寫指定信息在config.json文件中,通過定時(shí)訪問網(wǎng)頁(yè),獲取指定信息,從而達(dá)到對(duì)指定車程的監(jiān)控
1.分析網(wǎng)頁(yè)
按下F12,打開開發(fā)者工具,再刷新一下網(wǎng)頁(yè)
找到我們需要的信息
然后再分析一下它的請(qǐng)求方式
很直觀的就看到了幾條主要的信息
第一條和第三條是null不重要
第二條是起始站
第四條是終點(diǎn)站
第五條是個(gè)數(shù)字,經(jīng)過反復(fù)嘗試,發(fā)現(xiàn)是固定參數(shù)
第六條乍一看應(yīng)該是時(shí)間戳,經(jīng)過驗(yàn)證,的確是車票指定日期零點(diǎn)的時(shí)間戳
2.請(qǐng)求頭偽裝、帶參訪問指定網(wǎng)頁(yè),獲取信息:
def get_html(startStation, endStation, timeStamp): # 模擬請(qǐng)求 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', # 指定日期時(shí)間戳 '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ù)解析
因?yàn)檎?qǐng)求獲得的數(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)) # 檢測(cè)是否過期 out_data(item["發(fā)車日期"]) item["發(fā)車時(shí)間"] = jsonpath.jsonpath(html, '$..scheduleInfo..departureTimeDesc')[i] item["起始站"] = jsonpath.jsonpath(html, '$..departureStation..name')[i] # item["地址"] = jsonpath.jsonpath(html, '$..departureStation..addr')[i] item["終點(diǎn)站"] = jsonpath.jsonpath(html, '$..destinationStation..name')[i] item["余票"] = jsonpath.jsonpath(html, '$..scheduleInfo..remainSeatCnt')[i] item["票價(jià)"] = jsonpath.jsonpath(html, '$..scheduleInfo..fullTicketPrice')[i] item["車型"] = jsonpath.jsonpath(html, '$..scheduleInfo..busType')[i] item["車牌號(hào)"] = 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.篩選出有票的車次
這里是將已經(jīng)獲取過的車次保存到文件中,一旦檢測(cè)到新的車次,就準(zhǔn)備通知,如果檢測(cè)到?jīng)]有新車次,不做通知
def watch_ticks(bus_list): # 檢查目前還有票的車次 format_info(bus_list) has_ticks = [] filename = 'tick_log of ' + bus_list[0]["起始站"] + '-' + bus_list[0]["終點(diǎn)站"] + '.txt' # 如果log文件不存在,則新建一個(gè)空的文件 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ā)車時(shí)間"] not in alreald_send or not len(alreald_send): has_ticks.append(bus) with open('./logs/tick_log of ' + bus["起始站"] + '-' + bus["終點(diǎn)站"] + '.txt', 'a+', encoding='utf-8') as file: file.write(bus["發(fā)車時(shí)間"] + '\n') # print(has_ticks) return has_ticks
5.格式化終端輸出信息
輸出車程信息,這里改了終端車次顯示的顏色,有票的是綠色、沒票的是紅色,很快就能識(shí)別出自己想要的
def format_info(bus_list): print(bus_list[0]["發(fā)車日期"] + '\t' + bus_list[0]["起始站"] + '-' + bus_list[0]["終點(diǎn)站"]) print('-' * 120) # print("\t發(fā)車時(shí)間" # "\t\t\t起始站" # "\t\t\t終點(diǎn)站" # "\t\t余票" # "\t\t票價(jià)" # "\t\t路線" # "\t\t車型" # "\t\t車牌號(hào)") for bus in bus_list: print(bus["狀態(tài)"] + "\t" + bus["發(fā)車時(shí)間"], "\t\t" + bus["起始站"], "\t\t" + bus["終點(diǎn)站"], "\t\t" + str(bus["余票"]), "\t\t\t" + str(bus["票價(jià)"]), "\t\t" + bus["路線"], "\t\t" + bus["車型"], "\t\t" + bus["車牌號(hào)"] + '\033[0m') print('-' * 120)
6.設(shè)定郵件通知
這里代碼是以前的,我直接拿來改了一下
def send_email(sendUser, mail_user, mail_pass, receivers, start, end, tick_date, message): """發(fā)送郵件""" # 第三方 SMTP 服務(wù) mail_host = 'smtp.qq.com' # 設(shè)置服務(wù)器 sender = mail_user # 創(chuàng)建一個(gè)帶附件的案例 mail = MIMEMultipart() mail['From'] = Header(sendUser, 'utf-8') mail['To'] = ";".join(receivers) subject = '愉客行有新的票務(wù)情況:' + tick_date + '-' + start + '-' + end # 郵件標(biāo)題 mail['Subject'] = Header(subject, 'utf-8') # 郵件正文內(nèi)容 mail.attach(MIMEText(message, 'plain', 'utf-8')) try: smtpObj = smtplib.SMTP() smtpObj.connect(mail_host, 25) # 25為端口號(hào) 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è)定主函數(shù)
這里把用戶輸入的信息轉(zhuǎn)換一下,將日期轉(zhuǎn)為時(shí)間戳,并且可支持多車程的監(jiān)控,配置文件應(yīng)一一對(duì)應(yīng)。
將獲取到的車程信息保存
如果有變化,立刻發(fā)送郵件通知
設(shè)定了定時(shí)執(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) # 定時(shí)延遲 now = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime()) log_message = ("\n定時(shí)任務(wù)已觸發(fā)至:第%s輪\n當(dāng)前時(shí)間:%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ù),開始定時(shí)任務(wù)
if __name__ == '__main__': with open('config.json', 'r', encoding='utf-8') as file: config = json.load(file) startStation = config["起始站"] endStation = config["終點(diǎn)站"] 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["第三方客戶端授權(quán)碼"] receivers = config["接收方"] # 定時(shí)延遲 timer_times = 0 timer = threading.Timer(1800, main()) timer.start()
本來是想掛到服務(wù)器上,就做了一個(gè)檢測(cè)日期的函數(shù),如果車程日期在當(dāng)前日期之前,就直接退出程序,最后還是在本地上運(yùn)行的,就沒用的上
def out_data(date): # 檢查車票跟蹤是否過時(shí) # 是否過期一天 tomorrow = datetime.date.today() - datetime.timedelta(days=1) if date == tomorrow: print("車票跟蹤已過時(shí)!") os.exit(0)
9.結(jié)果圖
二、目錄結(jié)構(gòu)
三、完整代碼
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): # 模擬請(qǐng)求 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', # 指定日期時(shí)間戳 '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)) # 檢測(cè)是否過期 out_data(item["發(fā)車日期"]) item["發(fā)車時(shí)間"] = jsonpath.jsonpath(html, '$..scheduleInfo..departureTimeDesc')[i] item["起始站"] = jsonpath.jsonpath(html, '$..departureStation..name')[i] # item["地址"] = jsonpath.jsonpath(html, '$..departureStation..addr')[i] item["終點(diǎn)站"] = jsonpath.jsonpath(html, '$..destinationStation..name')[i] item["余票"] = jsonpath.jsonpath(html, '$..scheduleInfo..remainSeatCnt')[i] item["票價(jià)"] = jsonpath.jsonpath(html, '$..scheduleInfo..fullTicketPrice')[i] item["車型"] = jsonpath.jsonpath(html, '$..scheduleInfo..busType')[i] item["車牌號(hào)"] = 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]["終點(diǎn)站"] + '.txt' # 如果log文件不存在,則新建一個(gè)空的文件 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ā)車時(shí)間"] not in alreald_send or not len(alreald_send): has_ticks.append(bus) with open('./logs/tick_log of ' + bus["起始站"] + '-' + bus["終點(diǎn)站"] + '.txt', 'a+', encoding='utf-8') as file: file.write(bus["發(fā)車時(shí)間"] + '\n') # print(has_ticks) return has_ticks def out_data(date): # 檢查車票跟蹤是否過時(shí) # 是否過期一天 tomorrow = datetime.date.today() - datetime.timedelta(days=1) if date == tomorrow: print("車票跟蹤已過時(shí)!") os.exit(0) def format_info(bus_list): print(bus_list[0]["發(fā)車日期"] + '\t' + bus_list[0]["起始站"] + '-' + bus_list[0]["終點(diǎn)站"]) print('-' * 120) # print("\t發(fā)車時(shí)間" # "\t\t\t起始站" # "\t\t\t終點(diǎn)站" # "\t\t余票" # "\t\t票價(jià)" # "\t\t路線" # "\t\t車型" # "\t\t車牌號(hào)") for bus in bus_list: print(bus["狀態(tài)"] + "\t" + bus["發(fā)車時(shí)間"], "\t\t" + bus["起始站"], "\t\t" + bus["終點(diǎn)站"], "\t\t" + str(bus["余票"]), "\t\t\t" + str(bus["票價(jià)"]), "\t\t" + bus["路線"], "\t\t" + bus["車型"], "\t\t" + bus["車牌號(hào)"] + '\033[0m') print('-' * 120) def send_email(sendUser, mail_user, mail_pass, receivers, start, end, tick_date, message): """發(fā)送郵件""" # 第三方 SMTP 服務(wù) mail_host = 'smtp.qq.com' # 設(shè)置服務(wù)器 sender = mail_user # 創(chuàng)建一個(gè)帶附件的案例 mail = MIMEMultipart() mail['From'] = Header(sendUser, 'utf-8') mail['To'] = ";".join(receivers) subject = '愉客行有新的票務(wù)情況:' + tick_date + '-' + start + '-' + end # 郵件標(biāo)題 mail['Subject'] = Header(subject, 'utf-8') # 郵件正文內(nèi)容 mail.attach(MIMEText(message, 'plain', 'utf-8')) try: smtpObj = smtplib.SMTP() smtpObj.connect(mail_host, 25) # 25為端口號(hào) 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) # 定時(shí)延遲 now = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime()) log_message = ("\n定時(shí)任務(wù)已觸發(fā)至:第%s輪\n當(dāng)前時(shí)間:%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["終點(diǎn)站"] 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["第三方客戶端授權(quán)碼"] receivers = config["接收方"] # 定時(shí)延遲 timer_times = 0 timer = threading.Timer(1800, main()) timer.start()
四、config.json文件
{ "車票日期": [ "2021-4-30", "2021-5-5" ], "起始站": [ "萬(wàn)州", "彭水縣" ], "終點(diǎn)站": [ "涪陵", "萬(wàn)州" ], "發(fā)送人": [ "愉客行", "愉客行" ], "用戶名": [ "1*******27@qq.com", "1*******27@qq.com" ], "第三方客戶端授權(quán)碼": [ "oxms********iicj", "oxms********iicj" ], "接收方": [ "265******8@qq.com", "265******8@qq.com" ] }
到此這篇關(guān)于教你怎么用Python監(jiān)控愉客行車程的文章就介紹到這了,更多相關(guān)Python監(jiān)控愉客行車程內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家!
- Python實(shí)時(shí)監(jiān)控網(wǎng)站瀏覽記錄實(shí)現(xiàn)過程詳解
- python實(shí)現(xiàn)批量監(jiān)控網(wǎng)站
- 利用Python自動(dòng)監(jiān)控網(wǎng)站并發(fā)送郵件告警的方法
- python監(jiān)控網(wǎng)站運(yùn)行異常并發(fā)送郵件的方法
- 用python實(shí)現(xiàn)監(jiān)控視頻人數(shù)統(tǒng)計(jì)
- 用Python監(jiān)控NASA TV直播畫面的實(shí)現(xiàn)步驟
- Python實(shí)戰(zhàn)之能監(jiān)控文件變化的神器—看門狗
- Python實(shí)現(xiàn)用手機(jī)監(jiān)控遠(yuǎn)程控制電腦的方法
- python實(shí)現(xiàn)的web監(jiān)控系統(tǒng)
- 用Python監(jiān)控你的朋友都在瀏覽哪些網(wǎng)站?
相關(guān)文章
python使用minimax算法實(shí)現(xiàn)五子棋
這篇文章主要為大家詳細(xì)介紹了python使用minimax算法實(shí)現(xiàn)五子棋,具有一定的參考價(jià)值,感興趣的小伙伴們可以參考一下2019-07-07Python人工智能構(gòu)建簡(jiǎn)單聊天機(jī)器人示例詳解
這篇文章主要為大家介紹了Python人工智能構(gòu)建簡(jiǎn)單聊天機(jī)器人示例詳解,有需要的朋友可以借鑒參考下,希望能夠有所幫助,祝大家多多進(jìn)步,早日升職加薪2023-03-03python實(shí)現(xiàn)登錄與注冊(cè)系統(tǒng)
這篇文章主要為大家詳細(xì)介紹了python實(shí)現(xiàn)登錄與注冊(cè)系統(tǒng),文中示例代碼介紹的非常詳細(xì),具有一定的參考價(jià)值,感興趣的小伙伴們可以參考一下2020-11-11pip安裝時(shí)ReadTimeoutError的解決方法
今天小編就為大家分享一篇pip安裝時(shí)ReadTimeoutError的解決方法,具有很好的參考價(jià)值,希望對(duì)大家有所幫助。一起跟隨小編過來看看吧2018-06-06ansible作為python模塊庫(kù)使用的方法實(shí)例
ansible是一個(gè)python package,是個(gè)完全的unpack and play軟件,對(duì)客戶端唯一的要求是有ssh有python,并且裝了python-simplejson包,部署上簡(jiǎn)單到發(fā)指。下面這篇文章就給大家主要介紹了ansible作為python模塊庫(kù)使用的方法實(shí)例,需要的朋友可以參考借鑒。2017-01-01Python實(shí)現(xiàn)簡(jiǎn)易的限流器介紹
大家好,本篇文章主要講的是Python實(shí)現(xiàn)簡(jiǎn)易的限流器介紹,感興趣的同學(xué)趕快來看一看吧,對(duì)你有幫助的話記得收藏一下2022-01-01