nas/Auto_Bangumi
1
0
Fork 0
mirror of https://github.com/EstrellaXD/Auto_Bangumi.git synced 2026-05-12 02:56:13 +08:00
Code Issues Packages Projects Releases Wiki Activity

- Rewrite config class

Browse Source 
- Change multiprocessing to thread
- black all code
This commit is contained in:
EstrellaXD
2023-05-06 23:47:44 +08:00
parent d7a41d7404
commit 5c71b9d35d
38 changed files with 400 additions and 313 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
src/module/conf/__init__.py
View File

@@ -1,17 +1,8 @@
from .log import setup_logger, LOG_PATH
from .config import settings, VERSION
from .config import VERSION, settings
TMDB_API = "32b19d6a05b512190a056fa4e747cbbc"
DATA_PATH = "data/data.db"
class RSSLink(str):
def __new__(cls):
if "://" not in settings.rss_parser.custom_url:
return f"https://{settings.rss_parser.custom_url}/RSS/MyBangumi?token={settings.rss_parser.token}"
return f"{settings.rss_parser.custom_url}/RSS/MyBangumi?token={settings.rss_parser.token}"
PLATFORM = "Windows" if "\\" in settings.downloader.path else "Unix"
MIKANANI_URL = "mikanani.me"

128
src/module/conf/config.py
View File

@@ -3,80 +3,82 @@ import os
import logging
from dotenv import load_dotenv
from module.conf.const import ENV_TO_ATTR
from .const import ENV_TO_ATTR
from module.models.config import Config
logger = logging.getLogger(__name__)
try:
from ..__version__ import VERSION
from module.__version__ import VERSION
if VERSION == "DEV_VERSION":
logger.info("Can't find version info, use DEV_VERSION instead")
CONFIG_PATH = "config/config_dev.json"
else:
CONFIG_PATH = f"config/config.json"
except ImportError:
logger.info("Can't find version info, use DEV_VERSION instead")
VERSION = "DEV_VERSION"
CONFIG_PATH = "config/config_dev.json"
class Setting(Config):
@staticmethod
def reload():
load_config_from_file(CONFIG_PATH)
class Settings(Config):
def __init__(self):
super().__init__()
if os.path.exists(CONFIG_PATH):
self.load()
self.save()
else:
# load from env
load_dotenv(".env")
self.__load_from_env()
self.save()
def load(self):
with open(CONFIG_PATH, "r", encoding="utf-8") as f:
config = json.load(f)
config_obj = Config.parse_obj(config)
self.__dict__.update(config_obj.__dict__)
logger.info(f"Config loaded")
def save(self):
save_config_to_file(self, CONFIG_PATH)
def save_config_to_file(config: Config, path: str):
config_dict = config.dict()
with open(path, "w", encoding="utf-8") as f:
json.dump(config_dict, f, indent=4)
logger.info(f"Config saved")
def load_config_from_file(path: str) -> Config:
with open(path, "r", encoding="utf-8") as f:
config = json.load(f)
return Setting(**config)
def _val_from_env(env: str, attr: tuple):
if isinstance(attr, tuple):
conv_func = attr[1]
return conv_func(os.environ[env])
else:
return os.environ[env]
def env_to_config() -> Setting:
_settings = Setting().dict()
for key, section in ENV_TO_ATTR.items():
for env, attr in section.items():
if env in os.environ:
if isinstance(attr, list):
for _attr in attr:
attr_name = _attr[0] if isinstance(_attr, tuple) else _attr
_settings[key][attr_name] = _val_from_env(env, _attr)
else:
attr_name = attr[0] if isinstance(attr, tuple) else attr
_settings[key][attr_name] = _val_from_env(env, attr)
return Setting(**_settings)
if os.path.isdir("config") and VERSION == "DEV_VERSION":
CONFIG_PATH = "config/config_dev.json"
if os.path.isfile(CONFIG_PATH):
settings = load_config_from_file(CONFIG_PATH)
else:
load_dotenv(".env")
settings = env_to_config()
save_config_to_file(settings, CONFIG_PATH)
elif os.path.isdir("config") and VERSION != "DEV_VERSION":
CONFIG_PATH = "config/config.json"
if os.path.isfile(CONFIG_PATH):
settings = load_config_from_file(CONFIG_PATH)
else:
settings = env_to_config()
save_config_to_file(settings, CONFIG_PATH)
else:
settings = Setting()
config_dict = self.dict()
with open(CONFIG_PATH, "w", encoding="utf-8") as f:
json.dump(config_dict, f, indent=4)
logger.info(f"Config saved")
def rss_link(self):
if "://" not in self.rss_parser.custom_url:
return f"https://{self.rss_parser.custom_url}/RSS/MyBangumi?token={self.rss_parser.token}"
return (
f"{self.rss_parser.custom_url}/RSS/MyBangumi?token={self.rss_parser.token}"
)
def __load_from_env(self):
config_dict = self.dict()
for key, section in ENV_TO_ATTR.items():
for env, attr in section.items():
if env in os.environ:
if isinstance(attr, list):
for _attr in attr:
attr_name = _attr[0] if isinstance(_attr, tuple) else _attr
config_dict[key][attr_name] = self.__val_from_env(
env, _attr
)
else:
attr_name = attr[0] if isinstance(attr, tuple) else attr
config_dict[key][attr_name] = self.__val_from_env(env, attr)
config_obj = Config.parse_obj(config_dict)
self.__dict__.update(config_obj.__dict__)
logger.info(f"Config loaded from env")
@staticmethod
def __val_from_env(env: str, attr: tuple):
if isinstance(attr, tuple):
conv_func = attr[1]
return conv_func(os.environ[env])
else:
return os.environ[env]
settings = Settings()

22
src/module/conf/const.py
View File

@@ -7,7 +7,7 @@ DEFAULT_SETTINGS = {
"sleep_time": 7200,
"times": 20,
"webui_port": 7892,
"data_version": 4.0
"data_version": 4.0,
},
"downloader": {
"type": "qbittorrent",
@@ -15,7 +15,7 @@ DEFAULT_SETTINGS = {
"username": "admin",
"password": "adminadmin",
"path": "/downloads/Bangumi",
"ssl": False
"ssl": False,
},
"rss_parser": {
"enable": True,
@@ -24,14 +24,14 @@ DEFAULT_SETTINGS = {
"token": "",
"enable_tmdb": False,
"filter": ["720", "\\d+-\\d+"],
"language": "zh"
"language": "zh",
},
"bangumi_manage": {
"enable": True,
"eps_complete": False,
"rename_method": "pn",
"group_tag": False,
"remove_bad_torrent": False
"remove_bad_torrent": False,
},
"log": {
"debug_enable": False,
@@ -42,14 +42,9 @@ DEFAULT_SETTINGS = {
"host": "",
"port": 1080,
"username": "",
"password": ""
"password": "",
},
"notification": {
"enable": False,
"type": "telegram",
"token": "",
"chat_id": ""
}
"notification": {"enable": False, "type": "telegram", "token": "", "chat_id": ""},
}
@@ -80,7 +75,10 @@ ENV_TO_ATTR = {
"AB_METHOD": ("rename_method", lambda e: e.lower()),
"AB_GROUP_TAG": ("group_tag", lambda e: e.lower() in ("true", "1", "t")),
"AB_EP_COMPLETE": ("eps_complete", lambda e: e.lower() in ("true", "1", "t")),
"AB_REMOVE_BAD_BT": ("remove_bad_torrent", lambda e: e.lower() in ("true", "1", "t")),
"AB_REMOVE_BAD_BT": (
"remove_bad_torrent",
lambda e: e.lower() in ("true", "1", "t"),
),
},
"log": {
"AB_DEBUG_MODE": ("debug_enable", lambda e: e.lower() in ("true", "1", "t")),

12
src/module/conf/log.py
View File

@@ -1,15 +1,15 @@
import logging
from .config import settings
from module.models import Config
LOG_PATH = "data/log.txt"
def setup_logger():
def setup_logger(settings: Config):
level = logging.DEBUG if settings.log.debug_enable else logging.INFO
logging.addLevelName(logging.DEBUG, 'DEBUG:')
logging.addLevelName(logging.INFO, 'INFO:')
logging.addLevelName(logging.WARNING, 'WARNING:')
logging.addLevelName(logging.DEBUG, "DEBUG:")
logging.addLevelName(logging.INFO, "INFO:")
logging.addLevelName(logging.WARNING, "WARNING:")
LOGGING_FORMAT = "[%(asctime)s] %(levelname)-8s %(message)s"
logging.basicConfig(
level=level,
@@ -18,5 +18,5 @@ def setup_logger():
handlers=[
logging.FileHandler(LOG_PATH, encoding="utf-8"),
logging.StreamHandler(),
]
],
)

2
src/module/conf/parse.py
View File

@@ -12,4 +12,4 @@ def parse():
)
parser.add_argument("-d", "--debug", action="store_true", help="debug mode")
return parser.parse_args()
return parser.parse_args()