47.测试运行以及完整代码
2020-08-30 本文已影响0人
M_小七
测试运行以及完整代码
学习目标
- 了解 购票的整个逻辑
6.1 测试运行
执行12306.funk12306.py，购票成功最后显示的结果
输入要购票的乘车人的下标0
请输入要购买的坐席类型的拼音，如果输入错误，将强行购买无座，能回家就行了，还要tm什么自行车！：yingwo
{"validateMessagesShowId":"_validatorMessage","status":true,"httpstatus":200,"data":{"ifShowPassCode":"N","canChooseBeds":"N","canChooseSeats":"N","choose_Seats":"MOP9","isCanChooseMid":"N","ifShowPassCodeTime":"1","submitStatus":true,"smokeStr":""},"messages":[],"validateMessages":{}}
{"validateMessagesShowId":"_validatorMessage","status":true,"httpstatus":200,"data":{"count":"0","ticket":"16","op_2":"false","countT":"0","op_1":"false"},"messages":[],"validateMessages":{}}
此时排队买票的人数为：0
此时该车次的余票数量为：16
{'validateMessagesShowId': '_validatorMessage', 'status': True, 'httpstatus': 200, 'data': {'submitStatus': True}, 'messages': [], 'validateMessages': {}}
{"validateMessagesShowId":"_validatorMessage","status":true,"httpstatus":200,"data":{"queryOrderWaitTimeStatus":true,"count":0,"waitTime":4,"requestId":6478993261703780471,"waitCount":1,"tourFlag":"dc","orderId":null},"messages":[],"validateMessages":{}}
{"validateMessagesShowId":"_validatorMessage","status":true,"httpstatus":200,"data":{"errMsg":"网络传输过程中数据丢失，请查看未完成订单，继续支付！","submitStatus":false},"messages":[],"validateMessages":{}}
Process finished with exit code 0
6.2 项目地图
6.2.1-项目代码地图.png
6.3 项目文件结构
6.3.1-项目文件结构.png
6.3.1 12306.funk12306.py
import re
import os
import time
import json
import base64
import requests
from pprint import pprint
from utils.captcha import getCode
from utils.parse_date import parseDate
from utils.stations_dict import stations_dict
from utils.parse_passenger import parsePassenger
from utils.parse_seat_type import seat_type_dict
from utils.parse_trains_infos import parseTrainsInfos
redis_timeout = 180 # redis中cookies_dict的过期时间 单位秒
class Funk12306():
def __init__(self, username, password):
self.username = username
self.password = password
self.s = requests.session()
self.s.headers['User-Agent'] = 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/67.0.3396.87 Safari/537.36'
def get_cookies(self):
print('获取登录前的cookies')
"""以上都为获取cookies"""
url = 'https://www.12306.cn/index/'
self.s.get(url)
url = 'https://kyfw.12306.cn/otn/login/conf'
self.s.headers['Referer'] = 'https://kyfw.12306.cn/otn/resources/login.html'
self.s.post(url)
url = 'https://kyfw.12306.cn/otn/index12306/getLoginBanner'
resp = self.s.get(url)
print(resp.text)
# 检查用户是否登录
url = 'https://kyfw.12306.cn/passport/web/auth/uamtk-static'
resp = self.s.post(url, data={'appid': 'otn'})
print(resp.text)
# 获取登录二维码
url = 'https://kyfw.12306.cn/passport/web/create-qr64'
resp = self.s.post(url, data={'appid': 'otn'})
print(resp.text)
uuid = json.loads(resp.text)['uuid']
# 查询二维码状态
url = 'https://kyfw.12306.cn/passport/web/checkqr'
resp = self.s.post(url, data={'appid': 'otn', 'uuid': uuid})
print(resp.text)
"""获取图片验证码并手动输入或打码，发送验证请求"""
# 获取验证码图片
temp = str(time.time() * 1000)[:-5] # 15421 76058 853
url = 'https://kyfw.12306.cn/passport/captcha/captcha-image64?login_site=E&module=login&rand=sjrand&_={}'.format(temp)
resp = self.s.get(url)
print(resp.text)
captcha_img_b64 = json.loads(resp.text)['image']
captcha_img = base64.b64decode(captcha_img_b64)
with open('./imgs/{}.png'.format(temp), 'wb') as f:
f.write(captcha_img)
# 选择打码方式并获取结果
create_type = input(' 1 为手动打码 or 2 为收费平台自动打码，请选择：')
if create_type == '1': # 手动打码
answer_num = input('输入图片编号，从1开始：')
elif create_type == '2': # 接入打码平台
answer_num = getCode('./imgs/{}.png'.format(temp)) # 23
else:
raise Exception('输入错误！ 1 为手动打码 or 2 为收费平台自动打码')
# 改图片名字并获取点击坐标
answer_dict = { '1': '37,46,',
'2': '110,46,',
'3': '181,46,',
'4': '253,46,',
'5': '37,116,',
'6': '110,116,',
'7': '181,116,',
'8': '253,116,' }
os.rename('./imgs/{}.png'.format(temp),
'./imgs/{}_{}.png'.format(temp, answer_num))
answer = ''
for i in answer_num:
answer += answer_dict[i]
answer = answer[:-1]
# 验证图片验证码
url = 'https://kyfw.12306.cn/passport/captcha/captcha-check?answer={}&rand=sjrand&login_site=E&_={}'.format(
answer, str(time.time() * 1000)[:-4])
resp = self.s.get(url)
print(json.loads(resp.text))
"""发送登陆请求，以及后续登陆验证"""
# 登录
url = 'https://kyfw.12306.cn/passport/web/login'
data = {'username': self.username,
'password': self.password,
'appid': 'otn',
'answer': answer}
resp = self.s.post(url, data=data)
print(resp.text)
uamtk = json.loads(resp.text)['uamtk']
# 进入登录后的页面 发生302跳转
url = 'https://kyfw.12306.cn/otn/login/userLogin' # 302跳转
self.s.get(url)
url = 'https://kyfw.12306.cn/otn/passport?redirect=/otn/login/userLogin'
self.s.get(url)
# 获取newapptk
url = 'https://kyfw.12306.cn/passport/web/auth/uamtk'
self.s.headers['Referer'] = 'https://kyfw.12306.cn/otn/passport?redirect=/otn/login/userLogin'
resp = self.s.post(url, data={'appid': 'otn'})
print(json.loads(resp.text))
newapptk = json.loads(resp.text)['newapptk']
# 登录后验证
url = 'https://kyfw.12306.cn/otn/uamauthclient'
resp = self.s.post(url, data={'tk': newapptk})
print(json.loads(resp.text))
def buy_ticket(self):
self.s.headers.pop('Referer') # 清除referer
# 获取城市(车站)编码
from_station = input('输入出发城市或车站:')
to_station = input('输入到达城市或车站:')
train_date = input('输入出行日期,格式为2018-12-03:')
from_station_code = stations_dict.get(from_station, '')
to_station_code = stations_dict.get(to_station, '')
# 查询车量信息日志get
url = 'https://kyfw.12306.cn/otn/leftTicket/log?leftTicketDTO.train_date=%s&leftTicketDTO.from_station=%s&leftTicketDTO.to_station=%s&purpose_codes=ADULT' % (train_date, from_station_code, to_station_code)
resp = self.s.get(url)
print(resp.text)
# 查询车量具体信息query
url = 'https://kyfw.12306.cn/otn/leftTicket/query?leftTicketDTO.train_date=%s&leftTicketDTO.from_station=%s&leftTicketDTO.to_station=%s&purpose_codes=ADULT' % (train_date, from_station_code, to_station_code)
response = self.s.get(url)
# 解析获取trains_list
trains_list = parseTrainsInfos(json.loads(response.content)['data']['result'])
print('查询的列车信息如下：')
pprint(trains_list)
# 获取选择的列车
train_info_dict = trains_list[int(input('请输入选中车次的下标：'))]
print('选中了列车信息为：')
pprint(train_info_dict)
# 列车信息
secretStr = train_info_dict['secretStr']
leftTicket = train_info_dict['leftTicket']
train_location = train_info_dict['train_location']
# 检查用户是否保持登录成功
url = 'https://kyfw.12306.cn/otn/login/checkUser'
data = {'_json_att': ''}
resp = self.s.post(url, data=data)
print(json.loads(resp.text))
# 点击预定
url = 'https://kyfw.12306.cn/otn/leftTicket/submitOrderRequest'
data = {
'secretStr': secretStr,
'train_date': train_date,
'back_train_date': train_date,
'tour_flag': 'dc', # dc 单程 wf 往返
'purpose_codes': 'ADULT', # 成人
'query_from_station_name': from_station,
'query_to_station_name': to_station,
'undefined': ''
}
resp = self.s.post(url, data=data)
print(resp.text)
# 订单初始化 获取REPEAT_SUBMIT_TOKEN key_check_isChange
url = 'https://kyfw.12306.cn/otn/confirmPassenger/initDc'
data = {'_json_att': ''}
response = self.s.post(url, data=data)
repeat_submit_token = re.search(r"var globalRepeatSubmitToken = '([a-z0-9]+)';",
response.content.decode()).group(1)
key_check_isChange = re.search("'key_check_isChange':'([A-Z0-9]+)'", response.content.decode()).group(1)
# 获取用户信息
# 需要 REPEAT_SUBMIT_TOKEN
url = 'https://kyfw.12306.cn/otn/confirmPassenger/getPassengerDTOs'
data = {'_json_att': '',
'REPEAT_SUBMIT_TOKEN': repeat_submit_token}
response = self.s.post(url, data=data)
# 解析并构造乘客信息列表
passenger_list = parsePassenger(json.loads(response.content))
print('获取乘客信息有：')
pprint(passenger_list)
passenger_info_dict = passenger_list[int(input('输入要购票的乘车人的下标'))]
# 坐席类型
try:
seat_type = seat_type_dict[input('请输入要购买的坐席类型的拼音，如果输入错误，将强行购买无座，能回家就行了，还要tm什么自行车！：')]
except:
seat_type = seat_type_dict['wuzuo']
# 构造乘客信息
passengerTicketStr = '%s,0,1,%s,%s,%s,%s,N' % (
seat_type, passenger_info_dict['passenger_name'],
passenger_info_dict['passenger_id_type_code'],
passenger_info_dict['passenger_id_no'],
passenger_info_dict['passenger_mobile_no'])
oldPassengerStr = '%s,%s,%s,1_' % (
passenger_info_dict['passenger_name'],
passenger_info_dict['passenger_id_type_code'],
passenger_info_dict['passenger_id_no'])
# 检查选票人信息
url = 'https://kyfw.12306.cn/otn/confirmPassenger/checkOrderInfo'
data = {
'cancel_flag': '2', # 未知
'bed_level_order_num': '000000000000000000000000000000', # 未知
'passengerTicketStr': passengerTicketStr.encode('utf-8'), # O,0,1,靳文强,1,142303199512240614,18335456020,N
'oldPassengerStr': oldPassengerStr.encode('utf-8'), # 靳文强,1,142303199512240614,1_
'tour_flag': 'dc', # 单程
'randCode': '',
'whatsSelect': '1',
'_json_att': '',
'REPEAT_SUBMIT_TOKEN': repeat_submit_token
}
resp = self.s.post(url, data=data)
print(resp.text)
# 提交订单,并获取排队人数,和车票的真实余数
url = 'https://kyfw.12306.cn/otn/confirmPassenger/getQueueCount'
data = {
'train_date': parseDate(train_date), # Fri Nov 24 2017 00:00:00 GMT+0800 (中国标准时间)
'train_no': train_info_dict['train_no'], # 6c0000G31205
'stationTrainCode': train_info_dict['stationTrainCode'], # G312
'seatType': seat_type, # 席别
'fromStationTelecode': train_info_dict['from_station'], # one_train[6]
'toStationTelecode': train_info_dict['to_station'], # ? one_train[7]
'leftTicket': train_info_dict['leftTicket'], # one_train[12]
'purpose_codes': '00',
'train_location': train_info_dict['train_location'], # one_train[15]
'_json_att': '',
'REPEAT_SUBMIT_TOKEN': repeat_submit_token
}
resp = self.s.post(url, data=data)
print(resp.text)
print('此时排队买票的人数为：{}'.format(json.loads(resp.text)['data']['count']))
ticket = json.loads(resp.text)['data']['ticket']
print('此时该车次的余票数量为：{}'.format(ticket))
if ticket == '0':
print('没有余票，购票失败')
return '没有余票，购票失败'
# 确认订单,进行扣票 需要 key_check_isChange
url = 'https://kyfw.12306.cn/otn/confirmPassenger/confirmSingleForQueue'
data = {
'passengerTicketStr': passengerTicketStr.encode('utf-8'),
'oldPassengerStr': oldPassengerStr.encode('utf-8'),
'randCode': '',
'purpose_codes': '00',
'key_check_isChange': key_check_isChange,
'leftTicketStr': leftTicket,
'train_location': train_location, # one_train[15]
'choose_seats': '', # 选择坐席 ABCDEF 上中下铺 默认为空不选
'seatDetailType': '000',
'whatsSelect': '1',
'roomType': '00',
'dwAll': 'N', # ?
'_json_att': '',
'REPEAT_SUBMIT_TOKEN': repeat_submit_token
}
resp = self.s.post(url, data=data)
print(json.loads(resp.text))
if json.loads(resp.text)['status'] == False or json.loads(resp.text)['data']['submitStatus'] == False:
print('扣票失败')
return '扣票失败'
# 排队等待 返回waittime 获取 requestID 和 orderID
timestamp = str(int(time.time() * 1000)) # str(time.time() * 1000)[:-4]
url = 'https://kyfw.12306.cn/otn/confirmPassenger/queryOrderWaitTime?random=%s&tourFlag=dc&_json_att=&REPEAT_SUBMIT_TOKEN=%s' % (
timestamp, repeat_submit_token)
resp = self.s.get(url)
print(resp.text)
try:
orderID = json.loads(resp.text)['data']['orderId']
except:
# 排队等待 返回waittime 获取 requestID 和 orderID
timestamp = str(int(time.time() * 1000)) # str(time.time() * 1000)[:-4]
url = 'https://kyfw.12306.cn/otn/confirmPassenger/queryOrderWaitTime?random=%s&tourFlag=dc&_json_att=&REPEAT_SUBMIT_TOKEN=%s' % (
timestamp, repeat_submit_token)
resp = self.s.get(url)
print(resp.text)
try:
orderID = json.loads(resp.text)['data']['orderId']
except:
return '购票失败'
# 订单结果
url = 'https://kyfw.12306.cn/otn/confirmPassenger/resultOrderForDcQueue'
data = {
'orderSequence_no': orderID,
'_json_att': '',
'REPEAT_SUBMIT_TOKEN': repeat_submit_token
}
resp = self.s.post(url, data=data)
print(resp.text)
def run(self):
# 登录 获取cookies
self.get_cookies()
# 买票
self.buy_ticket()
if __name__ == '__main__':
username = input('请输入12306账号：')
password = input('请输入12306密码：')
funk = Funk12306(username, password)
funk.run()
6.3.2 12306.get_stations_dict.py
import re
import json
import requests
# 获取车站编号字符串 station_version=1.9076
url = 'https://kyfw.12306.cn/otn/resources/js/framework/station_name.js?station_version=1.9076'
resp = requests.get(url)
stations_str = re.search("'(.*)'", resp.text).group(1)
# 获取{城市(车站):编码, ...} 键值对
stations_dict = {}
for station in stations_str.split('@'):
if station == '': # 按@切会切出空字符串
continue
stations_dict[station.split('|')[1]] = station.split('|')[2]
with open('./utils/stations_dict.py', 'w', encoding='utf8') as f:
f.write('stations_dict = ')
json.dump(stations_dict, f, ensure_ascii=False, indent=4)
6.3.3 12306.utils.captcha.py
import hashlib
import requests
from datetime import datetime
RUOUSER = 'xxxx'
RUOPASS = 'xxxx'
# 若快 12306打码 直接传入本地文件路径
def getCode(img):
url = "http://api.ruokuai.com/create.json"
fileBytes = open(img, "rb").read()
paramDict = {
'username': RUOUSER,
'password': RUOPASS,
'typeid': 6113, # 专门用来识别12306图片验证的类型id
'timeout': 90,
'softid': 117157, # 推广用的
'softkey': '70acaa1e477a4374a7736264a24b974b' # 推广用的
}
paramKeys = ['username',
'password',
'typeid',
'timeout',
'softid',
'softkey'
]
result = http_upload_image(url, paramKeys, paramDict, fileBytes)
return result['Result']
# 若快12306打码 上传图片
def http_upload_image(url, paramKeys, paramDict, filebytes):
timestr = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
boundary = '------------' + hashlib.md5(timestr.encode("utf8")).hexdigest().lower()
boundarystr = '\r\n--%s\r\n' % (boundary)
bs = b''
for key in paramKeys:
bs = bs + boundarystr.encode('ascii')
param = "Content-Disposition: form-data; name=\"%s\"\r\n\r\n%s" % (key, paramDict[key])
# print param
bs = bs + param.encode('utf8')
bs = bs + boundarystr.encode('ascii')
header = 'Content-Disposition: form-data; name=\"image\"; filename=\"%s\"\r\nContent-Type: image/gif\r\n\r\n' % ('sample')
bs = bs + header.encode('utf8')
bs = bs + filebytes
tailer = '\r\n--%s--\r\n' % (boundary)
bs = bs + tailer.encode('ascii')
headers = {'Content-Type': 'multipart/form-data; boundary=%s' % boundary,
'Connection': 'Keep-Alive',
'Expect': '100-continue',
}
response = requests.post(url, params='', data=bs, headers=headers)
return response.json()
if __name__ == '__main__':
# 测试
ret = getCode('../imgs/1544627949970.png')
print(ret)
6.3.4 12306.utils.parse_date.py
import datetime
def parseDate(train_date):
"""
:param train_date: '2017-12-12'
:return:
"""
week_name = ["Mon", "Tue", "Wed", "Thu", "Fri", "Sat", "Sun"]
month_name = "Jan Feb Mar Apr May Jun Jul Aug Sep Oct Nov Dec".split()
y, m, d = map(int, train_date.split("-"))
weekday = datetime.datetime(y, m, d).weekday()
# Fri Nov 24 2017 00:00:00 GMT+0800 (中国标准时间)
return "{0} {1} {2} {3} 00:00:00 GMT+0800 (中国标准时间)".format(week_name[weekday], month_name[m - 1], d, y)
6.3.5 12306.utils.parse_passenger.py
def parsePassenger(passenger_dict):
passengers_infos_list = passenger_dict['data']['normal_passengers']
passenger_list = []
for passenger_info in passengers_infos_list:
passenger_info_dict = {}
passenger_info_dict['passenger_name'] = passenger_info.get('passenger_name', '')
passenger_info_dict['passenger_gender'] = passenger_info.get('sex_name', '')
passenger_info_dict['passenger_id_type_code'] = passenger_info.get('passenger_id_type_code', '')
passenger_info_dict['passenger_id_no'] = passenger_info.get('passenger_id_no', '')
passenger_info_dict['passenger_mobile_no'] = passenger_info.get('mobile_no', '')
passenger_list.append(passenger_info_dict)
return passenger_list
6.3.6 12306.utils.parse_seat_type.py
seat_type_dict = {
"erdengzuo": "O", # 二等座
"yingwo": "3", # 硬卧
"yingzuo": "1", # 硬座
"wuzuo": "1", # 无座
"ruanwo": "4", # 软卧
"ruanzuo": "2", # 软座
"dongwo": "F", # 动卧
"yidengzuo": "M", # 一等座
"gaojiruanwo": "6", # 高级软座
"shangwuzuo": "9", # 商务座
"tedengzuo": "P", # 特等座
}
6.3.7 12306.utils.parse_trains_infos.py
import urllib.parse
def parseTrainsInfos(trains_list):
"""
解析列车信息列表, 返回列车信息列表
"""
trains_infos_list = []
if trains_list == []:
return []
for train_info in trains_list:
train_info_list = train_info.split('|')
train_info_dict = {}
# 构造列车信息
train_info_dict['secretStr'] = urllib.parse.unquote(train_info_list[0]) # secretStr ;为''时无法购买车票
# train_info_list[1] 预定/列车停运
train_info_dict['train_no'] = urllib.parse.unquote(train_info_list[2]) # train_no
train_info_dict['stationTrainCode'] = urllib.parse.unquote(train_info_list[3]) # stationTrainCode 即车次 # 展示
train_info_dict['start_station'] = urllib.parse.unquote(train_info_list[4]) # 始发站 # 展示
train_info_dict['end_station'] = urllib.parse.unquote(train_info_list[5]) # 终点站 # 展示
train_info_dict['from_station'] = urllib.parse.unquote(train_info_list[6]) # 出发站 # 展示
train_info_dict['to_station'] = urllib.parse.unquote(train_info_list[7]) # 到达站 # 展示
train_info_dict['from_time'] = urllib.parse.unquote(train_info_list[8]) # 出发时间 # 展示
train_info_dict['to_time'] = urllib.parse.unquote(train_info_list[9]) # 到达时间 # 展示
train_info_dict['use_time'] = urllib.parse.unquote(train_info_list[10]) # 时长 # 展示
train_info_dict['buy_able'] = urllib.parse.unquote(train_info_list[11]) # 能否购买 Y 可以购买 N 不可以购买 IS_TIME_NOT_BUY 停运 # 展示
train_info_dict['leftTicket'] = urllib.parse.unquote(train_info_list[12]) # leftTicket
train_info_dict['start_time'] = urllib.parse.unquote(train_info_list[13]) # 车次始发日期 # 展示
train_info_dict['train_location'] = urllib.parse.unquote(train_info_list[15]) # train_location 不知道是啥??
train_info_dict['from_station_no'] = urllib.parse.unquote(train_info_list[16]) # 出发站编号
train_info_dict['to_station_no'] = urllib.parse.unquote(train_info_list[17]) # 到达站编号
# 14,18,19,20,27,34,35未知
train_info_dict['gaojiruanwo'] = urllib.parse.unquote(train_info_list[21]) # 高级软卧 # 展示
train_info_dict['qita'] = urllib.parse.unquote(train_info_list[22]) # 其他 # 展示
train_info_dict['ruanwo'] = urllib.parse.unquote(train_info_list[23]) # 软卧 # 展示
train_info_dict['ruanzuo'] = urllib.parse.unquote(train_info_list[24]) # 软座 # 展示
train_info_dict['tedengzuo'] = urllib.parse.unquote(train_info_list[25]) # 特等座 # 展示
train_info_dict['wuzuo'] = urllib.parse.unquote(train_info_list[26]) # 无座 # 展示
train_info_dict['yingwo'] = urllib.parse.unquote(train_info_list[28]) # 硬卧 # 展示
train_info_dict['yingzuo'] = urllib.parse.unquote(train_info_list[29]) # 硬座 # 展示
train_info_dict['erdengzuo'] = urllib.parse.unquote(train_info_list[30]) # 二等座 # 展示
train_info_dict['yidengzuo'] = urllib.parse.unquote(train_info_list[31]) # 一等座 # 展示
train_info_dict['shangwuzuo'] = urllib.parse.unquote(train_info_list[32]) # 商务座 # 展示
train_info_dict['dongwo'] = urllib.parse.unquote(train_info_list[33]) # 动卧 # 展示
trains_infos_list.append(train_info_dict)
return trains_infos_list
6.3.8 12306.utils.stations_dict.py
该文件由 12306.get_stations_dict.py运行生成
stations_dict = {
"北京北": "VAP",
"北京东": "BOP",
"北京": "BJP",
......
小结
- 了解 购票的整个逻辑
