引言
每逢过年就到了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 = resultMsg
if 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 = resultMsg
if 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.uamtk
response2 = 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 = resultMsg2
if resultCode2 == 0:
print('验证通过')
else:
return \"authFail\"
if 'newapptk' in dic2.keys():
Func12306.tk = dic2[\"newapptk\"]
# Func12306.cookies.pop('uamtk')
# Func12306.cookies['tk'] = Func12306.tk
url3 = '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 = resultMsg3
if 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'] = startDate
Func12306.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'] = startDate
Func12306.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('没票了')
break
elif Func12306.seatTypeList[b][trainIndex] == u\"无\" or Func12306.seatTypeList[b][trainIndex] == \"\" :
print(\"没票了\")
continue
elif Func12306.seatTypeList[b][trainIndex] == \"\":
print(\"还没开始售票\")
continue
elif Func12306.seatTypeList[b][trainIndex] != u\"有\" and len(passengersList) > int(Func12306.seatTypeList[b][trainIndex]):
print(\"票没人多\")
continue
else:
print(\"查询到有票\")
Func12306.trainIndexOfBuy = trainIndex
Func12306.seatIndexOfBuy = b
return 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 True
else:
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 True
elif 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 True
if 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 True
else:
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 True
elif 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 False
else:
return False
else:
return False
这里会有等待时间，我们获取到等待时间，然后再次发送这个请求，一直循环，直到等待时间为-1就是购票成功了 这个时候就可以开心的去12306上查询订单然后付款
结语
和大家分享一下，有兴趣的小伙伴们可以去试试哦！
一方面是检验一下自己的学习成果，另一方面也希望能帮你解决一下抢票的问题吧！
不过试试归试试，还是要做2手准备哦。不要全部寄托在软件的便利上面，要经过试验才知道能否成功。
很多细节在码代码的时候遇到，但是现在总结可能就忘了说，博客写的比较粗，没有写得那么详细。 最后，希望也在研究12306的朋友可以在这里有所收获
最后，喜欢学习python可以关注我们哦
