124 lines
5.0 KiB
Python
124 lines
5.0 KiB
Python
import logging
|
|
import os
|
|
import os.path
|
|
from collections import ChainMap
|
|
from configparser import ConfigParser
|
|
from shutil import copyfile
|
|
from typing import Optional, Any
|
|
|
|
from django.conf import settings as dj_settings
|
|
from django.contrib.auth.models import User
|
|
|
|
from .models import UserSettings, Subscription
|
|
from .utils.extended_interpolation_with_env import ExtendedInterpolatorWithEnv
|
|
|
|
_CONFIG_DIR = os.path.join(dj_settings.BASE_DIR, 'config')
|
|
_LOG_FILE = 'log.log'
|
|
_LOG_PATH = os.path.join(_CONFIG_DIR, _LOG_FILE)
|
|
_LOG_FORMAT = '%(asctime)s|%(process)d|%(thread)d|%(name)s|%(filename)s|%(lineno)d|%(levelname)s|%(message)s'
|
|
|
|
|
|
class AppSettings(ConfigParser):
|
|
_DEFAULT_INTERPOLATION = ExtendedInterpolatorWithEnv()
|
|
__DEFAULTS_FILE = 'defaults.ini'
|
|
__SETTINGS_FILE = 'config.ini'
|
|
|
|
def __init__(self, *args, **kwargs):
|
|
super().__init__(allow_no_value=True, *args, **kwargs)
|
|
self.__defaults_path = os.path.join(_CONFIG_DIR, AppSettings.__DEFAULTS_FILE)
|
|
self.__settings_path = os.path.join(_CONFIG_DIR, AppSettings.__SETTINGS_FILE)
|
|
|
|
def initialize(self):
|
|
self.read([self.__defaults_path, self.__settings_path])
|
|
|
|
def save(self):
|
|
if os.path.exists(self.__settings_path):
|
|
# Create a backup
|
|
copyfile(self.__settings_path, self.__settings_path + ".backup")
|
|
else:
|
|
# Ensure directory exists
|
|
settings_dir = os.path.dirname(self.__settings_path)
|
|
os.makedirs(settings_dir, exist_ok=True)
|
|
|
|
with open(self.__settings_path, 'w') as f:
|
|
self.write(f)
|
|
|
|
def __get_combined_dict(self, vars: Optional[Any], sub: Optional[Subscription], user: Optional[User]) -> ChainMap:
|
|
vars_dict = {}
|
|
sub_overloads_dict = {}
|
|
user_settings_dict = {}
|
|
|
|
if vars is not None:
|
|
vars_dict = vars
|
|
|
|
if sub is not None:
|
|
sub_overloads_dict = sub.get_overloads_dict()
|
|
|
|
if user is not None:
|
|
user_settings = UserSettings.find_by_user(user)
|
|
if user_settings is not None:
|
|
user_settings_dict = user_settings.to_dict()
|
|
|
|
return ChainMap(vars_dict, sub_overloads_dict, user_settings_dict)
|
|
|
|
def get_user(self, user: User, section: str, option: Any, vars=None, fallback=object()) -> str:
|
|
return super().get(section, option,
|
|
fallback=fallback,
|
|
vars=self.__get_combined_dict(vars, None, user))
|
|
|
|
def getboolean_user(self, user: User, section: str, option: Any, vars=None, fallback=object()) -> bool:
|
|
return super().getboolean(section, option,
|
|
fallback=fallback,
|
|
vars=self.__get_combined_dict(vars, None, user))
|
|
|
|
def getint_user(self, user: User, section: str, option: Any, vars=None, fallback=object()) -> int:
|
|
return super().getint(section, option,
|
|
fallback=fallback,
|
|
vars=self.__get_combined_dict(vars, None, user))
|
|
|
|
def getfloat_user(self, user: User, section: str, option: Any, vars=None, fallback=object()) -> float:
|
|
return super().getfloat(section, option,
|
|
fallback=fallback,
|
|
vars=self.__get_combined_dict(vars, None, user))
|
|
|
|
def get_sub(self, sub: Subscription, section: str, option: Any, vars=None, fallback=object()) -> str:
|
|
return super().get(section, option,
|
|
fallback=fallback,
|
|
vars=self.__get_combined_dict(vars, sub, sub.user))
|
|
|
|
def getboolean_sub(self, sub: Subscription, section: str, option: Any, vars=None, fallback=object()) -> bool:
|
|
return super().getboolean(section, option,
|
|
fallback=fallback,
|
|
vars=self.__get_combined_dict(vars, sub, sub.user))
|
|
|
|
def getint_sub(self, sub: Subscription, section: str, option: Any, vars=None, fallback=object()) -> int:
|
|
return super().getint(section, option,
|
|
fallback=fallback,
|
|
vars=self.__get_combined_dict(vars, sub, sub.user))
|
|
|
|
def getfloat_sub(self, sub: Subscription, section: str, option: Any, vars=None, fallback=object()) -> float:
|
|
return super().getfloat(section, option,
|
|
fallback=fallback,
|
|
vars=self.__get_combined_dict(vars, sub, sub.user))
|
|
|
|
|
|
settings = AppSettings()
|
|
|
|
|
|
def initialize_app_config():
|
|
settings.initialize()
|
|
__initialize_logger()
|
|
logging.info('Application started!')
|
|
|
|
|
|
def __initialize_logger():
|
|
log_level_str = settings.get('global', 'LogLevel', fallback='INFO')
|
|
|
|
try:
|
|
log_level = getattr(logging, log_level_str)
|
|
logging.basicConfig(filename=_LOG_PATH, level=log_level, format=_LOG_FORMAT)
|
|
|
|
except AttributeError:
|
|
logging.basicConfig(filename=_LOG_PATH, level=logging.INFO, format=_LOG_FORMAT)
|
|
logging.warning('Invalid log level "%s" in config file.', log_level_str)
|