handler/__init__.py

@@ -1,19 +1,20 @@
-from setting import logger
+from loguru import logger
 
 from .handle_event import HandlerEvent
 from .handle_paylist import HandlerPaylist
 from .handle_user import HandlerUser
 
 HANDLER_DICT = {
-    b'user': HandlerUser,
+    'user': HandlerUser,
-    b'paylist': HandlerPaylist,
+    'paylist': HandlerPaylist,
-    b'event': HandlerEvent,
+    'event': HandlerEvent,
 }
 
 
-def handle_factory(key, data):
+def handle_factory(data):
-    obj = HANDLER_DICT.get(key)
+    type_ = data['type']
+    obj = HANDLER_DICT.get(type_)
     if not obj:
-        logger.warning(f'未知类型{key}')
+        logger.warning(f'未知类型{type_}')
         return
     return obj(data)

handler/handler.py

@@ -1,9 +1,8 @@
 import hashlib
-import os
 from abc import ABCMeta, abstractmethod
 import traceback
 import arrow
-from setting import logger
+from loguru import logger
 
 from setting import settings
 
@@ -14,16 +13,14 @@ class Handler(metaclass=ABCMeta):
 
     def __init__(self, data):
         self._data = data
-        appid = data['appid']
+        self._appid = data['appid']
         self._event = self._data['event']
         self._game = self._data['project']
         self._type = self._data['type']
-        Handler._init_game_config()
+        Handler._init_game_config(self._appid)
-        Handler._init_mdb(f'{settings.DB_PREFIX}_{Handler.game_config[appid]["id_name"]}')
+        Handler._init_mdb(f'{settings.DB_PREFIX}_{Handler.game_config["id_name"]}')
-        self.secret_token = Handler.game_config[appid].get('secret_token', '')
+        self.secret_token = Handler.game_config.get('secret_token', '')
-        self._nt = arrow.now(tz=Handler.game_config[appid].get('timezone', 'local')).timestamp
+        self._nt = arrow.now(tz=Handler.game_config.get('timezone', 'local')).timestamp
-
-        print(Handler.game_config)
 
     @classmethod
     def _init_mdb(cls, dbname):
@@ -47,11 +44,9 @@ class Handler(metaclass=ABCMeta):
             coll.update_one(where, {'$set': update_data}, upsert=True)
 
     @classmethod
-    def _init_game_config(cls):
+    def _init_game_config(cls, appid):
         if not cls.game_config:
-            topic_list = settings.kafka_setting['topic_name']
+            cls.game_config = settings.admin_game_coll.find_one({'appid': appid})
-            data = settings.admin_game_coll.find({'appid': {'$in': topic_list}}, {'_id': False})
-            cls.game_config = {item['appid']:item for item in data}
         return cls.game_config
 
     def handler_event(self):

main.py

@@ -4,7 +4,7 @@ import time
 import ssl
 
 from kafka import KafkaConsumer
-from setting import logger
+from loguru import logger
 
 from handler import handle_factory
 from setting import settings
@@ -31,12 +31,11 @@ def main():
                              sasl_plain_username=conf['sasl_plain_username'],
                              sasl_plain_password=conf['sasl_plain_password'],
                              value_deserializer=json.loads)
-    consumer.subscribe(conf['topic_name'])
+    consumer.subscribe((conf['topic_name'],))
     for msg in consumer:
         st = time.time() * 1000
-        key = msg.key
         data = msg.value
-        obj = handle_factory(key, data)
+        obj = handle_factory(data)
         if not obj:
             continue
         obj.run()

setting.py

@@ -3,9 +3,6 @@ import os
 import pymongo
 from loguru import logger
 
-logger.add('log.log', format="{time} {level} {name}:{line} {message}", level="INFO",
-           rotation="100 MB", retention='7 days',
-           enqueue=True)
 
 
 class Config:
@@ -13,8 +10,8 @@ class Config:
         'sasl_plain_username': 'legu666666',
         'sasl_plain_password': '5EINbjyI',
         'bootstrap_servers': ["39.104.71.242:9093", "39.104.71.78:9093", "39.104.67.122:9093"],
-        'topic_name': os.environ.get('topic_name').split(','),
+        'topic_name': 'legu_geshouccs',
-        'consumer_id': os.environ.get('consumer_id')
+        'consumer_id': 'geshouccs_consumer'
     }
 
     local_mongo_uri = os.environ["local_mongo_uri"]
@@ -26,9 +23,16 @@ class Config:
 class Production(Config):
     DB_PREFIX = 'game'
 
+    logger.add('log.log', format="{time} {level} {name}:{line} {message}", level="INFO",
+               rotation="100 MB", retention='7 days',
+               enqueue=True)
+
 
 class Debug(Config):
     DB_PREFIX = 'debug'
+    logger.add('log.log', format="{time} {level} {name}:{line} {message}", level="DEBUG",
+               rotation="100 MB", retention='7 days',
+               enqueue=True)
 
 
 settings = Production if os.environ.get('xlegudata_env') == 'production' else Debug