import logging import os from pathlib import Path from typing import Dict, Any, List import uvicorn from config_file import read_yaml APP_VERSION="1.0" SCRIPT_PATH = os.path.dirname(__file__) CONFIGS_DIR = SCRIPT_PATH + "/configs" CONFIG_FILE_NAME = CONFIGS_DIR + "/config.yaml" SECRETS_NAME = 'secrets.yaml' CONFIG: Dict[str, Any] = {} API_KEY = '' IN_MEMORY_LOGS = False IN_MEMORY_LOGS_LEN = 100 HOST = '0.0.0.0' PORT: int = 8000 IS_DEBUG = False TG_TOKEN = "" TG_CHAT = "" USE_MATRIX = False MX_SERVER = "" MX_ROOM = "" MX_TOKEN = "" ACCOUNT = "" SETTLEMENT = "" STREET = "" HOUSE = "" BUILDING_PART_NUMBER = "" APARTMENT = "" PROCESS_START_HOUR = 6 PROCESS_STOP_HOUR = 23 PROCESSING_MINUTES: List[int] = [5, 15, 30, 45, 55] def read_app_config(): global CONFIG if not CONFIG: config_path = Path(CONFIG_FILE_NAME) if not config_path.exists(): LOG.error(f"{CONFIG_FILE_NAME} not found!") raise FileNotFoundError(f"{CONFIG_FILE_NAME} not found") CONFIG = read_yaml(CONFIG_FILE_NAME, SECRETS_NAME) def get_config_value(cfg, section, key, default): return cfg[section][key] if section in cfg else default def parse_config(): global API_KEY, IN_MEMORY_LOGS, IN_MEMORY_LOGS_LEN, HOST, PORT, IS_DEBUG, \ TG_TOKEN, TG_CHAT, \ USE_MATRIX, MX_SERVER, MX_ROOM, MX_TOKEN, \ ACCOUNT, SETTLEMENT, STREET, HOUSE, BUILDING_PART_NUMBER, APARTMENT, \ PROCESS_START_HOUR, PROCESS_STOP_HOUR HOST = CONFIG.get('application', {}).get('host', '') if not HOST: HOST = '0.0.0.0' PORT = CONFIG.get('application', {}).get('port', 8000) API_KEY = CONFIG.get('application').get('api-key', '') IN_MEMORY_LOGS = str(CONFIG.get('application', {}).get('logs_ep', {}).get('enabled', False)).lower() == 'true' IN_MEMORY_LOGS_LEN = CONFIG.get('application', {}).get('logs_ep', {}).get('max_records', 100) IS_DEBUG = CONFIG.get('application').get('debug', False) == 'true' TG_TOKEN = CONFIG.get('notifications', {}).get('tg', {}).get('token', '') TG_CHAT = CONFIG.get('notifications', {}).get('tg', {}).get('chat_id', '') USE_MATRIX = str(CONFIG.get('notifications', {}).get('matrix', {}).get('use_matrix', False)).lower() == 'true' MX_SERVER = CONFIG.get('notifications', {}).get('matrix', {}).get('mx_server', '') MX_ROOM = CONFIG.get('notifications', {}).get('matrix', {}).get('mx_room_id', '') MX_TOKEN = CONFIG.get('notifications', {}).get('matrix', {}).get('mx_access_token', '') ACCOUNT = CONFIG.get('data', {}).get('account', {}) SETTLEMENT = CONFIG.get('data', {}).get('settlement', {}) STREET = CONFIG.get('data', {}).get('street', {}) HOUSE = CONFIG.get('data', {}).get('house', {}) BUILDING_PART_NUMBER = CONFIG.get('data', {}).get('building_part_number', {}) APARTMENT = CONFIG.get('data', {}).get('apartment', {}) PROCESS_START_HOUR = CONFIG.get('application').get('process', {}).get('start_hour', 6) PROCESS_STOP_HOUR = CONFIG.get('application').get('process', {}).get('stop_hour', 23) if PROCESS_START_HOUR > PROCESS_STOP_HOUR: PROCESS_STOP_HOUR = 23 def get_uvicorn_config(): # global HOST, PORT if HOST and PORT: cfg = { "host": f"{HOST}", "port": PORT, "access_log": True, "log_level": "info", "log_config": f"{CONFIGS_DIR}/log.yaml" # "docs_url": "/api-docs" } else: cfg = { "host": "127.0.0.1", "port": 8000, "reload": True, "access_log": False, "log_level": "info", "log_config": f"{CONFIGS_DIR}/log.yaml" # "docs_url": "/api-docs" } LOG.debug(f'got uvicorn_config: {cfg}') return cfg def log_config(): # Not used due to not working cfg = uvicorn.config.LOGGING_CONFIG cfg["formatters"]["access"]["fmt"] = ( '%(asctime)s [%(levelname)s] - %(name)s: %(funcName)s[%(lineno)d] - %(message)s' ) return cfg LOG_LIST = [] class ListHandler(logging.Handler): def emit(self, record): if IN_MEMORY_LOGS: LOG_LIST.append(self.format(record)) while len(LOG_LIST) > IN_MEMORY_LOGS_LEN: LOG_LIST.pop(0) ### LOG = logging.getLogger(__name__) handler = ListHandler() handler.setFormatter(logging.Formatter('%(asctime)s [%(levelname)s] - %(name)s: %(funcName)s[%(lineno)d] - %(message)s')) LOG.addHandler(handler) def init_config(): read_app_config() parse_config() if __name__ == '__main__': init_config()