【Python】实现12306余票监控

前言

由于经常会遇到没票等现象，所以需要使用软件进行抢票，但由于一些软件有优先级问题，对于不舍得花钱的我来讲，并不是很好的体验，所以想自己尝试写一个类似的功能。

 

环境/插件

• Python3.6
• urllib（用于请求数据）
• smtplib（发送邮件）

 

文件目录

GrabTicket文件夹

• city.py（12306城市）
• GrabTicket.py（入口文件）
• GrabTicketOperation.py（主文件）
• GrabTicketSmtp.py（发送邮件文件）

 

分析

首先我们打开12306余票查询窗口

 

上图红色框的地方，就是表示列车有无车票的地方，我们需要根据这里边的数据来判断。

这里边有一些需要注意的就是，里边表示有票的有字符串“有”和数字“2”，所以我们需要对这两种情况进行判断。

接下来我们使用浏览器的开发者工具，来检查看看是否有接口可以使用：

这里我们可以看出，12306的列车查询是使用接口调用的，我们再来看看接口返回的数据：

 

我们可以清晰的看到，我们需要的列车数据是在data里边的result里边的数组，所以后面，我们只需要获取这里边的数据来判断就可以了。

 

分析数据

首先，我们得先分析数据，得出我们需要的数据字段，所以我们先写一段程序用来分析：

# 复制接口数据result里边的一条数据出来分析
results = ["null|23:00-06:00系统维护时间|6i000D312606|D3126|IOQ|NJH|IOQ|AOH|07:00|18:43|11:43|IS_TIME_NOT_BUY|qYF9CwzWBb4rPwv7Upcl6nOKai0yleG2FqmgmU4EFKXjmLhu|20180721|3|Q6|01|28|1|0|||||||有||||有|无|||O0M0O0|OMO|0"]
 
# 初始化数组键值
c = 0

# 对结果集进行循环
for i in results:
        # 将数据拆分成新的数组，并进行循环
	for n in i.split('|'):
                # 输出数组中每一个的数据n，以及下标值c
		print('[%s] %s' %( c,n ))
                # 下标值+1
		c += 1
        # 重置下标值c
	c = 0
        # 多个数据换行
	print('
	')

运行代码，我们来看看效果图：

测试多几次之后，我们可以得出我们需要的数据所在的位置，接下来我们修改下程序进行输出：

# 复制接口数据result里边的一条数据出来分析
results = ["null|23:00-06:00系统维护时间|6i000D312606|D3126|IOQ|NJH|IOQ|AOH|07:00|18:43|11:43|IS_TIME_NOT_BUY|qYF9CwzWBb4rPwv7Upcl6nOKai0yleG2FqmgmU4EFKXjmLhu|20180721|3|Q6|01|28|1|0|||||||有||||有|无|||O0M0O0|OMO|0"]
j = 1

# 初始化数组下标值
c = 0

# 初始化列车数组的下标值
index = 0

# 初始化列车数组
trains = []

# 对结果集进行循环
for i in results:

	# 为列车数组新增一个空数组元素
	trains.append([])

	# 将数据拆分成新的数组，并进行循环
	for n in i.split('|'):

		# 输出数组中每一个的数据n，以及下标值c
		print('[%s] %s' %( c,n ))

		# 将每一个数据依次放入到列车数组中
		trains[index].append(n)

		# 下标值+1
		c += 1

	# 重置下标值c
	c = 0

	# 多个数据换行
	print('
	')

	# 列车数组下标值+1
	index += 1

# 对处理好的列车数组进行循环遍历
for train in trains:
	
	# 打印我们所需要的数据
	print('火车：%s' %(train[3]))
	print('出发地：%s' %(train[6]))
	print('目的地：%s' %(train[7]))
	print('发车时间：%s' %(train[8]))
	print('到达时间：%s' %(train[9]))
	print('历时时间：%s' %(train[10]))
	print('商务座/特等座：%s' %(train[32]))
	print('一等座：%s' %(train[31]))
	print('二等座：%s' %(train[30]))
	print('高级软卧：%s' %(train[21]))
	print('软卧：%s' %(train[23]))
	print('硬卧：%s' %(train[28]))
	print('硬座：%s' %(train[29]))
	print('无座：%s' %(train[26]))
	print('
	')

运行，我们来看看结果：

这里边就是我们的结果集了，我们去12306页面对照一下：

看，我们的数据能对的上，证明我们已经分析对了数据，接下来，我们就可以实现我们的爬虫代码了：

from urllib import request
import ssl
import json

# 通过爬虫爬取数据
def getTrains():

	# 请求地址
	url = 'https://kyfw.12306.cn/otn/leftTicket/query?leftTicketDTO.train_date=2018-07-21&leftTicketDTO.from_station=SZQ&leftTicketDTO.to_station=SHH&purpose_codes=ADULT'
	# 请求头
	headers = {
		'User-Agent': r'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/59.0.3071.115 Safari/537.36'
	}
	# 设置请求
	req = request.Request(url, headers=headers)
	# 发送请求
	html = request.urlopen(req).read().decode('utf-8')

	# 格式化数据
	dict = json.loads(html) 
	# 获取想要的数据
	result= dict['data']['result']

	return result

# 复制接口数据result里边的一条数据出来分析
results = getTrains()

# 初始化数组下标值
c = 0

# 初始化列车数组的下标值
index = 0

# 初始化列车数组
trains = []

# 对结果集进行循环
for i in results:

	# 为列车数组新增一个空数组元素
	trains.append([])

	# 将数据拆分成新的数组，并进行循环
	for n in i.split('|'):

		# 将每一个数据依次放入到列车数组中
		trains[index].append(n)

		# 下标值+1
		c += 1

	# 重置下标值c
	c = 0

	# 列车数组下标值+1
	index += 1

# 对处理好的列车数组进行循环遍历
for train in trains:

	# 打印我们所需要的数据
	print('火车：%s' %(train[3]))
	print('出发地：%s' %(train[6]))
	print('目的地：%s' %(train[7]))
	print('发车时间：%s' %(train[8]))
	print('到达时间：%s' %(train[9]))
	print('历时时间：%s' %(train[10]))
	print('商务座/特等座：%s' %(train[32]))
	print('一等座：%s' %(train[31]))
	print('二等座：%s' %(train[30]))
	print('高级软卧：%s' %(train[21]))
	print('软卧：%s' %(train[23]))
	print('硬卧：%s' %(train[28]))
	print('硬座：%s' %(train[29]))
	print('无座：%s' %(train[26]))
	print('
	')

我们运行一下，看看结果：

这样，我们就能得到我们每一辆列车的数据了。

 

分析URL地址

第一步，我们已经分析出了我们的数据，现在我们开始写爬虫，再写之前，我们还需要分析一下12306列车接口URL的规律，这样才方便我们组合URL，查询不同城市、时间点的列车数据：

 

# 12306接口地址

https://kyfw.12306.cn/otn/leftTicket/query?leftTicketDTO.train_date=2018-07-21&leftTicketDTO.from_station=SZQ&leftTicketDTO.to_station=SHH&purpose_codes=ADULT

 

这里我们主要注意以下几个参数的用途：

• leftTicketDTO.train_date：出发时间
• leftTicketDTO.from_station：出发地
• leftTicketDTO.to_station：目的地

 

获取城市json文件

我们同时还注意到，我们的城市给转换成了大写的英文字母，这是12306自己的转换机制，所以我们需要试着来找一下12306是否有保存城市的json文件：

这样，我们就找到了12306的城市json文件了，我们可以将它保存下来，现在我们就准备就绪了。

 

根据城市名，获取城市代号

用户在输入了城市之后，我们需要获取到用户的城市代号，从上例的代码中，我们可以看到我们引入了一个city.py的文件，我就是将城市处理放在这个文件里边进行的：

# -*- coding: UTF-8 -*-
# @Time             : 2018/04/04 14:58
# @Author           : 小罗
# @File             : city.py
# @Software         : PyCharm
# @Python Version           : 3.6
# @About            : 12306城市文件
 
 
# 获取城市列表
def getCitys(city):
    # 城市数据（城市数据过多，这里只显示一部分，请自行去12306处获取完整的城市数据）
    favorite_names = '@bjb|北京北|VAP|beijingbei|bjb|0@bjd|北京东|BOP|beijingdong|bjd|1@bji|北京|BJP|beijing|bj|2@bjn|北京南|VNP|beijingnan|bjn|3@bjx|北京西|BXP|beijingxi|bjx|4@gzn|广州南|IZQ|guangzhounan|gzn|5@cqb|重庆北|CUW|chongqingbei|cqb|6@cqi|重庆|CQW|chongqing|cq|7@cqn|重庆南|CRW|chongqingnan|cqn|8@cqx|重庆西|CXW|chongqingxi|cqx|9@gzd|广州东|GGQ|guangzhoudong|gzd|10';
    # 遍历所有城市
    for i in favorite_names.split( '@'):
        if i:
            tmp = i.split( '|')
            if city == tmp[1]:
                return tmp[2]
 
    return False

这样，我们就可以通过调用city.getCitys()方法，传入我们输入的城市名称，就可以准确的获取我们的城市代号了。

实现用户输入

首先，我们需要让用户输入自己需要查询的出发地、目的地、出发时间，并对这些数据进行判断是否合格：

import GrabTicketOperation
import ssl
import city
import time
import re
from datetime import datetime

# 关闭ssh证书验证
ssl._create_default_https_context = ssl._create_unverified_context 

# 输入城市
def cityStation(ntype = 1):

        # 换行
	print('
	')

	# 初始化提示语
	passtext = ''
	# 判断是出发地还是目的地
	if (ntype == 1):
		passtext = '出发地'
	else:
		passtext = '目的地'
	# 开始无限循环，保证用户输对为止
	while(1):
		# 获取用户输入的数据
		city_station = input('请输入%s：' %( passtext ))
		# 检查输入的城市
		city_stations = city.getCitys(city_station)
		# 判断输入是否正确
		if (city_stations == False):
			# 不正确，提示，并且重新输入
			print('找不到 %s 这个城市' %(city_station))
		else:
			# 输入正确，跳出循环
			break
	# 返回正确的城市编号
	return city_stations

# 验证时间
def timeStation():

        # 换行
	print('
	')

	# 开始无限循环，保证用户输对为止
	while(1):
		# 获取用户输入的数据
		setOutTime = input('请输入出发时间（例：2018-04-04）：')
		# 判断时间格式是否正确
		if (checkTimeFormat(setOutTime) == False):
			print('请输入正确的时间格式，如：2018-04-04')
		else:
			# 将用户输入的日期转化为时间戳
			timeArray = time.strptime(setOutTime, "%Y-%m-%d")
		    # 转换为时间戳:
			timeStamp = int(time.mktime(timeArray))
			# 获取当前时间的时间戳
			nowtime = time.time()
			# 获取当天0点的时间戳
			nowtimeStamp = int(nowtime - nowtime % 86400 - 28800)
			# 判断时间大小
			if (timeStamp < nowtimeStamp):
				print('出发日期不能小于当前时间')
			else:
				break

	# 返回正确的城市编号
	return setOutTime

def checkTimeFormat(setOutTime):

	# 判断日期格式
	date_text = re.search(r"(d{4}-d{2}-d{2})",setOutTime)
	# 判断时间格式是否正确
	try:

		if date_text == None:
			return False

		date_text = date_text.group(0)

		if date_text != datetime.strptime(date_text, "%Y-%m-%d").strftime('%Y-%m-%d'):
			return False
		else:
			return True

	except ValueError:
		return False



# 出发地
from_station = cityStation(1)
# 目的地
to_station = cityStation(2)
# 出发时间
setOutTime = timeStation()
# 实例化类（后面需要编写这个类）
grabTicket = GrabTicketOperation.GrabTicket(from_station, to_station, setOutTime)
# 输入数据
grabTicket.callQueryTrains()

我们运行这个程序（需要将GrabTicketOperation.py，这个类的引入和使用给注释掉，后期会增加这个类），来看看结果：

 

编写操作类

接下来就是编写我们的操作类了，这也是主要的文件：

# -*- coding: UTF-8 -*-
# @Time 			: 2018/04/04 14:58
# @Author			: 小罗
# @File 			: GrabTicketOperation.py
# @Software			: PyCharm
# @Python Version	: 3.6
# @About 			: 12306抢票操作类

from splinter.browser import Browser
import urllib
from urllib import request
import ssl
import city
import json
from GrabTicketSmtp import GrabTicketSmtp

class GrabTicket:

	# 出发地
	from_station = ''

	# 目的地
	to_station = ''

	# 出发时间
	setOutTime = ''

	# 123056列车请求路径
	durl = 'https://kyfw.12306.cn/otn/leftTicket/query?'

	# 构造函数
	# @string from_station 出发地
	# @string to_station 目的地
	# @string time 时间，如：2018-04-04
	def __init__(self, from_station, to_station, setOutTime):

		# 出发城市
		self.from_station = from_station

		# 目的地城市
		self.to_station = to_station

		# 出发时间
		self.setOutTime = setOutTime

	# 拼接URL地址
	def getSplicingUrl(self):

		url = self.durl + 'leftTicketDTO.train_date=' + urllib.parse.quote(self.setOutTime) + '&leftTicketDTO.from_station=' + urllib.parse.quote(self.from_station) + '&leftTicketDTO.to_station=' + urllib.parse.quote(self.to_station) + '&purpose_codes=ADULT'

		return url

	# 抓取数据
	def curlTrainsInfo(self):
		
		# 获取链接
		url = self.getSplicingUrl()
		# 请求头
		headers = {
			# 'User-Agent': r'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/59.0.3071.115 Safari/537.36'
			'User-Agent': r'Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/55.0.2883.87 Safari/537.36'
		}
		# 设置请求
		req = request.Request(url, headers=headers)
		# 发送请求
		html = request.urlopen(req).read().decode('utf-8')

		# 格式化数据
		dicts = json.loads(html)
		# 获取想要的数据
		result = dicts['data']['result']

		return result

	# 处理数据
	def handleResultDatas(self, datas):
		j = 1
		c = 0
		index = 0
		trains = []
		for i in datas:
			trains.append([])
			for n in i.split('|'):
				#print('[%s] %s' %( c,n ))
				trains[index].append(n)
				c += 1
			c = 0
			#print('
	')
			j += 1
			index += 1

		return trains

	# 判断是否大于0
	def isCheckValueInt(self, value):

		value = int(value)

		if value > 0:
			# 内容
			self.intTrain = True
			return True

		return False

	# 将数据转化为整形
	def StringTurnInt(self, train):

		if train[32].isdigit():
			if self.isCheckValueInt(train[32]):
				return True

		if train[31].isdigit():
			if self.isCheckValueInt(train[31]):
				return True

		if train[30].isdigit():
			if self.isCheckValueInt(train[30]):
				return True

		if train[21].isdigit():
			if self.isCheckValueInt(train[21]):
				return True

		if train[23].isdigit():
			if self.isCheckValueInt(train[23]):
				return True

		if train[28].isdigit():
			if self.isCheckValueInt(train[28]):
				return True
	
		if train[29].isdigit():
			if self.isCheckValueInt(train[29]):
				return True

		if train[26].isdigit():
			if self.isCheckValueInt(train[26]):
				return True

	# 输出数据
	def outputResults(self, trains):

		content = ''

		# 初始化序号
		num = 1

		for train in trains:

			self.isIntTrain = False

			self.StringTurnInt(train)

			if train[32] == '有' or train[31] == '有' or train[30] == '有' or train[21] == '有' or train[23] == '有' or train[28] == '有' or train[29] == '有' or train[26] == '有' or self.isIntTrain == True:

				self.traincontents = []
				# 内容前缀
				traincontent_prefixs = [
					'<tr>',
					'<td>' + str(num) + '</td>',
					'<td>' + train[3] + '</td>',
					'<td>' + train[6] + '</td>',
					'<td>' + train[7] + '</td>',
					'<td>' + train[8] + '</td>',
					'<td>' + train[9] + '</td>',
					'<td>' + train[10] + '</td>'
				]
				traincontent_prefix = ''.join(traincontent_prefixs)
				# 内容后缀
				traincontent_suffix = '</tr>'

				num = num + 1

				# 商务座
				self.getIsStandbyTicket(train[32])

				# 一等座
				self.getIsStandbyTicket(train[31])

				# 二等座
				self.getIsStandbyTicket(train[30])

				# 高级软卧：
				self.getIsStandbyTicket(train[21])

				# 软卧：
				self.getIsStandbyTicket(train[23])

				# 硬卧：
				self.getIsStandbyTicket(train[28])

				# 硬座：
				self.getIsStandbyTicket(train[29])

				# 无座：
				self.getIsStandbyTicket(train[26])

				traincontent = ''.join(self.traincontents)

				content = content + traincontent_prefix + traincontent + traincontent_suffix

		if content == '':

			return False

		return content

	# 获取是否有无余票
	def getIsStandbyTicket(self, value):

		if value.isdigit():
			# 内容
			self.traincontents.append('<td style="color: #26a306;font-weight: 400;">' + value + '</td>')
		elif value == '有':
			# 内容
			self.traincontents.append('<td style="color: #26a306;font-weight: 400;">有</td>')
		else:
			self.traincontents.append('<td>无</td>')

	# 获取邮件内容 - 标题
	def getEmailContentTitle(self):

		emailTitle = '<tr><th colspan="30">12306余票监控</th></tr>'

		return emailTitle

	# 获取邮件内容 - 列表标题
	def getEmailContentListTitle(self):

		emaillistTitles = [
			'<tr>',
			'<td>序号</td>',
			'<td>列车</td>',
			'<td>出发地</td>',
			'<td>目的地</td>',
			'<td>发车时间</td>',
			'<td>到达时间</td>',
			'<td>历时时间</td>',
			'<td>商务座/特等座</td>',
			'<td>一等座</td>',
			'<td>二等座</td>',
			'<td>高级软卧</td>',
			'<td>软卧</td>',
			'<td>硬卧</td>',
			'<td>硬座</td>',
			'<td>无座</td>',
			'</tr>'
		]

		return ''.join(emaillistTitles)

	# 发送邮件
	def sentEmail(self, trains):

		# 获取标题
		emailTitle = self.getEmailContentTitle()

		# 获取列表标题
		emaillistTitle = self.getEmailContentListTitle()

		# 获取列表内容
		emailListContent = self.outputResults(trains)

		if emailListContent == False:

			return False

		# 拼接数据
		emailContents = [
			'<table>',
			'<thead>',
			emailTitle,
			emaillistTitle,
			'</thead>',
			'<tbody>',
			emailListContent,
			'</tbody>',
			'</table>'
		]

		emailContent = ''.join(emailContents)

		# 实例化邮件类
		grabTicket = GrabTicketSmtp('a710292863@qq.com', emailContent)
		# 输入数据
		grabTicket.sendEmail()

		return True


	# 调用函数
	def callQueryTrains(self):
		
		# 抓取数据
		result = self.curlTrainsInfo()

		# 处理数据
		trains = self.handleResultDatas(result)

		# 输出数据
		self.sentEmail(trains)

 

编写邮件发送类

# -*- coding: UTF-8 -*-
# @Time 			: 2018/04/04 14:58
# @Author			: 小罗
# @File 			: GrabTicketSmtp.py
# @Software			: PyCharm
# @Python Version	: 3.6
# @About 			: 发送邮件类

import smtplib  
from email.header import Header  
from email.mime.text import MIMEText

class GrabTicketSmtp:

    # SMTP服务器
    mail_host = "smtp.163.com"

    # 用户名
    mail_user = "kafeiwudeshaonian@163.com"

    # 授权密码，非登录密码
    mail_pass = ""

    # 发件人邮箱(最好写全, 不然会失败)  
    sender = "kafeiwudeshaonian@163.com"

    # 邮箱标题
    title = '好消息！列车有余票呀！'

    # 构造函数
    # @string receivers 收件人
    # @string content 邮箱内容
    def __init__(self, receivers, content):

        # 邮件内容
        self.content = content

        # 收件人
        self.receivers = [receivers]

    # 获取邮箱标题
    def getTitle(self):

        return self.title
  
    # 发送邮件
    def sendEmail(self): 

        print(self.mail_host)
        print(self.mail_user)
        print(self.mail_pass)
        print(self.sender)
        print(self.title)
        print(self.receivers)

        # 内容, 格式, 编码
        message = MIMEText(self.content, 'html', 'utf-8') 
        message['From'] = "{}".format(self.sender)  
        message['To'] = ",".join(self.receivers)  
        message['Subject'] = self.getTitle()  
      
        try: 
            # 启用SSL发信, 端口一般是465
            smtpObj = smtplib.SMTP_SSL(self.mail_host, 465)  
            # 登录验证
            smtpObj.login(self.mail_user, self.mail_pass)
            # 发送 
            smtpObj.sendmail(self.sender, self.receivers, message.as_string())
            # 发送成功
            print("发送成功")  
        except smtplib.SMTPException as e: 
            # 发送失败
            print(e)   
  
if __name__ == '__main__':  
    sendEmail()

 

完善调用类

import GrabTicketOperation
import ssl
import city
import time
import re
from datetime import datetime

# 关闭ssh证书验证
ssl._create_default_https_context = ssl._create_unverified_context 

# 输入城市
def cityStation(ntype = 1):
	# 初始化提示语
	passtext = ''
	# 判断是出发地还是目的地
	if (ntype == 1):
		passtext = '出发地'
	else:
		passtext = '目的地'
	# 开始无限循环，保证用户输对为止
	while(1):
		# 获取用户输入的数据
		city_station = input('请输入%s：' %( passtext ))
		# 检查输入的城市
		city_stations = city.getCitys(city_station)
		# 判断输入是否正确
		if (city_stations == False):
			# 不正确，提示，并且重新输入
			print('找不到 %s 这个城市' %(city_station))
		else:
			# 输入正确，跳出循环
			break
	# 返回正确的城市编号
	return city_stations

# 验证时间
def timeStation():

	# 开始无限循环，保证用户输对为止
	while(1):
		# 获取用户输入的数据
		setOutTime = input('请输入出发时间（例：2018-04-04）：')
		# 判断时间格式是否正确
		if (checkTimeFormat(setOutTime) == False):
			print('请输入正确的时间格式，如：2018-04-04')
		else:
			# 将用户输入的日期转化为时间戳
			timeArray = time.strptime(setOutTime, "%Y-%m-%d")
		    # 转换为时间戳:
			timeStamp = int(time.mktime(timeArray))
			# 获取当前时间的时间戳
			nowtime = time.time()
			# 获取当天0点的时间戳
			nowtimeStamp = int(nowtime - nowtime % 86400 - 28800)
			# 判断时间大小
			if (timeStamp < nowtimeStamp):
				print('出发日期不能小于当前时间')
			else:
				break

	# 返回正确的城市编号
	return setOutTime

def checkTimeFormat(setOutTime):
	# 判断日期格式
	date_text = re.search(r"(d{4}-d{2}-d{2})",setOutTime)
	# 判断时间格式是否正确
	try:

		if date_text == None:
			return False

		date_text = date_text.group(0)

		if date_text != datetime.strptime(date_text, "%Y-%m-%d").strftime('%Y-%m-%d'):
			return False
		else:
			return True

	except ValueError:
		return False



# 出发地
from_station = cityStation(1)
# 目的地
to_station = cityStation(2)
# 出发时间
setOutTime = timeStation()
# 实例化类
grabTicket = GrabTicketOperation.GrabTicket(from_station, to_station, setOutTime)
# 输入数据
grabTicket.callQueryTrains()

　　

执行程序

接下来我们执行调用类文件，来看看我们的结果：

这里边显示我们的邮件已经发送成功，接下来我们来看看我们的邮件：

我们再来对照一下12306的数据，看是否一致：

这样，我们就能简单的实现12306的余票监控了。

 

结语

• 此程序有许多改进的地方，大家可以根据自己的情况去完善；
• 后期还会有更多的文章供大家参考、讨论，让小编跟大家一起学习、进步；
• 此文章如需转载，请注明出处：https://www.cnblogs.com/kafeixiaoluo/p/9329500.html。