依稀记得以前寒假回家买不到票的尴尬,各种抢票软件,要么是分享得提速包,要么是开vvvip加速。于是写段Python来监控余票,并通过邮件发送通知。
这本来是一篇干货,但是距离最初写代码的时候已经有两年了,12306网站有了很大部分改动,加强了验证,尝试许久无果,就只能写一写查票和使用python发邮件来水一篇。
先确定车票方案。
这里以5月7号合肥-安庆西的火车票为例
构造请求
- 使用Fiddler或打开浏览器的开发者工具,选择Network监控网络请求。我这里使用Fiddler.
- 点击查询按钮,抓取到数据包如下:
其中只有两条高亮的需要利用,第一条是查询结果的显示页面,第二条是包含所有查询到的火车票信息的json数据。 - 这里可能就想,直接利用第二条请求的链接获取火车票信息不就行了?在两年前我第一次做这个的时候是可以的,但是写博客复现的时候发现,已经不行了,因为访问第一个链接的时候会产生cookies,没有该cookies访问第二个链接会跳转到网络错误的页面。
- 既然这样,那就按顺序访问这两条链接
# -*- coding: UTF-8 -*-
import requests
import json
import sys
from requests.exceptions import ReadTimeout,ConnectionError,RequestException,Timeout,ConnectTimeout,HTTPError
# 禁用安全请求警告
from urllib3 import disable_warnings
from urllib3.exceptions import InsecureRequestWarning
disable_warnings(InsecureRequestWarning)
headers = {
'User-agent':'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/81.0.4044.129 Safari/537.36',
'Host':'kyfw.12306.cn',
}
# 建立session会话
session = requests.Session()
# 设置不验证SSL,HTTPS
session.verify = False
# 这是第一条链接
response = session.get('https://kyfw.12306.cn/otn/leftTicket/init?linktypeid=dc&fs=%E5%90%88%E8%82%A5,HFH&ts=%E5%AE%89%E5%BA%86%E8%A5%BF,APH&date=2020-05-07&flag=N,N,Y', headers = headers)
# 得到火车票信息的json数据
response = session.get('https://kyfw.12306.cn/otn/leftTicket/query?leftTicketDTO.train_date=2020-05-07&leftTicketDTO.from_station=HFH&leftTicketDTO.to_station=APH&purpose_codes=ADULT', headers=headers, timeout=2)
if response.status_code != 200 :
# 网络响应超时
print("网络请求错误!错误:" + str(response.status_code))
sys.exit()
print(response.text)
disable_warnings(InsecureRequestWarning)
用于禁用https带来的安全警告。headers
可以查看抓取的数据包详情,复制其headers
的内容。session = requests.Session()
建立一个会话,这样的话访问过程中产生的cookies会自己保存并且在下一次访问时带上该cookies- 200是请求成功时返回的code,我们
print
一下结果看看
对比一下从浏览器访问时fiddler抓取到的数据包
python返回的数据和fiddler抓取到的数据包格式一致,说明获取成功。
分析余票
返回的火车票信息已经拿到,接下来就是分析这个json数据,提取出有用的信息
- 有效的信息在
data
对象里的result
数组中,result
数组中每一项都表示一个车次,具体信息使用“|”分隔开。 - 添加以下代码,即可查看每一个车次的具体信息。
train_info = json.loads(response.text).get('data').get('result')
for trian in train_info:
split_item = trian.split('|')
item_dict = {}
for index, item in enumerate(split_item, 0):
print('{}:\t{}'.format(index, item))
详细信息存储在列表
split_item
中。经过对比发现,split_item
中的第11项为是否开始售票的标志,第三项为车次等等,第29项为硬座剩余票数,有则显示“有”或者具体数字,无票则显示“无”,以此类推。现在来确定一下监控哪种票吧。不过平时的票好像还挺充足,那就选择无座。
现在可以筛选判断余票了,先拿硬座试一试
输出正确,改为无座,上代码
train_info = json.loads(response.text).get('data').get('result')
train_list = []
for trian in train_info:
split_item = trian.split('|')
item_dict = {}
for index, item in enumerate(split_item, 0):
pass
# print('{}:\t{}'.format(index, item))
if split_item[11] == 'Y': # 已经开始卖票
item_dict['车次'] = split_item[3]
item_dict['出发时间'] = split_item[8]
item_dict['到站时间'] = split_item[9]
item_dict['经历时长'] = split_item[10]
item_dict['硬座'] = split_item[29]
item_dict['硬卧'] = split_item[28]
item_dict['无座'] = split_item[26]
item_dict['高级软卧'] = split_item[21]
item_dict['软卧'] = split_item[23]
item_dict['特等座'] = split_item[32]
item_dict['一等座'] = split_item[31]
item_dict['二等座'] = split_item[30]
item_dict['动卧'] = split_item[33]
# train_list.append(item_dict)
if split_item[26] != '无':
print('车次' + split_item[3] + split_item[26] + '无座')
发送邮件
这里我是用网易邮箱发送,QQ邮箱接收,需要有邮箱的授权码而非邮箱账号密码。使用以下代码来测试发送邮件。
import smtplib
from email.mime.text import MIMEText
import time
msg = MIMEText('查询到12306余票', 'plain', 'utf-8')
msg_From = '***********@163.com'
msg_To = '***********@qq.com'
smtpSever = 'smtp.163.com' # 163邮箱的smtp Sever地址
smtpPort = '25' # 端口
sqm = '**********' # 在登录smtp时需要login中的密码应当使用授权码而非账户密码
msg['from'] = msg_From
msg['to'] = msg_To
msg['subject'] = 'Python自动邮件-12306查询到余票' % time.ctime()
smtp = smtplib
smtp = smtplib.SMTP()
smtp.connect(smtpSever, smtpPort)
smtp.login(msg_From, sqm)
smtp.sendmail(msg_From, msg_To, str(msg))
smtp.quit()
综合
完整代码如下,当检测到无座有票时,发送邮件通知指定邮箱
import requests
import json
import smtplib
import time
from email.mime.text import MIMEText
from time import sleep
from urllib import parse
# 禁用安全请求警告
from urllib3 import disable_warnings
from urllib3.exceptions import InsecureRequestWarning
disable_warnings(InsecureRequestWarning)
# 请求头
headers = {
'User-agent': 'Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/60.0.3112.113 Safari/537.36',
'Host': 'kyfw.12306.cn'
}
# 建立session会话
session = requests.Session()
# 设置不验证SSL,HTTPS
session.verify = False
train_list = []
def queryticket():
global train_list
response = session.get(
'https://kyfw.12306.cn/otn/leftTicket/init?linktypeid=dc&fs=%E5%90%88%E8%82%A5,HFH&ts=%E5%AE%89%E5%BA%86%E8%A5%BF,APH&date=2020-05-07&flag=N,N,Y',
headers=headers)
# 得到火车票信息
response = session.get(
'https://kyfw.12306.cn/otn/leftTicket/query?leftTicketDTO.train_date=2020-05-07&leftTicketDTO.from_station=HFH&leftTicketDTO.to_station=APH&purpose_codes=ADULT',
headers=headers, timeout=2)
if response.status_code != 200:
# 网络响应超时
print("网络请求错误!错误:" + str(response.status_code))
sys.exit()
train_info = json.loads(response.text).get('data').get('result')
for trian in train_info:
split_item = trian.split('|')
item_dict = {}
if split_item[11] == 'Y': # 已经开始卖票
item_dict['车次'] = split_item[3]
item_dict['出发时间'] = split_item[8]
item_dict['到站时间'] = split_item[9]
item_dict['经历时长'] = split_item[10]
item_dict['硬座'] = split_item[29]
item_dict['硬卧'] = split_item[28]
item_dict['无座'] = split_item[26]
item_dict['高级软卧'] = split_item[21]
item_dict['软卧'] = split_item[23]
item_dict['特等座'] = split_item[32]
item_dict['一等座'] = split_item[31]
item_dict['二等座'] = split_item[30]
item_dict['动卧'] = split_item[33]
if split_item[26] != '无':
train_list.append(item_dict)
print('车次' + split_item[3] + split_item[26] + '无座')
if train_list:
return True
else:
return False
# ------------------------发送邮件--------------------
def sendemail():
global train_list
msg = MIMEText('查票已成功%s' % str(train_list), 'plain', 'utf-8')
msg_From = '*********@163.com'
msg_To = '**********@qq.com'
smtpSever = 'smtp.163.com' # 164邮箱的smtp Sever地址
smtpPort = '25' # 端口
sqm = '**************' # 在登录smtp时需要login中的密码应当使用授权码而非账户密码
msg['from'] = msg_From
msg['to'] = msg_To
msg['subject'] = 'Python自动邮件-12306查询到余票%s' % time.ctime()
smtp = smtplib.SMTP()
smtp.connect(smtpSever, smtpPort)
smtp.login(msg_From, sqm)
smtp.sendmail(msg_From, msg_To, str(msg))
smtp.quit()
# --------------------------------------------------
for i in range(1000000):
token = queryticket()
print(i, end=" " if i % 100 != 0 else '\n')
sleep(0.3)
if token:
print("第" + str(i) + "次查票成功!!")
sendemail()
break