add task logger

config.toml

@@ -0,0 +1,16 @@
+[app]
+env = "dev"
+
+[web]
+ip = "0.0.0.0"
+port = 8081
+
+[users]
+user = "admin"
+password  = "admin"
+
+[redis]
+host = "127.0.0.1"
+port = 6379
+db = 1
+password = "local_redis_dd"

py12306/app/app.py

@@ -0,0 +1,59 @@
+import logging
+import os
+
+
+class Config:
+    class AppEnvType:
+        DEV = 'dev'
+        PRODUCTION = 'production'
+
+    APP_NAME = 'py12306'
+    APP_ENV = AppEnvType.PRODUCTION
+    LOADED = False
+    TEST_MODE = False
+
+    PROJECT_DIR = os.path.abspath(__file__ + '/../../../') + '/'
+    CONFIG_FILE = PROJECT_DIR + 'config.toml'
+
+    # Config
+    REDIS = {
+        'host': '127.0.0.1',
+        'port': 6379,
+        'db': 0,
+        'password': None,
+        'decode_responses': True
+    }
+
+    # Redis keys
+    REDIS_PREFIX_KEY_TASKS = 'tasks:'
+
+    # REDIS_KEY_USER_TASKS = 'user_jobs'
+
+    @classmethod
+    def load(cls):
+        """
+        Load configs from toml file
+        :return:
+        """
+        import toml
+        configs = toml.load(cls.CONFIG_FILE)
+
+        redis = configs.get('redis')
+        if redis:
+            cls.REDIS.update(redis)
+
+        app = configs.get('app')
+        if app:
+            cls.APP_ENV = app.get('env', cls.APP_ENV)
+
+
+if not Config.LOADED:
+    Config.load()
+
+# Logger
+Logger = logging.getLogger(Config.APP_NAME)
+Logger.setLevel('DEBUG' if Config.APP_ENV == Config.AppEnvType.DEV else 'ERROR')
+handler = logging.StreamHandler()
+formatter = logging.Formatter('%(asctime)s %(levelname)s: %(message)s')
+handler.setFormatter(formatter)
+Logger.addHandler(handler)

py12306/app/task.py

@@ -0,0 +1,34 @@
+from py12306.app.app import Logger, Config
+from py12306.lib.func import new_thread_with_jobs
+from py12306.lib.redis_lib import Redis
+
+
+def get_routes() -> dict:
+    from py12306.app.user import User
+    return {
+        'user': User.task_user,
+        'query': User.task_user,
+    }
+
+
+class Task:
+    routes: dict = None
+
+    @classmethod
+    def listen(cls):
+        routes = get_routes()
+        keys = [Config.REDIS_PREFIX_KEY_TASKS + key for key, _ in routes.items()]
+        while True:
+            key, job = Redis.share().get_task_sync(keys)
+            Logger.info('获得新任务 %s' % key)
+            if Config.TEST_MODE:  # ignore when in test env
+                return job
+            self = cls()
+            self.routes = routes
+            self.deal_job(key, job)
+
+    def deal_job(self, key: str, task: dict):
+        handler = self.routes.get(key)
+        if not handler:
+            return
+        new_thread_with_jobs(handler, wait=True, kwargs={'task': task})

py12306/app/user.py

@@ -0,0 +1,12 @@
+from py12306.lib.helper import ShareInstance
+
+
+class User(ShareInstance):
+
+    @classmethod
+    def task_user(cls, task: dict):
+        print(111)
+        a = 1
+        pass
+
+    pass

py12306/lib/func.py

@@ -0,0 +1,24 @@
+import threading
+
+
+def new_thread_with_jobs(jobs, wait=True, daemon=True, args=(), kwargs={}):
+    """
+    Run each job with a new thread
+    :param jobs:
+    :param wait:
+    :param daemon:
+    :param args:
+    :param kwargs:
+    :return:
+    """
+    threads = []
+    if not isinstance(jobs, list):
+        jobs = [jobs]
+    for job in jobs:
+        thread = threading.Thread(target=job, args=args, kwargs=kwargs)
+        thread.setDaemon(daemon)
+        thread.start()
+        threads.append(thread)
+    if wait:
+        for thread in threads:
+            thread.join()

py12306/lib/helper.py

@@ -0,0 +1,10 @@
+class ShareInstance():
+    __session = None
+
+    @classmethod
+    def share(cls):
+        if not cls.__session:
+            cls.__session = cls()
+        return cls.__session
+
+

py12306/lib/redis_lib.py

@@ -0,0 +1,27 @@
+from redis import Redis as PyRedis
+
+from py12306.app.app import Config
+from py12306.lib.helper import ShareInstance
+
+import json
+
+
+class Redis(PyRedis, ShareInstance):
+
+    def __init__(self, **kwargs):
+        if not kwargs:
+            kwargs = Config.REDIS
+        super().__init__(**kwargs)
+
+    def push_task(self, key: str, tasks: dict):
+        return self.rpush(key, json.dumps(tasks))
+
+    def get_task_sync(self, keys: list) -> tuple:
+        tasks = self.brpop(keys)
+        return tasks[0][len(Config.REDIS_PREFIX_KEY_TASKS):], json.loads(tasks[1])
+
+
+if __name__ == '__main__':
+    res = Redis.share().keys('*')
+    print(res)
+    pass

requirements.txt

@@ -1,3 +1,4 @@
+toml
 -i http://mirrors.aliyun.com/pypi/simple/  --trusted-host mirrors.aliyun.com
 appdirs==1.4.3
 beautifulsoup4==4.7.0

tests/helper.py

@@ -0,0 +1,15 @@
+from unittest import TestCase
+
+from py12306.app.app import Config
+from py12306.lib.redis_lib import Redis
+
+
+class BaseTest(TestCase):
+    redis: Redis = None
+    config: Config = None
+
+    def setUp(self) -> None:
+        super().setUp()
+        Config.TEST_MODE = True
+        self.config = Config
+        self.redis = Redis.share()

tests/test_redis.py

@@ -0,0 +1,9 @@
+from py12306.lib.redis_lib import Redis
+from tests.helper import BaseTest
+
+
+class TestRedis(BaseTest):
+
+    def test_connection(self):
+        res = Redis.share().info()
+        self.assertIsInstance(res, dict)

tests/test_task.py

@@ -0,0 +1,20 @@
+from tests.helper import BaseTest
+from py12306.app.task import Task
+
+
+class TestTask(BaseTest):
+    def test_push_task(self):
+        tasks = {
+            'query': {
+                'name': 'admin',
+            },
+            'user': {
+                'name': 'admin',
+                'password': 'password'
+            }
+        }
+        for key, task in tasks.items():
+            self.redis.push_task(self.config.REDIS_PREFIX_KEY_TASKS + key, task)
+
+        res = Task.listen()
+        self.assertIsInstance(res, dict)

tests/test_user.py

@@ -0,0 +1,5 @@
+from tests.helper import BaseTest
+
+
+class TestUser(BaseTest):
+    pass