🔧 使用 dotenv 重构 config

* 🔧 使用 dotenv 重构 config

默认配置从 config.json 移动到 config.py 中。如果要覆盖默认配置，在根目录创建
.env 文件按照 .env.example 的例子编辑。

这个方案的优点是：

* 支持写注释
* 以后如果新增配置项，如果用默认值就可以，不需要修改 .env 文件
* 如果通过 serverless、docker 或者 k8s 部署，方便不用修改文件，直接注入环境变量
  修改配置

.env.example

@@ -0,0 +1,26 @@
+# debug 开关
+DEBUG=false
+
+# MySQL
+DB_HOST=127.0.0.1
+DB_PORT=3306
+DB_USERNAME=user
+DB_PASSWORD="password"
+DB_DATABASE=paimon
+
+# Redis
+REDIS_HOST=127.0.0.1
+REDIS_PORT=6379
+REDIS_DB=0
+
+# 联系 https://t.me/BotFather 使用 /newbot 命令创建机器人并获取 token
+BOT_TOKEN="xxxxxxx"
+
+# 记录错误并发送消息通知开发人员
+ERROR_NOTIFICATION_CHAT_ID=chat_id
+
+# 文章推送群组
+CHANNELS=[{ "name": "", "chat_id": 1}]
+
+# bot 管理员
+ADMINS=[{ "username": "", "user_id": 1 }]

.gitignore

@@ -27,8 +27,11 @@ __pycache__/
 
 ### Customize ###
 
-config/config.json
 **_test.html
 test_**.html
 logs/
 /resources/*/*/test/
+
+### DotEnv ###
+.env
+

config.py

@@ -1,29 +1,60 @@
 import os
+from typing import Any
 
 import ujson
+from dotenv import load_dotenv
 
+from utils.storage import Storage
 
-class Config:
+# take environment variables from .env.
-    def __init__(self):
+load_dotenv()
-        project_path = os.path.dirname(__file__)
-        config_file = os.path.join(project_path, './config', 'config.json')
-        if not os.path.exists(config_file):
-            config_file = os.path.join(project_path, './config', 'config.example.json')
 
-        with open(config_file, 'r', encoding='utf-8') as f:
+env = os.getenv
-            self._config_json: dict = ujson.load(f)
 
-        self.DEBUG = self.get_config("debug")
+def str_to_bool(value: Any) -> bool:
-        if not isinstance(self.DEBUG, bool):
+    """Return whether the provided string (or any value really) represents true. Otherwise false.
-            self.DEBUG = False
+    Just like plugin server stringToBoolean.
-        self.ADMINISTRATORS = self.get_config("administrators")
+    """
-        self.MYSQL = self.get_config("mysql")
+    if not value:
-        self.REDIS = self.get_config("redis")
+        return False
-        self.TELEGRAM = self.get_config("telegram")
+    return str(value).lower() in ("y", "yes", "t", "true", "on", "1")
-        self.FUNCTION = self.get_config("function")
 
-    def get_config(self, name: str):
+_config = {
-        return self._config_json.get(name, {})
+    "debug": str_to_bool(os.getenv('DEBUG', 'True')),
 
+    "mysql": {
+        "host": env("DB_HOST", "127.0.0.1"),
+        "port": int(env("DB_PORT", "3306")),
+        "user": env("DB_USERNAME"),
+        "password": env("DB_PASSWORD"),
+        "database": env("DB_DATABASE"),
+    },
 
-config = Config()
+    "redis": {
+        "host": env("REDIS_HOST", "127.0.0.1"),
+        "port": int(env("REDIS_PORT", "6369")),
+        "database": int(env("REDIS_DB", "0")),
+    },
+
+    # 联系 https://t.me/BotFather 使用 /newbot 命令创建机器人并获取 token
+    "bot_token": env("BOT_TOKEN"),
+
+    # 记录错误并发送消息通知开发人员
+    "error_notification_chat_id": env("ERROR_NOTIFICATION_CHAT_ID"),
+
+    # 文章推送群组
+    "channels": [
+        # {"name": "", "chat_id": 1},
+        # 在环境变量里的格式是 json: [{"name": "", "chat_id": 1}]
+        *ujson.loads(env('CHANNELS', '[]'))
+    ],
+
+    # bot 管理员
+    "admins": [
+        # {"username": "", "user_id": 123},
+        # 在环境变量里的格式是 json: [{"username": "", "user_id": 1}]
+        *ujson.loads(env('ADMINS', '[]'))
+    ],
+}
+
+config = Storage(_config)

config/config.json.example

@@ -1,38 +0,0 @@
-{
-  "mysql": {
-    "host": "127.0.0.1",
-    "port": 3306,
-    "user": "",
-    "password": "",
-    "database": ""
-  },
-  "redis": {
-    "host": "127.0.0.1",
-    "port": 6379,
-    "database": 0
-  },
-  "telegram": {
-    "token": "",
-    "notice": {
-      "ERROR": {
-        "name": "",
-        "chat_id":
-      }
-    },
-    "channel": {
-      "POST": [
-        {
-            "name": "",
-            "chat_id":
-        }
-      ]
-    }
-  },
-  "administrators":
-    [
-      {
-        "username": "",
-        "user_id":
-      }
-    ]
-}

core/admin/services.py

@@ -18,7 +18,7 @@ class BotAdminService:
         admin_list = await self._cache.get_list()
         if len(admin_list) == 0:
             admin_list = await self._repository.get_all_user_id()
-            for config_admin in config.ADMINISTRATORS:
+            for config_admin in config.admins:
                 admin_list.append(config_admin["user_id"])
             await self._cache.set_list(admin_list)
         return admin_list
@@ -29,7 +29,7 @@ class BotAdminService:
         except IntegrityError as error:
             Log.warning(f"{user_id} 已经存在数据库 \n", error)
         admin_list = await self._repository.get_all_user_id()
-        for config_admin in config.ADMINISTRATORS:
+        for config_admin in config.admins:
             admin_list.append(config_admin["user_id"])
         await self._cache.set_list(admin_list)
         return True
@@ -40,7 +40,7 @@ class BotAdminService:
         except ValueError:
             return False
         admin_list = await self._repository.get_all_user_id()
-        for config_admin in config.ADMINISTRATORS:
+        for config_admin in config.admins:
             admin_list.append(config_admin["user_id"])
         await self._cache.set_list(admin_list)
         return True

core/template/services.py

@@ -22,7 +22,7 @@ class TemplateService:
         self._jinja2_template = {}
 
     def get_template(self, package_path: str, template_name: str, auto_escape: bool = True) -> Template:
-        if config.DEBUG:
+        if config.debug:
             # DEBUG下 禁止复用 方便查看和修改模板
             loader = PackageLoader(self._template_package_name, package_path)
             jinja2_env = Environment(loader=loader, enable_async=True, autoescape=auto_escape)

jobs/sign.py

@@ -31,7 +31,7 @@ class SignJob:
     @classmethod
     def build_jobs(cls, job_queue: JobQueue):
         sign = cls()
-        if config.DEBUG:
+        if config.debug:
             job_queue.run_once(sign.sign, 3, name="SignJobTest")
         # 每天凌晨一点执行
         job_queue.run_daily(sign.sign, datetime.time(hour=1, minute=0, second=0), name="SignJob")

logger.py

@@ -26,7 +26,7 @@ class Logger:
         self.logger = logging.getLogger("TGPaimonBot")
         root_logger = logging.getLogger()
         root_logger.setLevel(logging.CRITICAL)
-        if config.DEBUG:
+        if config.debug:
             self.logger.setLevel(logging.DEBUG)
         else:
             self.logger.setLevel(logging.INFO)

main.py

@@ -24,12 +24,12 @@ def main() -> None:
 
     # 初始化数据库
     Log.info("初始化数据库")
-    mysql = MySQL(host=config.MYSQL["host"], user=config.MYSQL["user"], password=config.MYSQL["password"],
+    mysql = MySQL(host=config.mysql["host"], user=config.mysql["user"], password=config.mysql["password"],
-                  port=config.MYSQL["port"], database=config.MYSQL["database"])
+                  port=config.mysql["port"], database=config.mysql["database"])
 
     # 初始化Redis缓存
     Log.info("初始化Redis缓存")
-    redis = RedisDB(host=config.REDIS["host"], port=config.REDIS["port"], db=config.REDIS["database"])
+    redis = RedisDB(host=config.redis["host"], port=config.redis["port"], db=config.redis["database"])
 
     # 初始化Playwright
     Log.info("初始化Playwright")
@@ -49,7 +49,7 @@ def main() -> None:
 
     application = Application\
         .builder()\
-        .token(config.TELEGRAM["token"])\
+        .token(config.bot_token)\
         .defaults(defaults)\
         .build()
 

plugins/genshin/help.py

@@ -35,7 +35,7 @@ class Help:
         message = update.message
         user = update.effective_user
         Log.info(f"用户 {user.full_name}[{user.id}] 发出help命令")
-        if self.file_id is None or config.DEBUG:
+        if self.file_id is None or config.debug:
             await message.reply_chat_action(ChatAction.TYPING)
             help_png = await self.template_service.render('bot/help', "help.html", {}, {"width": 768, "height": 768})
             await message.reply_chat_action(ChatAction.UPLOAD_PHOTO)

plugins/genshin/post.py

@@ -178,7 +178,7 @@ class Post(BasePlugins):
         message = update.message
         reply_keyboard = []
         try:
-            for channel_info in config.TELEGRAM["channel"]["POST"]:
+            for channel_info in config.channels:
                 name = channel_info["name"]
                 reply_keyboard.append([f"{name}"])
         except KeyError as error:
@@ -195,7 +195,7 @@ class Post(BasePlugins):
         message = update.message
         channel_id = -1
         try:
-            for channel_info in config.TELEGRAM["channel"]["POST"]:
+            for channel_info in config.channels:
                 if message.text == channel_info["name"]:
                     channel_id = channel_info["chat_id"]
         except KeyError as error:
@@ -252,7 +252,7 @@ class Post(BasePlugins):
         channel_id = post_handler_data.channel_id
         channel_name = None
         try:
-            for channel_info in config.TELEGRAM["channel"]["POST"]:
+            for channel_info in config.channels:
                 if post_handler_data.channel_id == channel_info["chat_id"]:
                     channel_name = channel_info["name"]
         except KeyError as error:

plugins/system/errorhandler.py

@@ -12,7 +12,7 @@ from config import config
 from logger import Log
 
 try:
-    notice_chat_id = config.TELEGRAM["notice"]["ERROR"]["chat_id"]
+    notice_chat_id = config.error_notification_chat_id
 except KeyError as error:
     Log.warning("错误通知Chat_id获取失败或未配置，BOT发生致命错误时不会收到通知 错误信息为\n", error)
     notice_chat_id = None

requirements.txt

@@ -20,4 +20,5 @@ pytz>=2021.3
 Pillow>=9.0.1
 SQLAlchemy>=1.4.39
 sqlmodel>=0.0.6
-asyncmy>=0.2.5
+asyncmy>=0.2.5
+python-dotenv>=0.20.0

utils/helpers.py

@@ -89,3 +89,4 @@ def region_server(uid: Union[int, str]) -> RegionEnum:
         return region
     else:
         raise TypeError(f"UID {uid} isn't associated with any region")
+

utils/storage.py

@@ -0,0 +1,39 @@
+# Storage is from web.py utils
+# https://github.com/webpy/webpy/blob/d69a49eb3c593be21fa4a5275ca9f028245678fd/web/utils.py#L81
+
+class Storage(dict):
+    """
+    A Storage object is like a dictionary except `obj.foo` can be used
+    in addition to `obj['foo']`.
+        >>> o = storage(a=1)
+        >>> o.a
+        1
+        >>> o['a']
+        1
+        >>> o.a = 2
+        >>> o['a']
+        2
+        >>> del o.a
+        >>> o.a
+        Traceback (most recent call last):
+            ...
+        AttributeError: 'a'
+    """
+
+    def __getattr__(self, key):
+        try:
+            return self[key]
+        except KeyError as k:
+            raise AttributeError(k)
+
+    def __setattr__(self, key, value):
+        self[key] = value
+
+    def __delattr__(self, key):
+        try:
+            del self[key]
+        except KeyError as k:
+            raise AttributeError(k)
+
+    def __repr__(self):
+        return "<Storage " + dict.__repr__(self) + ">"