You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
155 lines
4.8 KiB
155 lines
4.8 KiB
""" |
|
配置文件读取工具 |
|
支持从 config.toml 文件中读取配置信息 |
|
""" |
|
import toml |
|
import os |
|
import sys |
|
|
|
|
|
def get_config_path(): |
|
""" |
|
获取配置文件路径 |
|
|
|
查找顺序： |
|
1. 如果是打包后的 exe，优先查找 exe 同级目录的 config.toml |
|
2. 开发环境：查找 factory_sliceing 目录下的 config.toml |
|
3. 当前目录下的 config.toml |
|
""" |
|
# 判断是否在 PyInstaller 打包后的环境中 |
|
if getattr(sys, 'frozen', False): |
|
# 打包后的 exe 环境 |
|
# sys.executable 是 exe 文件的完整路径 |
|
exe_dir = os.path.dirname(os.path.abspath(sys.executable)) |
|
config_path = os.path.join(exe_dir, 'config.toml') |
|
|
|
# 如果 exe 同级目录存在配置文件，直接返回 |
|
if os.path.exists(config_path): |
|
return config_path |
|
|
|
# 如果不存在，也返回这个路径（让调用者知道应该在哪里创建配置文件） |
|
return config_path |
|
|
|
# 开发环境：查找 factory_sliceing 目录下的 config.toml |
|
current_dir = os.path.dirname(os.path.abspath(__file__)) |
|
# 向上查找 factory_sliceing 目录 |
|
factory_sliceing_dir = os.path.dirname(current_dir) |
|
config_path = os.path.join(factory_sliceing_dir, 'config.toml') |
|
|
|
# 如果文件存在，返回 |
|
if os.path.exists(config_path): |
|
return config_path |
|
|
|
# 如果不存在，尝试当前目录 |
|
config_path = os.path.join(current_dir, 'config.toml') |
|
if os.path.exists(config_path): |
|
return config_path |
|
|
|
# 都不存在，返回 factory_sliceing 目录下的路径（作为默认路径） |
|
return os.path.join(factory_sliceing_dir, 'config.toml') |
|
|
|
|
|
def cfg(key_name, default=None): |
|
""" |
|
读取配置文件中的值 |
|
|
|
:param key_name: 配置项的键名，支持点号分隔的嵌套键，如 "redis.host" 或 "redis.retry.interval" |
|
:param default: 如果配置项不存在，返回的默认值 |
|
:return: 配置项的值，如果不存在且未提供默认值则返回 None |
|
""" |
|
try: |
|
config_path = get_config_path() |
|
|
|
if not os.path.exists(config_path): |
|
print(f"警告: 配置文件不存在，路径: {config_path}") |
|
return default |
|
|
|
with open(config_path, 'r', encoding='utf-8') as f: |
|
config = toml.load(f) |
|
|
|
# 处理嵌套键名，如 "redis.host" |
|
if "." in key_name: |
|
keys = key_name.split(".") |
|
value = config |
|
for key in keys: |
|
if isinstance(value, dict) and key in value: |
|
value = value[key] |
|
else: |
|
if default is not None: |
|
return default |
|
print(f"配置项 {key_name} 不存在") |
|
return None |
|
return value |
|
else: |
|
# 单层键名 |
|
if key_name in config: |
|
return config[key_name] |
|
else: |
|
if default is not None: |
|
return default |
|
print(f"配置项 {key_name} 不存在") |
|
return None |
|
except FileNotFoundError: |
|
print(f"配置文件不存在，路径: {config_path}") |
|
return default |
|
except Exception as e: |
|
print(f"读取配置文件时发生错误: {e}") |
|
return default |
|
|
|
|
|
def get_redis_config(): |
|
""" |
|
获取 Redis 配置字典 |
|
:return: Redis 配置字典 |
|
""" |
|
return { |
|
'host': cfg('redis.host', 'localhost'), |
|
'password': cfg('redis.password', ''), |
|
'port': cfg('redis.port', 6379), |
|
'db': cfg('redis.db', 0), |
|
'socket_timeout': cfg('redis.socket_timeout', 30), |
|
'socket_connect_timeout': cfg('redis.socket_connect_timeout', 10), |
|
'socket_keepalive': cfg('redis.socket_keepalive', True), |
|
'health_check_interval': cfg('redis.health_check_interval', 30) |
|
} |
|
|
|
|
|
def get_oss_config(): |
|
""" |
|
获取 OSS 配置字典 |
|
:return: OSS 配置字典 |
|
""" |
|
return { |
|
'access_key_id': cfg('oss.access_key_id'), |
|
'access_key_secret': cfg('oss.access_key_secret'), |
|
'endpoint': cfg('oss.endpoint'), |
|
'bucket': cfg('oss.bucket') |
|
} |
|
|
|
|
|
def get_api_url(): |
|
""" |
|
获取 API 基础 URL |
|
:return: API URL 字符串 |
|
""" |
|
env = cfg('api.env', 'prod') |
|
base_url = cfg('api.base_url', 'https://mp.api.suwa3d.com') |
|
|
|
# 如果环境是 dev，可以覆盖 URL |
|
if env == 'dev': |
|
dev_url = cfg('api.dev_url', 'http://mp.api.dev.com') |
|
return dev_url |
|
|
|
return base_url |
|
|
|
|
|
def get_work_dir(): |
|
""" |
|
获取默认工作目录 |
|
:return: 工作目录路径，如果未配置则返回 None |
|
""" |
|
return cfg('work.default_dir', None) |
|
|
|
|
|
def get_exe_path(typeExe): |
|
return cfg('exe.'+typeExe+'_exe', None) |