多种类型的文件读取,并范围dict格式数据,常用场景为config文件读取,只需要输入文件,自动根据格式类型进行读取
目前支持格式:.py .ini .json .yaml

import configparser
import importlib.util
import json
import os
from collections import defaultdict
from pathlib import Path

import logging
import yaml
import base64


def base64_utf8(value):
    if not isinstance(value, str):
        return value

    try:
        decode = base64.b64decode(value)
        decode.decode('utf-8')
        return decode

    except Exception:
        return value


class ConfigReader:
    def __init__(self, file_path: Path.is_file = None, config_dir: Path.is_dir = None):
        """
        配置文件读取
        :param file_path: 配置文件地址,默认为当前地址
        :param config_dir: 配置文件路径,如填写则读取该路径下所有文件
        """
        super().__init__()
        self.config = defaultdict(lambda: defaultdict(any))
        self.current_file = Path(file_path).resolve() if file_path else None
        self.config_dir = Path(config_dir) if config_dir else Path(__file__).resolve().parent

        if file_path:
            self._load_single_file(self.current_file)
        else:
            self._load_directory()

    def _load_single_file(self, file_path: Path.is_file):
        """
        根据文件后缀名选择对应的读取方法
        :param file_path:
        :return:
        """
        logging.debug(f"Read config file: {str(file_path)}")
        suffix = file_path.suffix.lower()
        if suffix == '.json':
            self._load_json(file_path)
        elif suffix in ('.yaml', '.yml'):
            self._load_yaml(file_path)
        elif suffix == '.ini':
            self._load_ini(file_path)
        elif suffix == '.env':
            self._load_env(file_path)
        elif suffix == '.py':
            self._load_py(file_path)

    def _load_directory(self):
        """
        获取指定路径下的所有配置文件
        :return:
        """
        for config_file in self.config_dir.iterdir():
            if config_file.is_file() and config_file.name not in [os.path.basename(__file__), "__init__.py"]:
                try:
                    self._load_single_file(config_file)
                except ValueError:
                    continue

    def _load_json(self, file_path):
        """
        读取json文件
        :param file_path: json文件路径
        :return:
        """
        with open(file_path, 'r', encoding='utf-8') as f:
            self.config.update(json.load(f))

    def _load_yaml(self, file_path):
        """
        读取yaml配置文件
        :param file_path: yaml文件路径
        :return:
        """
        if yaml is None:
            raise ImportError("PyYAML required for YAML support")
        with open(file_path, 'r', encoding='utf-8') as f:
            self.config.update(yaml.safe_load(f) or {})

    def _load_ini(self, file_path):
        """
        读取ini类型配置文件
        :param file_path: ini 文件路径
        :return:
        """
        parser = configparser.ConfigParser()
        parser.read(file_path, encoding="utf-8")
        for section in parser.sections():
            for key, value in parser.items(section):
                if ',' in value:
                    self.config[section][key] = value.split(',')
                else:
                    self.config[section][key] = value

    def _load_env(self, file_path):
        """
        读取env类型配置文件
        :param file_path: env 文件路径
        :return:
        """
        with open(file_path, 'r', encoding='utf-8') as f:
            for line in f:
                line = line.strip()
                if line and not line.startswith('#'):
                    key, value = line.split('=', 1)
                    self.config[key.strip()] = value.strip()

    def _load_py(self, file_path):
        """
        读取py类型配置文件
        :param file_path: py 文件路径
        :return:
        """
        module_name = file_path.stem
        spec = importlib.util.spec_from_file_location(module_name, str(file_path))
        module = importlib.util.module_from_spec(spec)
        spec.loader.exec_module(module)
        self.config.update({
            k: v for k, v in vars(module).items()
            if not k.startswith('_')
        })

    def get(self, key, default=None) -> configparser:
        """
        根据对应的key,返回对应的配置信息
        :param key: 徐亚获取的配置
        :param default: 默认返回内容
        :return:
        """
        keys = key.split('.')
        value = self.config
        try:
            for k in keys:
                value = value[k]
            return base64_utf8(value)
        except (KeyError, TypeError):
            logging.debug(f"The configuration information of {key} is not obtained")
            return default


if __name__ == '__main__':
    dir_reader = ConfigReader(config_dir=r"E:\kali_python\furate-cve\config")
    print(f"合并后的配置项: {dir_reader.config}")

"""
from config_reader import ConfigReader

# 1. 单文件读取示例
print("=== JSON配置读取 ===")
json_config = ConfigReader("config.json")
print(f"数据库主机: {json_config.get('database.host')}")
print(f"缓存端口: {json_config.get('cache.port', 6379)}")

# 2. 目录扫描示例
print("\n=== 目录配置扫描 ===")
dir_reader = ConfigReader(config_dir="conf")
print(f"合并后的配置项: {dir_reader.config}")

# 3. 环境变量覆盖示例
print("\n=== 环境变量优先 ===")
env_config = ConfigReader(".env")
print(f"API密钥: {env_config.get('API_KEY')[:3]}***")

# 4. Python模块配置示例
print("\n=== PY配置读取 ===")
py_config = ConfigReader("settings.py")
print(f"调试模式: {py_config.get('DEBUG')}")
print(f"日志等级: {py_config.get('LOG_LEVEL')}")
Logo

Agent 垂直技术社区,欢迎活跃、内容共建。

更多推荐