feat: support parse config file with Variable Expansion for some secret key

backend/src/module/api/config.py

@@ -16,7 +16,7 @@ async def get_config(current_user=Depends(get_current_user)):
         raise HTTPException(
             status_code=status.HTTP_401_UNAUTHORIZED, detail="invalid token"
         )
-    return settings
+    return settings.dict()
 
 
 @router.post("/updateConfig")

backend/src/module/database/orm/connector.py

@@ -1,3 +1,4 @@
+import os
 import sqlite3
 
 from .delete import Delete
@@ -10,6 +11,10 @@ from module.conf import DATA_PATH
 
 class Connector:
     def __init__(self, table_name: str, data: dict, database: str = DATA_PATH):
+        # Create folder if not exists
+        if not os.path.exists(os.path.dirname(DATA_PATH)):
+            os.makedirs(os.path.dirname(DATA_PATH))
+
         self._conn = sqlite3.connect(database)
         self._cursor = self._conn.cursor()
         self.update = Update(self, table_name, data)

backend/src/module/models/config.py

@@ -1,3 +1,4 @@
+from os.path import expandvars
 from pydantic import BaseModel, Field
 
 # Sub config
@@ -12,21 +13,31 @@ class Program(BaseModel):
 class Downloader(BaseModel):
     type: str = Field("qbittorrent", description="Downloader type")
     host: str = Field("172.17.0.1:8080", description="Downloader host")
-    username: str = Field("admin", description="Downloader username")
-    password: str = Field("adminadmin", description="Downloader password")
+    username_: str = Field("admin", alias="username", description="Downloader username")
+    password_: str = Field("adminadmin", alias="password", description="Downloader password")
     path: str = Field("/downloads/Bangumi", description="Downloader path")
     ssl: bool = Field(False, description="Downloader ssl")
 
+    @property
+    def username(self):
+        return expandvars(self.username_)
+
+    @property
+    def password(self):
+        return expandvars(self.password_)
 
 class RSSParser(BaseModel):
     enable: bool = Field(True, description="Enable RSS parser")
     type: str = Field("mikan", description="RSS parser type")
-    token: str = Field("token", description="RSS parser token")
+    token_: str = Field("token", alias="token", description="RSS parser token")
     custom_url: str = Field("mikanani.me", description="Custom RSS host url")
     parser_type: str = Field("parser", description="Parser type")
     filter: list[str] = Field(["720", r"\d+-\d"], description="Filter")
     language: str = "zh"
 
+    @property
+    def token(self):
+        return expandvars(self.token_)
 
 class BangumiManage(BaseModel):
     enable: bool = Field(True, description="Enable bangumi manage")
@@ -45,16 +56,31 @@ class Proxy(BaseModel):
     type: str = Field("http", description="Proxy type")
     host: str = Field("", description="Proxy host")
     port: int = Field(0, description="Proxy port")
-    username: str = Field("", description="Proxy username")
-    password: str = Field("", description="Proxy password")
+    username_: str = Field("", alias="username", description="Proxy username")
+    password_: str = Field("", alias="password", description="Proxy password")
+
+    @property
+    def username(self):
+        return expandvars(self.username_)
+
+    @property
+    def password(self):
+        return expandvars(self.password_)
 
 
 class Notification(BaseModel):
     enable: bool = Field(False, description="Enable notification")
     type: str = Field("telegram", description="Notification type")
-    token: str = Field("", description="Notification token")
-    chat_id: str = Field("", description="Notification chat id")
+    token_: str = Field("", alias="token", description="Notification token")
+    chat_id_: str = Field("", alias="chat_id", description="Notification chat id")
 
+    @property
+    def token(self):
+        return expandvars(self.token_)
+
+    @property
+    def chat_id(self):
+        return expandvars(self.chat_id_)
 
 class Config(BaseModel):
     program: Program = Program()
@@ -64,3 +90,6 @@ class Config(BaseModel):
     log: Log = Log()
     proxy: Proxy = Proxy()
     notification: Notification = Notification()
+
+    def dict(self, *args, by_alias=True, **kwargs):
+        return super().dict(*args, by_alias=by_alias, **kwargs)