引言
每逢过年就到了12306抢票高峰期,自己总想研究一下12306购票的流程,虽然网上已经很多资料,但是总比不过自己的亲身体会,于是便琢磨着写一个抢票软件,本人比较熟悉python,所以软件是用python写的。
使用工具和库
开发环境是python3.6.2
开发工具是pycharm
辅助工具fiddler(神器)
使用到的重要库:
界面(tkinter)
http请求(requests库)
打包(pyinstaller库)
思考过程
其实本人职业并不是开发人员,任职是测试,但是喜欢平时用python写点小东西,所以开发大大们莫见笑。不废话,说说我才开始做的思考过程。
1.首先代码需要涉及前端和后台两个部分,前端我查了PyQt和Tkinter,觉得我这小东西没必要用PyQt,画个简单的前端即可,所以选择使用Tkinter
2.后台代码就是模拟12306订票流程,所以选择requests库做http请求
3.12306订票流程怎么去分解?fiddler神器帮了大忙,我就去12306官网正常登录购票,把整个流程的包全部抓到,然后分析请求数据和返回数据,后台代码就比较容易写了
4.根据后台代码的逻辑和返回,编写前端的用户提示和跳转
模拟12306购票流程
第一步登录
在你登录12306网站的时候,网页会get一个验证码图片,这个步骤封装方法如下:
def get_img(self):url = "https://kyfw.12306.cn/passport/captcha/captcha-image?login_site=E&module=login&rand=sjrand&{}".format(random.random())response = self.session.get(url=url, headers=self.headers, cookies=Func12306.cookies, verify=False)path = os.path.abspath('..')with open(path+"\\img.jpg",'wb') as f:f.write(response.content)
值得注意的是在抓包的时候发现请求里有个随机数,这里get请求需要带上这个随机数,所以使用了random()
headers可以在初始化的时候写好
self.headers = {'Accept-Encoding': 'gzip, deflate, br','Accept-Language': 'zh-CN,zh;q=0.8','X-Requested-With': 'XMLHttpRequest','Origin': 'https://kyfw.12306.cn','User-Agent': 'Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/53.0.2785.104 Safari/537.36 Core/1.53.4355.400 QQBrowser/9.7.12672.400','Content-Type': 'application/x-www-form-urlencoded; charset=UTF-8','Accept': 'application/json, text/javascript, */*; q=0.01',}
cookies会在你登录的时候自动保存在session里面
验证码图片可以保存在本地文件夹,然后给前端调用
根据fiddler抓到包的顺序来看,12306是先验证验证码,再验证帐号和密码,所以我们第一步是发送验证码信息给12306
验证码是由8个图片组成,12306服务器是校验的用户点击坐标来识别的,这里我们直接固定给出每个图片的中心坐标,简化了验证逻辑
def verify(self, clickList):url = 'https://kyfw.12306.cn/passport/captcha/captcha-check'code = ['35,35', '105,35', '175,35', '245,35', '35,105', '105,105', '175,105', '245,105']verifyList = []for a in clickList:verifyList.append(code[int(a)])codeList = ','.join(verifyList)data = {'answer': codeList,'login_site': 'E','rand': 'sjrand','_json_att':"",}response = self.session.post(url=url, data=data, headers=self.headers, cookies=Func12306.cookies,verify=False)try:dic = loads(response.content)except:return "NetWorkError"resultCode = dic['result_code']resultMsg = dic['result_message']self.verifyInfo = resultMsgif str(resultCode) == "4":return "verifySuccessful"else:return False
这是封装好的验证码验证逻辑
接下来就是要验证帐号和密码,根据fiddler抓包来看,验证一共发了三个请求,获得了一些需要后续验证在线的key,下面给出代码
def login(self, account, password):url = 'https://kyfw.12306.cn/passport/web/login'data = {'username': account,'password': password,'appid': 'otn','_json_att': "",}response = self.session.post(url=url, data=data, headers=self.headers, cookies=Func12306.cookies,verify=False)try:dic = loads(response.content)except:return "NetWorkError"resultCode = dic['result_code']resultMsg = dic['result_message']self.loginInfo = resultMsgif resultCode == 0:print('登陆成功')else:return "loginFail"if 'uamtk' in dic.keys():Func12306.uamtk = dic['uamtk']url2 = 'https://kyfw.12306.cn/passport/web/auth/uamtk'data2 = {"appid": "otn",'_json_att':""}# Func12306.cookies['uamtk'] = Func12306.uamtkresponse2 = self.session.post(url=url2, data=data2, headers=self.headers, cookies=Func12306.cookies,verify=False)try:dic2 = loads(response2.content)except:return "NetWorkError"resultCode2 = dic['result_code']resultMsg2 = dic['result_message']self.loginInfo = resultMsg2if resultCode2 == 0:print('验证通过')else:return "authFail"if 'newapptk' in dic2.keys():Func12306.tk = dic2["newapptk"]# Func12306.cookies.pop('uamtk')# Func12306.cookies['tk'] = Func12306.tkurl3 = 'https://kyfw.12306.cn/otn/uamauthclient'data3 = {"tk": Func12306.tk,'_json_att': "",}response3 = self.session.post(url=url3, data=data3, headers=self.headers, cookies=Func12306.cookies,verify=False)try:dic3 = loads(response3.content)except:return "NetWorkError"resultCode3 = dic3['result_code']resultMsg3 = dic3['result_message']self.loginInfo = resultMsg3if resultCode3 == 0:return "LoginSuccessful"else:return False
登录成功后,我们需要前端跳转到我们自己设计的抢票UI上,UI的设计比较简陋,我给个图
查询用户联系人信息
这里和12306逻辑不一样的是,我们是抢票软件,联系是提前选择好的,而12306是在购票的时候填写的,所以我们要先提前获取到联系人然后插入到我们的前端里面,下面给出联系人的获取
def get_passenger_info(self):url = 'https://kyfw.12306.cn/otn/confirmPassenger/getPassengerDTOs'data = {"_json_att": "","REPEAT_SUBMIT_TOKEN": Func12306.reSubmitTk}response = self.session.post(url=url, data=data, headers=self.headers, cookies=Func12306.cookies,verify=False)try:dic = loads(response.content)except:return "NetWorkError"if dic['messages'] != []:if dic['messages'][0] == '系统忙,请稍后重试':return 'systembusy'Func12306.passengerAllInfoList = dic['data']['normal_passengers']for a in Func12306.passengerAllInfoList:Func12306.passengerNameList.append(a['passenger_name'])Func12306.passengerIdList.append(a['passenger_id_no'])Func12306.passengerPhoneList.append(a['mobile_no'])return Func12306.passengerNameList
查询车次
联系人获取完毕,座位类型是自己研究12306后固定写上去的
这个时候下一步就是查询车次
这里给出代码
def search_ticket(self, startStation, endStation, startDate):try:Func12306.cookies['_jc_save_fromDate'] = startDateFunc12306.cookies['_jc_save_fromStation'] = (parse.quote(startStation.encode('unicode_escape').decode('latin-1') + ',' + self.stationCodeDict[startStation]).replace('\\','%')).upper().replace('%5CU', '%u')Func12306.cookies['_jc_save_toDate'] = startDateFunc12306.cookies['_jc_save_toStation'] = (parse.quote(endStation.encode('unicode_escape').decode('latin-1') + ',' + self.stationCodeDict[endStation]).replace('\\','%')).upper().replace('%5CU', '%u')Func12306.cookies['_jc_save_wfdc_flag'] = "dc"except:return "wrongtype"try:url1 = 'https://kyfw.12306.cn/otn/leftTicket/log?leftTicketDTO.train_date={}&leftTicketDTO.from_station={}&leftTicketDTO.to_station={}&purpose_codes=ADULT'.format(startDate, self.stationCodeDict[startStation], self.stationCodeDict[endStation])except:return "wrongtype"response1 = self.session.get(url=url1, headers=self.headers, cookies=Func12306.cookies, verify=False)try:dic1 = loads(response1.content)except:return "NetWorkError"if dic1['status']:print("OK")else:return "searchFail"try:url2 = 'https://kyfw.12306.cn/otn/{}?leftTicketDTO.train_date={}&leftTicketDTO.from_station={}&leftTicketDTO.to_station={}&purpose_codes=ADULT'.format(Func12306.query, startDate, self.stationCodeDict[startStation], self.stationCodeDict[endStation])except:return "wrongtype"response2 = self.session.get(url=url2, headers=self.headers, cookies=Func12306.cookies, verify=False)try:dic2 = loads(response2.content)except:return "NetWorkError"if dic2 == "":return "NetWorkError"if dic2['status'] is False:if 'c_url' in dic2.keys():Func12306.query = dic2['c_url']return "statusError"return "statusError"# elif dic2["messages"] != []:# # if dic["messages"][0] == u"选择的查询日期不在预售日期范围内":# return "search_error002"else:print("查询车成功")Func12306.trainInfoStartTimeList, Func12306.trainInfoEndTimeList, Func12306.trainInfoSecretStrList, Func12306.trainInfoNameList, Func12306.trainInfoLocationList, Func12306.trainInfoNoList = [], [], [], [], [], []Func12306.dw, Func12306.swz, Func12306.ydz, Func12306.edz, Func12306.yz, Func12306.yw, Func12306.wz, Func12306.rw, Func12306.gjrw, Func12306.tdz, Func12306.rz = [], [], [], [], [], [], [], [], [], [], []Func12306.seatTypeList = (Func12306.edz, Func12306.ydz, Func12306.yz, Func12306.rz, Func12306.yw, Func12306.rw, Func12306.dw, Func12306.wz, Func12306.swz, Func12306.tdz, Func12306.gjrw)for a in dic2['data']['result']:Func12306.trainInfoSecretStrList.append(a.split("|")[0])Func12306.trainInfoNoList.append(a.split("|")[2])Func12306.trainInfoNameList.append(a.split("|")[3])Func12306.trainInfoStartTimeList.append(a.split("|")[8])Func12306.trainInfoEndTimeList.append(a.split("|")[9])Func12306.trainInfoLocationList.append(a.split("|")[15])Func12306.dw.append(a.split("|")[33])Func12306.swz.append(a.split("|")[32])Func12306.ydz.append(a.split("|")[31])Func12306.edz.append(a.split("|")[30])Func12306.yz.append(a.split("|")[29])Func12306.yw.append(a.split("|")[28])Func12306.wz.append(a.split("|")[26])Func12306.tdz.append(a.split("|")[25])Func12306.rz.append(a.split("|")[24])Func12306.rw.append(a.split("|")[23])Func12306.gjrw.append(a.split("|")[21])Func12306.seatTypeList = (Func12306.edz, Func12306.ydz, Func12306.yz, Func12306.rz, Func12306.yw, Func12306.rw, Func12306.dw,Func12306.wz, Func12306.swz, Func12306.tdz, Func12306.gjrw)return Func12306.trainInfoNameList
这里值得注意的是查询出来的结果是用“|分隔的很多信息,需要自己研究每个位置是什么信息,可以对照12306页面研究,然后把获取的信息返回给前端调用
还有重要一点要注意就是12306的url不是固定的,它会带一个随机的大字字母在url里,我们可以先随便写一个,然后从返回值里获取到这个大写字母
抢票逻辑
这个时候我们就可以让用户选择车次和联系人以及座位类型,然后就可以进入抢票逻辑
抢票需要发送很多请求
首先我们要知道我们要买的票到底有还是没有
def check_ticket(self, startStation, endStation, startDate, seatType, passengersList, trainName):searchResult = self.search_ticket(startStation, endStation, startDate)# print(searchResult)# print(Func12306.edz)if searchResult == "wrongtype":return "wrongtype"if searchResult == "NetWorkError":return "NetWorkError"if searchResult == "searchFail":return "searchFail"if searchResult == "statusError":return "statusError"if searchResult == "search_error002":return "search_error002"for a in trainName:try:trainIndex = Func12306.trainInfoNameList.index(a)except:return "listNeedRefresh"for b in seatType:# print(trainIndex)# print(b)# print(Func12306.seatTypeList)# print(Func12306.seatTypeList[b])if Func12306.trainInfoSecretStrList[trainIndex] == 'null':print('没票了')breakelif Func12306.seatTypeList[b][trainIndex] == u"无" or Func12306.seatTypeList[b][trainIndex] == "" :print("没票了")continueelif Func12306.seatTypeList[b][trainIndex] == "*":print("还没开始售票")continueelif Func12306.seatTypeList[b][trainIndex] != u"有" and len(passengersList) > int(Func12306.seatTypeList[b][trainIndex]):print("票没人多")continueelse:print("查询到有票")Func12306.trainIndexOfBuy = trainIndexFunc12306.seatIndexOfBuy = breturn Func12306.seatTypeList[b][trainIndex]
这里面还涉及多人购买的时候,票不够人多的情况,我这边处理是没足够的票,大家就都不买,要买一起买
如果票足够了,我们就开始进入购票环节
首先需要验证用户信息
def check_user(self):url = 'https://kyfw.12306.cn/otn/login/checkUser'data = {"_json_att": ""}# self.headers["Cache-Control"] = "no-cache"# self.headers["If-Modified-Since"] = "0"response = self.session.post(url=url, data=data, headers=self.headers, cookies=Func12306.cookies,verify=False)try:dic = loads(response.content)except:return "NetWorkError"if dic['data']['flag']:print("用户在线验证成功")return Trueelse:print('检查到用户不在线,请重新登陆')return False
验证完之后需要开始提交订单
def submit_order(self, startStation, endStation, startDate):url = 'https://kyfw.12306.cn/otn/leftTicket/submitOrderRequest'data = {"secretStr": parse.unquote(Func12306.trainInfoSecretStrList[Func12306.trainIndexOfBuy]),"train_date": startDate,"back_train_date": startDate,"tour_flag": "dc","purpose_codes": "ADULT","query_from_station_name": startStation,"query_to_station_name": endStation,"undefined": ""}response = self.session.post(url=url, data=data, headers=self.headers,cookies=Func12306.cookies, verify=False)try:dic = loads(response.content)except:return "NetWorkError"if dic['status']:print('提交订单成功')return Trueelif dic['messages'] != []:if dic['messages'][0] == "车票信息已过期,请重新查询最新车票信息":print('车票信息已过期,请重新查询最新车票信息')return "ticketInfoOutData"else:print("提交失败")return False
提交完然后开始确认联系人信息
def confirm_passenger(self):url = 'https://kyfw.12306.cn/otn/confirmPassenger/initDc'data = {"_json_att": ''}response = self.session.post(url=url, data=data, headers=self.headers, cookies=Func12306.cookies,verify=False)try:Func12306.reSubmitTk = re.findall(u'globalRepeatSubmitToken = \'(\S+?)\'',response.text)[0]Func12306.keyIsChange = re.findall(u'key_check_isChange\':\'(\S+?)\'',response.text)[0]Func12306.leftTicketStr = re.findall(u'leftTicketStr\':\'(\S+?)\'',response.text)[0]except:print("获取KEY失败")return 'NetWorkError'def get_passenger_info(self):url = 'https://kyfw.12306.cn/otn/confirmPassenger/getPassengerDTOs'data = {"_json_att": "","REPEAT_SUBMIT_TOKEN": Func12306.reSubmitTk}response = self.session.post(url=url, data=data, headers=self.headers, cookies=Func12306.cookies,verify=False)try:dic = loads(response.content)except:return "NetWorkError"if dic['messages'] != []:if dic['messages'][0] == '系统忙,请稍后重试':return 'systembusy'Func12306.passengerAllInfoList = dic['data']['normal_passengers']for a in Func12306.passengerAllInfoList:Func12306.passengerNameList.append(a['passenger_name'])Func12306.passengerIdList.append(a['passenger_id_no'])Func12306.passengerPhoneList.append(a['mobile_no'])return Func12306.passengerNameList
确认好联系人之后,需要开始确认订单
def check_order(self, passengersList):url = 'https://kyfw.12306.cn/otn/confirmPassenger/checkOrderInfo'passengerTicketStr = ""oldPassengerStr = ""for a in passengersList:passengerTicketStr += Func12306.seatCodeList[Func12306.seatIndexOfBuy] + ",0,1,{},1,{},{},N_".format(Func12306.passengerNameList[a], Func12306.passengerIdList[a], Func12306.passengerPhoneList[a])oldPassengerStr += "{},1,{},1_".format(Func12306.passengerNameList[a], Func12306.passengerIdList[a])data = {"cancel_flag": "2","bed_level_order_num": "000000000000000000000000000000","passengerTicketStr": passengerTicketStr,"oldPassengerStr": oldPassengerStr,"tour_flag": "dc","randCode": "","whatsSelect": "1","_json_att": "","REPEAT_SUBMIT_TOKEN": Func12306.reSubmitTk}response = self.session.post(url=url, data=data, headers=self.headers,cookies=Func12306.cookies, verify=False)try:dic = loads(response.content)except:return "NetWorkError"if dic['data']['submitStatus'] is True:if dic['data']['ifShowPassCode'] == 'N':return Trueif dic['data']['ifShowPassCode'] == 'Y':return "Need Random Code"else:print("checkOrderFail")return False
这里有几点需要注意:
1.在这个过程之前,12306会get一张新验证码图片,在购票紧张的时候会在购票时候弹出给你填,如果购票不紧张就不会有但是我们要get到这张图
2.判断要不要填这个验证的key在上面代码里’ifShowPassCode’ == ‘Y’就是要填,我们要做判断
这里给出新验证码的获取代码
def get_buy_image(self):url = 'https://kyfw.12306.cn/otn/passcodeNew/getPassCodeNew?module=passenger&rand=randp&{}'.format(random.random())response = self.session.get(url=url, headers=self.headers, cookies=Func12306.cookies, verify=False)path = os.path.abspath('..')with open(path + "\\img.jpg", 'wb') as f:f.write(response.content)
确认订单成功之后,我们就要开始进入购票队列
def get_queue_count(self, startStation, endStation, startDate, seatType):url = 'https://kyfw.12306.cn/otn/confirmPassenger/getQueueCount'thatdaydata = datetime.datetime.strptime(startDate, "%Y-%m-%d")train_date = "{} {} {} {} 00:00:00 GMT+0800 (中国标准时间)".format(thatdaydata.strftime('%a'),thatdaydata.strftime('%b'), startDate.split('-')[2],startDate.split('-')[0])data = {"train_date": train_date,"train_no": Func12306.trainInfoNoList[Func12306.trainIndexOfBuy],"stationTrainCode": Func12306.trainInfoNameList[Func12306.trainIndexOfBuy],"seatType": Func12306.seatCodeList[Func12306.seatIndexOfBuy],"fromStationTelecode": self.stationCodeDict[startStation],"toStationTelecode": self.stationCodeDict[endStation],"leftTicket": Func12306.leftTicketStr,"purpose_codes": "00","train_location": Func12306.trainInfoLocationList[Func12306.trainIndexOfBuy],"_json_att": "","REPEAT_SUBMIT_TOKEN": Func12306.reSubmitTk}response = self.session.post(url=url, data=data, headers=self.headers, cookies=Func12306.cookies, verify=False)try:dic = loads(response.content)except:return "NetWorkError"if dic['status']:print("进入队列成功")return Trueelse:print("进入队列失败")return False
然后确认单人队列
def confirm_single_for_queue(self, seatType, passengersList, clickList = None):url = 'https://kyfw.12306.cn/otn/confirmPassenger/confirmSingleForQueue'passengerTicketStr = ""oldPassengerStr = ""for a in passengersList:passengerTicketStr += Func12306.seatCodeList[Func12306.seatIndexOfBuy] + ",0,1,{},1,{},{},N_".format(Func12306.passengerNameList[a], Func12306.passengerIdList[a], Func12306.passengerPhoneList[a])oldPassengerStr += "{},1,{},1_".format(Func12306.passengerNameList[a], Func12306.passengerIdList[a])if clickList is not None:code = ['35,35', '105,35', '175,35', '245,35', '35,105', '105,105', '175,105', '245,105']verifyList = []for a in clickList:verifyList.append(code[int(a)])codeList = ','.join(verifyList)print(codeList)else:codeList = ''data = {"passengerTicketStr": passengerTicketStr,"oldPassengerStr": oldPassengerStr,"randCode": codeList,"purpose_codes": "00","key_check_isChange": Func12306.keyIsChange,"leftTicketStr": Func12306.leftTicketStr,"train_location": Func12306.trainInfoLocationList[Func12306.trainIndexOfBuy],"choose_seats": "","seatDetailType": "000","whatsSelect": "1","roomType": "00","dwAll": "N","_json_att": "","REPEAT_SUBMIT_TOKEN": Func12306.reSubmitTk}response = self.session.post(url=url, data=data, headers=self.headers, cookies=Func12306.cookies, verify=False)try:dic = loads(response.content)except:return "NetWorkError"if 'data' in dic.keys():if dic['data']['submitStatus'] is True:print("提交订单成功")return Trueelif dic['data']['errMsg'] == u"验证码输入错误!":return "wrongCode"else:print("提交订单失败")return False
如果以上都成功了,就会进入一个等待订单生成了过程
def wait_time(self):url = 'https://kyfw.12306.cn/otn/confirmPassenger/queryOrderWaitTime?random={}&tourFlag=dc&_json_att=&REPEAT_SUBMIT_TOKEN={}'.format(round(time.time()*1000),Func12306.reSubmitTk)response = self.session.get(url=url, headers=self.headers, cookies=Func12306.cookies,verify=False)try:dic = loads(response.content)except:return "NetWorkError"if dic['status']:if dic['data']['queryOrderWaitTimeStatus']:if dic['data']['waitTime'] > 0 :return dic['data']['waitTime']elif dic['data']['waitTime'] == -1:Func12306.orderId = ''Func12306.orderId = dic['data']['orderId']return dic['data']['waitTime']else:return Falseelse:return Falseelse:return False
这里会有等待时间,我们获取到等待时间,然后再次发送这个请求,一直循环,直到等待时间为-1就是购票成功了
这个时候就可以开心的去12306上查询订单然后付款
结语
软件是年前写的,年后才想着写个博客记录一下,一是可以和大家分享一下,二是欢迎大牛指正不足
很多细节在码代码的时候遇到,但是现在总结可能就忘了说,而且博客写的比较粗,没有写得那么详细,有什么问题可以评论
以上代码全部是后台代码,前端太丑,就不拿出来了
最后,希望也在研究12306的朋友可以在这里有所收获