优化代码，增加测试模式

Author: pjialin

main.py

@@ -1,20 +1,41 @@
 # encoding=utf8
-import os
-from threading import Thread
+import sys
+from time import sleep
 
 from py12306.helpers.func import *
+from py12306.helpers.app import *
+from py12306.log.common_log import CommonLog
 from py12306.query.query import Query
 from py12306.user.user import User
 
 
 def main():
-    # Thread(target=Query.run).start()  # 余票查询
-    # create_thread_and_run(User, 'run', wait=False)
+    if '--test' in sys.argv or '-t' in sys.argv: test()
+    CommonLog.print_welcome().print_configs()
+
+    App.run_check()
     User.run()
     Query.run()
-    # Query.run()
-    while True:
-        sleep(1)
+    if not Const.IS_TEST:
+        while True:
+            sleep(1)
+
+    CommonLog.test_complete()
+
+
+def test():
+    """
+    功能检查
+    包含：
+        账号密码验证 (打码)
+        座位验证
+        乘客验证
+        语音验证码验证
+    :return:
+    """
+    Const.IS_TEST = True
+    if '--test-notification' in sys.argv or '-n' in sys.argv:
+        Const.IS_TEST_NOTIFICATION = True
     pass
 
 

py12306/config.py

@@ -52,8 +52,13 @@
 QUERY_DATA_DIR = RUNTIME_DIR + 'query/'
 USER_DATA_DIR = RUNTIME_DIR + 'user/'
 
-STATION_FILE = 'data/stations.txt'
-CONFIG_FILE = 'env.py'
+STATION_FILE = PROJECT_DIR + 'data/stations.txt'
+CONFIG_FILE = PROJECT_DIR + 'env.py'
+
+# 语音验证码
+NOTIFICATION_BY_VOICE_CODE = 0
+NOTIFICATION_VOICE_CODE_PHONE = ''
+NOTIFICATION_API_APP_CODE = ''
 
 if path.exists(CONFIG_FILE):
     exec(open(CONFIG_FILE, encoding='utf8').read())
@@ -71,7 +76,3 @@ class UserType:
         '学生': STUDENT,
         '残疾军人、伤残人民警察': SOLDIER,
     }
-
-
-def get(key, default=None):
-    return eval(key)

py12306/exceptions/MemberInvalidException.py

@@ -50,6 +50,8 @@
 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_NOTIFICATION_BY_VOICE_CODE = 'http://ali-voice.showapi.com/sendVoice?'
+
 urls = {
     "auth": {  # 登录接口
         "req_url": "/passport/web/auth/uamtk",

py12306/helpers/api.py

@@ -1,8 +1,12 @@
 from py12306.helpers.func import *
+from py12306.config import *
+from py12306.helpers.notification import Notification
 from py12306.log.common_log import CommonLog
+from py12306.log.order_log import OrderLog
 
 
 def app_available_check():
+    # return True  # Debug
     now = time_now()
     if now.hour >= 23 or now.hour < 6:
         CommonLog.add_quick_log(CommonLog.MESSAGE_12306_IS_CLOSED.format(time_now())).flush()
@@ -11,3 +15,44 @@ def app_available_check():
             open_time += datetime.timedelta(1)
         sleep((open_time - now).seconds)
     return True
+
+
+class App:
+    """
+    程序主类
+    TODO 需要完善
+    """
+
+    @classmethod
+    def check_auto_code(cls):
+        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 True
+        return False
+
+    @classmethod
+    def test_send_notifications(cls):
+        if config.NOTIFICATION_BY_VOICE_CODE:  # 语音通知
+            CommonLog.add_quick_log(CommonLog.MESSAGE_TEST_SEND_VOICE_CODE).flush()
+            Notification.voice_code(config.NOTIFICATION_VOICE_CODE_PHONE, '张三',
+                                    OrderLog.MESSAGE_ORDER_SUCCESS_NOTIFICATION_OF_VOICE_CODE_CONTENT.format('北京',
+                                                                                                             '深圳'))
+
+    @classmethod
+    def run_check(cls):
+        """
+        待优化
+        :return:
+        """
+        if not cls.check_auto_code():
+            CommonLog.add_quick_log(CommonLog.MESSAGE_CHECK_AUTO_CODE_FAIL).flush(exit=True)
+        if not cls.check_user_account_is_empty():
+            CommonLog.add_quick_log(CommonLog.MESSAGE_CHECK_EMPTY_USER_ACCOUNT).flush(exit=True)
+        if Const.IS_TEST_NOTIFICATION: cls.test_send_notifications()

py12306/helpers/app.py

@@ -1,10 +1,12 @@
 import datetime
 import random
 import threading
+import functools
+
 from time import sleep
 
 from py12306 import config
-import functools
+
 
 
 def singleton(cls):
@@ -65,6 +67,15 @@ def stay_second(second, call_back=None):
         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()
 
@@ -99,4 +110,17 @@ 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 test:
+
+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()
+
+
+@singleton
+class Const:
+    IS_TEST = False
+    IS_TEST_NOTIFICATION = False

py12306/helpers/func.py

@@ -0,0 +1,64 @@
+import urllib
+
+from py12306 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()
+        self.send_voice_code_of_yiyuan(phone, name=name, content=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)
+            })
+        response_message = '-'
+        result = {}
+        try:
+            result = response.json()
+            response_message = result['showapi_res_body']['remark']
+        except:
+            pass
+        if response.status_code == 401 or response.status_code == 403:
+            return CommonLog.add_quick_log(CommonLog.MESSAGE_VOICE_API_FORBID).flush()
+        if response.status_code == 200 and 'showapi_res_body' in result and result['showapi_res_body'].get('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()
+
+
+if __name__ == '__main__':
+    Notification.voice_code('13800138000', '张三', '你的车票 广州 到 深圳 购买成功，请登录 12306 进行支付')

py12306/helpers/notification.py

@@ -7,7 +7,6 @@ class Station:
     stations = []
 
     def __init__(self):
-        print('Station 初始化')
         if path.exists(config.STATION_FILE):
             result = open(config.STATION_FILE, encoding='utf-8').read()
             result = result.lstrip('@').split('@')

py12306/helpers/station.py

@@ -3,14 +3,13 @@
 
 from py12306.helpers.func import *
 
-
 class BaseLog:
     logs = []
     thread_logs = {}
     quick_log = []
 
     @classmethod
-    def add_log(cls, content):
+    def add_log(cls, content=''):
         self = cls()
         # print('添加 Log 主进程{} 进程ID{}'.format(is_main_thread(), current_thread_id()))
         if is_main_thread():
@@ -31,7 +30,6 @@ def flush(cls, sep='\n', end='\n', file=None, exit=False):
                 logs = self.logs
             else:
                 logs = self.thread_logs.get(current_thread_id())
-        # for i in logs:
         print(*logs, sep=sep, end=end, file=file)
         if self.quick_log:
             self.quick_log = []
@@ -40,12 +38,11 @@ def flush(cls, sep='\n', end='\n', file=None, exit=False):
                 self.logs = []
             else:
                 if logs: del self.thread_logs[current_thread_id()]
-        # print(self.logs)
         if exit:
             sys.exit()
 
     @classmethod
-    def add_quick_log(cls, content):
+    def add_quick_log(cls, content = ''):
         self = cls()
         self.quick_log.append(content)
         return self

py12306/log/base.py

@@ -1,18 +1,67 @@
 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 = '正在测试发送语音验证码...'
+
     def __init__(self):
         super().__init__()
         self.init_data()
 
     def init_data(self):
-        print('Common Log 初始化')
+        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('当前为测试模式，程序运行完成后自动结束')
+        self.add_quick_log()
+        self.flush()
+        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('语音验证码: {}'.format(get_true_false_text(config.QUERY_JOB_THREAD_ENABLED, enable, disable)))
+        self.add_quick_log('查询间隔: {} 秒'.format(config.QUERY_INTERVAL))
+        self.add_quick_log('用户心跳检测间隔: {} 秒'.format(config.USER_HEARTBEAT_INTERVAL))
+        self.add_quick_log()
+        self.flush()
+        return self
+
+    @classmethod
+    def test_complete(cls):
+        self = cls()
+        self.add_quick_log('# 测试完成，请检查输出是否正确 #')
+        self.flush()
+        return self
 
     @classmethod
     def print_auto_code_fail(cls, reason):