python寫一個簡單的12306搶票
阿新 • • 發佈：2019-01-05
引言
每逢過年就到了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 值得注意的是在抓包的時候發現請求裡有個隨機數，這裡get請求需要帶上這個隨機數，所以使用了random()
headers可以在初始化的時候寫好
self.headers = {
'Accept-Encoding': 'gzip, deflate, br',
'Accept-Language' 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上查詢訂單然後付款
結語
軟體是年前寫的，年後才想著寫個部落格記錄一下，一是可以和大家分享一下，二是歡迎大牛指正不足
很多細節在碼程式碼的時候遇到，但是現在總結可能就忘了說，而且部落格寫的比較粗，沒有寫得那麼詳細，有什麼問題可以評論
以上程式碼全部是後臺程式碼，前端太醜，就不拿出來了
最後，希望也在研究12306的朋友可以在這裡有所收穫