提示:该代码仅供学习使用，切勿滥用！！！
先来一个git地址:https://gitee.com/wang_li/li_wang
效果图:
1.获取Json文件的内容 2.根据信息生成URL 3.获取URL的内容，根据时间进行分析，如果有票则发送邮件，如果没有，则继续监听
内容如下:
{
"_Nodes" : "定义起始站",
"from_address" : "成都东",
"_Nodes" : "定义终点站",
"to_address" : "遂宁",
"_Nodes" : "定义车次",
"departure_date" : "2018-12-30",
"_Nodes" : "定义时间检查区间",
"start_time" : "09:00",
"_Nodes" : "定义时间结束区间",
"stop_time" : "12:00",
"_Nodes" : "定义列车类型 [ D:动车 Z:直达 T:特快 K:快速 A:全部 ]",
"type" : "A",
"_Nodes" : "定义是否发送邮件 true为发送 false为不发送",
"send_mail" : "false",
"_Nodes" : "如果上述定义为true,则下面则定义邮箱的信息包括，SMTP 和 认证信息",
"mail_smtp" : "smtp.qq.com",
"user_name" : "to_user@qq.com",
"user_password" : "password",
"mail_subject" : "仅仅是为了好玩",
"_Nodes" : "将信息发送给谁",
"mail_to_address" : "user1@qq.com;user2@163.com",
"_Nodes" : "有票时，提示的最大次数，当有间隔时，则重新计数 , 这里最大值为100",
"send_mail_max" : 3,
"_Nodes" : "刷新间隔时间,单位:秒",
"interval_time" : 30
}
代码如下:
#!/usr/bin/env python3 import requests import json import time import sys import email.mime.text import smtplib #解析json文件 def Dealwith_jsonfile(jsonfile) : return_dirct = {} #定义json标题 json_title = ['from_address' , 'to_address' , 'departure_date' , 'start_time' , 'stop_time' , 'type' , 'send_mail' , 'mail_smtp' , 'user_name' , 'user_password' , 'mail_subject', 'mail_to_address' , 'send_mail_max','interval_time'] #读取文件 open_json_file = open(jsonfile , 'rb') #初始化send_mail参数 send_mail_on = 0 #开启try except 捕获json错误的信息 try: dict_json_file = json.load(open_json_file) #查询是否开启发送邮件 if 'true' == dict_json_file['send_mail'] : send_mail_on = 1 # 0 为不开启，则不需要记录smtp/username/password/send_mail的信息 if 0 == send_mail_on : for title_configure in json_title : if 'mail_smtp' == title_configure or 'user_name' == title_configure or 'user_password' == title_configure or 'send_mail_max' == title_configure or 'mail_subject' == title_configure or 'mail_to_address' == title_configure: continue else : return_dirct[title_configure] = dict_json_file[title_configure] else : for title_configure in json_title : return_dirct[title_configure] = dict_json_file[title_configure] #关闭文件 open_json_file.close() print(return_dirct) return return_dirct except Exception as e: print("处理Json文件失败，详情如下:") print(e) if __name__ == "__main__" : Dealwith_jsonfile("go_gome.json")
执行的结果如下:
{'from_address': '成都东', 'to_address': '遂宁', 'departure_date': '2019-01-04', 'start_time': '09:30', 'stop_time': '12:00', 'type': 'D', 'send_mail': 'true', 'mail_smtp': 'smtp.qq.com', 'user_name': '2081795560@QQ.COM', 'user_password': 'idfqdnnrfjttbjbe', 'mail_subject': '仅仅是为了好玩', 'mail_to_address': '2859413527@QQ.COM', 'send_mail_max': 2, 'interval_time': 10}
网址:https://kyfw.12306.cn/otn/resources/js/framework/station_name.js?station_version=1.9085
具体内容如下:
通过分析得到，名词的后一位就是站别ID
Python获取站别ID如下:
#!/usr/bin/env python3 import requests def Get_Address_resolution(from_address,to_address , Address_type) : return_station_name = [] # 站点信息可以在12306网站上找到 station_name_url = 'https://kyfw.12306.cn/otn/resources/js/framework/station_name.js?station_version=1.9085' #获取网页信息 station_information = requests.get(station_name_url).text #以 | 分割 list_station_information = station_information.split("|") # 获取from_address 和 to_address 地址的ID if 1 == Address_type : address_com = 1 elif 2 == Address_type : address_com = -1 else : return return_station_name for address in from_address , to_address : try: if list_station_information.index(address): return_station_name.append( list_station_information[list_station_information.index(address) + address_com]) except Exception as e: print(e) return return_station_name if __name__ == "__main__" : print(Get_Address_resolution('成都东','遂宁',1)) print(Get_Address_resolution('ICW', 'NIW', 2))
执行结果如下:
['ICW', 'NIW'] ['成都东', '遂宁']
通过分析，我们可以得知，每次请求网页，实际上实在请求一个数据包，例如:
https://kyfw.12306.cn/otn/leftTicket/queryZ?leftTicketDTO.train_date=2019-01-05&leftTicketDTO.from_station=BJP&leftTicketDTO.to_station=SHH&purpose_codes=ADULT
我们只需要将数据拼接上就可以了
代码如下:
#!/usr/bin/env python3 def Generate_Url () : try: station_code = ['ICW', 'NIW'] #获取最新的地址解析时间(此信息也是固定的) return_url = "https://kyfw.12306.cn/otn/leftTicket/queryA?leftTicketDTO.train_date=%s&leftTicketDTO.from_station=%s&leftTicketDTO.to_station=%s&purpose_codes=ADULT" \ %('2019-01-03' ,station_code[0] , station_code[1] ) return return_url except Exception as e: print("生成URL失败") print(e) if __name__ == "__main__" : print(Generate_Url())
https://kyfw.12306.cn/otn/leftTicket/queryA?leftTicketDTO.train_date=2019-01-03&leftTicketDTO.from_station=ICW&leftTicketDTO.to_station=NIW&purpose_codes=ADULT
#!/usr/bin/env python3 import requests import json import time import sys import email.mime.text import smtplib #解析json文件 def Dealwith_jsonfile(jsonfile) : return_dirct = {} #定义json标题 json_title = ['from_address' , 'to_address' , 'departure_date' , 'start_time' , 'stop_time' , 'type' , 'send_mail' , 'mail_smtp' , 'user_name' , 'user_password' , 'mail_subject', 'mail_to_address' , 'send_mail_max','interval_time'] #读取文件 open_json_file = open(jsonfile , 'rb') #初始化send_mail参数 send_mail_on = 0 #开启try except 捕获json错误的信息 try: dict_json_file = json.load(open_json_file) #查询是否开启发送邮件 if 'true' == dict_json_file['send_mail'] : send_mail_on = 1 # 0 为不开启，则不需要记录smtp/username/password/send_mail的信息 if 0 == send_mail_on : for title_configure in json_title : if 'mail_smtp' == title_configure or 'user_name' == title_configure or 'user_password' == title_configure or 'send_mail_max' == title_configure or 'mail_subject' == title_configure or 'mail_to_address' == title_configure: continue else : return_dirct[title_configure] = dict_json_file[title_configure] else : for title_configure in json_title : return_dirct[title_configure] = dict_json_file[title_configure] #关闭文件 open_json_file.close() return return_dirct except Exception as e: print("处理Json文件失败，详情如下:") print(e) #抓取城市名称对应的ID def Get_Address_resolution(from_address,to_address , Address_type) : #定义返回列表 return_station_name = [] # 站点信息可以在12306网站上找到 station_name_url = 'https://kyfw.12306.cn/otn/resources/js/framework/station_name.js?station_version=1.9085' #获取网页信息 station_information = requests.get(station_name_url).text #以 | 分割 list_station_information = station_information.split("|") # 获取from_address 和 to_address 地址的ID if 1 == Address_type : address_com = 1 elif 2 == Address_type : address_com = -1 else : return return_station_name for address in from_address , to_address : try: if list_station_information.index(address): return_station_name.append( list_station_information[list_station_information.index(address) + address_com]) except Exception as e: print(e) return return_station_name # try: # if list_station_information.index(from_address) : # return_station_name.append(list_station_information[list_station_information.index(from_address) + 1]) # except Exception as e: # print(e) # # try: # if list_station_information.index(to_address): # return_station_name.append(list_station_information[list_station_information.index(to_address) + 1]) # except Exception as e: # print(e) # # #将ID返回回去 # return return_station_name # # #代码和上述一致，可以合并为到一块 # elif 2 == Address_type : # try: # if list_station_information.index(from_address) : # return_station_name.append(list_station_information[list_station_information.index(from_address) - 1]) # except Exception as e: # print(e) # # try: # if list_station_information.index(to_address): # return_station_name.append(list_station_information[list_station_information.index(to_address) - 1]) # except Exception as e: # print(e) # # #将ID返回回去 # return return_station_name #生成URL连接 def Generate_Url (operation_parameters) : try: station_code = Get_Address_resolution(operation_parameters['from_address'],operation_parameters['to_address'],1) #获取最新的地址解析时间(此信息也是固定的) return_url = "https://kyfw.12306.cn/otn/leftTicket/queryA?leftTicketDTO.train_date=%s&leftTicketDTO.from_station=%s&leftTicketDTO.to_station=%s&purpose_codes=ADULT" \ %(operation_parameters['departure_date'] ,station_code[0] , station_code[1] ) return return_url except Exception as e: print("生成URL失败") print(e) #爬取是否有票 def Crawling_Informations (url , operation_parameters) : #初始化检查失败参数 check_failed = 0 #初始化票send-mail次数 send_mail_number = {} #定义无限循环 while True: if check_failed > 10 : print("连续10次抓取失败，程序退出") break try: #获取内容并且转换为json crawling_text = requests.get(url).text crawling_json = json.loads(crawling_text) #定义空的列表用于存放票务信息 checkmessageall = [] #便利json文件的['data']['result']信息 for crawling_list in crawling_json['data']['result'] : checkmessage = [] #将票务信息按 | 分割 for split_list_json in crawling_list.split('|'): #排除长度问20字符的行 if len(split_list_json) > 20: pass else: #将其他值压进checkmessage checkmessage.append(split_list_json) #将整个checkmessage 押进checkmessageall checkmessageall.append(checkmessage) #开始处理我们爬过来的票务信息 for dealwithforlist in checkmessageall: # 计算列车类型 train_type = operation_parameters['type'] # 判断 [ D:动车 Z:直达 T:特快 K:快速 A:全部 ] 如果都不是的话，默认设置为A 全部 if train_type != 'D' and train_type != 'Z' and train_type != 'T' and train_type != 'K' and train_type != 'A' : train_type = 'A' # 开始匹配列车类型 if train_type in dealwithforlist[dealwithforlist.index('预订') + 2] or train_type == 'A' : #获取设置的开始时间和结束时间 start_time = operation_parameters['start_time'] stop_time = operation_parameters['stop_time'] # 判断开始时间是否大于结束时间 if stop_time <= start_time : print("开始时间不能超过结束时间") sys.exit(-1) # 开始匹配我们的 开始时间 和 结束时间 if (start_time <= dealwithforlist[dealwithforlist.index('预订') + 7] ) and (stop_time >= dealwithforlist[dealwithforlist.index('预订') + 7] ) : # 判断是否有票 if 'Y' == (dealwithforlist[dealwithforlist.index('预订') + 10]): print("有票") print(dealwithforlist) # 判断是否发送邮件 if 'true' == operation_parameters['send_mail']: #获取邮件发送最大次数 send_mail_max = operation_parameters['send_mail_max'] # 如果没有邮件发送计数，则重新生成 if (dealwithforlist[dealwithforlist.index('预订') + 2]) not in send_mail_number: send_mail_number[dealwithforlist[dealwithforlist.index('预订') + 2]] = 0 # 判断发送邮件 if (send_mail_number[dealwithforlist[dealwithforlist.index('预订') + 2]]) <= send_mail_max : Send_Mail_Only_You_Yan(dealwithforlist , operation_parameters) pass if send_mail_number[dealwithforlist[dealwithforlist.index('预订') + 2]] <= 1200 : send_mail_number[dealwithforlist[dealwithforlist.index('预订') + 2]] = send_mail_number[dealwithforlist[dealwithforlist.index('预订') + 2]] + 1 else: print("没有票") print(dealwithforlist) #将sned_mail_number至0 send_mail_number[dealwithforlist[dealwithforlist.index('预订') + 2]] = 0 # 获取间隔时间 interval_time = operation_parameters['interval_time'] #定义睡眠时间 time.sleep(interval_time) check_failed = 0 except Exception as e: print(e) check_failed = check_failed + 1 def Send_Mail_Only_You_Yan (dealwithforlist , operation_parameters): #获取票的信息 network_list_info_station_address = Get_Address_resolution(dealwithforlist[dealwithforlist.index('预订') + 3],dealwithforlist[dealwithforlist.index('预订') + 4],2) network_list_info_train_number = dealwithforlist[dealwithforlist.index('预订') + 2] network_list_info_start_time = dealwithforlist[dealwithforlist.index('预订') + 7] network_list_info_stop_time = dealwithforlist[dealwithforlist.index('预订') + 8] network_list_info_run_time = dealwithforlist[dealwithforlist.index('预订') + 9] network_list_info_date = dealwithforlist[dealwithforlist.index('预订') + 11] local_list_info_start_address = operation_parameters['from_address'] local_list_info_to_address = operation_parameters['to_address'] #获取邮件信息 local_list_info_mail_smtp = operation_parameters['mail_smtp'] local_list_info_mail_user = operation_parameters['user_name'] local_list_info_mail_password = operation_parameters['user_password'] local_list_info_mail_subject = operation_parameters['mail_subject'] local_list_info_mail_to_address = operation_parameters['mail_to_address'] HOST = local_list_info_mail_smtp SUBJECT = local_list_info_mail_subject TO = local_list_info_mail_to_address FROM = local_list_info_mail_user msg = email.mime.text.MIMEText( """""" %(network_list_info_train_number ,local_list_info_start_address ,local_list_info_to_address , network_list_info_station_address[0] , network_list_info_station_address[1] ,network_list_info_date , network_list_info_start_time , network_list_info_stop_time , network_list_info_run_time ), "html","utf-8") msg['Subject'] = SUBJECT msg['From'] = FROM msg['TO'] = TO server = smtplib.SMTP_SSL(HOST,'465') server.login(local_list_info_mail_user,local_list_info_mail_password) server.sendmail(FROM,TO.split(';'),msg.as_string()) server.quit if __name__ == "__main__" : # #解析json文件 # operation_parameters = Dealwith_jsonfile("go_gome.json") # # #生成URL连接 # url = Generate_Url(operation_parameters) # # #开始查询票票 # Crawling_Informations(url , operation_parameters) Crawling_Informations(Generate_Url(Dealwith_jsonfile("go_gome.json")) , Dealwith_jsonfile("go_gome.json"))
车次 需求出发站 需求到达站 车票出发站 车票到达站 时间 发车时间 到达时间 历时 %s %s %s %s %s %s %s %s %s
京公网安备 11010802041100号 | 京ICP备19059560号-4 | PHP1.CN 第一PHP社区 版权所有 