File: main.py

# -*- coding: utf-8 -*-
import sys

from py12306.app import *
from py12306.helpers.cdn import Cdn
from py12306.log.common_log import CommonLog
from py12306.query.query import Query
from py12306.user.user import User
from py12306.web.web import Web


def main():
    load_argvs()
    CommonLog.print_welcome()
    App.run()
    CommonLog.print_configs()
    App.did_start()

    App.run_check()
    Query.check_before_run()

    ####### 运行任务
    Web.run()
    Cdn.run()
    User.run()
    Query.run()
    if not Const.IS_TEST:
        while True:
            sleep(10000)
    else:
        if Config().is_cluster_enabled(): stay_second(5)  # 等待接受完通知
    CommonLog.print_test_complete()


def test():
    """
    功能检查
    包含：
        账号密码验证 (打码)
        座位验证
        乘客验证
        语音验证码验证
        通知验证
    :return:
    """
    Const.IS_TEST = True
    Config.OUT_PUT_LOG_TO_FILE_ENABLED = False
    if '--test-notification' in sys.argv or '-n' in sys.argv:
        Const.IS_TEST_NOTIFICATION = True
    pass


def load_argvs():
    if '--test' in sys.argv or '-t' in sys.argv: test()
    config_index = None

    if '--config' in sys.argv: config_index = sys.argv.index('--config')
    if '-c' in sys.argv: config_index = sys.argv.index('-c')
    if config_index:
        Config.CONFIG_FILE = sys.argv[config_index + 1:config_index + 2].pop()


if __name__ == '__main__':
    main()



File: py12306/config.py

# -*- coding: utf-8 -*-
import json
import re
from os import path

# 12306 账号
from py12306.helpers.func import *


@singleton
class Config:
    IS_DEBUG = False

    USER_ACCOUNTS = []
    # 查询任务
    QUERY_JOBS = []
    # 查询间隔
    QUERY_INTERVAL = 1
    # 查询重试次数
    REQUEST_MAX_RETRY = 5
    # 用户心跳检测间隔
    USER_HEARTBEAT_INTERVAL = 120
    # 多线程查询
    QUERY_JOB_THREAD_ENABLED = 0
    # 打码平台账号
    AUTO_CODE_PLATFORM = ''
    #用户打码平台地址
    API_USER_CODE_QCR_API = ''
    AUTO_CODE_ACCOUNT = {'user': '', 'pwd': ''}
    # 输出日志到文件
    OUT_PUT_LOG_TO_FILE_ENABLED = 0
    OUT_PUT_LOG_TO_FILE_PATH = 'runtime/12306.log'

    SEAT_TYPES = {'特等座': 25, '商务座': 32, '一等座': 31, '二等座': 30, '软卧': 23, '硬卧': 28, '硬座': 29, '无座': 26, }

    ORDER_SEAT_TYPES = {'特等座': 'P', '商务座': 9, '一等座': 'M', '二等座': 'O', '软卧': 4, '硬卧': 3, '硬座': 1, '无座': 1}

    PROJECT_DIR = path.dirname(path.dirname(path.abspath(__file__))) + '/'

    # Query
    RUNTIME_DIR = PROJECT_DIR + 'runtime/'
    QUERY_DATA_DIR = RUNTIME_DIR + 'query/'
    USER_DATA_DIR = RUNTIME_DIR + 'user/'
    USER_PASSENGERS_FILE = RUNTIME_DIR + 'user/%s_passengers.json'

    STATION_FILE = PROJECT_DIR + 'data/stations.txt'
    CONFIG_FILE = PROJECT_DIR + 'env.py'

    # 语音验证码
    NOTIFICATION_BY_VOICE_CODE = 0
    NOTIFICATION_VOICE_CODE_TYPE = ''
    NOTIFICATION_VOICE_CODE_PHONE = ''
    NOTIFICATION_API_APP_CODE = ''

    # 集群配置
    CLUSTER_ENABLED = 0
    NODE_SLAVE_CAN_BE_MASTER = 1
    NODE_IS_MASTER = 1
    NODE_NAME = ''
    REDIS_HOST = ''
    REDIS_PORT = '6379'
    REDIS_PASSWORD = ''

    # 钉钉配置
    DINGTALK_ENABLED = 0
    DINGTALK_WEBHOOK = ''

    # Telegram推送配置
    TELEGRAM_ENABLED = 0
    TELEGRAM_BOT_API_URL = ''

    # Bark 推送配置
    BARK_ENABLED = 0
    BARK_PUSH_URL = ''

    # ServerChan和PushBear配置
    SERVERCHAN_ENABLED = 0
    SERVERCHAN_KEY = '8474-ca071ADSFADSF'
    PUSHBEAR_ENABLED = 0
    PUSHBEAR_KEY = 'SCUdafadsfasfdafdf45234234234'

    # 邮箱配置
    EMAIL_ENABLED = 0
    EMAIL_SENDER = ''
    EMAIL_RECEIVER = ''
    EMAIL_SERVER_HOST = ''
    EMAIL_SERVER_USER = ''
    EMAIL_SERVER_PASSWORD = ''

    WEB_ENABLE = 0
    WEB_USER = {}
    WEB_PORT = 8080
    WEB_ENTER_HTML_PATH = PROJECT_DIR + 'py12306/web/static/index.html'

    # CDN
    CDN_ENABLED = 0
    CDN_CHECK_TIME_OUT = 2
    CDN_ITEM_FILE = PROJECT_DIR + 'data/cdn.txt'
    CDN_ENABLED_AVAILABLE_ITEM_FILE = QUERY_DATA_DIR + 'available.json'

    CACHE_RAIL_ID_ENABLED = 0
    RAIL_EXPIRATION = ''
    RAIL_DEVICEID = ''

    # Default time out
    TIME_OUT_OF_REQUEST = 5

    envs = []
    retry_time = 5
    last_modify_time = 0

    disallow_update_configs = [
        'CLUSTER_ENABLED',
        'NODE_IS_MASTER',
        'NODE_NAME',
        'REDIS_HOST',
        'REDIS_PORT',
        'REDIS_PASSWORD',
    ]

    def __init__(self):
        self.init_envs()
        self.last_modify_time = get_file_modify_time(self.CONFIG_FILE)
        if Config().is_slave():
            self.refresh_configs(True)
        else:
            create_thread_and_run(self, 'watch_file_change', False)

    @classmethod
    def run(cls):
        self = cls()
        self.start()

    # @classmethod
    # def keep_work(cls):
    #     self = cls()

    def start(self):
        self.save_to_remote()
        create_thread_and_run(self, 'refresh_configs', wait=Const.IS_TEST)

    def refresh_configs(self, once=False):
        if not self.is_cluster_enabled(): return
        while True:
            remote_configs = self.get_remote_config()
            self.update_configs_from_remote(remote_configs, once)
            if once or Const.IS_TEST: return
            stay_second(self.retry_time)

    def get_remote_config(self):
        if not self.is_cluster_enabled(): return
        from py12306.cluster.cluster import Cluster
        return Cluster().session.get_pickle(Cluster().KEY_CONFIGS, {})

    def save_to_remote(self):
        if not self.is_master(): return
        from py12306.cluster.cluster import Cluster
        Cluster().session.set_pickle(Cluster().KEY_CONFIGS, self.envs)

    def init_envs(self):
        self.envs = EnvLoader.load_with_file(self.CONFIG_FILE)
        self.update_configs(self.envs)

    def update_configs(self, envs):
        for key, value in envs:
            setattr(self, key, value)

    def watch_file_change(self):
        """
        监听配置文件修改
        :return:
        """
        if Config().is_slave(): return
        from py12306.log.common_log import CommonLog
        while True:
            value = get_file_modify_time(self.CONFIG_FILE)
            if value > self.last_modify_time:
                self.last_modify_time = value
                CommonLog.add_quick_log(CommonLog.MESSAGE_CONFIG_FILE_DID_CHANGED).flush()
                envs = EnvLoader.load_with_file(self.CONFIG_FILE)
                self.update_configs_from_remote(envs)
                if Config().is_master():  # 保存配置
                    self.save_to_remote()
            stay_second(self.retry_time)

    def update_configs_from_remote(self, envs, first=False):
        if envs == self.envs: return
        from py12306.query.query import Query
        from py12306.user.user import User
        from py12306.helpers.cdn import Cdn
        self.envs = envs
        for key, value in envs:
            if key in self.disallow_update_configs: continue
            if value != -1:
                old = getattr(self, key)
                setattr(self, key, value)
                if not first and old != value:
                    if key == 'USER_ACCOUNTS':
                        User().update_user_accounts(auto=True, old=old)
                    elif key == 'QUERY_JOBS':
                        Query().update_query_jobs(auto=True)  # 任务修改
                    elif key == 'QUERY_INTERVAL':
                        Query().update_query_interval(auto=True)
                    elif key == 'CDN_ENABLED':
                        Cdn().update_cdn_status(auto=True)

    @staticmethod
    def is_master():  # 是不是 主
        from py12306.cluster.cluster import Cluster
        return Config().CLUSTER_ENABLED and (Config().NODE_IS_MASTER or Cluster().is_master)

    @staticmethod
    def is_slave():  # 是不是 从
        return Config().CLUSTER_ENABLED and not Config.is_master()

    @staticmethod
    def is_cluster_enabled():
        return Config().CLUSTER_ENABLED

    @staticmethod
    def is_cdn_enabled():
        return Config().CDN_ENABLED

    @staticmethod
    def is_cache_rail_id_enabled():
        return Config().CACHE_RAIL_ID_ENABLED


class EnvLoader:
    envs = []

    def __init__(self):
        self.envs = []

    @classmethod
    def load_with_file(cls, file):
        self = cls()
        if path.exists(file):
            env_content = open(file, encoding='utf8').read()
            content = re.sub(r'^([A-Z]+)_', r'self.\1_', env_content, flags=re.M)
            exec(content)
        return self.envs

    def __setattr__(self, key, value):
        super().__setattr__(key, value)
        if re.search(r'^[A-Z]+_', key):
            self.envs.append(([key, value]))



File: py12306/__init__.py




File: py12306/app.py

# -*- coding: utf-8 -*-
import signal
import sys

from py12306.helpers.func import *
from py12306.config import Config
from py12306.helpers.notification import Notification
from py12306.log.common_log import CommonLog
from py12306.log.order_log import OrderLog


def app_available_check():
    if Config().IS_DEBUG:
        return True
    now = time_now()
    if now.weekday() == 1 and (now.hour > 23 and now.minute > 30 or now.hour < 5):
        CommonLog.add_quick_log(CommonLog.MESSAGE_12306_IS_CLOSED.format(time_now())).flush()
        open_time = datetime.datetime(now.year, now.month, now.day, 5)
        if open_time < now:
            open_time += datetime.timedelta(1)
        sleep((open_time - now).seconds)
    elif 1 < now.hour < 5:
        CommonLog.add_quick_log(CommonLog.MESSAGE_12306_IS_CLOSED.format(time_now())).flush()
        open_time = datetime.datetime(now.year, now.month, now.day, 5)
        sleep((open_time - now).seconds)
    return True


@singleton
class App:
    """
    程序主类
    TODO 代码需要优化
    """

    @classmethod
    def run(cls):
        self = cls()
        self.register_sign()
        self.start()

    def start(self):
        Config().run()
        self.init_class()

    @classmethod
    def did_start(cls):
        self = cls()
        from py12306.helpers.station import Station
        Station()  # 防止多线程时初始化出现问题
        # if Config.is_cluster_enabled():
        #     from py12306.cluster.cluster import Cluster
        #     Cluster().run()

    def init_class(self):
        from py12306.cluster.cluster import Cluster
        if Config.is_cluster_enabled():
            Cluster().run()

    def register_sign(self):
        is_windows = os.name == 'nt'
        # if is_windows:
        signs = [signal.SIGINT, signal.SIGTERM]
        # else:
        #     signs = [signal.SIGINT, signal.SIGHUP, signal.SIGTERM] # SIGHUP 会导致终端退出，程序也退出，暂时去掉
        for sign in signs:
            signal.signal(sign, self.handler_exit)

        pass

    def handler_exit(self, *args, **kwargs):
        """
        程序退出
        :param args:
        :param kwargs:
        :return:
        """
        if Config.is_cluster_enabled():
            from py12306.cluster.cluster import Cluster
            Cluster().left_cluster()

        sys.exit()

    @classmethod
    def check_auto_code(cls):
        if Config().AUTO_CODE_PLATFORM == 'free' or Config().AUTO_CODE_PLATFORM == 'user': return True
        if not Config().AUTO_CODE_ACCOUNT.get('user') or not Config().AUTO_CODE_ACCOUNT.get('pwd'):
            return False
        return True

    @classmethod
    def check_user_account_is_empty(cls):
        if Config().USER_ACCOUNTS:
            for account in Config().USER_ACCOUNTS:
                if account:
                    return False
        return True

    @staticmethod
    def check_data_dir_exists():
        os.makedirs(Config().QUERY_DATA_DIR, exist_ok=True)
        os.makedirs(Config().USER_DATA_DIR, exist_ok=True)
        touch_file(Config().OUT_PUT_LOG_TO_FILE_PATH)

    @classmethod
    def test_send_notifications(cls):
        if Config().NOTIFICATION_BY_VOICE_CODE:  # 语音通知
            CommonLog.add_quick_log(CommonLog.MESSAGE_TEST_SEND_VOICE_CODE).flush()
            if Config().NOTIFICATION_VOICE_CODE_TYPE == 'dingxin':
                voice_content = {'left_station': '广州', 'arrive_station': '深圳', 'set_type': '硬座', 'orderno': 'E123542'}
            else:
                voice_content = OrderLog.MESSAGE_ORDER_SUCCESS_NOTIFICATION_OF_VOICE_CODE_CONTENT.format('北京',
                                                                                                         '深圳')
            Notification.voice_code(Config().NOTIFICATION_VOICE_CODE_PHONE, '张三', voice_content)
        if Config().EMAIL_ENABLED:  # 邮件通知
            CommonLog.add_quick_log(CommonLog.MESSAGE_TEST_SEND_EMAIL).flush()
            Notification.send_email(Config().EMAIL_RECEIVER, '测试发送邮件', 'By py12306')

        if Config().DINGTALK_ENABLED:  # 钉钉通知
            CommonLog.add_quick_log(CommonLog.MESSAGE_TEST_SEND_DINGTALK).flush()
            Notification.dingtalk_webhook('测试发送信息')

        if Config().TELEGRAM_ENABLED:  # Telegram通知
            CommonLog.add_quick_log(CommonLog.MESSAGE_TEST_SEND_TELEGRAM).flush()
            Notification.send_to_telegram('测试发送信息')

        if Config().SERVERCHAN_ENABLED:  # ServerChan通知
            CommonLog.add_quick_log(CommonLog.MESSAGE_TEST_SEND_SERVER_CHAN).flush()
            Notification.server_chan(Config().SERVERCHAN_KEY, '测试发送消息', 'By py12306')

        if Config().PUSHBEAR_ENABLED:  # PushBear通知
            CommonLog.add_quick_log(CommonLog.MESSAGE_TEST_SEND_PUSH_BEAR).flush()
            Notification.push_bear(Config().PUSHBEAR_KEY, '测试发送消息', 'By py12306')

        if Config().BARK_ENABLED:  # Bark通知
            CommonLog.add_quick_log(CommonLog.MESSAGE_TEST_SEND_PUSH_BARK).flush()
            Notification.push_bark('测试发送信息')

    @classmethod
    def run_check(cls):
        """
        待优化
        :return:
        """
        cls.check_data_dir_exists()
        if not cls.check_user_account_is_empty():
            # CommonLog.add_quick_log(CommonLog.MESSAGE_CHECK_EMPTY_USER_ACCOUNT).flush(exit=True, publish=False) # 不填写用户则不自动下单
            if not cls.check_auto_code():
                CommonLog.add_quick_log(CommonLog.MESSAGE_CHECK_AUTO_CODE_FAIL).flush(exit=True, publish=False)
        if Const.IS_TEST_NOTIFICATION: cls.test_send_notifications()


# Expand
class Dict(dict):
    def get(self, key, default=None, sep='.'):
        keys = key.split(sep)
        for i, key in enumerate(keys):
            try:
                value = self[key]
                if len(keys[i + 1:]) and isinstance(value, Dict):
                    return value.get(sep.join(keys[i + 1:]), default=default, sep=sep)
                return value
            except:
                return self.dict_to_dict(default)

    def __getitem__(self, k):
        return self.dict_to_dict(super().__getitem__(k))

    @staticmethod
    def dict_to_dict(value):
        return Dict(value) if isinstance(value, dict) else value



File: py12306/order/order.py

import asyncio
import urllib

# from py12306.config import UserType
from pyppeteer import launch

from py12306.config import Config
from py12306.helpers.api import *
from py12306.helpers.func import *
from py12306.helpers.notification import Notification
from py12306.helpers.type import UserType, SeatType
from py12306.log.common_log import CommonLog
from py12306.log.order_log import OrderLog


class DomBounding:
    def __init__(self, rect: dict) -> None:
        super().__init__()
        self.x = rect['x']
        self.y = rect['y']
        self.width = rect['width']
        self.height = rect['height']


@singleton
class Browser:
    cookies = None
    post_data = None

    def __init__(self) -> None:
        super().__init__()

    def request_init_slide(self, session, html):
        """ 处理滑块，拿到 session_id, sig """
        OrderLog.add_quick_log('正在识别滑动验证码...').flush()
        return asyncio.get_event_loop_policy().new_event_loop().run_until_complete(
            self.__request_init_slide(session, html))

    async def __request_init_slide(self, session, html):
        """ 异步获取 """
        browser = await launch(headless=True, autoClose=True, handleSIGINT=False, handleSIGTERM=False,
                               handleSIGHUP=False)
        page = await browser.newPage()
        await page.setViewport({'width': 1200, 'height': 1080})
        await page.setRequestInterception(True)
        load_js = """() => {
            __old = navigator.userAgent; navigator.__defineGetter__('userAgent', () => __old.replace('Headless', ''));
            __old = navigator.appVersion; navigator.__defineGetter__('appVersion', () => __old.replace('Headless', ''));
            var __newProto = navigator.__proto__; delete __newProto.webdriver; navigator.__proto__ = __newProto;
        }"""
        source_url = 'https://kyfw.12306.cn/otn'
        html = html.replace('href="/otn', f'href="{source_url}').replace('src="/otn', f'src="{source_url}')

        @page.on('framenavigated')
        async def on_frame_navigated(_):
            await page.evaluate(load_js)

        @page.on('request')
        async def on_request(req):
            if req.url.startswith(API_INITDC_URL):
                if req.isNavigationRequest():
                    await page.setCookie(*session.dump_cookies())
                    return await req.respond({'body': html})
            return await req.continue_()

        await page.goto(API_INITDC_URL, timeout=30000)
        slide_btn = await page.waitForSelector('#slide_passcode .nc-lang-cnt', timeout=30000)
        rect = await slide_btn.boundingBox()
        pos = DomBounding(rect)
        pos.x += 5
        pos.y += 10
        await page.mouse.move(pos.x, pos.y)
        await page.mouse.down()
        await page.mouse.move(pos.x + pos.width, pos.y, steps=30)
        await page.mouse.up()
        # 等待获取 session id
        await page.evaluate(
            'async () => {let i = 3 * 10; while (!csessionid && i >= 0) await new Promise(resolve => setTimeout(resolve, 100), i--);}')
        ret = await page.evaluate('JSON.stringify({session_id: csessionid, sig: sig})')
        await page.close()
        await browser.close()
        return json.loads(ret)

    def request_init_slide2(self, session, data):
        """ 处理滑块，拿到 session_id, sig """
        OrderLog.add_quick_log('正在识别滑动验证码...').flush()
        return asyncio.get_event_loop_policy().new_event_loop().run_until_complete(
            self.__request_init_slide2(data))

    async def __request_init_slide2(self, data):
        from random import randint
        """ 异步获取 """
        try:
            browser = await launch(headless=True, autoClose=True, handleSIGINT=False, handleSIGTERM=False,
                                   handleSIGHUP=False, args=['--disable-infobars', '--no-sandbox'])
            page = await browser.newPage()
            await page.setViewport({'width': 1200, 'height': 1080})
            await page.setRequestInterception(True)
            load_js = """() => {
                __old = navigator.userAgent; navigator.__defineGetter__('userAgent', () => __old.replace('Headless', ''));
                __old = navigator.appVersion; navigator.__defineGetter__('appVersion', () => __old.replace('Headless', ''));
                var __newProto = navigator.__proto__; delete __newProto.webdriver; navigator.__proto__ = __newProto;
            }"""

            @page.on('framenavigated')
            async def on_frame_navigated(_):
                await page.evaluate(load_js)

            @page.on('request')
            async def on_request(req):
                if req.url == 'https://kyfw.12306.cn/passport/web/login':
                    self.post_data = req.postData
                    return await req.abort()
                return await req.continue_()

            await page.goto('https://kyfw.12306.cn/otn/resources/login.html', timeout=30000)
            await page.waitForSelector('#J-userName', timeout=30000)
            await page.type('#J-userName', data['username'], {'delay': randint(10, 30)})  # 账号
            await page.waitForSelector('#J-password', timeout=30000)
            await page.type('#J-password', data['password'], {'delay': randint(10, 30)})  # 密码
            await page.waitForSelector('#J-login', timeout=30000)
            await page.focus('#J-login')
            await page.click('#J-login')

            slide_btn = await page.waitForSelector('#nc_1_n1z', timeout=30000)
            rect = await slide_btn.boundingBox()
            pos = DomBounding(rect)
            pos.x += pos.width / 2
            pos.y += pos.height / 2
            await page.mouse.move(pos.x, pos.y)
            await page.mouse.down()
            await page.mouse.move(pos.x + pos.width * 10, pos.y, steps=30)
            await page.mouse.up()
            await page.evaluate(
            'async () => {let i = 3 * 10; while (!csessionid && i >= 0) await new Promise(resolve => setTimeout(resolve, 100), i--);}')
            await page.evaluate('JSON.stringify({session_id: csessionid, sig: sig})')
            self.cookies = await page.cookies()
            OrderLog.add_quick_log('滑动验证码识别成功').flush()
        except Exception as e:
            OrderLog.add_quick_log('滑动验证码识别失败').flush()
        try:
            await page.close()
        except:
            pass
        try:
            await browser.close()
        except:
            pass
        return self.cookies, self.post_data

class Order:
    """
    处理下单
    """
    session = None
    query_ins = None
    user_ins = None

    passenger_ticket_str = ''
    old_passenger_str = ''

    is_need_auth_code = False

    max_queue_wait = 60 * 5  # 最大排队时长
    current_queue_wait = 0
    retry_time = 3
    wait_queue_interval = 3

    order_id = 0

    notification_sustain_time = 60 * 30  # 通知持续时间 30 分钟
    notification_interval = 5 * 60  # 通知间隔

    def __init__(self, query, user):
        self.session = user.session
        from py12306.query.job import Job
        from py12306.user.job import UserJob
        assert isinstance(query, Job)
        assert isinstance(user, UserJob)
        self.query_ins = query
        self.user_ins = user
        self.is_slide = False

        self.make_passenger_ticket_str()

    def order(self):
        """
        开始下单
        下单模式  暂时不清楚，使用正常步骤下单
        :return:
        """
        # Debug
        if Config().IS_DEBUG:
            self.order_id = 'test'
            self.order_did_success()
            return random.randint(0, 10) > 7
        return self.normal_order()

    def normal_order(self):
        order_request_res = self.submit_order_request()
        if order_request_res == -1:
            return self.order_did_success()
        elif not order_request_res:
            return
        init_res, self.is_slide, init_html = self.user_ins.request_init_dc_page()
        if not init_res:
            return
        slide_info = {}
        if self.is_slide:
            try:
                slide_info = Browser().request_init_slide(self.session, init_html)
                if not slide_info.get('session_id') or not slide_info.get('sig'):
                    raise Exception()
            except Exception:
                OrderLog.add_quick_log('滑动验证码识别失败').flush()
                return
            OrderLog.add_quick_log('滑动验证码识别成功').flush()
        if not self.check_order_info(slide_info):
            return
        if not self.get_queue_count():
            return
        if not self.confirm_single_for_queue():
            return
        order_id = self.query_order_wait_time()
        if order_id:  # 发送通知
            self.order_id = order_id
            self.order_did_success()
            return True
        return False

    def order_did_success(self):
        OrderLog.print_ticket_did_ordered(self.order_id)
        OrderLog.notification(OrderLog.MESSAGE_ORDER_SUCCESS_NOTIFICATION_TITLE,
                              OrderLog.MESSAGE_ORDER_SUCCESS_NOTIFICATION_CONTENT.format(self.user_ins.user_name))
        self.send_notification()
        return True

    def send_notification(self):
        # num = 0  # 通知次数
        # sustain_time = self.notification_sustain_time
        info_message = OrderLog.get_order_success_notification_info(self.query_ins)
        normal_message = OrderLog.MESSAGE_ORDER_SUCCESS_NOTIFICATION_OF_EMAIL_CONTENT.format(self.order_id,
                                                                                             self.user_ins.user_name)
        if Config().EMAIL_ENABLED:  # 邮件通知
            Notification.send_email(Config().EMAIL_RECEIVER, OrderLog.MESSAGE_ORDER_SUCCESS_NOTIFICATION_TITLE,
                                    normal_message + info_message)
        if Config().DINGTALK_ENABLED:  # 钉钉通知
            Notification.dingtalk_webhook(normal_message + info_message)
        if Config().TELEGRAM_ENABLED:  # Telegram推送
            Notification.send_to_telegram(normal_message + info_message)
        if Config().SERVERCHAN_ENABLED:  # ServerChan通知
            Notification.server_chan(Config().SERVERCHAN_KEY, OrderLog.MESSAGE_ORDER_SUCCESS_NOTIFICATION_TITLE,
                                     normal_message + info_message)
        if Config().PUSHBEAR_ENABLED:  # PushBear通知
            Notification.push_bear(Config().PUSHBEAR_KEY, OrderLog.MESSAGE_ORDER_SUCCESS_NOTIFICATION_TITLE,
                                   normal_message + info_message)
        if Config().BARK_ENABLED:
            Notification.push_bark(normal_message + info_message)

        if Config().NOTIFICATION_BY_VOICE_CODE:  # 语音通知
            if Config().NOTIFICATION_VOICE_CODE_TYPE == 'dingxin':
                voice_info = {
                    'left_station': self.query_ins.left_station,
                    'arrive_station': self.query_ins.arrive_station,
                    'set_type': self.query_ins.current_seat_name,
                    'orderno': self.order_id
                }
            else:
                voice_info = OrderLog.MESSAGE_ORDER_SUCCESS_NOTIFICATION_OF_VOICE_CODE_CONTENT.format(
                    self.query_ins.left_station, self.query_ins.arrive_station)
            OrderLog.add_quick_log(OrderLog.MESSAGE_ORDER_SUCCESS_NOTIFICATION_OF_VOICE_CODE_START_SEND)
            Notification.voice_code(Config().NOTIFICATION_VOICE_CODE_PHONE, self.user_ins.get_name(), voice_info)
        # 取消循环发送通知
        # while sustain_time:  # TODO 后面直接查询有没有待支付的订单就可以
        #     num += 1
        #     else:
        #         break
        #     sustain_time -= self.notification_interval
        #     sleep(self.notification_interval)

        OrderLog.add_quick_log(OrderLog.MESSAGE_JOB_CLOSED).flush()
        return True

    def submit_order_request(self):
        data = {
            'secretStr': urllib.parse.unquote(self.query_ins.get_info_of_secret_str()),  # 解密
            'train_date': self.query_ins.left_date,  # 出发时间
            'back_train_date': self.query_ins.left_date,  # 返程时间
            'tour_flag': 'dc',  # 旅途类型
            'purpose_codes': 'ADULT',  # 成人 | 学生
            'query_from_station_name': self.query_ins.left_station,
            'query_to_station_name': self.query_ins.arrive_station,
        }
        response = self.session.post(API_SUBMIT_ORDER_REQUEST, data)
        result = response.json()
        if result.get('data') == '0':
            OrderLog.add_quick_log(OrderLog.MESSAGE_SUBMIT_ORDER_REQUEST_SUCCESS).flush()
            return True
        else:
            if (str(result.get('messages', '')).find('未处理') >= 0):  # 未处理订单
                # 0125 增加排队时长到 5 分钟之后，更多的是 排队失败，得通过拿到订单列表才能确认，再打个 TODO
                # self.order_id = 0  # 需要拿到订单号 TODO
                # return -1
                pass
            OrderLog.add_quick_log(
                OrderLog.MESSAGE_SUBMIT_ORDER_REQUEST_FAIL.format(
                    result.get('messages', CommonLog.MESSAGE_RESPONSE_EMPTY_ERROR))).flush()
        return False

    def check_order_info(self, slide_info=None):
        """
        cancel_flag=2
        bed_level_order_num=000000000000000000000000000000
        passengerTicketStr=
        tour_flag=dc
        randCode=
        whatsSelect=1
        _json_att=
        REPEAT_SUBMIT_TOKEN=458bf1b0a69431f34f9d2e9d3a11cfe9
        :return:
        """
        data = {  #
            'cancel_flag': 2,
            'bed_level_order_num': '000000000000000000000000000000',
            'passengerTicketStr': self.passenger_ticket_str,
            'oldPassengerStr': self.old_passenger_str,
            'tour_flag': 'dc',
            'randCode': '',
            'whatsSelect': '1',
            '_json_att': '',
            'REPEAT_SUBMIT_TOKEN': self.user_ins.global_repeat_submit_token
        }
        if self.is_slide:
            data.update({
                'sessionId': slide_info['session_id'],
                'sig': slide_info['sig'],
                'scene': 'nc_login',
            })
        response = self.session.post(API_CHECK_ORDER_INFO, data)
        result = response.json()
        if result.get('data.submitStatus'):  # 成功
            # ifShowPassCode 需要验证码
            OrderLog.add_quick_log(OrderLog.MESSAGE_CHECK_ORDER_INFO_SUCCESS).flush()
            if result.get('data.ifShowPassCode') != 'N':
                self.is_need_auth_code = True

            # if ( ticketInfoForPassengerForm.isAsync == ticket_submit_order.request_flag.isAsync & & ticketInfoForPassengerForm.queryLeftTicketRequestDTO.ypInfoDetail != "") { 不需要排队检测 js TODO
            return True
        else:
            error = CommonLog.MESSAGE_API_RESPONSE_CAN_NOT_BE_HANDLE
            if not result.get('data.isNoActive'):
                error = result.get('data.errMsg', CommonLog.MESSAGE_RESPONSE_EMPTY_ERROR)
            else:
                if result.get('data.checkSeatNum'):
                    error = '无法提交您的订单! ' + result.get('data.errMsg')
                else:
                    error = '出票失败! ' + result.get('data.errMsg')
            OrderLog.add_quick_log(OrderLog.MESSAGE_CHECK_ORDER_INFO_FAIL.format(error)).flush()
        return False

    def get_queue_count(self):
        """
        获取队列人数
        train_date	Mon Jan 01 2019 00:00:00 GMT+0800 (China Standard Time)
        train_no	630000Z12208
        stationTrainCode	Z122
        seatType	4
        fromStationTelecode	GZQ
        toStationTelecode	RXW
        leftTicket	CmDJZYrwUoJ1jFNonIgPzPFdMBvSSE8xfdUwvb2lq8CCWn%2Bzk1vM3roJaHk%3D
        purpose_codes	00
        train_location	QY
        _json_att
        REPEAT_SUBMIT_TOKEN	0977caf26f25d1da43e3213eb35ff87c
        :return:
        """
        data = {  #
            'train_date': '{} 00:00:00 GMT+0800 (China Standard Time)'.format(
                datetime.datetime.strptime(self.query_ins.left_date, '%Y-%m-%d').strftime("%a %h %d %Y")),
            'train_no': self.user_ins.ticket_info_for_passenger_form['queryLeftTicketRequestDTO']['train_no'],
            'stationTrainCode': self.user_ins.ticket_info_for_passenger_form['queryLeftTicketRequestDTO'][
                'station_train_code'],
            'seatType': self.query_ins.current_order_seat,
            'fromStationTelecode': self.user_ins.ticket_info_for_passenger_form['queryLeftTicketRequestDTO'][
                'from_station'],
            'toStationTelecode': self.user_ins.ticket_info_for_passenger_form['queryLeftTicketRequestDTO'][
                'to_station'],
            'leftTicket': self.user_ins.ticket_info_for_passenger_form['leftTicketStr'],
            'purpose_codes': self.user_ins.ticket_info_for_passenger_form['purpose_codes'],
            'train_location': self.user_ins.ticket_info_for_passenger_form['train_location'],
            '_json_att': '',
            'REPEAT_SUBMIT_TOKEN': self.user_ins.global_repeat_submit_token,
        }
        response = self.session.post(API_GET_QUEUE_COUNT, data)
        result = response.json()
        if result.get('status', False):  # 成功
            """
            "data": { 
                "count": "66",
                "ticket": "0,73", 
                "op_2": "false",
                "countT": "0",
                "op_1": "true"
            }
            
            """
            # if result.get('isRelogin') == 'Y': # 重新登录 TODO

            ticket = result.get('data.ticket').split(',')  # 余票列表
            # 这里可以判断 是真实是 硬座还是无座，避免自动分配到无座
            ticket_number = ticket[0]  # 余票
            if ticket_number != '充足' and int(ticket_number) <= 0:
                if self.query_ins.current_seat == SeatType.NO_SEAT:  # 允许无座
                    ticket_number = ticket[1]
                if not int(ticket_number):  # 跳过无座
                    OrderLog.add_quick_log(OrderLog.MESSAGE_GET_QUEUE_INFO_NO_SEAT).flush()
                    return False

            if result.get('data.op_2') == 'true':
                OrderLog.add_quick_log(OrderLog.MESSAGE_GET_QUEUE_LESS_TICKET).flush()
                return False

            current_position = int(result.get('data.countT', 0))
            OrderLog.add_quick_log(
                OrderLog.MESSAGE_GET_QUEUE_INFO_SUCCESS.format(current_position, ticket_number)).flush()
            return True
        else:
            # 加入小黑屋
            OrderLog.add_quick_log(OrderLog.MESSAGE_GET_QUEUE_COUNT_FAIL.format(
                result.get('messages', result.get('validateMessages', CommonLog.MESSAGE_RESPONSE_EMPTY_ERROR)))).flush()
        return False

    def confirm_single_for_queue(self):
        """
        确认排队
        passengerTicketStr
        oldPassengerStr
        randCode
        purpose_codes	00
        key_check_isChange	FEE6C6634A3EAA93E1E6CFC39A99E555A92E438436F18AFF78837CDB
        leftTicketStr	CmDJZYrwUoJ1jFNonIgPzPFdMBvSSE8xfdUwvb2lq8CCWn%2Bzk1vM3roJaHk%3D
        train_location	QY
        choose_seats
        seatDetailType	000
        whatsSelect	1
        roomType	00
        dwAll	N
        _json_att
        REPEAT_SUBMIT_TOKEN	0977caf26f25d1da43e3213eb35ff87c
        :return:
        """
        data = {  #
            'passengerTicketStr': self.passenger_ticket_str,
            'oldPassengerStr': self.old_passenger_str,
            'randCode': '',
            'purpose_codes': self.user_ins.ticket_info_for_passenger_form['purpose_codes'],
            'key_check_isChange': self.user_ins.ticket_info_for_passenger_form['key_check_isChange'],
            'leftTicketStr': self.user_ins.ticket_info_for_passenger_form['leftTicketStr'],
            'train_location': self.user_ins.ticket_info_for_passenger_form['train_location'],
            'choose_seats': '',
            'seatDetailType': '000',
            'whatsSelect': '1',
            'roomType': '00',
            'dwAll': 'N',
            '_json_att': '',
            'REPEAT_SUBMIT_TOKEN': self.user_ins.global_repeat_submit_token,
        }

        if self.is_need_auth_code:  # 目前好像是都不需要了，有问题再处理
            pass

        response = self.session.post(API_CONFIRM_SINGLE_FOR_QUEUE, data)
        result = response.json()

        if 'data' in result:
            """
           "data": {
                "submitStatus": true
            }
            """
            if result.get('data.submitStatus'):  # 成功
                OrderLog.add_quick_log(OrderLog.MESSAGE_CONFIRM_SINGLE_FOR_QUEUE_SUCCESS).flush()
                return True
            else:
                # 加入小黑屋 TODO
                OrderLog.add_quick_log(
                    OrderLog.MESSAGE_CONFIRM_SINGLE_FOR_QUEUE_ERROR.format(
                        result.get('data.errMsg', CommonLog.MESSAGE_RESPONSE_EMPTY_ERROR))).flush()
        else:
            OrderLog.add_quick_log(OrderLog.MESSAGE_CONFIRM_SINGLE_FOR_QUEUE_FAIL.format(
                result.get('messages', CommonLog.MESSAGE_RESPONSE_EMPTY_ERROR))).flush()
        return False

    def query_order_wait_time(self):
        """
        排队查询
        random	1546849953542
        tourFlag	dc
        _json_att
        REPEAT_SUBMIT_TOKEN	0977caf26f25d1da43e3213eb35ff87c
        :return:
        """
        self.current_queue_wait = self.max_queue_wait
        self.queue_num = 0
        while self.current_queue_wait:
            self.current_queue_wait -= self.wait_queue_interval
            self.queue_num += 1
            # TODO 取消超时订单，待优化
            data = {  #
                'random': str(random.random())[2:],
                'tourFlag': 'dc',
                '_json_att': '',
                'REPEAT_SUBMIT_TOKEN': self.user_ins.global_repeat_submit_token,
            }

            response = self.session.get(API_QUERY_ORDER_WAIT_TIME.format(urllib.parse.urlencode(data)))
            result = response.json()

            if result.get('status') and 'data' in result:
                """
               "data": {
                    "queryOrderWaitTimeStatus": true,
                    "count": 0,
                    "waitTime": -1,
                    "requestId": 6487958947291482523,
                    "waitCount": 0,
                    "tourFlag": "dc",
                    "orderId": "E222646122"
                }
                """
                result_data = result['data']
                order_id = result_data.get('orderId')
                if order_id:  # 成功
                    return order_id
                elif 'waitTime' in result_data:
                    # 计算等待时间
                    wait_time = int(result_data.get('waitTime'))
                    if wait_time == -1:  # 成功
                        # /otn/confirmPassenger/resultOrderForDcQueue 请求订单状态 目前不需要
                        # 不应该走到这
                        return order_id
                    elif wait_time == -100:  # 重新获取订单号
                        pass
                    elif wait_time >= 0:  # 等待
                        OrderLog.add_quick_log(
                            OrderLog.MESSAGE_QUERY_ORDER_WAIT_TIME_WAITING.format(result_data.get('waitCount', 0),
                                                                                  wait_time)).flush()
                    else:
                        if wait_time == -2 or wait_time == -3:  # -2 失败 -3 订单已撤销
                            OrderLog.add_quick_log(
                                OrderLog.MESSAGE_QUERY_ORDER_WAIT_TIME_FAIL.format(result_data.get('msg'))).flush()
                            return False
                        else:  # 未知原因
                            OrderLog.add_quick_log(
                                OrderLog.MESSAGE_QUERY_ORDER_WAIT_TIME_FAIL.format(
                                    result_data.get('msg', wait_time))).flush()
                            return False

                elif result_data.get('msg'):  # 失败 对不起，由于您取消次数过多，今日将不能继续受理您的订票请求。1月8日您可继续使用订票功能。
                    # TODO 需要增加判断 直接结束
                    OrderLog.add_quick_log(
                        OrderLog.MESSAGE_QUERY_ORDER_WAIT_TIME_FAIL.format(
                            result_data.get('msg', CommonLog.MESSAGE_RESPONSE_EMPTY_ERROR))).flush()
                    stay_second(self.retry_time)
                    return False
            elif result.get('messages') or result.get('validateMessages'):
                OrderLog.add_quick_log(OrderLog.MESSAGE_QUERY_ORDER_WAIT_TIME_FAIL.format(
                    result.get('messages', result.get('validateMessages')))).flush()
                return False
            else:
                pass
            OrderLog.add_quick_log(OrderLog.MESSAGE_QUERY_ORDER_WAIT_TIME_INFO.format(self.queue_num)).flush()
            stay_second(self.wait_queue_interval)

        return False

    def make_passenger_ticket_str(self):
        """
        生成提交车次的内容
        格式：
        1(seatType),0,1(车票类型:ticket_type_codes),张三(passenger_name),1(证件类型:passenger_id_type_code),xxxxxx(passenger_id_no),xxxx(mobile_no),N
        passengerTicketStr:
        张三(passenger_name),1(证件类型:passenger_id_type_code),xxxxxx(passenger_id_no),1_
        oldPassengerStr
        :return:
        """
        passenger_tickets = []
        old_passengers = []
        available_passengers = self.query_ins.passengers
        if len(available_passengers) > self.query_ins.member_num_take:  # 删除人数
            available_passengers = available_passengers[0:self.query_ins.member_num_take]
            OrderLog.print_passenger_did_deleted(available_passengers)

        for passenger in available_passengers:
            tmp_str = '{seat_type},0,{passenger_type},{passenger_name},{passenger_id_card_type},{passenger_id_card},{passenger_mobile},N,{enc_str}_'.format(
                seat_type=self.query_ins.current_order_seat, passenger_type=passenger['type'],
                passenger_name=passenger['name'],
                passenger_id_card_type=passenger['id_card_type'], passenger_id_card=passenger['id_card'],
                passenger_mobile=passenger['mobile'],
                enc_str=passenger['enc_str'],
            )
            passenger_tickets.append(tmp_str)

            if int(passenger['type']) != UserType.CHILD:
                tmp_old_str = '{passenger_name},{passenger_id_card_type},{passenger_id_card},{passenger_type}_'.format(
                    passenger_name=passenger['name'],
                    passenger_id_card_type=passenger['id_card_type'], passenger_id_card=passenger['id_card'],
                    passenger_type=passenger['type'],
                )
                old_passengers.append(tmp_old_str)

        self.passenger_ticket_str = ''.join(passenger_tickets).rstrip('_')
        self.old_passenger_str = ''.join(old_passengers).rstrip('_') + '__ _ _'  # 不加后面请求会出错



File: py12306/cluster/__init__.py




File: py12306/cluster/redis.py

import json
import pickle

import redis

from py12306.config import Config
from py12306.helpers.func import *
from py12306.log.redis_log import RedisLog
from redis import Redis as PyRedis


@singleton
class Redis(PyRedis):
    # session = None

    def __init__(self, *args):
        if Config.is_cluster_enabled():
            args = {
                'host': Config().REDIS_HOST,
                'port': Config().REDIS_PORT,
                'db': 0,
                'password': Config().REDIS_PASSWORD,
                'decode_responses': True
            }
            super().__init__(**args)
            RedisLog.add_quick_log(RedisLog.MESSAGE_REDIS_INIT_SUCCESS)
        else:
            super().__init__(**args)
        return self

    def get(self, name, default=None):
        res = super().get(name)
        # if decode: res = res.decode()
        return res if res else default

    def set(self, name, value, ex=None, px=None, nx=False, xx=False):
        return super().set(name, available_value(value), ex=ex, px=px, nx=nx, xx=xx)

    def set_dict(self, name, value):
        return self.set_pickle(name, value)
        # return self.set(name, json.dumps(value))

    def get_dict(self, name, default={}):
        return self.get_pickle(name, default)
        # res = self.get(name)
        # if res:
        #     return json.loads(res)
        # return default

    def set_pickle(self, name, value):
        return self.set(name, pickle.dumps(value, 0).decode())

    def get_pickle(self, name, default=None):
        res = self.get(name)
        return pickle.loads(res.encode()) if res else default

    # def smembers(self, name, default=[]):
    #     res = super().smembers(name)
    #     return [val.decode() for val in list(res)] if res else default



File: py12306/cluster/cluster.py

import json
import os
import pickle
import sys
import time

import redis
from redis.client import PubSub

from py12306.cluster.redis import Redis
from py12306.config import Config
from py12306.helpers.func import *
from py12306.log.cluster_log import ClusterLog


@singleton
class Cluster():
    KEY_PREFIX = 'py12306_'  # 目前只能手动
    KEY_QUERY_COUNT = KEY_PREFIX + 'query_count'
    KEY_QUERY_LAST_TIME = KEY_PREFIX + 'query_last_time'
    KEY_CONFIGS = KEY_PREFIX + 'configs'
    KEY_NODES = KEY_PREFIX + 'nodes'
    KEY_CHANNEL_LOG = KEY_PREFIX + 'channel_log'
    KEY_CHANNEL_EVENT = KEY_PREFIX + 'channel_even'
    KEY_USER_COOKIES = KEY_PREFIX + 'user_cookies'
    KEY_USER_INFOS = KEY_PREFIX + 'user_infos'
    KEY_USER_LAST_HEARTBEAT = KEY_PREFIX + 'user_last_heartbeat'
    KEY_NODES_ALIVE_PREFIX = KEY_PREFIX + 'nodes_alive_'

    KEY_CDN_AVAILABLE_ITEMS = KEY_PREFIX + 'cdn_available_items'
    KEY_CDN_LAST_CHECK_AT = KEY_PREFIX + 'cdn_last_check_at'

    # 锁
    KEY_LOCK_INIT_USER = KEY_PREFIX + 'lock_init_user'  # 暂未使用
    KEY_LOCK_DO_ORDER = KEY_PREFIX + 'lock_do_order'  # 订单锁
    lock_do_order_time = 60 * 1  # 订单锁超时时间

    lock_prefix = KEY_PREFIX + 'lock_'  # 锁键前缀
    lock_info_prefix = KEY_PREFIX + 'info_'

    KEY_MASTER = 1
    KEY_SLAVE = 0

    session: Redis = None
    pubsub: PubSub = None
    refresh_channel_time = 0.5
    retry_time = 2
    keep_alive_time = 3  # 报告存活间隔
    lost_alive_time = keep_alive_time * 2

    nodes = {}
    node_name = None
    is_ready = False
    is_master = False

    def __init__(self, *args):
        if Config.is_cluster_enabled():
            self.session = Redis()
        return self

    @classmethod
    def run(cls):
        self = cls()
        self.start()

    def start(self):
        self.pubsub = self.session.pubsub()
        self.pubsub.subscribe(self.KEY_CHANNEL_LOG, self.KEY_CHANNEL_EVENT)
        create_thread_and_run(self, 'subscribe', wait=False)
        self.is_ready = True
        self.get_nodes()  # 提前获取节点列表
        self.check_nodes()  # 防止 节点列表未清空
        self.join_cluster()
        create_thread_and_run(self, 'keep_alive', wait=False)
        create_thread_and_run(self, 'refresh_data', wait=False)

    def join_cluster(self):
        """
        加入到集群
        :return:
        """
        self.node_name = node_name = Config().NODE_NAME

        if Config().NODE_IS_MASTER:
            if self.node_name in self.nodes:  # 重复运行主节点
                ClusterLog.add_quick_log(ClusterLog.MESSAGE_MASTER_NODE_ALREADY_RUN.format(node_name)).flush(
                    publish=False)
                os._exit(1)
            if self.have_master():  # 子节点提升为主节点情况，交回控制
                message = ClusterLog.MESSAGE_NODE_BECOME_MASTER_AGAIN.format(node_name)
                self.publish_log_message(message)
                self.make_nodes_as_slave()
        elif not self.have_master():  # 只能通过主节点启动
            ClusterLog.add_quick_log(ClusterLog.MESSAGE_MASTER_NODE_NOT_FOUND).flush(publish=False)
            os._exit(1)

        if node_name in self.nodes:
            self.node_name = node_name = node_name + '_' + str(dict_count_key_num(self.nodes, node_name))
            ClusterLog.add_quick_log(ClusterLog.MESSAGE_NODE_ALREADY_IN_CLUSTER.format(node_name)).flush()

        self.session.hset(self.KEY_NODES, node_name, Config().NODE_IS_MASTER)
        message = ClusterLog.MESSAGE_JOIN_CLUSTER_SUCCESS.format(self.node_name, ClusterLog.get_print_nodes(
            self.get_nodes()))  # 手动 get nodes
        self.publish_log_message(message)

    def left_cluster(self, node_name=None):
        node_name = node_name if node_name else self.node_name
        self.session.hdel(self.KEY_NODES, node_name)
        message = ClusterLog.MESSAGE_LEFT_CLUSTER.format(node_name, ClusterLog.get_print_nodes(self.get_nodes()))
        self.publish_log_message(message, node_name)

    def make_nodes_as_slave(self):
        """
        将所有节点设为主节点
        :return:
        """
        for node in self.nodes:
            self.session.hset(self.KEY_NODES, node, self.KEY_SLAVE)

    def publish_log_message(self, message, node_name=None):
        """
        发布订阅消息
        :return:
        """
        node_name = node_name if node_name else self.node_name
        message = ClusterLog.MESSAGE_SUBSCRIBE_NOTIFICATION.format(node_name, message)
        self.session.publish(self.KEY_CHANNEL_LOG, message)

    def publish_event(self, name, data={}):
        """
        发布事件消息
        :return:
        """
        data = {'event': name, 'data': data}
        self.session.publish(self.KEY_CHANNEL_EVENT, json.dumps(data))

    def get_nodes(self) -> dict:
        res = self.session.hgetall(self.KEY_NODES)
        res = res if res else {}
        self.nodes = res
        return res

    def refresh_data(self):
        """
        单独进程处理数据同步
        :return:
        """
        while True:
            self.get_nodes()
            self.check_locks()
            self.check_nodes()
            self.check_master()
            stay_second(self.retry_time)

    def check_master(self):
        """
        检测主节点是否可用
        :return:
        """
        master = self.have_master()
        if master == self.node_name:  # 动态提升
            self.is_master = True
        else:
            self.is_master = False

        if not master:
            if Config().NODE_SLAVE_CAN_BE_MASTER:
                # 提升子节点为主节点
                slave = list(self.nodes)[0]
                self.session.hset(self.KEY_NODES, slave, self.KEY_MASTER)
                self.publish_log_message(ClusterLog.MESSAGE_ASCENDING_MASTER_NODE.format(slave,
                                                                                         ClusterLog.get_print_nodes(
                                                                                             self.get_nodes())))
                return True
            else:
                self.publish_log_message(ClusterLog.MESSAGE_MASTER_DID_LOST.format(self.retry_time))
                stay_second(self.retry_time)
                os._exit(1)  # 退出整个程序

    def have_master(self):
        return dict_find_key_by_value(self.nodes, str(self.KEY_MASTER), False)

    def check_nodes(self):
        """
        检查节点是否存活
        :return:
        """
        for node in self.nodes:
            if not self.session.exists(self.KEY_NODES_ALIVE_PREFIX + node):
                self.left_cluster(node)

    # def kick_out_from_nodes(self, node_name):
    #     pass

    def keep_alive(self):
        while True:
            if self.node_name not in self.get_nodes():  # 已经被 kict out  重新加下
                self.join_cluster()
            self.session.set(self.KEY_NODES_ALIVE_PREFIX + self.node_name, Config().NODE_IS_MASTER, ex=self.lost_alive_time)
            stay_second(self.keep_alive_time)

    def subscribe(self):
        while True:
            try:
                message = self.pubsub.get_message()
            except RuntimeError as err:
                if 'args' in dir(err) and err.args[0].find('pubsub connection not set') >= 0:  # 失去重连
                    self.pubsub.subscribe(self.KEY_CHANNEL_LOG, self.KEY_CHANNEL_EVENT)
                    continue
            if message:
                if message.get('type') == 'message' and message.get('channel') == self.KEY_CHANNEL_LOG and message.get(
                        'data'):
                    msg = message.get('data')
                    if self.node_name:
                        msg = msg.replace(ClusterLog.MESSAGE_SUBSCRIBE_NOTIFICATION_PREFIX.format(self.node_name), '')
                    ClusterLog.add_quick_log(msg).flush(publish=False)
                elif message.get('channel') == self.KEY_CHANNEL_EVENT:
                    create_thread_and_run(self, 'handle_events', args=(message,))
            stay_second(self.refresh_channel_time)

    def handle_events(self, message):
        # 这里应该分开处理，先都在这处理了
        if message.get('type') != 'message': return
        result = json.loads(message.get('data', {}))
        event_name = result.get('event')
        data = result.get('data')
        from py12306.helpers.event import Event
        method = getattr(Event(), event_name)
        if method:
            create_thread_and_run(Event(), event_name, Const.IS_TEST, kwargs={'data': data, 'callback': True})

    def get_lock(self, key: str, timeout=1, info={}):
        timeout = int(time.time()) + timeout
        res = self.session.setnx(key, timeout)
        if res:
            if info: self.session.set_dict(self.lock_info_prefix + key.replace(self.KEY_PREFIX, ''), info)  # 存储额外信息
            return True
        return False

    def get_lock_info(self, key, default={}):
        return self.session.get_dict(self.lock_info_prefix + key.replace(self.KEY_PREFIX, ''), default=default)

    def release_lock(self, key):
        self.session.delete(key)
        self.session.delete(self.lock_info_prefix + key.replace(self.KEY_PREFIX, ''))

    def check_locks(self):
        locks = self.session.keys(self.lock_prefix + '*')
        for key in locks:
            val = self.session.get(key)
            if val and int(val) <= time_int():
                self.release_lock(key)

    @classmethod
    def get_user_cookie(cls, key, default=None):
        self = cls()
        res = self.session.hget(Cluster.KEY_USER_COOKIES, key)
        return pickle.loads(res.encode()) if res else default

    @classmethod
    def set_user_cookie(cls, key, value):
        self = cls()
        return self.session.hset(Cluster.KEY_USER_COOKIES, key, pickle.dumps(value, 0).decode())

    @classmethod
    def set_user_info(cls, key, info):
        self = cls()
        return self.session.hset(Cluster.KEY_USER_INFOS, key, pickle.dumps(info, 0).decode())

    @classmethod
    def get_user_info(cls, key, default=None):
        self = cls()
        res = self.session.hget(Cluster.KEY_USER_INFOS, key)
        return pickle.loads(res.encode()) if res else default



File: py12306/web/web.py

# -*- coding: utf-8 -*-
import json
import logging
from datetime import timedelta

from flask import Flask, request
from flask_jwt_extended import (
    JWTManager)

from py12306.config import Config
from py12306.helpers.func import *


@singleton
class Web:
    session = None
    jwt = None
    log = None

    def __init__(self):
        self.session = Flask(__name__)
        self.log = logging.getLogger('werkzeug')
        self.log.setLevel(logging.ERROR)

        self.register_blueprint()
        self.session.config['JWT_SECRET_KEY'] = 'secret'  # 目前都是本地，暂不用放配置文件
        self.session.config['JWT_ACCESS_TOKEN_EXPIRES'] = timedelta(seconds=60 * 60 * 24 * 7)  # Token 超时时间 7 天
        self.jwt = JWTManager(self.session)

    def register_blueprint(self):
        from py12306.web.handler.user import user
        from py12306.web.handler.stat import stat
        from py12306.web.handler.app import app
        from py12306.web.handler.query import query
        from py12306.web.handler.log import log
        self.session.register_blueprint(user)
        self.session.register_blueprint(stat)
        self.session.register_blueprint(app)
        self.session.register_blueprint(query)
        self.session.register_blueprint(log)

    @classmethod
    def run(cls):
        self = cls()
        self.start()

    def start(self):
        if not Config().WEB_ENABLE or Config().is_slave(): return
        # if Config().IS_DEBUG:
        #     self.run_session()
        # else:
        create_thread_and_run(self, 'run_session', wait=False)

    def run_session(self):
        debug = False
        if is_main_thread():
            debug = Config().IS_DEBUG
        self.session.run(debug=debug, port=Config().WEB_PORT, host='0.0.0.0')


if __name__ == '__main__':
    Web.run()



File: py12306/web/__init__.py




File: py12306/web/handler/user.py

from flask import Blueprint, request
from flask.json import jsonify
from flask_jwt_extended import (jwt_required, create_access_token)

from py12306.config import Config
from py12306.helpers.func import str_to_time, timestamp_to_time
from py12306.user.job import UserJob
from py12306.user.user import User

user = Blueprint('user', __name__)


@user.route('/login', methods=['POST'])
def login():
    """
    用户登录
    :return:
    """
    username = request.json.get('username', None)
    password = request.json.get('password', None)
    if username and password and username == Config().WEB_USER.get('username') and password == Config().WEB_USER.get(
            'password'):
        access_token = create_access_token(identity=username)
        return jsonify(access_token=access_token)
    return jsonify({"msg": "用户名或密码错误"}), 422


@user.route('/users', methods=['GET'])
@jwt_required
def users():
    """
    用户任务列表
    :return:
    """
    jobs = User().users
    result = list(map(convert_job_to_info, jobs))
    return jsonify(result)


@user.route('/user/info', methods=['GET'])
@jwt_required
def user_info():
    """
    获取用户信息
    :return:
    """
    result = {
        'name': Config().WEB_USER.get('username')
    }
    return jsonify(result)


def convert_job_to_info(job: UserJob):
    return {
        'key': job.key,
        'user_name': job.user_name,
        'name': job.get_name(),
        'is_ready': job.is_ready,
        'is_loaded': job.user_loaded,  # 是否成功加载 ready 是当前是否可用
        'last_heartbeat': timestamp_to_time(job.last_heartbeat) if job.last_heartbeat else '-',
        'login_num': job.login_num
    }



File: py12306/web/handler/query.py

from flask import Blueprint, request
from flask.json import jsonify
from flask_jwt_extended import (jwt_required)

from py12306.config import Config
from py12306.query.job import Job
from py12306.query.query import Query

query = Blueprint('query', __name__)


@query.route('/query', methods=['GET'])
@jwt_required
def query_lists():
    """
    查询任务列表
    :return:
    """
    jobs = Query().jobs
    result = list(map(convert_job_to_info, jobs))
    return jsonify(result)


def convert_job_to_info(job: Job):
    return {
        'name': job.job_name,
        'left_dates': job.left_dates,
        'stations': job.stations,
        'members': job.members,
        'member_num': job.member_num,
        'allow_seats': job.allow_seats,
        'allow_train_numbers': job.allow_train_numbers,
        'except_train_numbers': job.except_train_numbers,
        'allow_less_member': job.allow_less_member,
        'passengers': job.passengers,
    }



File: py12306/web/handler/log.py

import linecache

from flask import Blueprint, request
from flask.json import jsonify
from flask_jwt_extended import (jwt_required)

from py12306.config import Config
from py12306.helpers.func import get_file_total_line_num, pick_file_lines
from py12306.log.common_log import CommonLog
from py12306.query.query import Query
from py12306.user.user import User

log = Blueprint('log', __name__)


@log.route('/log/output', methods=['GET'])
@jwt_required
def log_output():
    """
    日志
    :return:
    """
    last_line = int(request.args.get('line', 0))
    limit = int(request.args.get('limit', 10))
    max_old = 200  # 取最新时 往后再取的数
    file = Config().OUT_PUT_LOG_TO_FILE_PATH
    res = []

    if last_line == -1:
        total_line = get_file_total_line_num(file)
        last_line = total_line - max_old if total_line > max_old else 0
        ranges = range(last_line, last_line + max_old + limit)
        # limit = max_old + limit
    else:
        ranges = range(last_line, last_line + limit)

    if Config().OUT_PUT_LOG_TO_FILE_ENABLED:
        with open(Config().OUT_PUT_LOG_TO_FILE_PATH, 'r', encoding='utf-8') as f:
            res = pick_file_lines(f, ranges)

        # linecache.updatecache(file) # 使用 linecache windows 平台会出来编码问题 暂时弃用
        # for i in ranges:
        #     tmp = linecache.getline(file, last_line + i)
        #     if tmp != '': res.append(tmp)
        last_line += len(res)
    else:
        res = CommonLog.MESSAGE_OUTPUT_TO_FILE_IS_UN_ENABLE
    return jsonify({
        'last_line': last_line,
        'data': res
    })



File: py12306/web/handler/__init__.py




File: py12306/web/handler/stat.py

from flask import Blueprint, request
from flask.json import jsonify
from flask_jwt_extended import (jwt_required)

from py12306.config import Config
from py12306.query.query import Query
from py12306.user.user import User

stat = Blueprint('stat', __name__)


@stat.route('/stat/dashboard', methods=['GET'])
@jwt_required
def dashboard():
    """
    状态统计
    任务数量，用户数量，查询次数
    节点信息（TODO）
    :return:
    """
    from py12306.log.query_log import QueryLog
    query_job_count = len(Query().jobs)
    user_job_count = len(User().users)
    query_count = QueryLog().data.get('query_count')
    res = {
        'query_job_count': query_job_count,
        'user_job_count': user_job_count,
        'query_count': query_count,
    }
    if Config().CDN_ENABLED:
        from py12306.helpers.cdn import Cdn
        res['cdn_count'] = len(Cdn().available_items)
    return jsonify(res)


@stat.route('/stat/cluster', methods=['GET'])
@jwt_required
def clusters():
    """
    节点统计
    节点数量，主节点，子节点列表
    :return:
    """
    from py12306.cluster.cluster import Cluster
    nodes = Cluster().nodes
    count = len(nodes)
    node_lists = list(nodes)
    master = [key for key, val in nodes.items() if int(val) == Cluster.KEY_MASTER]
    master = master[0] if master else ''

    return jsonify({
        'master': master,
        'count': count,
        'node_lists': ', '.join(node_lists)
    })



File: py12306/web/handler/app.py

import json
import re

from flask import Blueprint, request, send_file
from flask.json import jsonify
from flask_jwt_extended import (jwt_required)

from py12306.config import Config
from py12306.query.query import Query
from py12306.user.user import User

app = Blueprint('app', __name__)


@app.route('/', methods=['GET', 'POST'])
def index():
    file = Config().WEB_ENTER_HTML_PATH
    result = ''
    with open(file, 'r', encoding='utf-8') as f:
        result = f.read()
        config = {
            'API_BASE_URL': ''  # TODO 自定义 Host
        }
        result = re.sub(r'<script>[\s\S]*?<\/script>', '<script>window.config={}</script>'.format(json.dumps(config)),
                        result)

    return result


@app.route('/app/menus', methods=['GET'])
@jwt_required
def menus():
    """
    菜单列表
    """
    menus = [
        {"id": 10, "name": "首页", "url": "/", "icon": "fa fa-tachometer-alt"},
        {"id": 20, "name": "用户管理", "url": "/user", "icon": "fa fa-user"},
        {"id": 30, "name": "查询任务", "url": "/query", "icon": "fa fa-infinity"},
        {"id": 40, "name": "实时日志", "url": "/log/realtime", "icon": "fa fa-signature"},
        {"id": 50, "name": "帮助", "url": "/help", "icon": "fa fa-search"}
    ]
    return jsonify(menus)


@app.route('/app/actions', methods=['GET'])
@jwt_required
def actions():
    """
    操作列表
    """
    actions = [
        {"text": "退出登录", "key": 'logout', "link": "", "icon": "fa fa-sign-out-alt"}
    ]
    return jsonify(actions)



File: py12306/user/user.py

from py12306.app import *
from py12306.cluster.cluster import Cluster
from py12306.helpers.event import Event
from py12306.helpers.func import *
from py12306.log.user_log import UserLog
from py12306.user.job import UserJob


@singleton
class User:
    users = []
    user_accounts = []

    retry_time = 3
    cluster = None

    def __init__(self):
        self.cluster = Cluster()
        self.update_interval()
        self.update_user_accounts()

    def update_user_accounts(self, auto=False, old=None):
        self.user_accounts = Config().USER_ACCOUNTS
        if auto:
            UserLog.add_quick_log(UserLog.MESSAGE_USERS_DID_CHANGED).flush()
            self.refresh_users(old)

    def update_interval(self, auto=False):
        self.interval = Config().USER_HEARTBEAT_INTERVAL
        if auto: jobs_do(self.users, 'update_user')

    @classmethod
    def run(cls):
        self = cls()
        # app_available_check() 用户系统不休息
        self.start()
        pass

    def start(self):
        self.init_users()
        UserLog.print_init_users(users=self.users)
        # 多线程维护用户
        create_thread_and_run(jobs=self.users, callback_name='run', wait=Const.IS_TEST)

    def init_users(self):
        for account in self.user_accounts:
            self.init_user(account)

    def init_user(self, info):
        user = UserJob(info=info)
        self.users.append(user)
        return user

    def refresh_users(self, old):
        for account in self.user_accounts:
            key = account.get('key')
            old_account = array_dict_find_by_key_value(old, 'key', key)
            if old_account and account != old_account:
                user = self.get_user(key)
                user.init_data(account)
            elif not old_account:  # 新用户 添加到 多线程
                new_user = self.init_user(account)
                create_thread_and_run(jobs=new_user, callback_name='run', wait=Const.IS_TEST)

        for account in old:  # 退出已删除的用户
            if not array_dict_find_by_key_value(self.user_accounts, 'key', account.get('key')):
                Event().user_job_destroy({'key': account.get('key')})

    @classmethod
    def is_empty(cls):
        self = cls()
        return not bool(self.users)

    @classmethod
    def get_user(cls, key) -> UserJob:
        self = cls()
        for user in self.users:
            if user.key == key: return user
        return None

    @classmethod
    def get_passenger_for_members(cls, members, key):
        """
        检测乘客信息
        :param passengers
        :return:
        """
        self = cls()

        for user in self.users:
            assert isinstance(user, UserJob)
            if user.key == key and user.wait_for_ready():
                return user.get_passengers_by_members(members)



File: py12306/user/job.py

import base64
import pickle
import re
from os import path

from py12306.cluster.cluster import Cluster
from py12306.helpers.api import *
from py12306.app import *
from py12306.helpers.auth_code import AuthCode
from py12306.helpers.event import Event
from py12306.helpers.func import *
from py12306.helpers.request import Request
from py12306.helpers.type import UserType
from py12306.helpers.qrcode import print_qrcode
from py12306.log.order_log import OrderLog
from py12306.log.user_log import UserLog
from py12306.log.common_log import CommonLog
from py12306.order.order import Browser


class UserJob:
    # heartbeat = 60 * 2  # 心跳保持时长
    is_alive = True
    check_interval = 5
    key = None
    user_name = ''
    password = ''
    type = 'qr'
    user = None
    info = {}  # 用户信息
    last_heartbeat = None
    is_ready = False
    user_loaded = False  # 用户是否已加载成功
    passengers = []
    retry_time = 3
    retry_count = 0
    login_num = 0  # 尝试登录次数
    sleep_interval = {'min': 0.1, 'max': 5}

    # Init page
    global_repeat_submit_token = None
    ticket_info_for_passenger_form = None
    order_request_dto = None

    cluster = None
    lock_init_user_time = 3 * 60
    cookie = False

    def __init__(self, info):
        self.cluster = Cluster()
        self.init_data(info)

    def init_data(self, info):
        self.session = Request()
        self.session.add_response_hook(self.response_login_check)
        self.key = str(info.get('key'))
        self.user_name = info.get('user_name')
        self.password = info.get('password')
        self.type = info.get('type')

    def update_user(self):
        from py12306.user.user import User
        self.user = User()
        self.load_user()

    def run(self):
        # load user
        self.update_user()
        self.start()

    def start(self):
        """
        检测心跳
        :return:
        """
        while True and self.is_alive:
            app_available_check()
            if Config().is_slave():
                self.load_user_from_remote()
            else:
                if Config().is_master() and not self.cookie: self.load_user_from_remote()  # 主节点加载一次 Cookie
                self.check_heartbeat()
            if Const.IS_TEST: return
            stay_second(self.check_interval)

    def check_heartbeat(self):
        # 心跳检测
        if self.get_last_heartbeat() and (time_int() - self.get_last_heartbeat()) < Config().USER_HEARTBEAT_INTERVAL:
            return True
        # 只有主节点才能走到这
        if self.is_first_time() or not self.check_user_is_login() or not self.can_access_passengers():
            if not self.load_user() and not self.handle_login(): return

        self.user_did_load()
        message = UserLog.MESSAGE_USER_HEARTBEAT_NORMAL.format(self.get_name(), Config().USER_HEARTBEAT_INTERVAL)
        UserLog.add_quick_log(message).flush()

    def get_last_heartbeat(self):
        if Config().is_cluster_enabled():
            return int(self.cluster.session.get(Cluster.KEY_USER_LAST_HEARTBEAT, 0))

        return self.last_heartbeat

    def set_last_heartbeat(self, time=None):
        time = time if time != None else time_int()
        if Config().is_cluster_enabled():
            self.cluster.session.set(Cluster.KEY_USER_LAST_HEARTBEAT, time)
        self.last_heartbeat = time

    # def init_cookies
    def is_first_time(self):
        if Config().is_cluster_enabled():
            return not self.cluster.get_user_cookie(self.key)
        return not path.exists(self.get_cookie_path())

    def handle_login(self, expire=False):
        if expire: UserLog.print_user_expired()
        self.is_ready = False
        UserLog.print_start_login(user=self)
        if self.type == 'qr':
            return self.qr_login()
        else:
            return self.login2()

    def login(self):
        """
        获取验证码结果
        :return 权限校验码
        """
        data = {
            'username': self.user_name,
            'password': self.password,
            'appid': 'otn'
        }
        answer = AuthCode.get_auth_code(self.session)
        data['answer'] = answer
        self.request_device_id()
        response = self.session.post(API_BASE_LOGIN.get('url'), data)
        result = response.json()
        if result.get('result_code') == 0:  # 登录成功
            """
            login 获得 cookie uamtk
            auth/uamtk      不请求，会返回 uamtk票据内容为空
            /otn/uamauthclient 能拿到用户名
            """
            new_tk = self.auth_uamtk()
            user_name = self.auth_uamauthclient(new_tk)
            self.update_user_info({'user_name': user_name})
            self.login_did_success()
            return True
        elif result.get('result_code') == 2:  # 账号之内错误
            # 登录失败，用户名或密码为空
            # 密码输入错误
            UserLog.add_quick_log(UserLog.MESSAGE_LOGIN_FAIL.format(result.get('result_message'))).flush()
        else:
            UserLog.add_quick_log(
                UserLog.MESSAGE_LOGIN_FAIL.format(result.get('result_message', result.get('message',
                                                                                          CommonLog.MESSAGE_RESPONSE_EMPTY_ERROR)))).flush()

        return False

    def qr_login(self):
        self.request_device_id()
        image_uuid, png_path = self.download_code()
        last_time = time_int()
        while True:
            data = {
                'RAIL_DEVICEID': self.session.cookies.get('RAIL_DEVICEID'),
                'RAIL_EXPIRATION': self.session.cookies.get('RAIL_EXPIRATION'),
                'uuid': image_uuid,
                'appid': 'otn'
            }
            response = self.session.post(API_AUTH_QRCODE_CHECK.get('url'), data)
            result = response.json()
            try:
                result_code = int(result.get('result_code'))
            except:
                if time_int() - last_time > 300:
                    last_time = time_int()
                    image_uuid, png_path = self.download_code()
                continue
            if result_code == 0:
                time.sleep(get_interval_num(self.sleep_interval))
            elif result_code == 1:
                UserLog.add_quick_log('请确认登录').flush()
                time.sleep(get_interval_num(self.sleep_interval))
            elif result_code == 2:
                break
            elif result_code == 3:
                try:
                    os.remove(png_path)
                except Exception as e:
                    UserLog.add_quick_log('无法删除文件: {}'.format(e)).flush()
                image_uuid, png_path = self.download_code()
            if time_int() - last_time > 300:
                last_time = time_int()
                image_uuid, png_path = self.download_code()
        try:
            os.remove(png_path)
        except Exception as e:
            UserLog.add_quick_log('无法删除文件: {}'.format(e)).flush()

        self.session.get(API_USER_LOGIN, allow_redirects=True)
        new_tk = self.auth_uamtk()
        user_name = self.auth_uamauthclient(new_tk)
        self.update_user_info({'user_name': user_name})
        self.session.get(API_USER_LOGIN, allow_redirects=True)
        self.login_did_success()
        return True

    def login2(self):
        data = {
            'username': self.user_name,
            'password': self.password,
        }
        headers = {
                    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/94.0.4606.61 Safari/537.36",
                }
        self.session.headers.update(headers)
        cookies, post_data = Browser().request_init_slide2(self.session, data)
        while not cookies or not post_data:
            cookies, post_data = Browser().request_init_slide2(self.session, data)
        for cookie in cookies:
            self.session.cookies.update({
                   cookie['name']: cookie['value']
            })
        response = self.session.post(API_BASE_LOGIN.get('url')+ '?' + post_data)
        result = response.json()
        if result.get('result_code') == 0:  # 登录成功
            """
            login 获得 cookie uamtk
            auth/uamtk      不请求，会返回 uamtk票据内容为空
            /otn/uamauthclient 能拿到用户名
            """
            new_tk = self.auth_uamtk()
            user_name = self.auth_uamauthclient(new_tk)
            self.update_user_info({'user_name': user_name})
            self.login_did_success()
            return True
        elif result.get('result_code') == 2:  # 账号之内错误
            # 登录失败，用户名或密码为空
            # 密码输入错误
            UserLog.add_quick_log(UserLog.MESSAGE_LOGIN_FAIL.format(result.get('result_message'))).flush()
        else:
            UserLog.add_quick_log(
                UserLog.MESSAGE_LOGIN_FAIL.format(result.get('result_message', result.get('message',
                                                                                          CommonLog.MESSAGE_RESPONSE_EMPTY_ERROR)))).flush()

        return False

    def download_code(self):
        try:
            UserLog.add_quick_log(UserLog.MESSAGE_QRCODE_DOWNLOADING).flush()
            response = self.session.post(API_AUTH_QRCODE_BASE64_DOWNLOAD.get('url'), data={'appid': 'otn'})
            result = response.json()
            if result.get('result_code') == '0':
                img_bytes = base64.b64decode(result.get('image'))
                try:
                    os.mkdir(Config().USER_DATA_DIR + '/qrcode')
                except FileExistsError:
                    pass
                png_path = path.normpath(Config().USER_DATA_DIR + '/qrcode/%d.png' % time.time())
                with open(png_path, 'wb') as file:
                    file.write(img_bytes)
                    file.close()
                if os.name == 'nt':
                    os.startfile(png_path)
                else:
                    print_qrcode(png_path)
                UserLog.add_log(UserLog.MESSAGE_QRCODE_DOWNLOADED.format(png_path)).flush()
                Notification.send_email_with_qrcode(Config().EMAIL_RECEIVER, '你有新的登录二维码啦!', png_path)
                self.retry_count = 0
                return result.get('uuid'), png_path
            raise KeyError('获取二维码失败: {}'.format(result.get('result_message')))
        except Exception as e:
            sleep_time = get_interval_num(self.sleep_interval)
            UserLog.add_quick_log(
                UserLog.MESSAGE_QRCODE_FAIL.format(e, sleep_time)).flush()
            time.sleep(sleep_time)
            self.request_device_id(self.retry_count % 20 == 0)
            self.retry_count += 1
            return self.download_code()

    def check_user_is_login(self):
        retry = 0
        while retry < Config().REQUEST_MAX_RETRY:
            retry += 1
            response = self.session.get(API_USER_LOGIN_CHECK)
            is_login = response.json().get('data.is_login', False) == 'Y'
            if is_login:
                self.save_user()
                self.set_last_heartbeat()
                return self.get_user_info()  # 检测应该是不会维持状态，这里再请求下个人中心看有没有用，01-10 看来应该是没用  01-22 有时拿到的状态 是已失效的再加上试试
            time.sleep(get_interval_num(self.sleep_interval))
        return is_login

    def auth_uamtk(self):
        retry = 0
        while retry < Config().REQUEST_MAX_RETRY:
            retry += 1
            response = self.session.post(API_AUTH_UAMTK.get('url'), {'appid': 'otn'}, headers={
                'Referer': 'https://kyfw.12306.cn/otn/passport?redirect=/otn/login/userLogin',
                'Origin': 'https://kyfw.12306.cn'
            })
            result = response.json()
            if result.get('newapptk'):
                return result.get('newapptk')
            # TODO 处理获取失败情况
        return False

    def auth_uamauthclient(self, tk):
        retry = 0
        while retry < Config().REQUEST_MAX_RETRY:
            retry += 1
            response = self.session.post(API_AUTH_UAMAUTHCLIENT.get('url'), {'tk': tk})
            result = response.json()
            if result.get('username'):
                return result.get('username')
            # TODO 处理获取失败情况
        return False

    def request_device_id(self, force_renew = False):
        """
        获取加密后的浏览器特征 ID
        :return:
        """
        # 判断cookie 是否过期，未过期可以不必下载
        expire_time =  self.session.cookies.get('RAIL_EXPIRATION')
        if not force_renew and expire_time and int(expire_time) - time_int_ms() > 0:
            return
        if 'pjialin' not in API_GET_BROWSER_DEVICE_ID:
            return self.request_device_id2()
        response = self.session.get(API_GET_BROWSER_DEVICE_ID)
        if response.status_code == 200:
            try:
                result = json.loads(response.text)
                headers = {
                    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/94.0.4606.61 Safari/537.36"
                }
                self.session.headers.update(headers)
                response = self.session.get(base64.b64decode(result['id']).decode())
                if response.text.find('callbackFunction') >= 0:
                    result = response.text[18:-2]
                result = json.loads(result)
                if not Config().is_cache_rail_id_enabled():
                   self.session.cookies.update({
                       'RAIL_EXPIRATION': result.get('exp'),
                       'RAIL_DEVICEID': result.get('dfp'),
                   })
                else:
                   self.session.cookies.update({
                       'RAIL_EXPIRATION': Config().RAIL_EXPIRATION,
                       'RAIL_DEVICEID': Config().RAIL_DEVICEID,
                   })
            except:
                return self.request_device_id()
        else:
            return self.request_device_id()

    def request_device_id2(self):
        headers = {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/94.0.4606.61 Safari/537.36"
        }
        self.session.headers.update(headers)
        response = self.session.get(API_GET_BROWSER_DEVICE_ID)
        if response.status_code == 200:
            try:
                if response.text.find('callbackFunction') >= 0:
                    result = response.text[18:-2]
                    result = json.loads(result)
                    if not Config().is_cache_rail_id_enabled():
                       self.session.cookies.update({
                           'RAIL_EXPIRATION': result.get('exp'),
                           'RAIL_DEVICEID': result.get('dfp'),
                       })
                    else:
                       self.session.cookies.update({
                           'RAIL_EXPIRATION': Config().RAIL_EXPIRATION,
                           'RAIL_DEVICEID': Config().RAIL_DEVICEID,
                       })
            except:
                return self.request_device_id2()
        else:
            return self.request_device_id2()

    def login_did_success(self):
        """
        用户登录成功
        :return:
        """
        self.login_num += 1
        self.welcome_user()
        self.save_user()
        self.get_user_info()
        self.set_last_heartbeat()
        self.is_ready = True

    def welcome_user(self):
        UserLog.print_welcome_user(self)
        pass

    def get_cookie_path(self):
        return Config().USER_DATA_DIR + self.user_name + '.cookie'

    def update_user_info(self, info):
        self.info = {**self.info, **info}

    def get_name(self):
        return self.info.get('user_name', '')

    def save_user(self):
        if Config().is_master():
            self.cluster.set_user_cookie(self.key, self.session.cookies)
            self.cluster.set_user_info(self.key, self.info)
        with open(self.get_cookie_path(), 'wb') as f:
            pickle.dump(self.session.cookies, f)

    def did_loaded_user(self):
        """
        恢复用户成功
        :return:
        """
        UserLog.add_quick_log(UserLog.MESSAGE_LOADED_USER.format(self.user_name)).flush()
        if self.check_user_is_login() and self.can_access_passengers():
            UserLog.add_quick_log(UserLog.MESSAGE_LOADED_USER_SUCCESS.format(self.user_name)).flush()
            UserLog.print_welcome_user(self)
            self.user_did_load()
            return True
        else:
            UserLog.add_quick_log(UserLog.MESSAGE_LOADED_USER_BUT_EXPIRED).flush()
            self.set_last_heartbeat(0)
            return False

    def user_did_load(self):
        """
        用户已经加载成功
        :return:
        """
        self.is_ready = True
        if self.user_loaded: return
        self.user_loaded = True
        Event().user_loaded({'key': self.key})  # 发布通知

    def get_user_info(self):
        retry = 0
        while retry < Config().REQUEST_MAX_RETRY:
            retry += 1
            response = self.session.get(API_USER_INFO.get('url'))
            result = response.json()
            user_data = result.get('data.userDTO.loginUserDTO')
            # 子节点访问会导致主节点登录失效 TODO 可快考虑实时同步 cookie
            if user_data:
                self.update_user_info({**user_data, **{'user_name': user_data.get('name')}})
                self.save_user()
                return True
            time.sleep(get_interval_num(self.sleep_interval))
        return False

    def load_user(self):
        if Config().is_cluster_enabled(): return
        cookie_path = self.get_cookie_path()

        if path.exists(cookie_path):
            with open(self.get_cookie_path(), 'rb') as f:
                cookie = pickle.load(f)
                self.cookie = True
                self.session.cookies.update(cookie)
                return self.did_loaded_user()
        return None

    def load_user_from_remote(self):
        cookie = self.cluster.get_user_cookie(self.key)
        info = self.cluster.get_user_info(self.key)
        if Config().is_slave() and (not cookie or not info):
            while True:  # 子节点只能取
                UserLog.add_quick_log(UserLog.MESSAGE_USER_COOKIE_NOT_FOUND_FROM_REMOTE.format(self.user_name)).flush()
                stay_second(self.retry_time)
                return self.load_user_from_remote()
        if info: self.info = info
        if cookie:
            self.session.cookies.update(cookie)
            if not self.cookie:  # 第一次加载
                self.cookie = True
                if not Config().is_slave():
                    self.did_loaded_user()
                else:
                    self.is_ready = True  # 设置子节点用户 已准备好
                    UserLog.print_welcome_user(self)
            return True
        return False

    def check_is_ready(self):
        return self.is_ready

    def wait_for_ready(self):
        if self.is_ready: return self
        UserLog.add_quick_log(UserLog.MESSAGE_WAIT_USER_INIT_COMPLETE.format(self.retry_time)).flush()
        stay_second(self.retry_time)
        return self.wait_for_ready()

    def destroy(self):
        """
        退出用户
        :return:
        """
        UserLog.add_quick_log(UserLog.MESSAGE_USER_BEING_DESTROY.format(self.user_name)).flush()
        self.is_alive = False

    def response_login_check(self, response, **kwargs):
        if Config().is_master() and response.json().get('data.noLogin') == 'true':  # relogin
            self.handle_login(expire=True)

    def get_user_passengers(self):
        if self.passengers: return self.passengers
        response = self.session.post(API_USER_PASSENGERS)
        result = response.json()
        if result.get('data.normal_passengers'):
            self.passengers = result.get('data.normal_passengers')
            # 将乘客写入到文件
            with open(Config().USER_PASSENGERS_FILE % self.user_name, 'w', encoding='utf-8') as f:
                f.write(json.dumps(self.passengers, indent=4, ensure_ascii=False))
            return self.passengers
        else:
            wait_time = get_interval_num(self.sleep_interval)
            UserLog.add_quick_log(
                UserLog.MESSAGE_GET_USER_PASSENGERS_FAIL.format(
                    result.get('messages', CommonLog.MESSAGE_RESPONSE_EMPTY_ERROR), wait_time)).flush()
            if Config().is_slave():
                self.load_user_from_remote()  # 加载最新 cookie
            stay_second(wait_time)
            return self.get_user_passengers()

    def can_access_passengers(self):
        retry = 0
        while retry < Config().REQUEST_MAX_RETRY:
            retry += 1
            response = self.session.post(API_USER_PASSENGERS)
            result = response.json()
            if result.get('data.normal_passengers'):
                return True
            else:
                wait_time = get_interval_num(self.sleep_interval)
                UserLog.add_quick_log(
                    UserLog.MESSAGE_TEST_GET_USER_PASSENGERS_FAIL.format(
                        result.get('messages', CommonLog.MESSAGE_RESPONSE_EMPTY_ERROR), wait_time)).flush()
                if Config().is_slave():
                    self.load_user_from_remote()  # 加载最新 cookie
                stay_second(wait_time)
        return False

    def get_passengers_by_members(self, members):
        """
        获取格式化后的乘客信息
        :param members:
        :return:
        [{
            name: '项羽',
            type: 1,
            id_card: 0000000000000000000,
            type_text: '成人',
            enc_str: 'aaaaaa'
        }]
        """
        self.get_user_passengers()
        results = []
        for member in members:
            is_member_code = is_number(member)
            if not is_member_code:
                if member[0] == "*":
                    audlt = 1
                    member = member[1:]
                else:
                    audlt = 0
                child_check = array_dict_find_by_key_value(results, 'name', member)
            if not is_member_code and child_check:
                new_member = child_check.copy()
                new_member['type'] = UserType.CHILD
                new_member['type_text'] = dict_find_key_by_value(UserType.dicts, int(new_member['type']))
            else:
                if is_member_code:
                    passenger = array_dict_find_by_key_value(self.passengers, 'code', member)
                else:
                    passenger = array_dict_find_by_key_value(self.passengers, 'passenger_name', member)
                    if audlt:
                        passenger['passenger_type'] = UserType.ADULT
                if not passenger:
                    UserLog.add_quick_log(
                        UserLog.MESSAGE_USER_PASSENGERS_IS_INVALID.format(self.user_name, member)).flush()
                    return False
                new_member = {
                    'name': passenger.get('passenger_name'),
                    'id_card': passenger.get('passenger_id_no'),
                    'id_card_type': passenger.get('passenger_id_type_code'),
                    'mobile': passenger.get('mobile_no'),
                    'type': passenger.get('passenger_type'),
                    'type_text': dict_find_key_by_value(UserType.dicts, int(passenger.get('passenger_type'))),
                    'enc_str': passenger.get('allEncStr')
                }
            results.append(new_member)

        return results

    def request_init_dc_page(self):
        """
        请求下单页面 拿到 token
        :return:
        """
        data = {'_json_att': ''}
        response = self.session.post(API_INITDC_URL, data)
        html = response.text
        token = re.search(r'var globalRepeatSubmitToken = \'(.+?)\'', html)
        form = re.search(r'var ticketInfoForPassengerForm *= *(\{.+\})', html)
        order = re.search(r'var orderRequestDTO *= *(\{.+\})', html)
        # 系统忙，请稍后重试
        if html.find('系统忙，请稍后重试') != -1:
            OrderLog.add_quick_log(OrderLog.MESSAGE_REQUEST_INIT_DC_PAGE_FAIL).flush()  # 重试无用，直接跳过
            return False, False, html
        try:
            self.global_repeat_submit_token = token.groups()[0]
            self.ticket_info_for_passenger_form = json.loads(form.groups()[0].replace("'", '"'))
            self.order_request_dto = json.loads(order.groups()[0].replace("'", '"'))
        except:
            return False, False, html  # TODO Error

        slide_val = re.search(r"var if_check_slide_passcode.*='(\d?)'", html)
        is_slide = False
        if slide_val:
            is_slide = int(slide_val[1]) == 1
        return True, is_slide, html



File: py12306/user/__init__.py




File: py12306/exceptions/__init__.py




File: py12306/log/query_log.py

# -*- coding:utf-8 -*-
import datetime
import json
import sys
from os import path

from py12306.config import Config
from py12306.cluster.cluster import Cluster
from py12306.log.base import BaseLog
from py12306.helpers.func import *


@singleton
class QueryLog(BaseLog):
    # 这里如果不声明，会出现重复打印，目前不知道什么原因
    logs = []
    thread_logs = {}
    quick_log = []

    data = {
        'query_count': 0,
        'last_time': '',
    }
    data_path = None

    LOG_INIT_JOBS = ''

    MESSAGE_GIVE_UP_CHANCE_CAUSE_TICKET_NUM_LESS_THAN_SPECIFIED = '余票数小于乘车人数，放弃此次提交机会'
    MESSAGE_QUERY_LOG_OF_EVERY_TRAIN = '{}'
    MESSAGE_QUERY_LOG_OF_TRAIN_INFO = '{} {}'
    MESSAGE_QUERY_START_BY_DATE = '出发日期 {}: {} - {}'

    MESSAGE_JOBS_DID_CHANGED = '任务已更新，正在重新加载...\n'

    MESSAGE_SKIP_ORDER = '跳过本次请求，节点 {} 用户 {} 正在处理该订单\n'

    MESSAGE_QUERY_JOB_BEING_DESTROY = '查询任务 {} 已结束\n'

    MESSAGE_INIT_PASSENGERS_SUCCESS = '初始化乘客成功'
    MESSAGE_CHECK_PASSENGERS = '查询任务 {} 正在验证乘客信息'

    MESSAGE_USER_IS_EMPTY_WHEN_DO_ORDER = '未配置自动下单账号，{} 秒后继续查询\n'
    MESSAGE_ORDER_USER_IS_EMPTY = '未找到下单账号，{} 秒后继续查询'

    cluster = None

    def __init__(self):
        super().__init__()
        self.data_path = Config().QUERY_DATA_DIR + 'status.json'
        self.cluster = Cluster()

    @classmethod
    def init_data(cls):
        self = cls()
        # 获取上次记录
        result = False
        if not Config.is_cluster_enabled() and path.exists(self.data_path):
            with open(self.data_path, encoding='utf-8') as f:
                result = f.read()
                try:
                    result = json.loads(result)
                except json.JSONDecodeError as e:
                    result = {}
                    # self.add_quick_log('加载status.json失败, 文件内容为: {}.'.format(repr(result)))
                    # self.flush()  # 这里可以用不用提示

        if Config.is_cluster_enabled():
            result = self.get_data_from_cluster()

        if result:
            self.data = {**self.data, **result}
            self.print_data_restored()

    def get_data_from_cluster(self):
        query_count = self.cluster.session.get(Cluster.KEY_QUERY_COUNT, 0)
        last_time = self.cluster.session.get(Cluster.KEY_QUERY_LAST_TIME, '')
        if query_count and last_time:
            return {'query_count': query_count, 'last_time': last_time}
        return False

    def refresh_data_of_cluster(self):
        return {
            'query_count': self.cluster.session.incr(Cluster.KEY_QUERY_COUNT),
            'last_time': self.cluster.session.set(Cluster.KEY_QUERY_LAST_TIME, time_now()),
        }

    @classmethod
    def print_init_jobs(cls, jobs):
        """
        输出初始化信息
        :return:
        """
        self = cls()
        self.add_log('# 发现 {} 个任务 #'.format(len(jobs)))
        index = 1
        for job in jobs:
            self.add_log('================== 任务 {} =================='.format(index))
            for station in job.stations:
                self.add_log('出发站：{}    到达站：{}'.format(station.get('left'), station.get('arrive')))

            self.add_log('乘车日期：{}'.format(job.left_dates))
            self.add_log('坐席：{}'.format('，'.join(job.allow_seats)))
            self.add_log('乘车人：{}'.format('，'.join(job.members)))
            if job.except_train_numbers:
                train_number_message = '排除 ' + '，'.join(job.allow_train_numbers)
            else:
                train_number_message = '，'.join(job.allow_train_numbers if job.allow_train_numbers else ['不筛选'])
            self.add_log('筛选车次：{}'.format(train_number_message))
            self.add_log('任务名称：{}'.format(job.job_name))
            # 乘车日期：['2019-01-24', '2019-01-25', '2019-01-26', '2019-01-27']
            self.add_log('')
            index += 1

        self.flush()
        return self

    @classmethod
    def print_ticket_num_less_than_specified(cls, rest_num, job):
        self = cls()
        self.add_quick_log(
            '余票数小于乘车人数，当前余票数: {rest_num}, 实际人数 {actual_num}, 删减人车人数到: {take_num}'.format(rest_num=rest_num,
                                                                                         actual_num=job.member_num,
                                                                                         take_num=job.member_num_take))
        self.flush()
        return self

    @classmethod
    def print_ticket_seat_available(cls, left_date, train_number, seat_type, rest_num):
        self = cls()
        self.add_quick_log(
            '[ 查询到座位可用 出发时间 {left_date} 车次 {train_number} 座位类型 {seat_type} 余票数量 {rest_num} ]'.format(
                left_date=left_date,
                train_number=train_number,
                seat_type=seat_type,
                rest_num=rest_num))
        self.flush()
        return self

    @classmethod
    def print_ticket_available(cls, left_date, train_number, rest_num):
        self = cls()
        self.add_quick_log('检查完成 开始提交订单 '.format())
        self.notification('查询到可用车票', '时间 {left_date} 车次 {train_number} 余票数量 {rest_num}'.format(left_date=left_date,
                                                                                               train_number=train_number,
                                                                                               rest_num=rest_num))
        self.flush()
        return self

    @classmethod
    def print_query_error(cls, reason, code=None):
        self = cls()
        self.add_quick_log('查询余票请求失败')
        if code:
            self.add_quick_log('状态码 {}   '.format(code))
        if reason:
            self.add_quick_log('错误原因 {}   '.format(reason))
        self.flush(sep='\t')
        return self

    @classmethod
    def print_job_start(cls, job_name):
        self = cls()
        message = '>> 第 {query_count} 次查询 {job_name} {time}'.format(
            query_count=int(self.data.get('query_count', 0)) + 1,
            job_name=job_name, time=time_now().strftime("%Y-%m-%d %H:%M:%S"))
        self.add_log(message)
        self.refresh_data()
        if is_main_thread():
            self.flush(publish=False)
        return self

    @classmethod
    def add_query_time_log(cls, time, is_cdn):
        return cls().add_log(('*' if is_cdn else '') + '耗时 %.2f' % time)

    @classmethod
    def add_stay_log(cls, second):
        self = cls()
        self.add_log('停留 {}'.format(second))
        return self

    def print_data_restored(self):
        self.add_quick_log('============================================================')
        self.add_quick_log('|=== 查询记录恢复成功 上次查询 {last_date} ===|'.format(last_date=self.data.get('last_time')))
        self.add_quick_log('============================================================')
        self.add_quick_log('')
        self.flush(publish=False)
        return self

    def refresh_data(self):
        if Config.is_cluster_enabled():
            self.data = {**self.data, **self.refresh_data_of_cluster()}
        else:
            self.data['query_count'] += 1
            self.data['last_time'] = str(datetime.datetime.now())
            self.save_data()

    def save_data(self):
        with open(self.data_path, 'w') as file:
            file.write(json.dumps(self.data))



File: py12306/log/cluster_log.py

from py12306.log.base import BaseLog
from py12306.helpers.func import *


@singleton
class ClusterLog(BaseLog):
    # 这里如果不声明，会出现重复打印，目前不知道什么原因
    logs = []
    thread_logs = {}
    quick_log = []

    MESSAGE_JOIN_CLUSTER_SUCCESS = '# 节点 {} 成功加入到集群，当前节点列表 {} #'

    MESSAGE_LEFT_CLUSTER = '# 节点 {} 已离开集群，当前节点列表 {} #'

    MESSAGE_NODE_ALREADY_IN_CLUSTER = '# 当前节点已存在于集群中，自动分配新的节点名称 {} #'

    MESSAGE_SUBSCRIBE_NOTIFICATION_PREFIX = '{} ）'
    MESSAGE_SUBSCRIBE_NOTIFICATION = MESSAGE_SUBSCRIBE_NOTIFICATION_PREFIX + '{}'

    MESSAGE_ASCENDING_MASTER_NODE = '# 已将 {} 提升为主节点，当前节点列表 {} #'

    MESSAGE_MASTER_DID_LOST = '# 主节点已退出，{} 秒后程序将自动退出 #'

    MESSAGE_MASTER_NODE_ALREADY_RUN = '# 启动失败，主节点 {} 已经在运行中 #'
    MESSAGE_MASTER_NODE_NOT_FOUND = '# 启动失败，请先启动主节点 #'

    MESSAGE_NODE_BECOME_MASTER_AGAIN = '# 节点 {} 已启动，已自动成为主节点 #'



    @staticmethod
    def get_print_nodes(nodes):
        message = ['{}{}'.format('*' if val == '1' else '', key) for key, val in nodes.items()]
        return '[ {} ]'.format(', '.join(message))



File: py12306/log/__init__.py




File: py12306/log/user_log.py

from py12306.log.base import BaseLog
from py12306.helpers.func import *


@singleton
class UserLog(BaseLog):
    # 这里如果不声明，会出现重复打印，目前不知道什么原因
    logs = []
    thread_logs = {}
    quick_log = []

    MESSAGE_DOWNLAOD_AUTH_CODE_FAIL = '验证码下载失败 错误原因: {} {} 秒后重试'
    MESSAGE_DOWNLAODING_THE_CODE = '正在下载验证码...'
    MESSAGE_CODE_AUTH_FAIL = '验证码验证失败 错误原因: {}'
    MESSAGE_CODE_AUTH_SUCCESS = '验证码验证成功 开始登录...'
    MESSAGE_QRCODE_DOWNLOADING = '正在下载二维码...'
    MESSAGE_QRCODE_DOWNLOADED = '二维码保存在: {}，请使用手机客户端扫描'
    MESSAGE_QRCODE_FAIL = '二维码获取失败: {}, {} 秒后重试'
    MESSAGE_LOGIN_FAIL = '登录失败 错误原因: {}'
    MESSAGE_LOADED_USER = '正在尝试恢复用户: {}'
    MESSAGE_LOADED_USER_SUCCESS = '用户恢复成功: {}'
    MESSAGE_LOADED_USER_BUT_EXPIRED = '用户状态已过期，正在重新登录'
    MESSAGE_USER_HEARTBEAT_NORMAL = '用户 {} 心跳正常，下次检测 {} 秒后'

    MESSAGE_GET_USER_PASSENGERS_FAIL = '获取用户乘客列表失败，错误原因: {} {} 秒后重试'
    MESSAGE_TEST_GET_USER_PASSENGERS_FAIL = '测试获取用户乘客列表失败，错误原因: {} {} 秒后重试'
    MESSAGE_USER_PASSENGERS_IS_INVALID = '乘客信息校验失败，在账号 {} 中未找到该乘客: {}\n'

    # MESSAGE_WAIT_USER_INIT_COMPLETE = '未找到可用账号或用户正在初始化，{} 秒后重试'

    MESSAGE_USERS_DID_CHANGED = '\n用户信息已更新，正在重新加载...'

    MESSAGE_USER_BEING_DESTROY = '用户 {} 已退出'
    MESSAGE_USER_COOKIE_NOT_FOUND_FROM_REMOTE = '用户 {} 状态加载中...'

    MESSAGE_WAIT_USER_INIT_COMPLETE = '账号正在登录中，{} 秒后自动重试'

    def __init__(self):
        super().__init__()
        self.init_data()

    def init_data(self):
        pass

    @classmethod
    def print_init_users(cls, users):
        """
        输出初始化信息
        :return:
        """
        self = cls()
        self.add_quick_log('# 发现 {} 个用户 #\n'.format(len(users)))
        self.flush()
        return self

    @classmethod
    def print_welcome_user(cls, user):
        self = cls()
        self.add_quick_log('# 欢迎回来，{} #\n'.format(user.get_name()))
        self.flush()
        return self

    @classmethod
    def print_start_login(cls, user):
        self = cls()
        self.add_quick_log('正在登录用户 {}'.format(user.user_name))
        self.flush()
        return self

    @classmethod
    def print_user_passenger_init_success(cls, passengers):
        self = cls()
        result = [passenger.get('name') + '(' + passenger.get('type_text') + ')' for passenger in passengers]
        self.add_quick_log('# 乘客验证成功 {} #\n'.format(', '.join(result)))
        self.flush()
        return self

    @classmethod
    def print_user_expired(cls):
        return cls().add_quick_log(cls.MESSAGE_LOADED_USER_BUT_EXPIRED).flush()



File: py12306/log/common_log.py

from py12306.log.base import BaseLog
from py12306.config import *
from py12306.helpers.func import *


@singleton
class CommonLog(BaseLog):
    # 这里如果不声明，会出现重复打印，目前不知道什么原因
    logs = []
    thread_logs = {}
    quick_log = []

    MESSAGE_12306_IS_CLOSED = '当前时间: {}     |       12306 休息时间，程序将在明天早上 6 点自动运行'
    MESSAGE_RETRY_AUTH_CODE = '{} 秒后重新获取验证码'

    MESSAGE_EMPTY_APP_CODE = '无法发送语音消息，未填写验证码接口 appcode'
    MESSAGE_VOICE_API_FORBID = '语音消息发送失败，请检查 appcode 是否填写正确或 套餐余额是否充足'
    MESSAGE_VOICE_API_SEND_FAIL = '语音消息发送失败，错误原因 {}'
    MESSAGE_VOICE_API_SEND_SUCCESS = '语音消息发送成功! 接口返回信息 {} '

    MESSAGE_CHECK_AUTO_CODE_FAIL = '请配置打码账号的账号密码'
    MESSAGE_CHECK_EMPTY_USER_ACCOUNT = '请配置 12306 账号密码'

    MESSAGE_TEST_SEND_VOICE_CODE = '正在测试发送语音验证码...'
    MESSAGE_TEST_SEND_EMAIL = '正在测试发送邮件...'
    MESSAGE_TEST_SEND_DINGTALK = '正在测试发送钉钉消息...'
    MESSAGE_TEST_SEND_TELEGRAM = '正在测试推送到Telegram...'
    MESSAGE_TEST_SEND_SERVER_CHAN = '正在测试发送ServerChan消息...'
    MESSAGE_TEST_SEND_PUSH_BEAR = '正在测试发送PushBear消息...'
    MESSAGE_TEST_SEND_PUSH_BARK = '正在测试发送Bark消息...'

    MESSAGE_CONFIG_FILE_DID_CHANGED = '配置文件已修改，正在重新加载中\n'
    MESSAGE_API_RESPONSE_CAN_NOT_BE_HANDLE = '接口返回错误'

    MESSAGE_SEND_EMAIL_SUCCESS = '邮件发送成功，请检查收件箱'
    MESSAGE_SEND_EMAIL_FAIL = '邮件发送失败，请手动检查配置，错误原因 {}'

    MESSAGE_SEND_EMAIL_WITH_QRCODE_SUCCESS = '二维码邮件发送成功，请检查收件箱扫描登陆'

    MESSAGE_SEND_TELEGRAM_SUCCESS = 'Telegram推送成功'
    MESSAGE_SEND_TELEGRAM_FAIL = 'Telegram推送失败，错误原因 {}'

    MESSAGE_SEND_SERVER_CHAN_SUCCESS = '发送成功，请检查微信'
    MESSAGE_SEND_SERVER_CHAN_FAIL = 'ServerChan发送失败，请检查KEY'

    MESSAGE_SEND_PUSH_BEAR_SUCCESS = '发送成功，请检查微信'
    MESSAGE_SEND_PUSH_BEAR_FAIL = 'PushBear发送失败，请检查KEY'

    MESSAGE_SEND_BARK_SUCCESS = 'Bark推送成功'
    MESSAGE_SEND_BARK_FAIL = 'Bark推送失败，错误原因 {}'

    MESSAGE_OUTPUT_TO_FILE_IS_UN_ENABLE = '请先打开配置项中的：OUT_PUT_LOG_TO_FILE_ENABLED ( 输出到文件 )'

    MESSAGE_GET_RESPONSE_FROM_FREE_AUTO_CODE = '从免费打码获取结果失败'

    MESSAGE_RESPONSE_EMPTY_ERROR = '网络错误'

    MESSAGE_CDN_START_TO_CHECK = '正在筛选 {} 个 CDN...'
    MESSAGE_CDN_START_TO_RECHECK = '正在重新筛选 {} 个 CDN...当前时间 {}\n'
    MESSAGE_CDN_RESTORE_SUCCESS = 'CDN 恢复成功，上次检测 {}\n'
    MESSAGE_CDN_CHECKED_SUCCESS = '# CDN 检测完成，可用 CDN {} #\n'
    MESSAGE_CDN_CLOSED = '# CDN 已关闭 #'

    def __init__(self):
        super().__init__()
        self.init_data()

    def init_data(self):
        pass

    @classmethod
    def print_welcome(cls):
        self = cls()
        self.add_quick_log('######## py12306 购票助手，本程序为开源工具，请勿用于商业用途 ########')
        if Const.IS_TEST:
            self.add_quick_log()
            self.add_quick_log('当前为测试模式，程序运行完成后自动结束')
        if not Const.IS_TEST and Config().OUT_PUT_LOG_TO_FILE_ENABLED:
            self.add_quick_log()
            self.add_quick_log('日志已输出到文件中: {}'.format(Config().OUT_PUT_LOG_TO_FILE_PATH))
        if Config().WEB_ENABLE:
            self.add_quick_log()
            self.add_quick_log('WEB 管理页面已开启，请访问 主机地址 + 端口 {} 进行查看'.format(Config().WEB_PORT))

        self.add_quick_log()
        self.flush(file=False, publish=False)
        return self

    @classmethod
    def print_configs(cls):
        # 打印配置
        self = cls()
        enable = '已开启'
        disable = '未开启'
        self.add_quick_log('**** 当前配置 ****')
        self.add_quick_log('多线程查询: {}'.format(get_true_false_text(Config().QUERY_JOB_THREAD_ENABLED, enable, disable)))
        self.add_quick_log('CDN 状态: {}'.format(get_true_false_text(Config().CDN_ENABLED, enable, disable))).flush()
        self.add_quick_log('通知状态:')
        if Config().NOTIFICATION_BY_VOICE_CODE:
            self.add_quick_log(
                '语音验证码: {}'.format(get_true_false_text(Config().NOTIFICATION_BY_VOICE_CODE, enable, disable)))
        if Config().EMAIL_ENABLED:
            self.add_quick_log('邮件通知: {}'.format(get_true_false_text(Config().EMAIL_ENABLED, enable, disable)))
        if Config().DINGTALK_ENABLED:
            self.add_quick_log('钉钉通知: {}'.format(get_true_false_text(Config().DINGTALK_ENABLED, enable, disable)))
        if Config().TELEGRAM_ENABLED:
            self.add_quick_log('Telegram通知: {}'.format(get_true_false_text(Config().TELEGRAM_ENABLED, enable, disable)))
        if Config().SERVERCHAN_ENABLED:
            self.add_quick_log(
                'ServerChan通知: {}'.format(get_true_false_text(Config().SERVERCHAN_ENABLED, enable, disable)))
        if Config().BARK_ENABLED:
            self.add_quick_log('Bark通知: {}'.format(get_true_false_text(Config().BARK_ENABLED, enable, disable)))
        if Config().PUSHBEAR_ENABLED:
            self.add_quick_log(
                'PushBear通知: {}'.format(get_true_false_text(Config().PUSHBEAR_ENABLED, enable, disable)))
        self.add_quick_log().flush(sep='\t\t')
        self.add_quick_log('查询间隔: {} 秒'.format(Config().QUERY_INTERVAL))
        self.add_quick_log('用户心跳检测间隔: {} 秒'.format(Config().USER_HEARTBEAT_INTERVAL))
        self.add_quick_log('WEB 管理页面: {}'.format(get_true_false_text(Config().WEB_ENABLE, enable, disable)))
        if Config().is_cluster_enabled():
            from py12306.cluster.cluster import Cluster
            self.add_quick_log('分布式查询: {}'.format(get_true_false_text(Config().is_cluster_enabled(), enable, enable)))
            self.add_quick_log('节点名称: {}'.format(Cluster().node_name))
            self.add_quick_log('节点是否主节点: {}'.format(get_true_false_text(Config().is_master(), '是', '否')))
            self.add_quick_log(
                '子节点提升为主节点: {}'.format(get_true_false_text(Config().NODE_SLAVE_CAN_BE_MASTER, enable, disable)))
        self.add_quick_log()
        self.flush()
        return self

    @classmethod
    def print_test_complete(cls):
        self = cls()
        self.add_quick_log('# 测试完成，请检查输出是否正确 #')
        self.flush(publish=False)
        return self

    @classmethod
    def print_auto_code_fail(cls, reason):
        self = cls()
        self.add_quick_log('打码失败: 错误原因 {reason}'.format(reason=reason))
        self.flush()
        return self

    @classmethod
    def print_auth_code_info(cls, reason):
        self = cls()
        self.add_quick_log('打码信息: {reason}'.format(reason=reason))
        self.flush()
        return self



File: py12306/log/order_log.py

from py12306.log.base import BaseLog
from py12306.helpers.func import *


@singleton
class OrderLog(BaseLog):
    # 这里如果不声明，会出现重复打印，目前不知道什么原因
    logs = []
    thread_logs = {}
    quick_log = []

    MESSAGE_REQUEST_INIT_DC_PAGE_FAIL = '请求初始化订单页面失败'

    MESSAGE_SUBMIT_ORDER_REQUEST_FAIL = '提交订单失败，错误原因 {} \n'
    MESSAGE_SUBMIT_ORDER_REQUEST_SUCCESS = '提交订单成功'
    MESSAGE_CHECK_ORDER_INFO_FAIL = '检查订单失败，错误原因 {} \n'
    MESSAGE_CHECK_ORDER_INFO_SUCCESS = '检查订单成功'

    MESSAGE_GET_QUEUE_INFO_SUCCESS = '获取排队信息成功，目前排队人数 {}, 余票还剩余 {} 张'
    MESSAGE_GET_QUEUE_INFO_NO_SEAT = '接口返回实际为无票，跳过本次排队'
    MESSAGE_GET_QUEUE_COUNT_SUCCESS = '排队成功，你当前排在第 {} 位, 余票还剩余 {} 张'
    MESSAGE_GET_QUEUE_LESS_TICKET = '排队失败，目前排队人数已经超过余票张数'
    MESSAGE_GET_QUEUE_COUNT_FAIL = '排队失败，错误原因 {}'

    MESSAGE_CONFIRM_SINGLE_FOR_QUEUE_SUCCESS = '# 提交订单成功！#'
    MESSAGE_CONFIRM_SINGLE_FOR_QUEUE_ERROR = '出票失败，错误原因 {}'
    MESSAGE_CONFIRM_SINGLE_FOR_QUEUE_FAIL = '提交订单失败，错误原因 {}'

    MESSAGE_QUERY_ORDER_WAIT_TIME_WAITING = '排队等待中，排队人数 {}，预计还需要 {} 秒'
    MESSAGE_QUERY_ORDER_WAIT_TIME_FAIL = '排队失败，错误原因 {}'
    MESSAGE_QUERY_ORDER_WAIT_TIME_INFO = '第 {} 次排队，请耐心等待'

    MESSAGE_ORDER_SUCCESS_NOTIFICATION_TITLE = '车票购买成功！'
    MESSAGE_ORDER_SUCCESS_NOTIFICATION_CONTENT = '请及时登录12306账号[{}]，打开 \'未完成订单\'，在30分钟内完成支付!'
    MESSAGE_ORDER_SUCCESS_NOTIFICATION_INFO = '\t\t车次信息： {} {}[{}] -> {}[{}]，乘车日期 {}，席位：{}，乘车人：{}'

    MESSAGE_ORDER_SUCCESS_NOTIFICATION_OF_VOICE_CODE_START_SEND = '正在发送语音通知...'
    MESSAGE_ORDER_SUCCESS_NOTIFICATION_OF_VOICE_CODE_CONTENT = '你的车票 {} 到 {} 购买成功，请登录 12306 进行支付'

    MESSAGE_ORDER_SUCCESS_NOTIFICATION_OF_EMAIL_CONTENT = '订单号 {}，请及时登录12306账号[{}]，打开 \'未完成订单\'，在30分钟内完成支付!'

    MESSAGE_JOB_CLOSED = '当前任务已结束'

    @classmethod
    def print_passenger_did_deleted(cls, passengers):
        self = cls()
        result = [passenger.get('name') + '(' + passenger.get('type_text') + ')' for passenger in passengers]
        self.add_quick_log('# 删减后的乘客列表 {} #'.format(', '.join(result)))
        self.flush()
        return self

    @classmethod
    def print_ticket_did_ordered(cls, order_id):
        self = cls()
        self.add_quick_log('# 车票购买成功，订单号 {} #'.format(order_id))
        self.flush()
        return self

    @classmethod
    def get_order_success_notification_info(cls, query):
        from py12306.query.job import Job
        assert isinstance(query, Job)
        passengers = [passenger.get(
            'name') + '(' + passenger.get('type_text') + ')' for passenger in query.passengers]
        return cls.MESSAGE_ORDER_SUCCESS_NOTIFICATION_INFO.format(query.get_info_of_train_number(),
                                                                  query.get_info_of_left_station(),
                                                                  query.get_info_of_train_left_time(),
                                                                  query.get_info_of_arrive_station(),
                                                                  query.get_info_of_train_arrive_time(),
                                                                  query.get_info_of_left_date(),
                                                                  query.current_seat_name,
                                                                  '，'.join(passengers))



File: py12306/log/redis_log.py

from py12306.log.base import BaseLog
from py12306.helpers.func import *


@singleton
class RedisLog(BaseLog):
    # 这里如果不声明，会出现重复打印，目前不知道什么原因
    logs = []
    thread_logs = {}
    quick_log = []

    MESSAGE_REDIS_INIT_SUCCESS = 'Redis 初始化成功'



File: py12306/log/base.py

import os
import sys
import io
from contextlib import redirect_stdout

from py12306.config import Config
from py12306.helpers.func import *


class BaseLog:
    logs = []
    thread_logs = {}
    quick_log = []

    @classmethod
    def add_log(cls, content=''):
        self = cls()
        # print('添加 Log 主进程{} 进程ID{}'.format(is_main_thread(), current_thread_id()))
        if is_main_thread():
            self.logs.append(content)
        else:
            tmp_log = self.thread_logs.get(current_thread_id(), [])
            tmp_log.append(content)
            self.thread_logs[current_thread_id()] = tmp_log
        return self

    @classmethod
    def flush(cls, sep='\n', end='\n', file=None, exit=False, publish=True):
        from py12306.cluster.cluster import Cluster
        self = cls()
        logs = self.get_logs()
        # 输出到文件
        if file == None and Config().OUT_PUT_LOG_TO_FILE_ENABLED and not Const.IS_TEST:  # TODO 文件无法写入友好提示
            file = open(Config().OUT_PUT_LOG_TO_FILE_PATH, 'a', encoding='utf-8')
        if not file: file = None
        # 输出日志到各个节点
        if publish and self.quick_log and Config().is_cluster_enabled() and Cluster().is_ready:  #
            f = io.StringIO()
            with redirect_stdout(f):
                print(*logs, sep=sep, end='' if end == '\n' else end)
            out = f.getvalue()
            Cluster().publish_log_message(out)
        else:
            print(*logs, sep=sep, end=end, file=file)
        self.empty_logs(logs)
        if exit: sys.exit()

    def get_logs(self):
        if self.quick_log:
            logs = self.quick_log
        else:
            if is_main_thread():
                logs = self.logs
            else:
                logs = self.thread_logs.get(current_thread_id())
        return logs

    def empty_logs(self, logs=None):
        if self.quick_log:
            self.quick_log = []
        else:
            if is_main_thread():
                self.logs = []
            else:
                if logs and self.thread_logs.get(current_thread_id()): del self.thread_logs[current_thread_id()]

    @classmethod
    def add_quick_log(cls, content=''):
        self = cls()
        self.quick_log.append(content)
        return self

    def notification(self, title, content=''):
        # if sys.platform == 'darwin': # 不太友好 先关闭，之前没考虑到 mac 下会请求权限
        #     os.system( 'osascript -e \'tell app "System Events" to display notification "{content}" with title "{title}"\''.format(
        #             title=title, content=content))
        pass



File: py12306/query/query.py

from base64 import b64decode
from py12306.config import Config
from py12306.cluster.cluster import Cluster
from py12306.app import app_available_check
from py12306.helpers.func import *
from py12306.helpers.request import Request
from py12306.log.query_log import QueryLog
from py12306.query.job import Job
from py12306.helpers.api import API_QUERY_INIT_PAGE, API_GET_BROWSER_DEVICE_ID


@singleton
class Query:
    """
    余票查询

    """
    jobs = []
    query_jobs = []
    session = {}

    # 查询间隔
    interval = {}
    cluster = None

    is_in_thread = False
    retry_time = 3
    is_ready = False
    api_type = None  # Query api url, Current know value  leftTicket/queryX | leftTicket/queryZ

    def __init__(self):
        self.session = Request()
        self.request_device_id()
        self.cluster = Cluster()
        self.update_query_interval()
        self.update_query_jobs()
        self.get_query_api_type()

    def update_query_interval(self, auto=False):
        self.interval = init_interval_by_number(Config().QUERY_INTERVAL)
        if auto:
            jobs_do(self.jobs, 'update_interval')

    def update_query_jobs(self, auto=False):
        self.query_jobs = Config().QUERY_JOBS
        if auto:
            QueryLog.add_quick_log(QueryLog.MESSAGE_JOBS_DID_CHANGED).flush()
            self.refresh_jobs()
            if not Config().is_slave():
                jobs_do(self.jobs, 'check_passengers')

    @classmethod
    def run(cls):
        self = cls()
        app_available_check()
        self.start()
        pass

    @classmethod
    def check_before_run(cls):
        self = cls()
        self.init_jobs()
        self.is_ready = True

    def start(self):
        # return # DEBUG
        QueryLog.init_data()
        stay_second(3)
        # 多线程
        while True:
            if Config().QUERY_JOB_THREAD_ENABLED:  # 多线程
                if not self.is_in_thread:
                    self.is_in_thread = True
                    create_thread_and_run(jobs=self.jobs, callback_name='run', wait=Const.IS_TEST)
                if Const.IS_TEST: return
                stay_second(self.retry_time)
            else:
                if not self.jobs: break
                self.is_in_thread = False
                jobs_do(self.jobs, 'run')
                if Const.IS_TEST: return

        # while True:
        #     app_available_check()
        #     if Config().QUERY_JOB_THREAD_ENABLED:  # 多线程
        #         create_thread_and_run(jobs=self.jobs, callback_name='run')
        #     else:
        #         for job in self.jobs: job.run()
        #     if Const.IS_TEST: return
        # self.refresh_jobs()  # 刷新任务

    def refresh_jobs(self):
        """
        更新任务
        :return:
        """
        allow_jobs = []
        for job in self.query_jobs:
            id = md5(job)
            job_ins = objects_find_object_by_key_value(self.jobs, 'id', id)  # [1 ,2]
            if not job_ins:
                job_ins = self.init_job(job)
                if Config().QUERY_JOB_THREAD_ENABLED:  # 多线程重新添加
                    create_thread_and_run(jobs=job_ins, callback_name='run', wait=Const.IS_TEST)
            allow_jobs.append(job_ins)

        for job in self.jobs:  # 退出已删除 Job
            if job not in allow_jobs: job.destroy()

        QueryLog.print_init_jobs(jobs=self.jobs)

    def init_jobs(self):
        for job in self.query_jobs:
            self.init_job(job)
        QueryLog.print_init_jobs(jobs=self.jobs)

    def init_job(self, job):
        job = Job(info=job, query=self)
        self.jobs.append(job)
        return job

    def request_device_id(self, force_renew = False):
        """
        获取加密后的浏览器特征 ID
        :return:
        """
        expire_time =  self.session.cookies.get('RAIL_EXPIRATION')
        if not force_renew and expire_time and int(expire_time) - time_int_ms() > 0:
            return
        if 'pjialin' not in API_GET_BROWSER_DEVICE_ID:
            return self.request_device_id2()
        response = self.session.get(API_GET_BROWSER_DEVICE_ID)
        if response.status_code == 200:
            try:
                result = json.loads(response.text)
                response = self.session.get(b64decode(result['id']).decode())
                if response.text.find('callbackFunction') >= 0:
                    result = response.text[18:-2]
                result = json.loads(result)
                if not Config().is_cache_rail_id_enabled():
                    self.session.cookies.update({
                        'RAIL_EXPIRATION': result.get('exp'),
                        'RAIL_DEVICEID': result.get('dfp'),
                    })
                else:
                    self.session.cookies.update({
                        'RAIL_EXPIRATION': Config().RAIL_EXPIRATION,
                        'RAIL_DEVICEID': Config().RAIL_DEVICEID,
                    })
            except:
                return self.request_device_id()
        else:
            return self.request_device_id()

    def request_device_id2(self):
        headers = {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/94.0.4606.61 Safari/537.36"
        }
        self.session.headers.update(headers)
        response = self.session.get(API_GET_BROWSER_DEVICE_ID)
        if response.status_code == 200:
            try:
                if response.text.find('callbackFunction') >= 0:
                    result = response.text[18:-2]
                    result = json.loads(result)
                    if not Config().is_cache_rail_id_enabled():
                       self.session.cookies.update({
                           'RAIL_EXPIRATION': result.get('exp'),
                           'RAIL_DEVICEID': result.get('dfp'),
                       })
                    else:
                       self.session.cookies.update({
                           'RAIL_EXPIRATION': Config().RAIL_EXPIRATION,
                           'RAIL_DEVICEID': Config().RAIL_DEVICEID,
                       })
            except:
                return self.request_device_id2()
        else:
            return self.request_device_id2()

    @classmethod
    def wait_for_ready(cls):
        self = cls()
        if self.is_ready: return self
        stay_second(self.retry_time)
        return self.wait_for_ready()

    @classmethod
    def job_by_name(cls, name) -> Job:
        self = cls()
        for job in self.jobs:
            if job.job_name == name: return job
        return None

    @classmethod
    def job_by_name(cls, name) -> Job:
        self = cls()
        return objects_find_object_by_key_value(self.jobs, 'job_name', name)

    @classmethod
    def job_by_account_key(cls, account_key) -> Job:
        self = cls()
        return objects_find_object_by_key_value(self.jobs, 'account_key', account_key)

    @classmethod
    def get_query_api_type(cls):
        import re
        self = cls()
        if self.api_type:
            return self.api_type
        response = self.session.get(API_QUERY_INIT_PAGE)
        if response.status_code == 200:
            res = re.search(r'var CLeftTicketUrl = \'(.*)\';', response.text)
            try:
                self.api_type = res.group(1)
            except Exception:
                pass
        if not self.api_type:
            QueryLog.add_quick_log('查询地址获取失败, 正在重新获取...').flush()
            sleep(get_interval_num(self.interval))
        self.request_device_id(True)
        return cls.get_query_api_type()

# def get_jobs_from_cluster(self):
#     jobs = self.cluster.session.get_dict(Cluster.KEY_JOBS)
#     return jobs
#
# def update_jobs_of_cluster(self):
#     if config.CLUSTER_ENABLED and config.NODE_IS_MASTER:
#         return self.cluster.session.set_dict(Cluster.KEY_JOBS, self.query_jobs)
#
# def refresh_jobs(self):
#     if not config.CLUSTER_ENABLED: return
#     jobs = self.get_jobs_from_cluster()
#     if jobs != self.query_jobs:
#         self.jobs = []
#         self.query_jobs = jobs
#         QueryLog.add_quick_log(QueryLog.MESSAGE_JOBS_DID_CHANGED).flush()
#         self.init_jobs()



File: py12306/query/job.py

from datetime import timedelta
from datetime import datetime

from py12306.app import app_available_check
from py12306.cluster.cluster import Cluster
from py12306.config import Config
from py12306.helpers.api import LEFT_TICKETS
from py12306.helpers.station import Station
from py12306.helpers.type import OrderSeatType, SeatType
from py12306.log.query_log import QueryLog
from py12306.helpers.func import *
from py12306.log.user_log import UserLog
from py12306.order.order import Order
from py12306.user.user import User
from py12306.helpers.event import Event


class Job:
    """
    查询任务
    """
    id = 0
    is_alive = True
    job_name = None
    left_dates = []
    left_date = None
    stations = []
    left_station = ''
    arrive_station = ''
    left_station_code = ''
    arrive_station_code = ''
    from_time = timedelta(hours=0)
    to_time = timedelta(hours=24)

    account_key = 0
    allow_seats = []
    current_seat = None
    current_seat_name = ''
    current_order_seat = None
    allow_train_numbers = []
    except_train_numbers = []
    members = []
    member_num = 0
    member_num_take = 0  # 最终提交的人数
    passengers = []
    allow_less_member = False
    retry_time = 3

    interval = {}
    interval_additional = 0
    interval_additional_max = 5

    query = None
    cluster = None
    ticket_info = {}
    is_cdn = False
    query_time_out = 3
    INDEX_TICKET_NUM = 11
    INDEX_TRAIN_NUMBER = 3
    INDEX_TRAIN_NO = 2
    INDEX_LEFT_DATE = 13
    INDEX_LEFT_STATION = 6  # 4 5 始发 终点
    INDEX_ARRIVE_STATION = 7
    INDEX_ORDER_TEXT = 1  # 下单文字
    INDEX_SECRET_STR = 0
    INDEX_LEFT_TIME = 8
    INDEX_ARRIVE_TIME = 9

    max_buy_time = 32

    def __init__(self, info, query):
        self.cluster = Cluster()
        self.query = query
        self.init_data(info)
        self.update_interval()

    def init_data(self, info):
        self.id = md5(info)
        self.left_dates = info.get('left_dates')
        self.stations = info.get('stations')
        self.stations = [self.stations] if isinstance(self.stations, dict) else self.stations
        if not self.job_name:  # name 不能被修改
            self.job_name = info.get('job_name',
                                     '{} -> {}'.format(self.stations[0]['left'], self.stations[0]['arrive']))

        self.account_key = str(info.get('account_key'))
        self.allow_seats = info.get('seats')
        self.allow_train_numbers = info.get('train_numbers')
        self.except_train_numbers = info.get('except_train_numbers')
        self.members = list(map(str, info.get('members')))
        self.member_num = len(self.members)
        self.member_num_take = self.member_num
        self.allow_less_member = bool(info.get('allow_less_member'))
        period = info.get('period')
        if isinstance(period, dict):
            if 'from' in period:
                parts = period['from'].split(':')
                if len(parts) == 2:
                    self.from_time = timedelta(
                        hours=int(parts[0]), seconds=int(parts[1]))
            if 'to' in period:
                parts = period['to'].split(':')
                if len(parts) == 2:
                    self.to_time = timedelta(
                        hours=int(parts[0]), seconds=int(parts[1]))

    def update_interval(self):
        self.interval = self.query.interval

    def run(self):
        self.start()

    def start(self):
        """
        处理单个任务
        根据日期循环查询, 展示处理时间
        :param job:
        :return:
        """
        while True and self.is_alive:
            app_available_check()
            QueryLog.print_job_start(self.job_name)
            for station in self.stations:
                self.refresh_station(station)
                for date in self.left_dates:
                    self.left_date = date
                    response = self.query_by_date(date)
                    self.handle_response(response)
                    QueryLog.add_query_time_log(time=response.elapsed.total_seconds(), is_cdn=self.is_cdn)
                    if not self.is_alive: return
                    self.safe_stay()
                    if is_main_thread():
                        QueryLog.flush(sep='\t\t', publish=False)
            if not Config().QUERY_JOB_THREAD_ENABLED:
                QueryLog.add_quick_log('').flush(publish=False)
                break
            else:
                QueryLog.add_log('\n').flush(sep='\t\t', publish=False)
            if Const.IS_TEST: return

    def judge_date_legal(self, date):
        date_now = datetime.datetime.now()
        date_query = datetime.datetime.strptime(str(date), "%Y-%m-%d")
        diff = (date_query - date_now).days
        if date_now.day == date_query.day:
            diff = 0
        if diff < 0:
            msg = '乘车日期错误，比当前时间还早！！'
            QueryLog.add_quick_log(msg).flush(publish=False)
            raise RuntimeError(msg)
        elif diff > self.max_buy_time:
            msg = '乘车日期错误，超出一个月预售期！！'
            QueryLog.add_quick_log(msg).flush(publish=False)
            raise RuntimeError(msg)
        else:
            return date_query.strftime("%Y-%m-%d")

    def query_by_date(self, date):
        """
        通过日期进行查询
        :return:
        """
        date = self.judge_date_legal(date)
        from py12306.helpers.cdn import Cdn
        QueryLog.add_log(('\n' if not is_main_thread() else '') + QueryLog.MESSAGE_QUERY_START_BY_DATE.format(date,
                                                                                                              self.left_station,
                                                                                                              self.arrive_station))
        url = LEFT_TICKETS.get('url').format(left_date=date, left_station=self.left_station_code,
                                             arrive_station=self.arrive_station_code, type=self.query.api_type)
        if Config.is_cdn_enabled() and Cdn().is_ready:
            self.is_cdn = True
            return self.query.session.cdn_request(url, timeout=self.query_time_out, allow_redirects=False)
        self.is_cdn = False
        return self.query.session.get(url, timeout=self.query_time_out, allow_redirects=False)

    def handle_response(self, response):
        """
        错误判断
        余票判断
        小黑屋判断
        座位判断
        乘车人判断
        :param result:
        :return:
        """
        results = self.get_results(response)
        if not results:
            return False
        for result in results:
            self.ticket_info = ticket_info = result.split('|')
            if not self.is_trains_number_valid():  # 车次是否有效
                continue
            QueryLog.add_log(QueryLog.MESSAGE_QUERY_LOG_OF_EVERY_TRAIN.format(self.get_info_of_train_number()))
            if not self.is_has_ticket(ticket_info):
                continue
            allow_seats = self.allow_seats if self.allow_seats else list(
                Config.SEAT_TYPES.values())  # 未设置 则所有可用 TODO  合法检测
            self.handle_seats(allow_seats, ticket_info)
            if not self.is_alive: return

    def handle_seats(self, allow_seats, ticket_info):
        for seat in allow_seats:  # 检查座位是否有票
            self.set_seat(seat)
            ticket_of_seat = ticket_info[self.current_seat]
            if not self.is_has_ticket_by_seat(ticket_of_seat):  # 座位是否有效
                continue
            QueryLog.print_ticket_seat_available(left_date=self.get_info_of_left_date(),
                                                 train_number=self.get_info_of_train_number(), seat_type=seat,
                                                 rest_num=ticket_of_seat)
            if not self.is_member_number_valid(ticket_of_seat):  # 乘车人数是否有效
                if self.allow_less_member:
                    self.member_num_take = int(ticket_of_seat)
                    QueryLog.print_ticket_num_less_than_specified(ticket_of_seat, self)
                else:
                    QueryLog.add_quick_log(
                        QueryLog.MESSAGE_GIVE_UP_CHANCE_CAUSE_TICKET_NUM_LESS_THAN_SPECIFIED).flush()
                    continue
            if Const.IS_TEST: return
            # 检查完成 开始提交订单
            QueryLog.print_ticket_available(left_date=self.get_info_of_left_date(),
                                            train_number=self.get_info_of_train_number(),
                                            rest_num=ticket_of_seat)
            if User.is_empty():
                QueryLog.add_quick_log(QueryLog.MESSAGE_USER_IS_EMPTY_WHEN_DO_ORDER.format(self.retry_time))
                return stay_second(self.retry_time)

            order_result = False
            user = self.get_user()
            if not user:
                QueryLog.add_quick_log(QueryLog.MESSAGE_ORDER_USER_IS_EMPTY.format(self.retry_time))
                return stay_second(self.retry_time)

            lock_id = Cluster.KEY_LOCK_DO_ORDER + '_' + user.key
            if Config().is_cluster_enabled():
                if self.cluster.get_lock(lock_id, Cluster.lock_do_order_time,
                                         {'node': self.cluster.node_name}):  # 获得下单锁
                    order_result = self.do_order(user)
                    if not order_result:  # 下单失败，解锁
                        self.cluster.release_lock(lock_id)
                else:
                    QueryLog.add_quick_log(
                        QueryLog.MESSAGE_SKIP_ORDER.format(self.cluster.get_lock_info(lock_id).get('node'),
                                                           user.user_name))
                    stay_second(self.retry_time)  # 防止过多重复
            else:
                order_result = self.do_order(user)

            # 任务已成功 通知集群停止任务
            if order_result:
                Event().job_destroy({'name': self.job_name})

    def do_order(self, user):
        self.check_passengers()
        order = Order(user=user, query=self)
        return order.order()

    def get_results(self, response):
        """
        解析查询返回结果
        :param response:
        :return:
        """
        if response.status_code != 200:
            QueryLog.print_query_error(response.reason, response.status_code)
            if self.interval_additional < self.interval_additional_max:
                self.interval_additional += self.interval.get('min')
        else:
            self.interval_additional = 0
        result = response.json().get('data.result')
        return result if result else False

    def is_has_ticket(self, ticket_info):
        return self.get_info_of_ticket_num() == 'Y' and self.get_info_of_order_text() == '预订'

    def is_has_ticket_by_seat(self, seat):
        return seat != '' and seat != '无' and seat != '*'

    def is_trains_number_valid(self):
        train_left_time = self.get_info_of_train_left_time()
        time_parts = train_left_time.split(':')
        left_time = timedelta(
            hours=int(time_parts[0]), seconds=int(time_parts[1]))
        if left_time < self.from_time or left_time > self.to_time:
            return False

        if self.except_train_numbers:
            return self.get_info_of_train_number().upper() not in map(str.upper, self.except_train_numbers)
        if self.allow_train_numbers:
            return self.get_info_of_train_number().upper() in map(str.upper, self.allow_train_numbers)
        return True

    def is_member_number_valid(self, seat):
        return seat == '有' or self.member_num <= int(seat)

    def destroy(self):
        """
        退出任务
        :return:
        """
        from py12306.query.query import Query
        self.is_alive = False
        QueryLog.add_quick_log(QueryLog.MESSAGE_QUERY_JOB_BEING_DESTROY.format(self.job_name)).flush()
        # sys.exit(1) # 无法退出线程...
        # 手动移出jobs 防止单线程死循环
        index = Query().jobs.index(self)
        Query().jobs.pop(index)

    def safe_stay(self):
        origin_interval = get_interval_num(self.interval)
        interval = origin_interval + self.interval_additional
        QueryLog.add_stay_log(
            '%s + %s' % (origin_interval, self.interval_additional) if self.interval_additional else origin_interval)
        stay_second(interval)

    def set_passengers(self, passengers):
        UserLog.print_user_passenger_init_success(passengers)
        self.passengers = passengers

    def set_seat(self, seat):
        self.current_seat_name = seat
        self.current_seat = SeatType.dicts.get(seat)
        self.current_order_seat = OrderSeatType.dicts.get(seat)

    def get_user(self):
        user = User.get_user(self.account_key)
        # if not user.check_is_ready(): # 这里不需要检测了，后面获取乘客时已经检测过
        #     #
        #     pass
        return user

    def check_passengers(self):
        if not self.passengers:
            QueryLog.add_quick_log(QueryLog.MESSAGE_CHECK_PASSENGERS.format(self.job_name)).flush()
            passengers = User.get_passenger_for_members(self.members, self.account_key)
            if passengers:
                self.set_passengers(passengers)
            else:  # 退出当前查询任务
                self.destroy()
        return True

    def refresh_station(self, station):
        self.left_station = station.get('left')
        self.arrive_station = station.get('arrive')
        self.left_station_code = Station.get_station_key_by_name(self.left_station)
        self.arrive_station_code = Station.get_station_key_by_name(self.arrive_station)

    # 提供一些便利方法
    def get_info_of_left_date(self):
        return self.ticket_info[self.INDEX_LEFT_DATE]

    def get_info_of_ticket_num(self):
        return self.ticket_info[self.INDEX_TICKET_NUM]

    def get_info_of_train_number(self):
        return self.ticket_info[self.INDEX_TRAIN_NUMBER]

    def get_info_of_train_no(self):
        return self.ticket_info[self.INDEX_TRAIN_NO]

    def get_info_of_left_station(self):
        return Station.get_station_name_by_key(self.ticket_info[self.INDEX_LEFT_STATION])

    def get_info_of_arrive_station(self):
        return Station.get_station_name_by_key(self.ticket_info[self.INDEX_ARRIVE_STATION])

    def get_info_of_order_text(self):
        return self.ticket_info[self.INDEX_ORDER_TEXT]

    def get_info_of_secret_str(self):
        return self.ticket_info[self.INDEX_SECRET_STR]

    def get_info_of_train_left_time(self):
        return self.ticket_info[self.INDEX_LEFT_TIME]

    def get_info_of_train_arrive_time(self):
        return self.ticket_info[self.INDEX_ARRIVE_TIME]



File: py12306/query/__init__.py




File: py12306/helpers/auth_code.py

import random
import time

from requests.exceptions import SSLError

from py12306.config import Config
from py12306.helpers.OCR import OCR
from py12306.helpers.api import *
from py12306.helpers.request import Request
from py12306.helpers.func import *
from py12306.log.common_log import CommonLog
from py12306.log.user_log import UserLog


class AuthCode:
    """
    验证码类
    """
    session = None
    data_path = None
    retry_time = 5

    def __init__(self, session):
        self.data_path = Config().RUNTIME_DIR
        self.session = session

    @classmethod
    def get_auth_code(cls, session):
        self = cls(session)
        img = self.download_code()
        position = OCR.get_img_position(img)
        if not position:  # 打码失败
            return self.retry_get_auth_code()

        answer = ','.join(map(str, position))

        if not self.check_code(answer):
            return self.retry_get_auth_code()
        return position

    def retry_get_auth_code(self): # TODO 安全次数检测
        CommonLog.add_quick_log(CommonLog.MESSAGE_RETRY_AUTH_CODE.format(self.retry_time)).flush()
        time.sleep(self.retry_time)
        return self.get_auth_code(self.session)

    def download_code(self):
        url = API_AUTH_CODE_BASE64_DOWNLOAD.format(random=random.random())
        # code_path = self.data_path + 'code.png'
        try:
            self.session.cookies.clear_session_cookies()
            UserLog.add_quick_log(UserLog.MESSAGE_DOWNLAODING_THE_CODE).flush()
            # response = self.session.save_to_file(url, code_path)  # TODO 返回错误情况
            response = self.session.get(url)
            result = response.json()
            if result.get('image'):
                return result.get('image')
            raise SSLError('返回数据为空')
        except SSLError as e:
            UserLog.add_quick_log(
                UserLog.MESSAGE_DOWNLAOD_AUTH_CODE_FAIL.format(e, self.retry_time)).flush()
            time.sleep(self.retry_time)
            return self.download_code()

    def check_code(self, answer):
        """
        校验验证码
        :return:
        """
        url = API_AUTH_CODE_CHECK.get('url').format(answer=answer, random=time_int())
        response = self.session.get(url)
        result = response.json()
        if result.get('result_code') == '4':
            UserLog.add_quick_log(UserLog.MESSAGE_CODE_AUTH_SUCCESS).flush()
            return True
        else:
            # {'result_message': '验证码校验失败', 'result_code': '5'}
            UserLog.add_quick_log(
                UserLog.MESSAGE_CODE_AUTH_FAIL.format(result.get('result_message'))).flush()
            self.session.cookies.clear_session_cookies()

        return False


if __name__ == '__main__':
    code_result = AuthCode.get_auth_code()



File: py12306/helpers/event.py

from py12306.helpers.func import *
from py12306.config import Config


@singleton
class Event():
    """
    处理事件
    """
    # 事件
    KEY_JOB_DESTROY = 'job_destroy'
    KEY_USER_JOB_DESTROY = 'user_job_destroy'
    KEY_USER_LOADED = 'user_loaded'
    cluster = None

    def __init__(self):
        from py12306.cluster.cluster import Cluster
        self.cluster = Cluster()

    def job_destroy(self, data={}, callback=False):  # 停止查询任务
        from py12306.query.query import Query
        if Config().is_cluster_enabled() and not callback:
            return self.cluster.publish_event(self.KEY_JOB_DESTROY, data)  # 通知其它节点退出

        job = Query.job_by_name(data.get('name'))
        if job:
            job.destroy()

    def user_loaded(self, data={}, callback=False):  # 用户初始化完成
        if Config().is_cluster_enabled() and not callback:
            return self.cluster.publish_event(self.KEY_USER_LOADED, data)  # 通知其它节点退出
        from py12306.query.query import Query

        if not Config().is_cluster_enabled() or Config().is_master():
            query = Query.wait_for_ready()
            for job in query.jobs:
                if job.account_key == data.get('key'):
                    create_thread_and_run(job, 'check_passengers', Const.IS_TEST)  # 检查乘客信息 防止提交订单时才检查
                    stay_second(1)

    def user_job_destroy(self, data={}, callback=False):
        from py12306.user.user import User
        if Config().is_cluster_enabled() and not callback:
            return self.cluster.publish_event(self.KEY_JOB_DESTROY, data)  # 通知其它节点退出

        user = User.get_user(data.get('key'))
        if user:
            user.destroy()



File: py12306/helpers/notification.py

import urllib

from py12306.config import Config
from py12306.helpers.api import *
from py12306.helpers.request import Request
from py12306.log.common_log import CommonLog


class Notification():
    """
    通知类
    """
    session = None

    def __init__(self):
        self.session = Request()

    @classmethod
    def voice_code(cls, phone, name='', content=''):
        self = cls()
        if Config().NOTIFICATION_VOICE_CODE_TYPE == 'dingxin':
            self.send_voice_code_of_dingxin(phone, name=name, info=content)
        else:
            self.send_voice_code_of_yiyuan(phone, name=name, content=content)

    @classmethod
    def dingtalk_webhook(cls, content=''):
        self = cls()
        self.send_dingtalk_by_webbook(content=content)

    @classmethod
    def send_email(cls, to, title='', content=''):
        self = cls()
        self.send_email_by_smtp(to, title, content)

    @classmethod
    def send_email_with_qrcode(cls, to, title='', qrcode_path=''):
        self = cls()
        self.send_email_by_smtp_with_qrcode(to, title, qrcode_path)

    @classmethod
    def send_to_telegram(cls, content=''):
        self = cls()
        self.send_to_telegram_bot(content=content)

    @classmethod
    def server_chan(cls, skey='', title='', content=''):
        self = cls()
        self.send_serverchan(skey=skey, title=title, content=content)

    @classmethod
    def push_bear(cls, skey='', title='', content=''):
        self = cls()
        self.send_pushbear(skey=skey, title=title, content=content)

    @classmethod
    def push_bark(cls, content=''):
        self = cls()
        self.push_to_bark(content)

    def send_voice_code_of_yiyuan(self, phone, name='', content=''):
        """
        发送语音验证码
        购买地址 https://market.aliyun.com/products/57126001/cmapi019902.html?spm=5176.2020520132.101.5.37857218O6iJ3n
        :return:
        """
        appcode = Config().NOTIFICATION_API_APP_CODE
        if not appcode:
            CommonLog.add_quick_log(CommonLog.MESSAGE_EMPTY_APP_CODE).flush()
            return False
        body = {
            'userName': name,
            'mailNo': content
        }
        params = {
            'content': body,
            'mobile': phone,
            'sex': 2,
            'tNum': 'T170701001056'
        }
        response = self.session.request(url=API_NOTIFICATION_BY_VOICE_CODE + urllib.parse.urlencode(params),
                                        method='GET', headers={'Authorization': 'APPCODE {}'.format(appcode)})
        result = response.json()
        response_message = result.get('showapi_res_body.remark')
        if response.status_code in [400, 401, 403]:
            return CommonLog.add_quick_log(CommonLog.MESSAGE_VOICE_API_FORBID).flush()
        if response.status_code == 200 and result.get('showapi_res_body.flag'):
            CommonLog.add_quick_log(CommonLog.MESSAGE_VOICE_API_SEND_SUCCESS.format(response_message)).flush()
            return True
        else:
            return CommonLog.add_quick_log(CommonLog.MESSAGE_VOICE_API_SEND_FAIL.format(response_message)).flush()

    def send_voice_code_of_dingxin(self, phone, name='', info={}):
        """
        发送语音验证码 ( 鼎信 )
        购买地址 https://market.aliyun.com/products/56928004/cmapi026600.html?spm=5176.2020520132.101.2.51547218rkAXxy
        :return:
        """
        appcode = Config().NOTIFICATION_API_APP_CODE
        if not appcode:
            CommonLog.add_quick_log(CommonLog.MESSAGE_EMPTY_APP_CODE).flush()
            return False
        data = {
            'tpl_id': 'TP1901174',
            'phone': phone,
            'param': 'name:{name},job_name:{left_station}到{arrive_station}{set_type},orderno:{orderno}'.format(
                name=name, left_station=info.get('left_station'), arrive_station=info.get('arrive_station'),
                set_type=info.get('set_type'), orderno=info.get('orderno'))
        }
        response = self.session.request(url=API_NOTIFICATION_BY_VOICE_CODE_DINGXIN, method='POST', data=data,
                                        headers={'Authorization': 'APPCODE {}'.format(appcode)})
        result = response.json()
        response_message = result.get('return_code')
        if response.status_code in [400, 401, 403]:
            return CommonLog.add_quick_log(CommonLog.MESSAGE_VOICE_API_FORBID).flush()
        if response.status_code == 200 and result.get('return_code') == '00000':
            CommonLog.add_quick_log(CommonLog.MESSAGE_VOICE_API_SEND_SUCCESS.format(response_message)).flush()
            return True
        else:
            return CommonLog.add_quick_log(CommonLog.MESSAGE_VOICE_API_SEND_FAIL.format(response_message)).flush()

    def send_email_by_smtp(self, to, title, content):
        import smtplib
        from email.message import EmailMessage
        to = to if isinstance(to, list) else [to]
        message = EmailMessage()
        message['Subject'] = title
        message['From'] = Config().EMAIL_SENDER
        message['To'] = to
        message.set_content(content)
        try:
            server = smtplib.SMTP(Config().EMAIL_SERVER_HOST)
            server.ehlo()
            server.starttls()
            server.login(Config().EMAIL_SERVER_USER, Config().EMAIL_SERVER_PASSWORD)
            server.send_message(message)
            server.quit()
            CommonLog.add_quick_log(CommonLog.MESSAGE_SEND_EMAIL_SUCCESS).flush()
        except Exception as e:
            CommonLog.add_quick_log(CommonLog.MESSAGE_SEND_EMAIL_FAIL.format(e)).flush()

    def send_email_by_smtp_with_qrcode(self, to, title, qrcode_path):
        import smtplib
        from email.mime.text import MIMEText
        from email.mime.multipart import MIMEMultipart
        from email.mime.image import MIMEImage
        to = to if isinstance(to, list) else [to]
        message = MIMEMultipart()
        message['Subject'] = title
        message['From'] = Config().EMAIL_SENDER
        message['To'] = ", ".join(to)
        htmlFile = """
        <html>
          <head></head>
            <body>
              <p>
                这是你的二维码
              </p>
              <p>
              <br /><img src="cid:0", width=200, height=200  ></p>
            </body>
        </html>
        """
        htmlApart = MIMEText(htmlFile, 'html')
        imageFile = qrcode_path
        imageApart = MIMEImage(open(imageFile, 'rb').read(), imageFile.split('.')[-1])
        imageApart.add_header('Content-ID', '<0>')
        message.attach(imageApart)
        message.attach(htmlApart)
        try:
            server = smtplib.SMTP(Config().EMAIL_SERVER_HOST)
            server.ehlo()
            server.starttls()
            server.login(Config().EMAIL_SERVER_USER, Config().EMAIL_SERVER_PASSWORD)
            server.send_message(message)
            server.quit()
            CommonLog.add_quick_log(CommonLog.MESSAGE_SEND_EMAIL_WITH_QRCODE_SUCCESS).flush()
            self.push_bark(CommonLog.MESSAGE_SEND_EMAIL_WITH_QRCODE_SUCCESS)
        except Exception as e:
            CommonLog.add_quick_log(CommonLog.MESSAGE_SEND_EMAIL_FAIL.format(e)).flush()

    def send_dingtalk_by_webbook(self, content):
        from dingtalkchatbot.chatbot import DingtalkChatbot
        webhook = Config().DINGTALK_WEBHOOK
        dingtalk = DingtalkChatbot(webhook)
        dingtalk.send_text(msg=content, is_at_all=True)
        pass

    def send_to_telegram_bot(self, content):
        bot_api_url = Config().TELEGRAM_BOT_API_URL
        if not bot_api_url:
            return False
        data = {
            'text': content
        }
        response = self.session.request(url=bot_api_url, method='POST', data=data)
        result = response.json().get('result')
        response_status = result.get('statusCode')
        if response_status == 200:
            CommonLog.add_quick_log(CommonLog.MESSAGE_SEND_TELEGRAM_SUCCESS).flush()
        else:
            response_error_message = result.get('description')
            CommonLog.add_quick_log(CommonLog.MESSAGE_SEND_TELEGRAM_FAIL.format(response_error_message)).flush()

    def push_to_bark(self, content):
        bark_url = Config().BARK_PUSH_URL
        if not bark_url:
            return False

        response = self.session.request(url=bark_url + '/' + content, method='get')
        result = response.json()
        response_status = result.get('code')
        if response_status == 200:
            CommonLog.add_quick_log(CommonLog.MESSAGE_SEND_BARK_SUCCESS).flush()
        else:
            response_error_message = result.get('message')
            CommonLog.add_quick_log(CommonLog.MESSAGE_SEND_BARK_FAIL.format(response_error_message)).flush()

    def send_serverchan(self, skey, title, content):
        from lightpush import lightpush
        lgp = lightpush()
        lgp.set_single_push(key=skey)
        try:
            lgp.single_push(title, content)
            CommonLog.add_quick_log(CommonLog.MESSAGE_SEND_SERVER_CHAN_SUCCESS).flush()
        except Exception as e:
            CommonLog.add_quick_log(CommonLog.MESSAGE_SEND_SERVER_CHAN_FAIL.format(e)).flush()

    def send_pushbear(self, skey, title, content):
        from lightpush import lightpush
        lgp = lightpush()
        lgp.set_group_push(key=skey)
        try:
            lgp.group_push(title, content)
            CommonLog.add_quick_log(CommonLog.MESSAGE_SEND_PUSH_BEAR_SUCCESS).flush()
        except Exception as e:
            CommonLog.add_quick_log(CommonLog.MESSAGE_SEND_PUSH_BEAR_SUCCESS.format(e)).flush()


if __name__ == '__main__':
    name = '张三4'
    content = '你的车票 广州 到 深圳 购买成功，请登录 12306 进行支付'
    # Notification.voice_code('13800138000', name, content)
    # Notification.send_email('user@email.com', name, content)
    # Notification.dingtalk_webhook(content)
    Notification.voice_code('13800138000', name, {
        'left_station': '广州',
        'arrive_station': '深圳',
        'set_type': '硬座',
        'orderno': 'E123542'
    })



File: py12306/helpers/request.py

import requests
from requests.exceptions import *

from py12306.helpers.func import *
from requests_html import HTMLSession, HTMLResponse

requests.packages.urllib3.disable_warnings()


class Request(HTMLSession):
    """
    请求处理类
    """

    # session = {}
    def save_to_file(self, url, path):
        response = self.get(url, stream=True)
        with open(path, 'wb') as f:
            for chunk in response.iter_content(chunk_size=1024):
                f.write(chunk)
        return response

    @staticmethod
    def _handle_response(response, **kwargs) -> HTMLResponse:
        """
        扩充 response
        :param response:
        :param kwargs:
        :return:
        """
        response = HTMLSession._handle_response(response, **kwargs)
        expand_class(response, 'json', Request.json)
        return response

    def add_response_hook(self, hook):
        hooks = self.hooks['response']
        if not isinstance(hooks, list):
            hooks = [hooks]
        hooks.append(hook)
        self.hooks['response'] = hooks
        return self

    def json(self, default={}):
        """
        重写 json 方法，拦截错误
        :return:
        """
        from py12306.app import Dict
        try:
            result = self.old_json()
            return Dict(result)
        except:
            return Dict(default)

    def request(self, *args, **kwargs):  # 拦截所有错误
        try:
            if not 'timeout' in kwargs:
                from py12306.config import Config
                kwargs['timeout'] = Config().TIME_OUT_OF_REQUEST
            response = super().request(*args, **kwargs)
            return response
        except RequestException as e:
            from py12306.log.common_log import CommonLog
            if e.response:
                response = e.response
            else:
                response = HTMLResponse(HTMLSession)
                # response.status_code = 500
                expand_class(response, 'json', Request.json)
            response.reason = response.reason if response.reason else CommonLog.MESSAGE_RESPONSE_EMPTY_ERROR
            return response

    def cdn_request(self, url: str, cdn=None, method='GET', **kwargs):
        from py12306.helpers.api import HOST_URL_OF_12306
        from py12306.helpers.cdn import Cdn
        if not cdn: cdn = Cdn.get_cdn()
        url = url.replace(HOST_URL_OF_12306, cdn)

        return self.request(method, url, headers={'Host': HOST_URL_OF_12306}, verify=False, **kwargs)

    def dump_cookies(self):
        cookies = []
        for _, item in self.cookies._cookies.items():
            for _, urls in item.items():
                for _, cookie in urls.items():
                    from http.cookiejar import Cookie
                    assert isinstance(cookie, Cookie)
                    if cookie.domain:
                        cookies.append({
                            'name': cookie.name,
                            'value': cookie.value,
                            'url': 'https://' + cookie.domain + cookie.path,
                        })
        return cookies



File: py12306/helpers/__init__.py




File: py12306/helpers/type.py

from py12306.helpers.func import *


@singleton
class UserType:
    ADULT = 1
    CHILD = 2
    STUDENT = 3
    SOLDIER = 4

    dicts = {
        '成人': ADULT,
        '儿童': CHILD,
        '学生': STUDENT,
        '残疾军人、伤残人民警察': SOLDIER,
    }


@singleton
class OrderSeatType:
    dicts = {
        '特等座': 'P',
        '商务座': 9,
        '一等座': 'M',
        '二等座': 'O',
        '软卧': 4,
        '硬卧': 3,
        '动卧': 1,
        '软座': 2,
        '硬座': 1,
        '无座': 1,
    }


@singleton
class SeatType:
    NO_SEAT = 26
    dicts = {
        '特等座': 25,
        '商务座': 32,
        '一等座': 31,
        '二等座': 30,
        '软卧': 23,
        '硬卧': 28,
        '动卧': 33,
        '软座': 24,
        '硬座': 29,
        '无座': NO_SEAT,
    }



File: py12306/helpers/qrcode.py

# -*- coding: utf-8 -*-

import png


def print_qrcode(path):
    """
    将二维码输出到控制台
    需要终端尺寸足够大才能显示

    :param path: 二维码图片路径 (PNG 格式)
    :return: None
    """
    reader = png.Reader(path)
    width, height, rows, info = reader.read()
    lines = list(rows)

    planes = info['planes']  # 通道数
    threshold = (2 ** info['bitdepth']) / 2  # 色彩阈值

    # 识别二维码尺寸
    x_flag = -1   # x 边距标志
    y_flag = -1   # y 边距标志
    x_white = -1  # 定位图案白块 x 坐标
    y_white = -1  # 定位图案白块 y 坐标

    i = y_flag
    while i < height:
        if y_white > 0 and x_white > 0:
            break
        j = x_flag
        while j < width:
            total = 0
            for k in range(planes):
                px = lines[i][j * planes + k]
                total += px
            avg = total / planes
            black = avg < threshold
            if y_white > 0 and x_white > 0:
                break
            if x_flag > 0 > x_white and not black:
                x_white = j
            if x_flag == -1 and black:
                x_flag = j
            if y_flag > 0 > y_white and not black:
                y_white = i
            if y_flag == -1 and black:
                y_flag = i
            if x_flag > 0 and y_flag > 0:
                i += 1
            j += 1
        i += 1

    assert y_white - y_flag == x_white - x_flag
    scale = y_white - y_flag

    assert width - x_flag == height - y_flag
    module_count = int((width - x_flag * 2) / scale)

    whole_white = '█'
    whole_black = ' '
    down_black = '▀'
    up_black = '▄'

    dual_flag = False
    last_line = []
    output = '\n'
    for i in range(module_count + 2):
        output += up_black
    output += '\n'
    i = y_flag
    while i < height - y_flag:
        if dual_flag:
            output += whole_white
        t = 0
        j = x_flag
        while j < width - x_flag:
            total = 0
            for k in range(planes):
                px = lines[i][j * planes + k]
                total += px
            avg = total / planes
            black = avg < threshold
            if dual_flag:
                last_black = last_line[t]
                if black and last_black:
                    output += whole_black
                elif black and not last_black:
                    output += down_black
                elif not black and last_black:
                    output += up_black
                elif not black and not last_black:
                    output += whole_white
            else:
                last_line[t:t+1] = [black]
            t = t + 1
            j += scale
        if dual_flag:
            output += whole_white + '\n'
        dual_flag = not dual_flag
        i += scale
    output += whole_white
    for i in range(module_count):
        output += up_black if last_line[i] else whole_white
    output += whole_white + '\n'
    print(output, flush=True)



File: py12306/helpers/OCR.py

import math
import random

from py12306.config import Config
from py12306.helpers.api import API_FREE_CODE_QCR_API
from py12306.helpers.request import Request
from py12306.log.common_log import CommonLog
from py12306.vender.ruokuai.main import RKClient


class OCR:
    """
    图片识别
    """
    session = None

    def __init__(self):
        self.session = Request()

    @classmethod
    def get_img_position(cls, img):
        """
        获取图像坐标
        :param img_path:
        :return:
        """
        self = cls()
        if Config().AUTO_CODE_PLATFORM == 'free' or Config().AUTO_CODE_PLATFORM == 'user':
            return self.get_image_by_free_site(img)
        return self.get_img_position_by_ruokuai(img)

    def get_img_position_by_ruokuai(self, img):
        ruokuai_account = Config().AUTO_CODE_ACCOUNT
        soft_id = '119671'
        soft_key = '6839cbaca1f942f58d2760baba5ed987'
        rc = RKClient(ruokuai_account.get('user'), ruokuai_account.get('pwd'), soft_id, soft_key)
        result = rc.rk_create(img, 6113)
        if "Result" in result:
            return self.get_image_position_by_offset(list(result['Result']))
        CommonLog.print_auto_code_fail(result.get("Error", CommonLog.MESSAGE_RESPONSE_EMPTY_ERROR))
        return None

    def get_image_position_by_offset(self, offsets):
        positions = []
        width = 75
        height = 75
        for offset in offsets:
            random_x = random.randint(-5, 5)
            random_y = random.randint(-5, 5)
            offset = int(offset)
            x = width * ((offset - 1) % 4 + 1) - width / 2 + random_x
            y = height * math.ceil(offset / 4) - height / 2 + random_y
            positions.append(int(x))
            positions.append(int(y))
        return positions

    def get_image_by_free_site(self, img):
        data = {
            'img': img
        }
        if Config().AUTO_CODE_PLATFORM == 'free':
           response = self.session.post(API_FREE_CODE_QCR_API, data=data, timeout=30)
        else:
           response = self.session.post(Config().API_USER_CODE_QCR_API, data=data, timeout=30)
        result = response.json()
        if result.get('msg') == 'success':
            pos = result.get('result')
            return self.get_image_position_by_offset(pos)

        CommonLog.print_auto_code_fail(CommonLog.MESSAGE_GET_RESPONSE_FROM_FREE_AUTO_CODE)
        return None


if __name__ == '__main__':
    pass
    # code_result = AuthCode.get_auth_code()



File: py12306/helpers/api.py

# coding=utf-8

HOST_URL_OF_12306 = 'kyfw.12306.cn'
BASE_URL_OF_12306 = 'https://' + HOST_URL_OF_12306

LEFT_TICKETS = {
    "url": BASE_URL_OF_12306 + "/otn/{type}?leftTicketDTO.train_date={left_date}&leftTicketDTO.from_station={left_station}&leftTicketDTO.to_station={arrive_station}&purpose_codes=ADULT",
}

API_BASE_LOGIN = {
    "url": BASE_URL_OF_12306 + '/passport/web/login',
}

API_USER_LOGIN_CHECK = BASE_URL_OF_12306 + '/otn/login/conf'

API_AUTH_QRCODE_BASE64_DOWNLOAD = {
    'url': BASE_URL_OF_12306 + '/passport/web/create-qr64'
}

API_AUTH_QRCODE_CHECK = {
    'url': BASE_URL_OF_12306 + '/passport/web/checkqr'
}

API_USER_LOGIN = {
    'url': BASE_URL_OF_12306 + '/otn/login/userLogin'
}

API_AUTH_CODE_DOWNLOAD = {
    'url': BASE_URL_OF_12306 + '/passport/captcha/captcha-image?login_site=E&module=login&rand=sjrand&_={random}'
}
API_AUTH_CODE_BASE64_DOWNLOAD = BASE_URL_OF_12306 + '/passport/captcha/captcha-image64?login_site=E&module=login&rand=sjrand&_={random}'
API_AUTH_CODE_CHECK = {
    'url': BASE_URL_OF_12306 + '/passport/captcha/captcha-check?answer={answer}&rand=sjrand&login_site=E&_={random}'
}
API_AUTH_UAMTK = {
    'url': BASE_URL_OF_12306 + '/passport/web/auth/uamtk'
}
API_AUTH_UAMAUTHCLIENT = {
    'url': BASE_URL_OF_12306 + '/otn/uamauthclient'
}

API_USER_INFO = {
    'url': BASE_URL_OF_12306 + '/otn/modifyUser/initQueryUserInfoApi'
}
API_USER_PASSENGERS = BASE_URL_OF_12306 + '/otn/confirmPassenger/getPassengerDTOs'
API_SUBMIT_ORDER_REQUEST = BASE_URL_OF_12306 + '/otn/leftTicket/submitOrderRequest'
API_CHECK_ORDER_INFO = BASE_URL_OF_12306 + '/otn/confirmPassenger/checkOrderInfo'
API_INITDC_URL = BASE_URL_OF_12306 + '/otn/confirmPassenger/initDc'  # 生成订单时需要先请求这个页面
API_GET_QUEUE_COUNT = BASE_URL_OF_12306 + '/otn/confirmPassenger/getQueueCount'
API_CONFIRM_SINGLE_FOR_QUEUE = BASE_URL_OF_12306 + '/otn/confirmPassenger/confirmSingleForQueue'
API_QUERY_ORDER_WAIT_TIME = BASE_URL_OF_12306 + '/otn/confirmPassenger/queryOrderWaitTime?{}'  # 排队查询
API_QUERY_INIT_PAGE = BASE_URL_OF_12306 + '/otn/leftTicket/init'
# API_GET_BROWSER_DEVICE_ID = BASE_URL_OF_12306 + '/otn/HttpZF/logdevice'
API_GET_BROWSER_DEVICE_ID = 'https://12306-rail-id-v2.pjialin.com/'
API_FREE_CODE_QCR_API = 'https://12306-ocr.pjialin.com/check/'

API_NOTIFICATION_BY_VOICE_CODE = 'http://ali-voice.showapi.com/sendVoice?'
API_NOTIFICATION_BY_VOICE_CODE_DINGXIN = 'http://yuyin2.market.alicloudapi.com/dx/voice_notice'

API_CHECK_CDN_AVAILABLE = 'https://{}/otn/dynamicJs/omseuuq'



File: py12306/helpers/station.py

from os import path

from py12306.config import Config
from py12306.helpers.func import *


@singleton
class Station:
    stations = []
    station_kvs = {}

    def __init__(self):
        if path.exists(Config().STATION_FILE):
            result = open(Config().STATION_FILE, encoding='utf-8').read()
            result = result.lstrip('@').split('@')
            for i in result:
                tmp_info = i.split('|')
                self.stations.append({
                    'key': tmp_info[2],
                    'name': tmp_info[1],
                    'pinyin': tmp_info[3],
                    'id': tmp_info[5]
                })
                self.station_kvs[tmp_info[1]] = tmp_info[2]

    @classmethod
    def get_station_by_name(cls, name):
        return cls.get_station_by(name, 'name')

    @classmethod
    def get_station_by(cls, value, field):
        self = cls()
        for station in self.stations:
            if station.get(field) == value:
                return station
        return None

    @classmethod
    def get_station_key_by_name(cls, name):
        self = cls()
        return self.station_kvs[name]

    @classmethod
    def get_station_name_by_key(cls, key):
        return cls.get_station_by(key, 'key').get('name')




File: py12306/helpers/cdn.py

import random
import json
from datetime import timedelta
from os import path

from py12306.cluster.cluster import Cluster
from py12306.config import Config
from py12306.app import app_available_check
from py12306.helpers.api import API_CHECK_CDN_AVAILABLE, HOST_URL_OF_12306
from py12306.helpers.func import *
from py12306.helpers.request import Request
from py12306.log.common_log import CommonLog


@singleton
class Cdn:
    """
    CDN 管理
    """
    items = []
    available_items = []
    unavailable_items = []
    recheck_available_items = []
    recheck_unavailable_items = []
    retry_time = 3
    is_ready = False
    is_finished = False
    is_ready_num = 10  # 当可用超过 10，已准备好
    is_alive = True
    is_recheck = False

    safe_stay_time = 0.2
    retry_num = 1
    thread_num = 5
    check_time_out = 3

    last_check_at = 0
    save_second = 5
    check_keep_second = 60 * 60 * 24

    def __init__(self):
        self.cluster = Cluster()
        self.init_config()
        create_thread_and_run(self, 'watch_cdn', False)

    def init_data(self):
        self.items = []
        self.available_items = []
        self.unavailable_items = []
        self.is_finished = False
        self.is_ready = False
        self.is_recheck = False

    def init_config(self):
        self.check_time_out = Config().CDN_CHECK_TIME_OUT

    def update_cdn_status(self, auto=False):
        if auto:
            self.init_config()
            if Config().is_cdn_enabled():
                self.run()
            else:
                self.destroy()

    @classmethod
    def run(cls):
        self = cls()
        app_available_check()
        self.is_alive = True
        self.start()
        pass

    def start(self):
        if not Config.is_cdn_enabled(): return
        self.load_items()
        CommonLog.add_quick_log(CommonLog.MESSAGE_CDN_START_TO_CHECK.format(len(self.items))).flush()
        self.restore_items()
        for i in range(self.thread_num):  # 多线程
            create_thread_and_run(jobs=self, callback_name='check_available', wait=False)

    def load_items(self):
        with open(Config().CDN_ITEM_FILE, encoding='utf-8') as f:
            for line, val in enumerate(f):
                self.items.append(val.rstrip('\n'))

    def restore_items(self):
        """
        恢复已有数据
        :return: bool
        """
        result = False
        if path.exists(Config().CDN_ENABLED_AVAILABLE_ITEM_FILE):
            with open(Config().CDN_ENABLED_AVAILABLE_ITEM_FILE, encoding='utf-8') as f:
                result = f.read()
                try:
                    result = json.loads(result)
                except json.JSONDecodeError as e:
                    result = {}

        # if Config.is_cluster_enabled(): # 集群不用同步 cdn
        #     result = self.get_data_from_cluster()

        if result:
            self.last_check_at = result.get('last_check_at', '')
            if self.last_check_at: self.last_check_at = str_to_time(self.last_check_at)
            self.available_items = result.get('items', [])
            self.unavailable_items = result.get('fail_items', [])
            CommonLog.add_quick_log(CommonLog.MESSAGE_CDN_RESTORE_SUCCESS.format(self.last_check_at)).flush()
            return True
        return False

    # def get_data_from_cluster(self):
    #     available_items = self.cluster.session.smembers(Cluster.KEY_CDN_AVAILABLE_ITEMS)
    #     last_time = self.cluster.session.get(Cluster.KEY_CDN_LAST_CHECK_AT, '')
    #     if available_items and last_time:
    #         return {'items': available_items, 'last_check_at': last_time}
    #     return False

    def is_need_to_recheck(self):
        """
        是否需要重新检查 cdn
        :return:
        """
        if self.last_check_at and (
                time_now() - self.last_check_at).seconds > self.check_keep_second:
            return True
        return False

    def get_unchecked_item(self):
        if not self.is_recheck:
            items = list(set(self.items) - set(self.available_items) - set(self.unavailable_items))
        else:
            items = list(set(self.items) - set(self.recheck_available_items) - set(self.recheck_unavailable_items))
        if items: return random.choice(items)
        return None

    def check_available(self):
        while True and self.is_alive:
            item = self.get_unchecked_item()
            if not item: return self.check_did_finished()
            self.check_item_available(item)

    def watch_cdn(self):
        """
        监控 cdn 状态，自动重新检测
        :return:
        """
        while True:
            if self.is_alive and not self.is_recheck and self.is_need_to_recheck():  # 重新检测
                self.is_recheck = True
                self.is_finished = False
                CommonLog.add_quick_log(
                    CommonLog.MESSAGE_CDN_START_TO_RECHECK.format(len(self.items), time_now())).flush()
                for i in range(self.thread_num):  # 多线程
                    create_thread_and_run(jobs=self, callback_name='check_available', wait=False)
            stay_second(self.retry_num)

    def destroy(self):
        """
        关闭 CDN
        :return:
        """
        CommonLog.add_quick_log(CommonLog.MESSAGE_CDN_CLOSED).flush()
        self.is_alive = False
        self.init_data()

    def check_item_available(self, item, try_num=0):
        session = Request()
        response = session.get(API_CHECK_CDN_AVAILABLE.format(item), headers={'Host': HOST_URL_OF_12306},
                               timeout=self.check_time_out,
                               verify=False)

        if response.status_code == 200:
            if not self.is_recheck:
                self.available_items.append(item)
            else:
                self.recheck_available_items.append(item)
            if not self.is_ready: self.check_is_ready()
        elif try_num < self.retry_num:  # 重试
            stay_second(self.safe_stay_time)
            return self.check_item_available(item, try_num + 1)
        else:
            if not self.is_recheck:
                self.unavailable_items.append(item)
            else:
                self.recheck_unavailable_items.append(item)
        if not self.is_recheck and (
                not self.last_check_at or (time_now() - self.last_check_at).seconds > self.save_second):
            self.save_available_items()
        stay_second(self.safe_stay_time)

    def check_did_finished(self):
        self.is_ready = True
        if not self.is_finished:
            self.is_finished = True
            if self.is_recheck:
                self.is_recheck = False
                self.available_items = self.recheck_available_items
                self.unavailable_items = self.recheck_unavailable_items
                self.recheck_available_items = []
                self.recheck_unavailable_items = []
            CommonLog.add_quick_log(CommonLog.MESSAGE_CDN_CHECKED_SUCCESS.format(len(self.available_items))).flush()
            self.save_available_items()

    def save_available_items(self):
        self.last_check_at = time_now()
        data = {'items': self.available_items, 'fail_items': self.unavailable_items,
                'last_check_at': str(self.last_check_at)}
        with open(Config().CDN_ENABLED_AVAILABLE_ITEM_FILE, 'w') as f:
            f.write(json.dumps(data))

        # if Config.is_master():
        #     self.cluster.session.sadd(Cluster.KEY_CDN_AVAILABLE_ITEMS, self.available_items)
        #     self.cluster.session.set(Cluster.KEY_CDN_LAST_CHECK_AT, time_now())

    def check_is_ready(self):
        if len(self.available_items) > self.is_ready_num:
            self.is_ready = True
        else:
            self.is_ready = False

    @classmethod
    def get_cdn(cls):
        self = cls()
        if self.is_ready and self.available_items:
            return random.choice(self.available_items)
        return None


if __name__ == '__main__':
    # Const.IS_TEST = True
    Cdn.run()
    while not Cdn().is_finished:
        stay_second(1)



File: py12306/helpers/func.py

# -*- coding: utf-8 -*-
import datetime
import hashlib
import json
import os
import random
import threading
import functools
import time

from time import sleep
from types import MethodType


def singleton(cls):
    """
    将一个类作为单例
    来自 https://wiki.python.org/moin/PythonDecoratorLibrary#Singleton
    """

    cls.__new_original__ = cls.__new__

    @functools.wraps(cls.__new__)
    def singleton_new(cls, *args, **kw):
        it = cls.__dict__.get('__it__')
        if it is not None:
            return it

        cls.__it__ = it = cls.__new_original__(cls, *args, **kw)
        it.__init_original__(*args, **kw)
        return it

    cls.__new__ = singleton_new
    cls.__init_original__ = cls.__init__
    cls.__init__ = object.__init__

    return cls


# 座位 # TODO
# def get_number_by_name(name):
#     return config.SEAT_TYPES[name]


# def get_seat_name_by_number(number): # TODO remove config
# return [k for k, v in config.SEAT_TYPES.items() if v == number].pop()


# 初始化间隔
def init_interval_by_number(number):
    if isinstance(number, dict):
        min = float(number.get('min'))
        max = float(number.get('max'))
    else:
        min = number / 2
        max = number
    return {
        'min': min,
        'max': max
    }


def get_interval_num(interval, decimal=2):
    return round(random.uniform(interval.get('min'), interval.get('max')), decimal)


def stay_second(second, call_back=None):
    sleep(second)
    if call_back:
        return call_back()


def sleep_forever():
    """
    当不是主线程时，假象停止
    :return:
    """
    if not is_main_thread():
        while True: sleep(10000000)


def is_main_thread():
    return threading.current_thread() == threading.main_thread()


def current_thread_id():
    return threading.current_thread().ident


def time_now():
    return datetime.datetime.now()


def timestamp_to_time(timestamp):
    time_struct = time.localtime(timestamp)
    return time.strftime('%Y-%m-%d %H:%M:%S', time_struct)


def get_file_modify_time(filePath):
    timestamp = os.path.getmtime(filePath)
    return timestamp_to_time(timestamp)


def get_file_total_line_num(file, encoding='utf-8'):
    with open(file, 'r', encoding=encoding) as f:
        return len(f.readlines())


def touch_file(path):
    with open(path, 'a'): pass


def pick_file_lines(file, lines):
    return [x for i, x in enumerate(file) if i in lines]


def str_to_time(str):
    return datetime.datetime.strptime(str, '%Y-%m-%d %H:%M:%S.%f')


def time_int():
    return int(time.time())

def time_int_ms():
    return int(time.time() * 1000)

def is_number(val):
    if isinstance(val, int): return val
    if isinstance(val, str): return val.isdigit()
    return False


def create_thread_and_run(jobs, callback_name, wait=True, daemon=True, args=(), kwargs={}):
    threads = []
    if not isinstance(jobs, list): jobs = [jobs]
    for job in jobs:
        thread = threading.Thread(target=getattr(job, callback_name), args=args, kwargs=kwargs)
        thread.setDaemon(daemon)
        thread.start()
        threads.append(thread)
    if wait:
        for thread in threads: thread.join()


def jobs_do(jobs, do):
    if not isinstance(jobs, list): jobs = [jobs]
    for job in jobs:
        getattr(job, do)()


def dict_find_key_by_value(data, value, default=None):
    result = [k for k, v in data.items() if v == value]
    return result.pop() if len(result) else default


def objects_find_object_by_key_value(objects, key, value, default=None):
    result = [obj for obj in objects if getattr(obj, key) == value]
    return result.pop() if len(result) else default


def dict_count_key_num(data: dict, key, like=False):
    count = 0
    for k in data.keys():
        if like:
            if k.find(key) >= 0: count += 1
        elif k == key:
            count += 1
    return count


def array_dict_find_by_key_value(data, key, value, default=None):
    result = [v for k, v in enumerate(data) if key in v and v[key] == value]
    return result.pop() if len(result) else default


def get_true_false_text(value, true='', false=''):
    if value: return true
    return false


def sleep_forever_when_in_test():
    if Const.IS_TEST: sleep_forever()


def expand_class(cls, key, value, keep_old=True):
    if (keep_old):
        setattr(cls, 'old_' + key, getattr(cls, key))
    setattr(cls, key, MethodType(value, cls))
    return cls


def available_value(value):
    if isinstance(value, str) or isinstance(value, bytes):
        return value
    return str(value)


def md5(value):
    return hashlib.md5(json.dumps(value).encode()).hexdigest()


@singleton
class Const:
    IS_TEST = False
    IS_TEST_NOTIFICATION = False



File: py12306/vender/ruokuai/main.py

import requests
from hashlib import md5


class RKClient(object):

    def __init__(self, username, password, soft_id, soft_key):
        self.username = username
        self.password = md5(password.encode('utf-8')).hexdigest()
        self.soft_id = soft_id
        self.soft_key = soft_key
        self.base_params = {
            'username': self.username,
            'password': self.password,
            'softid': self.soft_id,
            'softkey': self.soft_key,
        }
        self.headers = {
            'Connection': 'Keep-Alive',
            'Expect': '100-continue',
            'User-Agent': 'ben',
        }

    def rk_create(self, image, im_type, timeout=20):
        """
        im: 图片字节
        im_type: 题目类型
        """
        params = {
            'typeid': im_type,
            'timeout': timeout,
            'image': image
        }
        params.update(self.base_params)
        r = requests.post('http://api.ruokuai.com/create.json', data=params, timeout=timeout)
        return r.json()

    def rk_report_error(self, im_id):
        """
        im_id:报错题目的ID
        """
        params = {
            'id': im_id,
        }
        params.update(self.base_params)
        r = requests.post('http://api.ruokuai.com/reporterror.json', data=params, headers=self.headers)
        return r.json()


if __name__ == '__main__':
    rc = RKClient('username', 'password', 'soft_id', 'soft_key')
    # im = open('a.jpg', 'rb').read()
    # print rc.rk_create(im, 3040)




# 🚂 py12306 购票助手
分布式，多账号，多任务购票

## Features
- [x] 多日期查询余票
- [x] 自动打码下单
- [x] 用户状态恢复
- [x] 电话语音通知
- [x] 多账号、多任务、多线程支持
- [x] 单个任务多站点查询 
- [x] 分布式运行
- [x] Docker 支持
- [x] 动态修改配置文件
- [x] 邮件通知
- [x] Web 管理页面
- [x] 微信消息通知
- [ ] 代理池支持 ([pyproxy-async](https://github.com/pjialin/pyproxy-async))

## 使用
py12306 需要运行在 python 3.6 以上版本（其它版本暂未测试)

**1. 安装依赖**
```bash
git clone https://github.com/pjialin/py12306

pip install -r requirements.txt
```

**2. 配置程序**
```bash
cp env.py.example env.py
```
自动打码

（若快已停止服务，目前只能设置**free**打码模式）
free 已对接到打码共享平台，[https://py12306-helper.pjialin.com](https://py12306-helper.pjialin.com/)，欢迎参与分享

语音通知

语音验证码使用的是阿里云 API 市场上的一个服务商，需要到 [https://market.aliyun.com/products/56928004/cmapi026600.html](https://market.aliyun.com/products/56928004/cmapi026600.html) 购买后将 appcode 填写到配置中

**3. 启动前测试**

目前提供了一些简单的测试，包括用户账号检测，乘客信息检测，车站检测等

开始测试 -t 
```bash
python main.py -t
```

测试通知消息 (语音, 邮件) -t -n
```bash
# 默认不会进行通知测试，要对通知进行测试需要加上 -n 参数 
python main.py -t -n
```

**4. 运行程序**
```bash
python main.py
```

### 参数列表

- -t 测试配置信息
- -t -n 测试配置信息以及通知消息
- -c 指定自定义配置文件位置

### 分布式集群

集群依赖于 redis，目前支持情况
- 单台主节点多个子节点同时运行
- 主节点宕机后自动切换提升子节点为主节点
- 主节点恢复后自动恢复为真实主节点
- 配置通过主节点同步到所有子节点
- 主节点配置修改后无需重启子节点，支持自动更新
- 子节点消息实时同步到主节点

**使用**

将配置文件的中 `CLUSTER_ENABLED` 打开即开启分布式

目前提供了一个单独的子节点配置文件 `env.slave.py.example` 将文件修改为 `env.slave.py`， 通过 `python main.py -c env.slave.py` 即可快速启动


## Docker 使用
**1. 将配置文件下载到本地**
```bash
docker run --rm pjialin/py12306 cat /config/env.py > env.py
# 或
curl https://raw.githubusercontent.com/pjialin/py12306/master/env.docker.py.example -o env.py
```

**2. 修改好配置后运行**
```bash
docker run --rm --name py12306 -p 8008:8008 -d -v $(pwd):/config -v py12306:/data pjialin/py12306
```
当前目录会多一个 12306.log 的日志文件， `tail -f 12306.log`

### Docker-compose 中使用
**1. 复制配置文件**
```
cp docker-compose.yml.example docker-compose.yml
```

**2. 从 docker-compose 运行**

在`docker-compose.yml`所在的目录使用命令
```
docker-compose up -d
```

## Web 管理页面

目前支持用户和任务以及实时日志查看，更多功能后续会不断加入

**使用**

打开 Web 功能需要将配置中的 `WEB_ENABLE` 打开，启动程序后访问当前主机地址 + 端口号 (默认 8008) 即可，如 http://127.0.0.1:8008

## 更新
- 19-01-10
    - 支持分布式集群
- 19-01-11
    - 配置文件支持动态修改
- 19-01-12
    - 新增免费打码
- 19-01-14
    - 新增 Web 页面支持
- 19-01-15
    - 新增 钉钉通知
    - 新增 Telegram 通知
    - 新增 ServerChan 和 PushBear 微信推送
- 19-01-18
    - 新增 CDN 查询

## 截图
### Web 管理页面
![Web 管理页面图片](https://github.com/pjialin/py12306/blob/master/data/images/web.png)

### 下单成功
![下单成功图片](https://github.com/pjialin/py12306/blob/master/data/images/order_success.png)

### 关于防封
目前查询和登录操作是分开的，查询是不依赖用户是否登录，放在 A 云 T 云容易被限制 ip，建议在其它网络环境下运行

QQ 交流群 [780289875](https://jq.qq.com/?_wv=1027&k=5PgzDwV)，TG 群 [Py12306 交流](https://t.me/joinchat/F3sSegrF3x8KAmsd1mTu7w)

### Online IDE
[![在 Gitpod 中打开](https://gitpod.io/button/open-in-gitpod.svg)](https://gitpod.io#https://github.com/pjialin/py12306)

## Thanks
- 感谢大佬 [testerSunshine](https://github.com/testerSunshine/12306)，借鉴了部分实现
- 感谢所有提供 pr 的大佬 
- 感谢大佬 [zhaipro](https://github.com/zhaipro/easy12306) 的验证码本地识别模型与算法

## License

[Apache License.](https://github.com/pjialin/py12306/blob/master/LICENSE)

