157 lines
5.3 KiB
Python
157 lines
5.3 KiB
Python
# config/config_manager.py
|
||
import logging
|
||
import os
|
||
import json
|
||
import configparser
|
||
import time
|
||
import threading
|
||
from common.logging_service import logging_service
|
||
from common.ini_handler import IniHandler
|
||
|
||
|
||
class IniManager:
|
||
"""
|
||
settings.init配置管理类，读取API相关配置
|
||
暂不支持多个配置文件
|
||
"""
|
||
|
||
def __init__(self):
|
||
"""初始化实例变量"""
|
||
self._DEFAULT_CONFIG_FILE = os.path.join(
|
||
os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
|
||
'settings.ini'
|
||
)
|
||
self.ini_handler = IniHandler(self._DEFAULT_CONFIG_FILE)
|
||
# 默认配置值
|
||
self._DEFAULT_API_MAX_RETRIES = 3
|
||
self._DEFAULT_API_RETRY_INTERVAL = 1.0
|
||
self._DEFAULT_API_TIMEOUT = 30.0
|
||
self._DEFAULT_API_AUTH_TIMEOUT = 12 * 60 * 60 # 12小时
|
||
self._DEFAULT_API_BASE_URL = "https://www.shnthy.com:9154"
|
||
self._DEFAULT_API_LOGIN_MODEL = {
|
||
"Program": 11,
|
||
"SC": "1000000001",
|
||
"loginName": "leduser",
|
||
"password": "bfcda35cf4eba92d4583931bbe4ff72ffdfa8b5c9c4b72611bd33f5babee069d"
|
||
}
|
||
|
||
# 参数缓存字典，格式：{参数名: {'value': 值, 'mtime': 文件修改时间}}
|
||
self._param_cache = {}
|
||
# 缓存锁，确保多线程安全
|
||
self._cache_lock = threading.RLock()
|
||
# 文件修改时间缓存
|
||
self._file_mtime_cache = None
|
||
|
||
def _get_file_mtime(self):
|
||
"""获取文件修改时间，带缓存"""
|
||
try:
|
||
current_mtime = os.path.getmtime(self._DEFAULT_CONFIG_FILE)
|
||
return current_mtime
|
||
except (FileNotFoundError, OSError):
|
||
raise FileNotFoundError(f"配置文件不存在或无法访问: settings.ini")
|
||
|
||
def _read_config_value(self, section, key, default_value, value_type=str):
|
||
"""读取配置值，带参数级缓存
|
||
|
||
Args:
|
||
section: 配置段
|
||
key: 配置键
|
||
default_value: 默认值
|
||
value_type: 值类型
|
||
|
||
Returns:
|
||
配置值
|
||
"""
|
||
cache_key = f"{section}.{key}"
|
||
current_mtime = self._get_file_mtime()
|
||
# 检查缓存
|
||
if cache_key in self._param_cache:
|
||
cached_data = self._param_cache[cache_key]
|
||
# 如果文件未修改，直接返回缓存值
|
||
if cached_data['value'] is not None and cached_data['mtime'] == current_mtime:
|
||
return cached_data['value']
|
||
|
||
try:
|
||
# 获取值
|
||
if value_type == int:
|
||
value = self.ini_handler.get_int_value(section, key)
|
||
elif value_type == float:
|
||
value = self.ini_handler.get_float_value(section, key)
|
||
else:
|
||
value = self.ini_handler.get_value(section, key)
|
||
|
||
except Exception as e:
|
||
value = default_value
|
||
logging_service.log_message(logging.ERROR,str(e))
|
||
|
||
# 使用锁确保缓存操作的线程安全
|
||
with self._cache_lock:
|
||
self._param_cache[cache_key] = {
|
||
'value': value,
|
||
'mtime': current_mtime
|
||
}
|
||
return value
|
||
|
||
@property
|
||
def api_max_retries(self):
|
||
"""获取API最大重试次数"""
|
||
return self._read_config_value('api', 'max_retries', self._DEFAULT_API_MAX_RETRIES, int)
|
||
|
||
@property
|
||
def api_retry_interval(self):
|
||
"""获取API重试间隔(秒)"""
|
||
return self._read_config_value('api', 'retry_interval', self._DEFAULT_API_RETRY_INTERVAL, float)
|
||
|
||
@property
|
||
def api_timeout(self):
|
||
"""获取API请求超时时间(秒)"""
|
||
return self._read_config_value('api', 'timeout', self._DEFAULT_API_TIMEOUT, float)
|
||
|
||
@property
|
||
def api_auth_timeout(self):
|
||
"""获取API授权失效前最大时间(秒)"""
|
||
return self._read_config_value('api', 'auth_timeout', self._DEFAULT_API_AUTH_TIMEOUT, int)
|
||
|
||
@property
|
||
def api_base_url(self):
|
||
"""获取API基础URL"""
|
||
return self._read_config_value('api', 'base_url', self._DEFAULT_API_BASE_URL, str)
|
||
|
||
@property
|
||
def api_login_url(self):
|
||
"""获取API登录URL"""
|
||
return f"{self.api_base_url}/api/user/perlogin"
|
||
|
||
@property
|
||
def api_login_model(self):
|
||
"""获取API登录模型(字典格式)"""
|
||
login_model_str = self._read_config_value('api', 'login_model', None, str)
|
||
if login_model_str:
|
||
try:
|
||
value = json.loads(login_model_str)
|
||
except Exception as e:
|
||
print(f"解析API登录模型失败: {e}")
|
||
value = self._DEFAULT_API_LOGIN_MODEL
|
||
else:
|
||
value = self._DEFAULT_API_LOGIN_MODEL
|
||
|
||
return value
|
||
|
||
@property
|
||
def log_path(self):
|
||
"""获取日志文件路径"""
|
||
return self._read_config_value('app', 'LOG_PATH', 'logs/app.log', str)
|
||
|
||
ini_manager = IniManager()
|
||
|
||
|
||
# if __name__ == '__main__':
|
||
# logging_service.init_log(ini_manager.log_path)
|
||
# try:
|
||
|
||
# logging_service.log_message(logging.INFO, "测试INI配置管理类")
|
||
# logging_service.log_message(logging.ERROR, f"api_max_retries")
|
||
# except Exception as e:
|
||
# logging_service.log_message(logging.ERROR,str(e))
|
||
|
||
# time.sleep(1000) |