Python 12306验证码自动验证、用户登录和查询余票
2018-08-09 本文已影响6人
东东隆东抢
1,查票结果效果
余票查询效果图.png
2,本文使用到的库
- re
- os
- time
- datetime
- requests
- prettytable
3,获取验证码图片
使用Google浏览器并开启开发者模式(使用F12快捷键)，在地址栏里输入如下链接:
https://kyfw.12306.cn/otn/login/init
在Network—All里查看相关请求，下拉发现了验证码图片的请求，查看Headers—Request URL如下：
https://kyfw.12306.cn/passport/captcha/captcha-image?login_site=E&module=login&rand=sjrand&0.05013721011282968
链接里最后一个参数0.05013721011282968每次请求时都不一样。但我发现没有此参数同样能够请求到验证码图片，故可以直接使用如下链接请求验证码图片:
https://kyfw.12306.cn/passport/captcha/captcha-image?login_site=E&module=login&rand=sjrand
获取验证码图片
定义一个名称为Login的类并进行定义其
init方法：
class Login:
def __init__(self):
self.headers = {
"User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_12_3) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/59.0.3071.115 Safari/537.36"
}
self.session = requests.session() #创建session会话
self.session.headers = self.headers
self.session.verify = False #跳过SSL验证
self.loginUrl = 'https://kyfw.12306.cn/passport/web/login' # 登录Url
self.captchaCheckUrl = 'https://kyfw.12306.cn/passport/captcha/captcha-check' #验证码验证链接
self.captchaImgUrl = 'https://kyfw.12306.cn/passport/captcha/captcha-image?login_site=E&module=login&rand=sjrand' #获取验证码图片
self.captchaFilePath = 'captcha.jpg' #验证码图片路径
定义获取验证码图片的方法，如获取失败则使用递归再次请求：
def getCaptCha(self):
response = self.session.get(self.captchaImgUrl)
if response.status_code == 200:
print('验证码图片请求成功')
with open(self.captchaFilePath, 'wb') as f:
f.write(response.content) # 验证码图片写入当前文件夹
else:
print('验证码下载失败, 正在重试...')
self.getCaptCha() # 递归
return
在登录页面输入用户名和密码后点击验证码图片，点击登录按钮。在开发者模式下查看验证码验证请求的链接为:https://kyfw.12306.cn/passport/captcha/captcha-check
Post请求
同时我们发现使用的是Post请求，那么提交的数据Data有哪些呢？通过下图发现提交的有3个数据，分别是answer、login_site和rand。
验证码校验请求参数
其中login_site和rand参数值是固定的，而answer是我们点击图片时鼠标在图片上的坐标，坐标基准(0,0)点如下图红色箭头处：
坐标原点
将验证码图片按照如下的方式进行分割，按照位置标号为1-8，从每个图里定义一个固定的点坐标用于验证码验证：
分割图片
3.1 手动输入验证码位置
def captchaCheck(self):
self.getCaptcha() # 获取验证码图片
imgLocation = input("请输入验证码图片位置，以英文状态下的分号','分割:\n")
# 验证码坐标字典
coordinates = {'1':'35,35', '2':'105,35', '3':'175,35', '4':'245,35',
'5':'35,105', '6':'105,105', '7':'175,105','8':'245,105'}
rightImgCoordinates =[] # 创建数组存放正确的验证码坐标
for i in imgLocation.split(','):
rightImgCoordinates.append(coordinates[i])
answer = ','.join(rightImgCoordinates)
data = {
'login_site':'E', # 固定的
'rand': 'sjrand', # 固定的
'answer': answer # 验证码对应的坐标
}
result = self.session.post(self.captchaCheckUrl,data=data).json()
if result['result_code'] == '4':
print('验证码验证成功')
else:
print('出错啦：{}'.format(result['result_message']))
self.captchaCheck()
return
3.2 通过打码平台自动验证
所谓的打码平台自动验证是指用户给打码平台传入一张验证码图片，平台通过码工去人工识别验证码(码工有出错可能)，平台再将其结果返回给用户，这个过程一般也就2-3秒时间。12306验证码是多个坐标拼接成的字符串，因此我们需要平台返回多个坐标字符串。
百度搜索打码平台关键字能够找到很多相关平台，其中包含打码兔、超级鹰等。写本文的时发现打码兔平台已经转型，不再提供打码服务，于是我只能去注册超级鹰账户。平台网站上有如何使用Python进行打码的相关文档，使用时需要注意验证码图片的类型，返回多个坐标对应的codetype为9004，具体请参考验证码类型。
如下代码是超级鹰官网提供的，我做了一些改动，原因是平台返回的坐标是以图片的左上角为原点，这与12306坐标基准不一致。
import requests
from hashlib import md5
class Chaojiying_Client:
def __init__(self, username, password, soft_id):
#平台账号
self.username = username
#平台密码
self.password = md5(password.encode('utf-8')).hexdigest()
# 软件ID号，注册创建软件时会自动生成
self.soft_id = soft_id
self.base_params = {
'user' : self.username,
'pass2' : self.password,
'softid': self.soft_id,
}
self.headers = {
'Connection': 'Keep-Alive',
'User-Agent': 'Mozilla/4.0 (compatible; MSIE 8.0; Windows NT 5.1; Trident/4.0)',
}
def PostPic(self, im, codetype):
params = {
'codetype': codetype,
}
params.update(self.base_params)
files = {'userfile': ('ccc.jpg', im)}
result = requests.post('http://upload.chaojiying.net/Upload/Processing.php', data=params, files=files, headers=self.headers).json()
# 如下代码是我修改的部分，原来代码请去官网查看
answerList = result['pic_str'].replace('|',',').split(',')
for index in range(len(answerList)):
#answerList类似为 [10,135,70,200]，将纵坐标减去30并返回字符串
if index % 2 != 0:
answerList[index] = str(int(answerList[index])-30)
else:
answerList[index] = str(answerList[index])
answerStr = ','.join(answerList) #用逗号拼接成'10,105,70,170'
print(answerStr)
return answerStr # 返回验证码坐标
将获取的验证码图片提交给打码平台获取验证码坐标字符串：
def getCaptchaAnswer(self):
response= self.session.get(self.captchaImgUrl)
if response.status_code ==200:
print('验证码图片请求成功')
with open(self.captchaFilePath,'wb') as f:
f.write(response.content)
else:
print('验证码下载失败, 正在重试...')
self.getCaptchaAnswer() #递归
try:
img = open(self.captchaFilePath, 'rb').read()
answerStr = self.chaoJiYing.PostPic(img, 9004)
return answerStr
except Exception as e:
print(e)
然后根据获取的坐标字符串再进行自动验证：
def captchaCheck(self):
self.getCaptcha()
answer = self.getCaptchaAnswer()
data = {
'login_site':'E', # 固定的
'rand': 'sjrand', # 固定的
'answer': answer # 验证码对应的坐标字符串
}
result = self.session.post(self.captchaCheckUrl,data=data).json()
if result['result_code'] == '4':
print('验证码验证成功')
else:
print('出错啦：{}'.format(result['result_message']))
self.captchaCheck()
return
4,登录验证
实际我们使用浏览器进行登录时会先检查用户名和密码是否为空，不为空时才进行验证码验证(为空时直接报错提示)，验证码验证通过后再对用户名和密码进行验证。
登录同样采用的是Post请求，提交用户名、密码以及固定的appid 3个数据：
def login(self):
self.captchaCheck() # 先进行验证码验证
loginData = {
'username': Config.userName, # 用户名，我自己写了个Config文件，将用户名、密码以及其他常量放里面
'password': Config.password, # 密码
'appid': 'otn' # 固定
}
result = self.session.post(self.loginUrl,data=loginData).json()
if result['result_code'] == 0:
print('用户登录成功') # 实际并没有成功，登录过程很复杂，还有很多请求
else:
print('登录失败，请重试')
5,工具类
在写查询余票代码时发现有很多共用的方法在代码里，于是我自己将其抽出来放在一个类文件里。增加复用率，避免大量垃圾代码。
from datetime import datetime
import time
# 工具类
class Utilitys():
# 反转字典(key与value对调，生成新dict)
def reversalDict(self, dict):
return {v: k for k, v in dict.items()}
# 将时间字符串(如38:50)转化为小时和分钟的形式
def getDuration(self, timeStr):
duration = timeStr.replace(':', '时') + '分'
# 如果时间格式是00时25分，则只显示多少分钟
if duration.startswith('00'):
return duration[4:]
return duration
# 获取一个格式为2018-08-01的日期是周几
def getWeekDay(self, date):
weekDayDict = {
0: '周一',
1: '周二',
2: '周三',
3: '周四',
4: '周五',
5: '周六',
6: '周天',
}
day = datetime.strptime(date, '%Y-%m-%d').weekday()
return weekDayDict[day]
# 转化日期格式，返回 X月X日
def getDateFormat(self, date):
# date 格式为2018-08-08
dateList = date.split('-')
if dateList[1].startswith('0'):
month = dateList[1].replace('0', '')
else:
month = dateList[1]
if dateList[2].startswith('0'):
day = dateList[1].replace('0', '')
else:
day = dateList[2]
return '{}月{}日'.format(month, day)
# 检查购票日期是否合理
def checkDate(self, date):
'''
功能:检测乘车日期的正确性
:param trainDate: 乘车时间（2018-08-04）
:return: 返回时间是否为标准的形式的标志
'''
localTime = time.localtime()
localDate = '%04d-%02d-%02d' % (localTime.tm_year, localTime.tm_mon, localTime.tm_mday)
# 获得当前时间时间戳
currentTimeStamp = int(time.time())
# 预售时长的时间戳
deltaTimeStamp = '2505600'
# 截至日期时间戳
deadTimeStamp = currentTimeStamp + int(deltaTimeStamp)
# 获取预售票的截止日期时间
deadTime = time.localtime(deadTimeStamp)
deadDate = '%04d-%02d-%02d' % (deadTime.tm_year, deadTime.tm_mon, deadTime.tm_mday)
print('请注意合理的乘车日期范围是:{} 至 {}'.format(localDate, deadDate))
# 判断输入的乘车时间是否在合理乘车时间范围内
# 将购票日期转换为时间数组
trainTimeStruct = time.strptime(date, "%Y-%m-%d")
# 转换为时间戳:
trainTimeStamp = int(time.mktime(trainTimeStruct))
# 将购票时间修改为12306可接受格式 ，如用户输入2018-8-7则格式改为2018-08-07
trainTime = time.localtime(trainTimeStamp)
trainDate = '%04d-%02d-%02d' % (trainTime.tm_year, trainTime.tm_mon, trainTime.tm_mday)
# 比较购票日期时间戳与当前时间戳和预售截止日期时间戳
if currentTimeStamp <= trainTimeStamp and trainTimeStamp <= deadTimeStamp:
return True, trainDate
else:
print('出错啦: 您输入的乘车日期:{}, 当前系统日期:{}, 预售截止日期:{}'.format(trainDate, localDate, deadDate))
return False, None
# 一个日期转成2018-08-01这样的格式
def getDate(self,dateStr):
# date格式为20180801
year = time.strptime(dateStr,'%Y%m%d').tm_year
month = time.strptime(dateStr,'%Y%m%d').tm_mon
day = time.strptime(dateStr,'%Y%m%d').tm_mday
return '%04d-%02d-%02d' % (year,month,day)
6,余票查询与打印
首先明确一点，即用户在不登录情况下也是可以查车票信息的。打开浏览器进入12306余票查询页面查询链接，然后打开开发者模式，在页面上输入出发地为上海，目的地为成都，出发日期为2018-08-22，车票类型选择成人，点击查询按钮。我们发现只有如下的1个Get请求:
https://kyfw.12306.cn/otn/leftTicket/query?leftTicketDTO.train_date=2018-08-22&leftTicketDTO.from_station=SHH&leftTicketDTO.to_station=CDW&purpose_codes=ADULT
此请求包含leftTicketDTO.train_date,leftTicketDTO.from_station,leftTicketDTO.to_station及purpose_codes几个参数。从参数的英文含义上不难判断它们分别代表出发日期、出发地、目的地和车票类型。
但出发地怎么是SHH，目的地又怎么是CDW？这些都是什么鬼？我百度了一下，这些字符是指车站的电报码。可这些数据从何而来呢？
在开发者模式打开的情况下刷新查询页面，发现多了很多请求。仔细查看每个请求都在做些什么操作？服务器又返回了什么？Oh my god，竟然在刚打开查询页面的时候就请求到了。
我们把数据请求下来并加以保存，保存的原因是这些数据很难变动，请求一次，下次直接使用。
class Stations:
def __init__(self):
self.headers = {
"User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_12_3) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/59.0.3071.115 Safari/537.36"
}
self.stationCodeUrl = 'https://kyfw.12306.cn/otn/resources/js/framework/station_name.js'
self.queryUrl ='https://kyfw.12306.cn/otn/leftTicket/query'
self.queryPriceUrl ='https://kyfw.12306.cn/otn/leftTicket/queryTicketPrice'
self.session = requests.session()
self.session.verify = False
self.session.headers = self.headers
self.filePath = 'stations.txt' # 获取到的电报码存放文件
self.color = Colored()
self.utility = Utilitys()
从服务器上请求电报码，根据正则表达式找出符合条件的所有数据并加以保存:
def getStationCodes(self):
# 若文件存在，则直接推出，无需每次都执行请求
if os.path.exists(self.filePath):
return
res = self.session.get(self.stationNameUrl)
# [\u4e00-\u95fa5]是汉字的头和尾
stations = re.findall(r'([\u4e00-\u95fa5]+)\|([A-Z]+)',res.text)
# 注意编码格式utf-8
with open(self.filePath,'w',encoding='utf-8') as f:
# ensure_ascii = False 防止乱码
f.write(json.dumps(dict(stations),ensure_ascii = False))
本地电报码文件部分内容.png
检查用户输入的日期是否在预售期内，同时检查输入的出发地和目的地是否正确（能否在本地
stations.txt文件里查询到）。
def queryPrompt(self):
trainDate = input('请输入购票时间，格式为2018-01-01:\n')
# 工具类文件里定义了checkDate 方法
timeFlag,trainDate = self.utility.checkDate(trainDate)
if timeFlag == False:
print('日期格式错误或者不在预售期内，请从新输入')
self.queryPrompt()
fromStation = input('请输入出发站：\n')
toStation = input('请输入到达站：\n')
#读取文件并转化为字典
with open(self.filePath, "r", encoding='utf-8') as file:
stationsDict = json.load(file)
# 判断出发站和达到站是否存在
if fromStation not in stationsDict.keys() or toStation not in stationsDict.keys():
print('出错啦: {}或{}不在站点列表中'.format(fromStation, toStation))
# 从字典里获取站点的电报码
fromStationCode = stationsDict[fromStation]
toStationCode = stationsDict[toStation]
queryData = {
'fromStation':fromStation, # 上海
'toStation':toStation, # 成都
'trainDate':trainDate, # 2018-08-22
'fromStationCode':fromStationCode, # SHH
'toStationCode':toStationCode # CDW
}
return queryData # 返回查询数据
车票查询返回的结果
请求查询数据：
def queryTickets(self):
# 获取trainDate,fromStationCode,toStationCode,fromStation和toStation
queryData = self.queryPrompt()
# print(queryData)
params = {
'leftTicketDTO.train_date':queryData['trainDate'],
'leftTicketDTO.from_station':queryData['fromStationCode'],
'leftTicketDTO.to_station':queryData['toStationCode'],
'purpose_codes' : 'ADULT'
}
res = self.session.get(self.queryUrl,params = params)
if res.status_code ==200:
self.getTrainInfo(res.json(),queryData)
def getTrainInfo(self,result,queryData):
trainDict = {} #单个车次信息字典
trains = [] # 存放车辆信息
results = result['data']['result'] #result是查询请求返回的数据（json）
maps = result['data']['map'] #车站名称与电报码
for item in results:
trainInfo = item.split('|')
# for index, item in enumerate(trainInfo, 0):
# print('{}:\t{}'.format(index, item))
# 已经开始卖票
if trainInfo[11] =='Y':
# 备注
# trainDict['remark'] = trainInfo[2]
# 车次
trainDict['trainName'] = self.color.magenta(trainInfo[3])
# 始发站
trainDict['fromStation'] = self.color.green(maps[trainInfo[6]])
# 终点站
trainDict['toStation'] = self.color.red(maps[trainInfo[7]])
# 出发时间
trainDict['departTime'] = self.color.green(trainInfo[8])
# 到达时间
trainDict['arriveTime'] = self.color.red(trainInfo[9])
# 总用时
trainDict['totalTime'] = self.utility.getDuration(trainInfo[10])
# 商务特等座
trainDict['businessSeat'] = trainInfo[32]
# 一等座
trainDict['firstClassSeat'] = trainInfo[31]
# 二等座
trainDict['secondClassSeat'] = trainInfo[30]
# 高级软卧
trainDict['superSoftBerth'] = trainInfo[21]
# 软卧
trainDict['softBerth'] = trainInfo[23]
# 动卧
trainDict['moveBerth'] = trainInfo[33]
# 硬卧
trainDict['hardBerth'] = trainInfo[26]
# 硬座
trainDict['hardSeat'] = trainInfo[28]
# 无座
trainDict['noSeat'] = trainInfo[29]
# 其他
trainDict['otherSeat'] = trainInfo[22]
# 备注
trainDict['remark'] = trainInfo[1].replace('<br/>','')
# 如果值为空，则将值修改为'--',有票则‘有’字显示为绿色，无票‘无’字红色显示
for key in trainDict.keys():
if trainDict[key] == '':
trainDict[key] = '--'
if trainDict[key] =='有':
trainDict[key] = self.color.green('有')
if trainDict[key] =='无':
trainDict[key] = self.color.red('无')
# 价格字典，可根据‘wuzuo’等 key查询车票价格
priceDict = self.getPrice(self.queryPrice(trainInfo))
# 凭2代身份证直接进站 trainInfo[18] =1 的话
train = [trainDict['trainName']+ self.color.green('[ID]') if trainInfo[18] == '1' else trainDict['trainName'], trainDict['fromStation'] + '\n' + trainDict['toStation'],
trainDict['departTime'] + '\n' + trainDict['arriveTime'],
trainDict['totalTime'], trainDict['businessSeat']+ '\n' + priceDict['shangwuzuo'] , trainDict['firstClassSeat']+ '\n' + priceDict['yidengzuo'],
trainDict['secondClassSeat']+ '\n' + priceDict['erdengzuo'], trainDict['superSoftBerth']+ '\n' + priceDict['gaojiruanwo'], trainDict['softBerth']+ '\n' + priceDict['ruanwo'],
trainDict['moveBerth']+ '\n' + priceDict['dongwo'],
trainDict['hardBerth']+ '\n' + priceDict['yingwo'], trainDict['hardSeat']+ '\n' + priceDict['yingzuo'], trainDict['noSeat']+ '\n' + priceDict['wuzuo'], trainDict['otherSeat'],
trainDict['remark']]
# 直接使用append方法将字典添加到列表中，如果需要更改字典中的数据，那么列表中的内容也会发生改变，这是因为dict在Python里是object，不属于primitive
# type（即int、float、string、None、bool)。这意味着你一般操控的是一个指向object（对象）的指针，而非object本身。下面是改善方法：使用copy()
trains.append(train) # 注意trainDict.copy()
# 打印查询到的余票信息
self.prettyPrint(trains,queryData)
def getPrice(self, priceDict):
price ={}
# 无座票价
price['wuzuo'] = self.checkPrice('WZ', priceDict)
# 一等座票价
price['yidengzuo'] = self.checkPrice('M', priceDict)
# 二等座票价
price['erdengzuo'] = self.checkPrice('O', priceDict)
# 高级软卧票价
price['gaojiruanwo'] = self.checkPrice('A6', priceDict)
# 硬卧票价
price['yingwo'] = self.checkPrice('A3', priceDict)
# 软卧票价
price['ruanwo'] = self.checkPrice('A4', priceDict)
# 动卧票价
price['dongwo'] = self.checkPrice('F', priceDict)
# 硬座票价
price['yingzuo'] = self.checkPrice('A1', priceDict)
# 商务座票价
price['shangwuzuo'] = self.checkPrice('A9', priceDict)
return price
def checkPrice(self,key,priceDict):
if key in priceDict.keys():
return self.color.green(priceDict[key])
else:
return ''
def queryPrice(self,trainInfo):
# 注意参数顺序
parameters = {
'train_no': trainInfo[2],
'from_station_no':trainInfo[16],
'to_station_no': trainInfo[17],
'seat_types' :trainInfo[35],
'train_date' : self.utility.getDate(trainInfo[13])
}
res = self.session.get(self.queryPriceUrl,params = parameters)
time.sleep(1) #此处加个延迟
# print(res.url)
# 有时候请求一次可能出现失败的情况，然后url会自动跳转到 http://www.12306.cn/mormhweb/logFiles/error.html
if res.url != 'http://www.12306.cn/mormhweb/logFiles/error.html':
result = res.json()['data']
return result
else:
self.queryPrice(trainInfo)
return
PrettyTable 在terminal 里使用才能看到效果，在IDE里可能会出现|符号不对齐的现象。
def prettyPrint(self,trains,queryData):
header = ["车次", "车站", "时间", "历时", "商务座","一等座", "二等座",'高级软卧',"软卧", "动卧", "硬卧", "硬座", "无座",'其他','备注']
pt = PrettyTable(header)
date = queryData['trainDate']
pt.title = self.color.cyan('{}——>{}({} {}),共查询到{}个可购票的车次'.format(queryData['fromStation'],queryData['toStation'],self.utility.getDateFormat(date),self.utility.getWeekDay(date),len(trains)))
pt.align["车次"] = "l"
for train in trains:
pt.add_row(train)
print(pt)
定义一个Colored 类
from colorama import init, Fore, Back, Style
init(convert=True)
init(autoreset=False)
class Colored(object):
# 前景色:红色 背景色:默认
def red(self, s):
return Fore.RED + s + Fore.RESET
# 前景色:绿色 背景色:默认
def green(self, s):
return Fore.GREEN + s + Fore.RESET
# 前景色:黄色 背景色:默认
def yellow(self, s):
return Fore.YELLOW + s + Fore.RESET
# 前景色:蓝色 背景色:默认
def blue(self, s):
return Fore.BLUE + s + Fore.RESET
# 前景色:洋红色 背景色:默认
def magenta(self, s):
return Fore.MAGENTA + s + Fore.RESET
# 前景色:青色 背景色:默认
def cyan(self, s):
return Fore.CYAN + s + Fore.RESET
# 前景色:白色 背景色:默认
def white(self, s):
return Fore.WHITE + s + Fore.RESET
# 前景色:黑色 背景色:默认
def black(self, s):
return Fore.BLACK
# 前景色:白色 背景色:绿色
def white_green(self, s):
return Fore.WHITE + Back.GREEN + s + Fore.RESET + Back.RESET
后记
在实际项目中查询各车次不同类型座位车票价格是没有多大意义的，我们只需要查询到余票信息即可。南京到上海每天大概有280列车次，每列车次都进行一次价格请求，一共进行280次，耗时久。本想用多进程完成的，代码也写了，但在运行过程中发现请求一会儿服务器就拒绝请求了，可能是12306做了反爬的措施。
另外，登录并没有真正地成功，登录过程中有很多其他请求。后期准备把完整的登录写出来，再增加订票等内容。
