payment/payment_backend/config/utils.py

#!/usr/bin/env python
# -*- coding: UTF-8 -*-
"""
@Project ：payment
@File    ：__init__.py.py
@IDE     ：PyCharm
@Author  ：rengengchen
@Time    ：2024/11/06 16:11
"""
import argparse
import os
import random
import sys
from argparse import Namespace
from configparser import ConfigParser

import requests
from loguru import logger

random_seed = 20240717
ROOT_DIR = os.path.abspath(os.path.dirname(os.path.dirname(__file__)))


def setup_seed(seed):
    # import torch
    # torch.manual_seed(seed)
    # torch.cuda.manual_seed_all(seed)
    # torch.backends.cudnn.deterministic = True
    # import numpy as np
    # np.random.seed(seed)
    random.seed(seed)


setup_seed(random_seed)


class Setting:
    def __init__(self,
                 settings:dict = None,
                 config_parser: ConfigParser = None,
                 argument_parser: Namespace = None,
                 _visited=None,
                 _parent=None,
                 **kwargs):
        self._parent = _parent

        self.update(_visited=_visited, **settings)

        if config_parser:
            self.update_config_parser(config_parser)

        if argument_parser:
            self.update_argument_parser(argument_parser)

        self.update(_visited=_visited, **kwargs)

    def update(self, _visited=None, **kwargs):
        if _visited is None:
            _visited = {}
        for k, v in kwargs.items():
            cls_attr = getattr(self.__class__, k, None)
            if callable(cls_attr):
                raise KeyError(f"The key '{k}' conflicts with an existing class method. you can use {k}_ instead.")
            if isinstance(v, dict):
                obj_id = id(v)
                if obj_id in _visited:
                    logger.warning(f"Circular reference detected in key: '{k}'.")
                    v = _visited[obj_id]
                else:
                    v = Setting(_visited=_visited, _parent=self, **v)
                    _visited[obj_id] = v
            self.__dict__[k] = v

    def update_config_parser(self, config_parser: ConfigParser):
        for section in config_parser.sections():
            section_config = Setting(_parent=self)
            for option in config_parser.options(section):
                section_config[option] = config_parser.get(section, option)
            self.__dict__[section] = section_config

    def update_argument_parser(self, argument_parser: Namespace):
        for k in vars(argument_parser):
            self.__dict__[k] = getattr(argument_parser, k)

    def get(self, item, default=None):
        if item not in self.__dict__ and self._parent is not None:
            return self._parent.get(item, default)
        return default

    def get_int(self, item):
        return int(self.get(item))

    def get_float(self, item):
        return float(self.get(item))

    def get_bool(self, item):
        return bool(self.get(item))

    def set(self, item, value):
        self.__dict__[item] = value

    def __getitem__(self, item):
        return self.__dict__[item]

    def __setitem__(self, key, value):
        self.__dict__[key] = value

    def __getattr__(self, key):
        return self.get(key)

    def __str__(self):
        def _str_helper(config, indent=0, visited=None):
            if visited is None:
                visited = set()
            lines = []
            indent_str = '    ' * indent
            for key, value in config.__dict__.items():
                if key.startswith('_'):
                    continue
                if isinstance(value, Setting):
                    if id(value) in visited:
                        lines.append(f"{indent_str}{key}: <Circular Reference>")
                    else:
                        visited.add(id(value))
                        lines.append(f"{indent_str}{key}:")
                        lines.append(_str_helper(value, indent + 1, visited))
                else:
                    lines.append(f"{indent_str}{key}: {value}")
            return '\n'.join(lines)

        return _str_helper(self)


def log_config(config):
    # fmt = '%(asctime)s [%(name)s] %(levelname)s: %(message)s'
    # datefmt = "%Y-%m-%d %H:%M:%S"
    logger.remove()
    logger.add(sys.stdout, level=config.log_level)
    logger.add(sys.stderr, level="ERROR")
    logger.add(os.path.join(ROOT_DIR, "logs", "{time}.log"), level="DEBUG", encoding='utf8', rotation="100 MB",
               retention=3)


def get_config(yaml_file=fr'{ROOT_DIR}/config/param.yaml', ini_file=fr'{ROOT_DIR}/config/param.ini') -> Setting:
    requests.adapters.DEFAULT_RETRIES = 3

    config = Setting()
    if os.path.exists(yaml_file):
        configparser = ConfigParser()
        configparser.read(ini_file)
        config.update_config_parser(configparser)
    parser = argparse.ArgumentParser(description='payment system')
    parser.add_argument("--seed", type=int, default=2024)
    args = parser.parse_args()
    config.update_argument_parser(args)
    if os.path.exists(yaml_file):
        with open('config.yaml', 'r') as file:
            data = yaml.safe_load(file)
        config.update(data)
    return config