12306车票自购
import datetime
import json
import time

from api_login import check_user, check_verify
from api_with_cookie import update_cookie
from global_var import get_value
from log.log import log
from passenger import get_passengers, get_order_passengers
from qr_login import qr_login
from util import is_success


def candidate_process(train_detail):
    log('开始候补流程')
    check_pass = False
    submit_after_login = False
    login_with_psw = get_value('login') == 'psw'
    while not check_pass:
        check_pass = check_face(train_detail)
        if not check_pass:
            if login_with_psw:
                is_login = False
                log('开启账号密码登录流程')
                check_verify()
                while not is_login:
                    can_go_next = get_value('can_go_next')
                    if can_go_next:
                        break
                    time.sleep(0.3)
                submit_after_login = True
                log('账号密码登录成功')
                break
            else:
                log('开启二维码登录流程')
                rsp = qr_login()
                if is_success(rsp) and rsp['data']['is_login'] == 'Y':
                    log('二维码登录成功')
                    submit_after_login = True
                    break
        else:
            log('check_face 通过，检查用户是否已登录')
            check_pass = check_user()
            if check_pass:
                submit_after_login = True
            break

    if submit_after_login:
        log('准备提交候补订单')
        seat_type = get_value('config_dict')['seatType']
        secret_list = train_detail['secretStr'] + '#' + seat_type + '|'
        rsp = submit_candidate_order(secret_list)
        if is_success(rsp) and rsp['data']['flag']:
            rsp = after_nate_passenger_init()
            if is_success(rsp):
                log('获取乘客列表')
                rsp = get_passengers()
                passenger_list = get_order_passengers(rsp)
            else:
                log('passengerInitApi 失败')
                if isinstance(rsp, dict):
                    log(json.dumps(rsp))
                passenger_list = None
            if passenger_list is not None and len(passenger_list) > 0:
                passenger_info = ''
                for passenger in passenger_list:
                    born_date = int(passenger['born_date'].split('-')[0])
                    age = datetime.date.today().year - born_date
                    is_old_than_60 = age >= 60
                    passenger_info += (passenger['passenger_type'] + '#' + passenger['passenger_name'] + '#' +
                                       passenger['passenger_id_type_code'] + '#' + passenger['passenger_id_no'] + '#' + passenger['allEncStr'] + '#')
                    if is_old_than_60:
                        passenger_info += '1;'
                    else:
                        passenger_info += '0;'
                if len(passenger_info) > 0 and check_candidate_queue_can_goto_next():
                    log('开始提交候补订单')
                    rsp = confirm_candidate_order(passenger_info, train_detail)
                    if is_success(rsp) and rsp['data']['flag']:
                        if rsp['data']['isAsync']:
                            while True:
                                next_request_time = query_candidate_queue()
                                if next_request_time > 0:
                                    time.sleep(next_request_time)
                                else:
                                    break
                    else:
                        log('confirmHB接口返回失败')
                        log(json.dumps(rsp))
        else:
            if isinstance(rsp, dict):
                log('submit_candidate_order --- ' + rsp['messages'][0])


def confirm_candidate_order(passenger_info, train_detail):
    url = 'https://kyfw.12306.cn/otn/afterNate/confirmHB'
    seat_type = get_value('config_dict')['seatType']
    rsp = update_cookie(url, post_data={
        'passengerInfo': passenger_info,
        'jzParam': '',
        # 候补车次信息
        'hbTrain': train_detail['queryLeftNewDTO']['train_no'] + ',' + seat_type + '#',
        'lkParam': '',
        'sessionId': '',
        'sig': '',
        'scene': 'nc_login',
        'encryptedData': '',
        # 是否接受无座，默认不接受
        'if_receive_wseat': 'N',
        # 截止兑换时间，默认开车前24小时
        'realize_limit_time_diff': 1440
    }, other={
        'Content-Type': 'application/x-www-form-urlencoded;charset=UTF-8'
    })
    return rsp

def after_nate_passenger_init():
    url = 'https://kyfw.12306.cn/otn/afterNate/passengerInitApi'
    return update_cookie(url, other={
        'Content-Type': 'application/x-www-form-urlencoded;charset=UTF-8'
    })

# 候补队列查询
def check_candidate_queue_can_goto_next():
    url = 'https://kyfw.12306.cn/otn/afterNate/getQueueNum'
    rsp = update_cookie(url)
    if is_success(rsp) and rsp['data']['flag']:
        return True
    else:
        log('getQueueNum失败')
        log(json.dumps(rsp))
    return False

def query_candidate_queue():
    url = 'https://kyfw.12306.cn/otn/afterNate/queryQueue'
    rsp = update_cookie(url)
    if is_success(rsp) and rsp['data']['flag'] and rsp['data']['isAsync']:
        status = rsp['data']['status']
        if status == 1:
            log('候补排队成功，待支付')
            return 0
        elif status == -1:
            log('候补排队失败')
        else:
            wait_time = rsp['data']['waitTime']
            n = int(wait_time / 3)
            if n < 1:
                n = 1
            return n
    return -1


def check_face(train_detail):
    seat_type = get_value('config_dict')['seatType']
    secret_list = train_detail['secretStr'] + '#' + seat_type + '|'
    success_secret = train_detail['secretStr'] + '#' + seat_type
    url = 'https://kyfw.12306.cn/otn/afterNate/chechFace'
    rsp = update_cookie(url, post_data={
        'secretList': secret_list,
        '_json_attr': '',
    }, other={
        'Content-Type': 'application/x-www-form-urlencoded;charset=UTF-8'
    })
    if is_success(rsp) and rsp['data']['login_flag'] and rsp['data']['face_flag']:
        return get_success_rate(success_secret)
    else:
        log('check_face 失败...login_flag = ' + str(rsp['data']['login_flag']) + ', face_flag = ' + str(rsp['data']['face_flag']))
    return False

def get_success_rate(success_secret):
    url = 'https://kyfw.12306.cn/otn/afterNate/getSuccessRate'
    rsp = update_cookie(url, post_data={
        'successSecret': success_secret,
        '_json_attr': '',
    }, other={
        'Content-Type': 'application/x-www-form-urlencoded;charset=UTF-8'
    })
    if is_success(rsp) and rsp['data']['flag'][0] is not None:
        level = int(rsp['data']['flag'][0]['level'])
        log(rsp['data']['flag'][0]['info'])
        # 候补人数少时可以尝试候补
        if level <= 2:
            return True
    return False


def submit_candidate_order(secret_list):
    url = 'https://kyfw.12306.cn/otn/afterNate/submitOrderRequest'
    return update_cookie(url, post_data={
        'secretList': secret_list,
        '_json_attr': '',
    }, other={
        'Content-Type': 'application/x-www-form-urlencoded;charset=UTF-8'
    })
after_nate.py
johnshazhu/12306
import json

import handle_sms_code
import notification_util
from api_with_cookie import update_cookie
from global_var import set_value, get_value
from log.log import log
from util import is_success

def user_login():
    return update_cookie('https://kyfw.12306.cn/otn/login/userLogin', post_method=False)

def redirect_user_login():
    return update_cookie('https://kyfw.12306.cn/otn/passport?redirect=/otn/login/userLogin', post_method=False)

def uamtk():
    return update_cookie('https://kyfw.12306.cn/passport/web/auth/uamtk', post_data={
        'appid': 'otn'
    }, other={
        'Referer': 'https://kyfw.12306.cn/otn/passport?redirect=/otn/login/userLogin',
        'Content-Type': 'application/x-www-form-urlencoded;application/json;charset=UTF-8'
    })

def uamauthclient(new_app_tk):
    return update_cookie('https://kyfw.12306.cn/otn/uamauthclient', post_data={
        'tk': new_app_tk,
    }, other={
        'Referer': 'https://kyfw.12306.cn/otn/passport?redirect=/otn/login/userLogin',
        'Content-Type': 'application/x-www-form-urlencoded;application/json;charset=UTF-8'
    })

def web_auth_uamtk_static():
    log('查看是否已登录')
    return update_cookie('https://kyfw.12306.cn/passport/web/auth/uamtk-static', post_data={
            'appid': 'otn'
        }, other={
            'Content-Type': 'application/x-www-form-urlencoded'
        })

def conf():
    return update_cookie('https://kyfw.12306.cn/index/otn/login/conf')

def get_login_config():
    log('获取登录配置')
    rsp = update_cookie('https://kyfw.12306.cn/otn/login/conf')
    if is_success(rsp):
        data = rsp['data']
        set_value('is_uam_login', data['is_uam_login'] == 'Y')
        set_value('is_login_passCode', data['is_login_passCode'] == 'Y')
        set_value('is_login', data['is_login'] == 'Y')
        set_value('queryUrl', data['queryUrl'])
    return rsp

def check_verify():
    rsp = check_login_verify()
    if rsp['login_check_code'] == '0':
        process_start_from_login()
    elif rsp['login_check_code'] == '2':
        log('滑块方式验证暂不支持')
    elif rsp['login_check_code'] == '3':
        log('短信验证码方式验证')
        rsp = get_sms_code()
        if is_success(rsp):
            notification_util.show_notification('短信验证码', '请注意查看短信验证码并输入')
            handle_sms_code.input_sms_code(submit_sms_code_callback)
        else:
            log(rsp['result_message'])
    else:
        log(json.dumps(rsp))

def check_login_verify():
    return update_cookie('https://kyfw.12306.cn/passport/web/checkLoginVerify', post_data={
        'username': get_value('config_obj').username,
        'appid': 'otn'
    }, other={
        'Referer': 'https://kyfw.12306.cn/otn/resources/login.html',
        'Content-Type': 'application/x-www-form-urlencoded;application/json;charset=UTF-8',
        'Origin': 'https://kyfw.12306.cn',
        'Accept': '*/*'
    })

def get_sms_code():
    log('获取短信验证码')
    url = 'https://kyfw.12306.cn/passport/web/getMessageCode'
    check_data = {
        'username': get_value('config_obj').username,
        'appid': 'otn',
        'castNum': get_value('config_dict')['castNum'],
    }
    return update_cookie(url, post_data=check_data, other={
        'Referer': 'https://kyfw.12306.cn/otn/resources/login.html',
        'Content-Type': 'application/x-www-form-urlencoded;application/json;charset=UTF-8',
        'Origin': 'https://kyfw.12306.cn',
        'Accept': '*/*'
    })

def submit_sms_code_callback():
    sms_code = handle_sms_code.get_sms_code()
    config = get_value('config_obj')
    config.checkMode = '0'
    config.randCode = sms_code
    set_value('config_obj', config)

    handle_sms_code.destroy()
    process_start_from_login()


def to_user_auth_center():
    log('to_user_auth_center --- 用户中心统一认证')
    rsp = uamtk()
    if is_success(rsp):
        tk = rsp['newapptk']
        if tk is None:
            tk = rsp['apptk']
        rsp = uamauthclient(tk)
        if is_success(rsp):
            user_login()
            log('user_login --- 更新cookie uKey')
            rsp = conf()
            log('conf --- 获取配置信息，is_login = ' + rsp['data']['is_login'])
            set_value('can_go_next', is_success(rsp) and rsp['data']['is_login'] == 'Y')
            return rsp
        else:
            log('uamauthclient : ' +  rsp['result_message'] + ', result_code = ' + str(rsp['result_code']))
    else:
        log('uamtk : ' + rsp['result_message'] + ', result_code = ' + str(rsp['result_code']))


def process_start_from_login():
    log('开始调用web/login接口登录')
    rsp = web_login()
    session = get_value('session')
    if 'isPasswordCopy' in session.headers:
        del session.headers['isPasswordCopy']
        set_value('session', session)
    if rsp['result_code'] == 0:
        log('跳转otn/login/userLogin')
        rsp = user_login()
        if rsp == 302:
            redirect_user_login()
        to_user_auth_center()
    elif rsp['result_code'] == 91 or rsp['result_code'] == 92 or rsp['result_code'] == 94 or rsp['result_code'] == 95 or rsp['result_code'] == 97:
        is_uam_login = get_value('is_uam_login')
        if is_uam_login:
            to_user_auth_center()
    elif rsp['result_code'] == 101:
        log('您的密码很久没有修改了，为降低安全风险，请您重新设置密码后再登录')
    else:
        log(rsp['result_message'])

def web_login():
    return update_cookie('https://kyfw.12306.cn/passport/web/login', post_data=get_value('config_obj').__dict__, other={
        'isPasswordCopy': 'N',
        'Referer': 'https://kyfw.12306.cn/otn/resources/login.html',
        'Content-Type': 'application/x-www-form-urlencoded;application/json;charset=UTF-8',
        'Origin': 'https://kyfw.12306.cn',
    })

def check_user():
    url = 'https://kyfw.12306.cn/otn/login/checkUser'
    post_data = {
        '_json_att': ''
    }
    rsp = update_cookie(url, post_data=post_data, other={
        'Content-Type': 'application/x-www-form-urlencoded; charset=UTF-8'
    })
    if is_success(rsp) and rsp['data']['flag']:
        return True
    return False
api_login.py
johnshazhu/12306
import json
from urllib.parse import urlencode

import requests
import urllib3

from global_var import set_value, get_value

def init():
    set_value('session', requests.session())
    set_value('cookie', {})

def get_cookie():
    temp_cookies_list = ''
    for k in get_value('cookie').keys():
        temp_cookies_list += (k + '=' + get_value('cookie').get(k) + ';')
    temp_cookies_list = temp_cookies_list[:(len(temp_cookies_list) - 1)]
    return temp_cookies_list

def update_cookie(url, post_data=None, other=None, post_method=True):
    urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
    session = get_value('session')
    headers = {
        'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/117.0.0.0 Safari/537.36',
        'Host': 'kyfw.12306.cn',
    }
    for k in headers.keys():
        session.headers[k] = headers[k]

    if other is not None:
        for k in other.keys():
            session.headers[k] = other[k]

    cookie_str = get_cookie()
    if len(cookie_str) > 0:
        session.headers['Cookie'] = cookie_str

    if post_method:
        if post_data is not None and 'Content-Type' in session.headers and session.headers['Content-Type'].find('application/x-www-form-urlencoded') > -1:
            post_data = urlencode(post_data, safe='/*').encode()
        session_rsp = session.post(url, data=post_data, verify=False)
    else:
        if post_data is not None:
            param = urlencode(post_data)
            finally_url = url + '?%s' % param
        else:
            finally_url = url
        session_rsp = session.get(finally_url, verify=False)
    if session_rsp.status_code == 200:
        session_set_cookie = session_rsp.headers.get('Set-Cookie')
        cookie = get_value('cookie')
        if session_set_cookie is not None:
            # Path=,; route=/cfg
            # Path=/abc, route=/cfg; cmd=test
            # Expire=Thu, 1970
            cookie_items = session_set_cookie.split('; ')
            for cookie_item in cookie_items:
                equals_count = cookie_item.count('=')
                if equals_count > 1:
                    sub_items = cookie_item.split(', ')
                    for child in sub_items:
                        item_key = child.split('=')[0]
                        item_value = child.split('=')[1]
                        if item_key.lower() == 'path':
                            continue
                        else:
                            cookie[item_key] = item_value
                else:
                    item_key = cookie_item.split('=')[0]
                    if item_key.lower() != 'path':
                        item_value = cookie_item.split('=')[1]
                        cookie[item_key] = item_value
        set_value('cookie', cookie)
        # if (post_method or session.headers['Content-Type'].find('application/json') > -1) and is_not_html_content(session_rsp):
        if is_not_html_content(session_rsp):
            return json.loads(session_rsp.content)
        else:
            return session_rsp.content
    else:
        return session_rsp.status_code


def is_not_html_content(rsp):
    return rsp.headers['Content-Type'].find('text/html') == -1
api_with_cookie.py
johnshazhu/12306
import json

import api_with_cookie
import global_var
from js.js_util import exec_js
from log.log import init_log, log
from station import init_station_names

from global_var import set_value
from unescape import escape

SM4_key = 'tiekeyuankp12306'

class Config:
    def __init__(self, username, password):
        self.username = username
        self.password = password
        self.appid = 'otn'
        self.sessionId = ''
        self.sig = ''
        self.if_check_slide_passcode_token = ''
        self.scene = ''
        # 0
        self.checkMode = ''
        # SMS验证码
        self.randCode = ''
        self._json_att = ''

    @staticmethod
    def object_hook(dict_data):
        return Config(dict_data['username'], dict_data['password'])


def init_config():
    global_var.init()

    api_with_cookie.init()

    config_dict = {}

    init_log(False)

    seats = ['1', '2', '3', '4', '6', '9', 'A', 'F', 'I', 'J', 'M', 'O', 'P']
    can_choose_seats_type = ['9', 'M', 'O', 'P']
    can_choose_seat_detail_type = ['3', '4', 'F']
    other_seats_type = ['1', '2', '6', 'A', 'I', 'J']
    keys_check_ignore = 'timesBetweenTwoQuery, chooseSeats, seatDetailType, trainCode, candidate, aftertime'
    config_check_pass = True
    with open('config.properties', 'r') as f:
        for line in f.readlines():
            line = line.strip()
            if not len(line):
                continue
            key = line.split('=')[0]
            config_dict[key] = str(line)[(len(key) + 1):]
            if len(config_dict[key]) == 0:
                if keys_check_ignore.find(key) == -1:
                    config_check_pass = False
                    log(f'请在config.properties中配置{key}的值=xxx')

    seat_type = config_dict['seatType']
    # 检查坐席设置是否正常
    if config_check_pass:
        if seat_type not in set(seats):
            config_check_pass = False
            log(f'seatType的值应该在{seats}中')
        elif seat_type in set(other_seats_type):
            # 是否是不可选座类型
            config_dict['seatDetailType'] = '000'
            config_dict['chooseSeats'] = ''
        elif seat_type in set(can_choose_seats_type):
            # 检查是动车/高铁坐席
            # 动车或高铁时，seatDetailType设置无效，默认为'000'
            config_dict['seatDetailType'] = '000'
            if len(config_dict['chooseSeats']) == 0:
                config_check_pass = False
                log('请在config.properties中配置chooseSeats的值')
        elif seat_type in set(can_choose_seat_detail_type):
            # 检查是否是卧铺类坐席
            # 卧铺类时，chooseSeats设置无效，默认为''
            config_dict['chooseSeats'] = ''
            if len(config_dict['seatDetailType']) == 0:
                config_check_pass = False
                log('请在config.properties中配置seatDetailType的值')

        if config_check_pass:
            set_value('config_dict', config_dict)
            station_name_list = init_station_names()

            if station_name_list is not None:
                from_station_code = get_station_code(config_dict['from'], station_name_list)
                to_station_code = get_station_code(config_dict['to'], station_name_list)
                set_value('from_station_code', from_station_code)
                set_value('to_station_code', to_station_code)
                set_value('_jc_save_fromStation', escape(config_dict['from'] + ',' + from_station_code))
                set_value('_jc_save_toStation', escape(config_dict['to'] + ',' + to_station_code))
            config_content = json.dumps(config_dict)
            config_obj = json.loads(config_content, object_hook=Config.object_hook)
            config_obj.password = get_encrypt_content(config_obj.password)
            log('username = ' + config_obj.username)
            set_value('config_obj', config_obj)
    return config_check_pass


def get_station_code(name, station_name_list):
    for station in station_name_list:
        if station[1] == name:
            print(station)
            return station[2]


def get_encrypt_content(plain_text):
    encrypt_result = exec_js('js/SM4.js', "encrypt_ecb", plain_text, SM4_key)
    return f'@{encrypt_result}'
config.py
johnshazhu/12306
def init():
    global global_dict
    global_dict = {}

def set_value(key, value):
    global_dict[key] = value

def get_value(key):
    return global_dict[key]



global_var.py
johnshazhu/12306
from tkinter import *

root = None
sms_code = None


def check_sms_code(content):
    global sms_code
    print(content)
    sms_code = content
    if (content.isdigit() and len(content) <= 6) or len(content) == 0:
        return True
    return False


def get_sms_code():
    return sms_code


def destroy():
    if isinstance(root, Tk):
        print('submit then destroy')
        root.destroy()
    # win.protocol("WM_DELETE_WINDOW", lambda: sys.exit(0))


def input_sms_code(submit_callback):
    global root
    global value
    root = Tk()
    root.title('请输入短信验证码')

    Label(root, text='短信验证码 :').grid(row=0, column=0)
    value = StringVar()
    cmd = root.register(check_sms_code)
    editor = Entry(root, width=10, textvariable=value, validate='key', validatecommand=(cmd, '%P'))
    editor.grid(row=0, column=1, padx=10, pady=5)
    root.geometry('400x200')

    Button(root, text='确定', width=6, highlightbackground='#55CEAC', command=submit_callback) \
        .grid(row=0, column=2, padx=10, pady=5)

    root.mainloop()

handle_sms_code.py
johnshazhu/12306
import execjs

from log.log import log


def exec_js(path, func, *args):
    with open(path, 'r', encoding='UTF-8') as f:
        js_code = f.read()
    context = execjs.compile(js_code)
    try:
        exec_result = context.call(func, *args)
    except Exception as err:
        print(type(err))
        exec_result = err.__getattribute__('doc')
        print(err)
    log(f'javascript {path} {func} execute result = {exec_result}')
    return exec_result
js/js_util.py
johnshazhu/12306
import datetime

file = None

def log_file_name():
    return datetime.datetime.now().strftime('%Y%m%d') + '.txt'

def init_log(save_log):
    global file
    if save_log:
        file = open('log/' + log_file_name(), 'wb')

def log(msg):
    now = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
    print(f'{now}\t{msg}')
    if file is not None:
        file.write((msg + '\n').encode())
    if msg == 'exit':
        log_exit()

def log_exit():
    if file is not None:
        file.close()
log/log.py
johnshazhu/12306
from api_login import get_login_config, web_auth_uamtk_static, check_verify
from config import init_config
from global_var import get_value, set_value
from log.log import log
from query_and_order import start
from util import is_success

if __name__ == '__main__':
    if init_config():
        rsp = get_login_config()
        if is_success(rsp):
            rsp = web_auth_uamtk_static()
            is_login = False
            if is_success(rsp):
                is_login = True
                log('已登录')
            else:
                log('未登录')
                check_verify()

            while not is_login:
                can_go_next = get_value('can_go_next')
                if can_go_next:
                    log('登录成功')
                    is_login = True
                    break

            set_value('login', 'psw')
            start()

    log('exit')
login.py
johnshazhu/12306
import os
import platform
if 'Windows' == platform.system():
    import win32api
    import win32con


def show_notification(title, message):
    if 'Darwin' == platform.system():
        command = f'''
            osascript -e 'display notification "{message}" with title "{title}"'
        '''
        os.system(command)
    elif 'Windows' == platform.system():
        win32api.MessageBox(0, message,
                            "MessageBox",
                            win32con.MB_OK | win32con.MB_ICONWARNING)
notification_util.py
johnshazhu/12306
import json

from api_with_cookie import update_cookie
from global_var import get_value
from log.log import log


def get_passengers(token=None):
    url = 'https://kyfw.12306.cn/otn/confirmPassenger/getPassengerDTOs'
    if token is None:
        return update_cookie(url)

    return update_cookie(url, post_data={
        '_json_att': '',
        'REPEAT_SUBMIT_TOKEN': token
    }, other={
        'Content-Type': 'application/x-www-form-urlencoded; charset=UTF-8'
    })

def get_order_passengers(rsp):
    if rsp['status'] and len(rsp['data']['normal_passengers']) > 0:
        order_passengers = []
        for passenger in rsp['data']['normal_passengers']:
            if get_value('config_dict')['passengers'].find(passenger['passenger_name']) != -1:
                order_passengers.append(passenger)
        return order_passengers
    else:
        log('获取乘客列表失败')
        log(json.dumps(rsp))
passenger.py
johnshazhu/12306
import ssl

from config import init_config
from global_var import set_value
from log.log import log
from qr_login import qr_login
from query_and_order import start
from util import is_success


if __name__ == '__main__':
    ssl._create_default_https_context = ssl._create_unverified_context
    if init_config():
        rsp = qr_login()
        set_value('login', 'qr')
        if is_success(rsp) and rsp['data']['is_login'] == 'Y':
            start()
    log('exit')

qr.py
johnshazhu/12306
import base64
import os
import time

from api_login import user_login, redirect_user_login, to_user_auth_center
from api_with_cookie import update_cookie
from log.log import log


def create_qr64():
    url = 'https://kyfw.12306.cn/passport/web/create-qr64'
    rsp = update_cookie(url, post_data={
        'appid': 'otn'
    }, other={
        'Content-Type': 'application/x-www-form-urlencoded; charset=UTF-8'
    })
    if rsp['result_code'] == '0':
        image = base64.b64decode(rsp['image'])
        if os.path.exists("resource/qr_image.jpg"):
            os.remove("resource/qr_image.jpg")
        file = open('resource/qr_image.jpg', 'wb')
        file.write(image)
        file.close()
        log(rsp['result_message'])
        return rsp['uuid']
    else:
        log('登录二维码生成失败，code = ' + rsp['result_code'] + ', result_message = ' + rsp['result_message'])
    return None

def check_qr(uuid):
    log('二维码扫描检测中...')
    scan_success = False
    while True:
        url = 'https://kyfw.12306.cn/passport/web/checkqr'
        rsp = update_cookie(url, post_data={
            'appid': 'otn',
            'uuid': uuid
        }, other={
            'Content-Type': 'application/x-www-form-urlencoded; charset=UTF-8'
        })
        if rsp == 302:
            time.sleep(1)
            continue
        elif rsp['result_code'] == '0' or rsp['result_code'] == '1':
            time.sleep(1)
            continue
        else:
            log(rsp['result_message'] + ', result_code = ' + rsp['result_code'])
            scan_success = rsp['result_code'] == '2'
            break

    return scan_success

def qr_login():
    uuid = create_qr64()
    if uuid is not None:
        if check_qr(uuid):
            rsp = user_login()

            if rsp == 302:
                redirect_user_login()
                log('redirect_user_login --- 登录成功后更新cookie中的JSESSIONID')

            return to_user_auth_center()
    return None
qr_login.py
johnshazhu/12306
import json
import threading
import time
import urllib

from after_nate import candidate_process
from api_with_cookie import update_cookie
from global_var import get_value, set_value
from js.js_util import exec_js
from log.log import log
from passenger import get_passengers, get_order_passengers
from ticket.query_left_ticket import get_detail, can_buy_seat
from util import get_today_str, is_candidate_config, is_success


def query_left_tickets():
    return update_cookie('https://kyfw.12306.cn/otn/leftTicket/queryZ', post_data={
        'leftTicketDTO.train_date': get_value('config_dict')['date'],
        'leftTicketDTO.from_station': get_value('from_station_code'),
        'leftTicketDTO.to_station': get_value('to_station_code'),
        'purpose_codes': get_value('config_dict')['purpose_codes'],
    }, other={
        'Referer': 'https://kyfw.12306.cn/otn/leftTicket/init',
        'If-Modified-Since': '0',
        'X-Requested-With': 'XMLHttpRequest',
        'Accept': '*/*',
        'Sec-Fetch-Mode': 'cors',
        'Sec-Fetch-Dest': 'empty',
        'Sec-Fetch-Site': 'same-origin',
        'Cache-Control': 'no-cache',
        'Accept-Encoding': 'gzip, deflate, br',
        'Content-Type': 'application/x-www-form-urlencoded;application/json;charset=UTF-8'
    }, post_method=False)


def submit_order(train_detail, back_date):
    order_date = get_value('config_dict')['date']
    submit_order_url = 'https://kyfw.12306.cn/otn/leftTicket/submitOrderRequest'
    post_data = {
        'secretStr': urllib.parse.unquote(train_detail['secretStr'], encoding='utf-8', errors='replace'),
        'train_date': order_date,
        'back_train_date': get_today_str(),
        'purpose_codes': get_value('config_dict')['purpose_codes'],
        'tour_flag': 'dc',
        'query_from_station_name': get_value('config_dict')['from'],
        'query_to_station_name': get_value('config_dict')['to'],
        'undefined': ''
    }
    return update_cookie(submit_order_url, post_data=post_data, other={
        'Content-Type': 'application/x-www-form-urlencoded; charset=UTF-8'
    }, post_method=True)


def get_repeat_submit_token():
    url = 'https://kyfw.12306.cn/otn/confirmPassenger/initDc'
    content = update_cookie(url, other={
        'Content-Type': 'application/x-www-form-urlencoded'
    })
    return get_token_from_content(content.decode('utf-8'))

def check_order_info(passengers, token):
    check_order_info_url = 'https://kyfw.12306.cn/otn/confirmPassenger/checkOrderInfo'
    return update_cookie(check_order_info_url, post_data={
        'cancel_flag': 2,
        'bed_level_order_num': '000000000000000000000000000000',
        'passengerTicketStr': get_passenger_tickets(passengers),
        'oldPassengerStr': get_old_passengers(passengers),
        'tour_flag': 'dc',
        'whatsSelect': str(len(passengers)),
        'sessionId': '',
        'sig': '',
        'scene': 'nc_login',
        '_json_att': '',
        'REPEAT_SUBMIT_TOKEN': token
    }, other={
        'Content-Type': 'application/x-www-form-urlencoded; charset=UTF-8'
    })

def get_queue_count(token, train_info):
    url = 'https://kyfw.12306.cn/otn/confirmPassenger/getQueueCount'
    config_dict = get_value('config_dict')
    ticket_info_for_passenger_form = get_value('ticket_info_for_passenger_form')

    return update_cookie(url, post_data={
        'train_date': exec_js('js/date.js', 'getTrainDate', config_dict['date']),
        'train_no': train_info['train_no'],
        'stationTrainCode': train_info['station_train_code'],
        'seatType': config_dict['seatType'],
        'fromStationTelecode': train_info['from_station_telecode'],
        'toStationTelecode': train_info['to_station_telecode'],
        'leftTicket': ticket_info_for_passenger_form['queryLeftTicketRequestDTO']['ypInfoDetail'],
        'purpose_codes': ticket_info_for_passenger_form['purpose_codes'],
        'train_location': train_info['location_code'],
        'REPEAT_SUBMIT_TOKEN': token
    }, other={
        'Referer': 'https://kyfw.12306.cn/otn/confirmPassenger/initDc',
        'Content-Type': 'application/x-www-form-urlencoded; charset=UTF-8'
    })

def confirm_single_for_queue(token, confirm_order_passengers, train_info, encrypted_data):
    cookie = get_value('cookie')
    ticket_info_for_passenger_form = get_value('ticket_info_for_passenger_form')
    config_dict = get_value('config_dict')
    url = 'https://kyfw.12306.cn/otn/confirmPassenger/confirmSingleForQueue'
    if '_passport_session' in cookie:
        del cookie['_passport_session']

    cookie['highContrastMode'] = 'defaltMode'
    cookie['cursorStatus'] = 'off'
    cookie['guidesStatus'] = 'off'
    cookie['_jc_save_wfdc_flag'] = 'dc'
    cookie['_jc_save_showIns'] = 'true'
    cookie['_jc_save_fromStation'] = get_value('_jc_save_fromStation')
    cookie['_jc_save_toStation'] = get_value('_jc_save_toStation')
    cookie['_jc_save_fromDate'] = config_dict['date']
    cookie['_jc_save_toDate'] = get_today_str()

    return update_cookie(url, post_data={
        'passengerTicketStr': get_passenger_tickets(confirm_order_passengers),
        'oldPassengerStr': get_old_passengers(confirm_order_passengers),
        'purpose_codes': ticket_info_for_passenger_form['purpose_codes'],
        'key_check_isChange': ticket_info_for_passenger_form['key_check_isChange'],
        'leftTicketStr': ticket_info_for_passenger_form['leftTicketStr'],
        'train_location': ticket_info_for_passenger_form['train_location'],
        'choose_seats': config_dict['chooseSeats'],
        # 卧铺席， 2个下铺200，两个中铺020，其他 101（上下铺）
        'seatDetailType': config_dict['seatDetailType'],
        # 静音车厢
        'is_jy': 'N',
        # 残疾人
        'is_cj': 'Y',
        # js 中...window.json_ua.toString()
        # 'encryptedData': get_encrypted_data(),
        'encryptedData': encrypted_data,
        'whatsSelect': len(confirm_order_passengers),
        # 动卧？
        'roomType': '00',
        # 单程往返全选
        'dwAll': 'N',
        '_json_att': '',
        'REPEAT_SUBMIT_TOKEN': token
    }, other={
        'Referer': 'https://kyfw.12306.cn/otn/confirmPassenger/initDc',
        'Sec-Ch-Ua': '"Chromium";v="116", "Not)A;Brand";v="24", "Google Chrome";v="116"',
        'Sec-Ch-Ua-Mobile': '?0',
        'Sec-Ch-Ua-Platform': "macOS",
        'Sec-Fetch-Dest': 'empty',
        'Sec-Fetch-Mode': 'cors',
        'Sec-Fetch-Site': 'same-origin',
        'Pragma': 'no-cache',
        'X-Requested-With': 'XMLHttpRequest',
        'Cache-Control': 'no-cache',
        'Content-Type': 'application/x-www-form-urlencoded; charset=UTF-8'
    })

def query_order_wait_time(token):
    url = 'https://kyfw.12306.cn/otn/confirmPassenger/queryOrderWaitTime'
    return update_cookie(url, post_data={
        # 'random': exec_js('js/date.js', 'getDatetime'),
        'random': int(time.time() * 1000),
        'tourFlag': 'dc',
        '_json_att': '',
        'REPEAT_SUBMIT_TOKEN': token
    }, other={
        'Referer': 'https://kyfw.12306.cn/otn/confirmPassenger/initDc'
    }, post_method=False)

def result_order_for_dc_queue(order_id, token):
    url = 'https://kyfw.12306.cn/otn/confirmPassenger/resultOrderForDcQueue'
    return update_cookie(url, post_data={
        'orderSequence_no': order_id,
        '_json_att': '',
        'REPEAT_SUBMIT_TOKEN': token
    }, other={
        'Referer': 'https://kyfw.12306.cn/otn/confirmPassenger/initDc'
    })

def get_old_passengers(passenger_list):
    old_passengers_str = ''
    for passenger_info in passenger_list:
        name = passenger_info['passenger_name']
        id_number = passenger_info['passenger_id_no']
        id_type_code = passenger_info['passenger_id_type_code']
        passenger_type = passenger_info['passenger_type']
        old_passengers_str += f'{name},{id_type_code},{id_number},{passenger_type}_'
    return old_passengers_str

def get_token_from_content(html_content):
    start = html_content.find('var ticketInfoForPassengerForm=')
    if start > -1:
        content = html_content[start:]
        start_index = content.find('=')
        end_index = content.find(';')
        if start_index > -1 and end_index > -1:
            content = content[start_index + 1:end_index]
            content = content.replace('\'', '\"')
            set_value('ticket_info_for_passenger_form', json.loads(content))

    start = html_content.find('var globalRepeatSubmitToken = ')
    if start > -1:
        content = html_content[start:]
        start_index = content.find('\'')
        end_index = content.find(';')
        if start_index > -1 and end_index > -1:
            token = content[(start_index + 1):(end_index - 1)]
            log(f'token = {token}')
            set_value('token', token)
            return token


def update_cookie_without_uamtk():
    cookie = get_value('cookie')
    if 'uamtk' in cookie:
        del cookie['uamtk']
        set_value('cookie', cookie)

def get_selected_train_detail(rsp):
    detail_list = get_detail(rsp['data']['result'], rsp['data']['map'])

    # 是否只看设置的车次
    train_code = get_value('config_dict')['trainCode']
    check_train_code = False
    train_set = None
    if train_code is not None and len(str(train_code)) > 0:
        check_train_code = True
        train_list = train_code.split(',')
        train_set = set(train_list)

    # 是否只看几点以后的车次
    after_time = get_value('config_dict')['aftertime']
    check_start_time = False
    if train_code is not None and len(str(after_time)) > 0:
        check_start_time = True

    # 只看有对应坐席的车次
    seat_type = get_value('config_dict')['seatType']

    candidate = is_candidate_config()
    candidate_train = None
    for detail in detail_list:
        train_info = detail['queryLeftNewDTO']
        # 若设置了车次，当前车次不在预定车次列表里
        if check_train_code and train_info['station_train_code'] not in train_set:
            continue

        # 非候补下单，不能预订 或 对应坐席不能预订
        if not candidate and (train_info['canWebBuy'] != 'Y' or not can_buy_seat(train_info, get_value('config_dict')['seatType'])):
            continue

        # 车次未设置，过滤设置时间以前的车次
        if not check_train_code and check_start_time and str(train_info['start_time']) < after_time:
            continue

        # 过滤掉不支持坐席的车次
        if str(train_info['seat_types']).find(seat_type) == -1:
            continue

        # 可候补车次
        if candidate and train_info['houbu_train_flag'] == '1' and train_info['houbu_seat_limit'] == '':
            log('车次' + train_info['station_train_code'] + '可候补，发车时间为 ' + train_info['start_time'] + ', 历时 ' + train_info['lishi'] + '小时')
            candidate_train = detail
            update_cookie_without_uamtk()
            break

        log('车次' + train_info['station_train_code'] + '可购票，发车时间为 ' + train_info['start_time'] + ', 历时 ' +
            train_info['lishi'] + '小时')
        log(json.dumps(train_info))
        update_cookie_without_uamtk()
        return detail

    if candidate_train is not None:
        candidate_process(candidate_train)

def get_passenger_tickets(passenger_list):
    passengers_tickets = ''
    for passenger_info in passenger_list:
        name = passenger_info['passenger_name']
        id_number = passenger_info['passenger_id_no']
        mobile = passenger_info['mobile_no']
        id_type_code = passenger_info['passenger_id_type_code']
        all_enc_str = passenger_info['allEncStr']
        passenger_type = passenger_info['passenger_type']
        seat_type = get_value('config_dict')['seatType']
        passengers_tickets += f'{seat_type},0,{passenger_type},{name},{id_type_code},{id_number},{mobile},N,{all_enc_str}_'
    return passengers_tickets[0:(len(passengers_tickets) - 1)]

def check_is_can_select_seat_or_bed(rsp):
    if rsp['status'] and rsp['data']['submitStatus']:
        seat_type = get_value('config_dict')['seatType']
        if rsp['data']['choose_Seats'].find(seat_type) > -1:
            # 是否可选座位或铺位
            if rsp['data']['canChooseSeats'] == 'Y' or rsp['data']['canChooseBeds'] == 'Y':
                return True
        else:
            return True
    else:
        set_value('ordering', False)
        log('checkOrderInfo失败')
        log(json.dumps(rsp))
    return False

def check_can_confirm_order(rsp):
    if rsp['status']:
        log('余票 : ' + rsp['data']['ticket'] + ', 排队人数 : ' + rsp['data']['count'])
        if rsp['data']['op_2'] != 'false':
            log('目前排队人数已经超过余票张数，请您选择其他席别或车次')
            return False
        return True
    else:
        set_value('ordering', False)
        log('getQueueCount失败')
        log(json.dumps(rsp))
    return False

def check_wait_time(rsp):
    if rsp['status'] and rsp['data']['submitStatus']:
        if rsp['data']['isAsync'] or rsp['data']['isAsync'] == '1':
            return True
    else:
        set_value('ordering', False)
        log('confirmSingleForQueue失败')
        log(json.dumps(rsp))
    return False

def need_check_order(rsp):
    return rsp['status'] and rsp['data']['orderId'] is not None

def need_resend_wait_time_req(rsp):
    return rsp['status'] and rsp['data']['orderId'] is None

def check_order_success(rsp):
    return rsp['status'] and rsp['data']['submitStatus']

def start_timer_job(token):
    t1 = threading.Thread(target=timer_job, args=(token, 0))
    t2 = threading.Thread(target=delay_timer_job, args=(token,))
    log('start_timer_job')
    t1.start()
    t2.start()
    t1.join()
    t2.join()

def delay_timer_job(token):
    log('delay_timer_job')
    timer = threading.Timer(1.0, function=timer_job, args=(token, 1))
    timer.start()

def timer_job(token, index):
    disp_time = get_value('disp_time')
    next_request_time = get_value('next_request_time')
    log('execute timer_job index = ' + str(index) + ', disp_time = ' + str(disp_time) + ', next_request_time = ' + str(next_request_time))

    if disp_time <= 0:
        rsp = result_order_for_dc_queue(get_value('orderId'), token)
        if check_order_success(rsp):
            set_value('order_success', True)
            log('订单确认成功, 订单号 = ' + get_value('orderId') + ', 可到待支付订单中确认支付')
            return True
        else:
            set_value('ordering', False)
            log('订单确认失败')
            return False

    if disp_time == next_request_time:
        rsp = query_order_wait_time(token)
        if rsp['status'] and rsp['data']['queryOrderWaitTimeStatus']:
            wait_time = rsp['data']['waitTime']
            if wait_time != -100:
                log(f'等待时长 = {wait_time} 秒')
                set_value('disp_time', wait_time)
                set_value('orderId', rsp['data']['orderId'])
                d = int(wait_time / 1.5)
                if d > 60:
                    d = 60
                b = wait_time - d
                if b <= 0:
                    b = 1
                set_value('next_request_time', b)

    if disp_time > 1:
        disp_time -= 1
        set_value('disp_time', disp_time)
    if index > 0:
        time.sleep(1)
        timer_job(token, index + 1)

def process_from_query_start():
    log('---开始查询---')
    rsp = query_left_tickets()
    if is_success(rsp):
        detail = get_selected_train_detail(rsp)
        # 可能需要调用checkUser接口
        # rsp = check_user()
        # if is_success(rsp) and rsp['data']['flag']:
        if detail is None:
            return False
        log('开始准备提交订单')
        set_value('ordering', True)
        rsp = submit_order(detail, '')
        if is_success(rsp):
            log('获取token')
            token = get_repeat_submit_token()
            log('获取乘客信息列表')
            rsp = get_passengers(token)
            passenger_list = get_order_passengers(rsp)
            if passenger_list is not None and len(passenger_list) > 0:
                log('检查订单信息')
                rsp = check_order_info(passenger_list, token)
                if check_is_can_select_seat_or_bed(rsp):
                    log('开始查看排队数')
                    selected_train_info = detail['queryLeftNewDTO']
                    rsp = get_queue_count(token, selected_train_info)
                    if check_can_confirm_order(rsp):
                        # encrypted_data = exec_js('js/suit.js', 'init', get_cookie())
                        log('确定加入排队队列')
                        rsp = confirm_single_for_queue(token, passenger_list, selected_train_info, '')
                        if check_wait_time(rsp):
                            set_value('disp_time', 1)
                            set_value('next_request_time', 1)
                            log('等待中...')
                            start_timer_job(token)
                            return True
            else:
                set_value('ordering', False)
                log('乘客列表信息获取失败')
                log(json.dumps(rsp))
        else:
            set_value('ordering', False)
            log('submitOrderRequest失败')
            log(json.dumps(rsp))
    else:
        log('余票查询失败 : ' + rsp['messages'])

    return False

def start():
    config_dict = get_value('config_dict')
    delay_time = 0
    # 配置候补下单时，自动查询无效
    if not is_candidate_config():
        if 'timesBetweenTwoQuery' in config_dict:
            delay_time = float(config_dict['timesBetweenTwoQuery'])
            log(f'余票自动查询，间隔为 {delay_time} 秒')

    if delay_time > 0.3:
        set_value('order_success', False)
        set_value('ordering', False)
        while not get_value('order_success'):
            # 仅未提交订单时开始自动查询
            if not get_value('ordering'):
                success = process_from_query_start()
                if not success:
                    time.sleep(delay_time)
            else:
                # 提交订单中，等待时间延长点
                time.sleep(2 * delay_time)
    else:
        log('单次查询开始')
        process_from_query_start()
query_and_order.py
johnshazhu/12306
import execjs

from log.log import log


def init_station_names():
    with open('js/station_names.js', 'r', encoding='UTF-8') as f1:
        js_code = f1.read()
        context = execjs.compile(js_code)
        result = context.call("get_station_list")
        log('站点列表初始化完毕')
        return result
station.py
johnshazhu/12306
from log.log import log


class TicketResultData:
    def __init__(self, result, flag, map):
        self.result = result
        self.flag = flag
        self.map = map

    @staticmethod
    def object_hook(dict_data):
        return TicketResultData(
            dict_data['result'],
            dict_data['flag'],
            dict_data['map']
        )


class TicketResult:
    def __init__(self, httpstatus, data, messages, status):
        self.httpstatus = httpstatus
        self.data = data
        self.messages = messages
        self.status = status

    @staticmethod
    def object_hook(dict_data):
        print(type(dict_data))
        print(dict_data)
        return TicketResult(
            dict_data['httpstatus'],
            dict_data['data'],
            # json.loads(dict_data['data'], cls=TicketResultData.object_hook),
            dict_data['messages'],
            dict_data['status']
        )


def get_dj_by_seat(seat_type):
    seat_type_map = {
        '9': 'SWZ',
        'M': 'ZY',
        'O': 'ZE',

        '6': 'GR',
        'I': 'RW',
        'F': 'SRRB',
        'J': 'YW',

        '1': 'YZ',
        '2': 'RZ',
        '3': 'YW',
        '4': 'RW',
        'A': 'YYRW',
        'P': 'TZ'
    }
    return seat_type_map[seat_type]

def get_detail(result, data_map):
    keys = [
        'train_no',
        # 车次号
        'station_train_code',
        'start_station_telecode',
        'end_station_telecode',
        'from_station_telecode',
        'to_station_telecode',
        # 发车时间
        'start_time',
        # 到达时间
        'arrive_time',
        # 历时
        'lishi',
        # 是否可预订
        'canWebBuy',

        'yp_info',
        'start_train_date',
        'train_seat_feature',
        'location_code',
        'from_station_no',
        'to_station_no',
        'is_support_card',
        'controlled_train_flag',
        'gg_num',
        # 高级软卧
        'gr_num',

        # 其他
        'qt_num',
        # 软卧
        'rw_num',
        # 软座
        'rz_num',
        # 特等座?
        'tz_num',
        # 无座
        'wz_num',
        'yb_num',
        # 硬卧
        'yw_num',
        # 硬座
        'yz_num',
        # 二等座数
        'ze_num',
        # 一等座数
        'zy_num',

        # 商务座数
        'swz_num',
        # 动卧
        'srrb_num',
        'yp_ex',
        'seat_types',
        'exchange_train_flag',
        'houbu_train_flag',
        'houbu_seat_limit',
        'yp_info_new',
        'dw_flag',
        'stopcheckTime',

        'country_flag',
        'local_arrive_time',
        'local_start_time',
    ]
    result_list = []
    for item in result:
        contents = item.split('|')
        start = 0
        secret_str = contents[start]
        start += 1
        button_text_info = contents[start]
        start += 1
        child = {
            'secretStr': secret_str,
            'buttonTextInfo': button_text_info,
            'queryLeftNewDTO': {

            }
        }
        i = 0
        num_start = keys.index('gg_num')
        num_end = keys.index('srrb_num')
        while i < len(keys):
            if num_end >= i >= num_start:
                child['queryLeftNewDTO'][keys[i]] = contents[start + i]
            else:
                child['queryLeftNewDTO'][keys[i]] = contents[start + i]
            i = i + 1
        child['queryLeftNewDTO']['from_station_name'] = data_map[contents[6]],
        child['queryLeftNewDTO']['to_station_name'] = data_map[contents[7]],

        result_list.append(child)

    return result_list


def can_buy_seat(detail, seat_type):
    seat_type_filed_map = {
        '9': 'swz_num',
        'M': 'zy_num',
        'O': 'ze_num',
        'P': 'tz_num',

        '6': 'gr_num',
        'I': 'rw_num',
        # 动车的高级软卧下单确认页面就是高级动卧
        'A': 'gr_num',
        'F': 'srrb_num',
        'J': 'yw_num',

        '1': 'yz_num',
        '2': 'rz_num',
        '3': 'yw_num',
        '4': 'rw_num',
    }
    num = detail[seat_type_filed_map[seat_type]]
    log(f'can_buy_seat num = {num}')
    return num == '有' or (len(num) > 0 and num.isdigit())

ticket/query_left_ticket.py
johnshazhu/12306
import urllib.parse
import sys
import html
import re

def escape(string):
    string = string.encode('unicode_escape').decode("utf-8")
    string = urllib.parse.quote(string, safe='\\')
    string = string.upper().replace('\\U', '%u')
    return string

def unescape(string):
    string = urllib.parse.unquote(string)
    quoted = html.unescape(string).encode(sys.getfilesystemencoding()).decode('utf-8')
    return re.sub(r'%u([a-fA-F0-9]{4}|[a-fA-F0-9]{2})', lambda m: chr(int(m.group(1), 16)), quoted)
unescape.py
johnshazhu/12306
import datetime

from global_var import get_value


def is_candidate_config():
    candidate = False
    candidate_config = get_value('config_dict')['candidate']
    if candidate_config is not None and candidate_config == '1':
        candidate = True
    return candidate

def get_today_str():
    return datetime.date.today().strftime('%Y-%m-%d')

def is_success(response):
    if response is None:
        return False
    if 'result_code' in response:
        return str(response['result_code']) == '0'
    if 'status' in response:
        return response['status']
    return False


util.py
johnshazhu/12306