mirror of
https://github.com/Nanaloveyuki/py-logiliteal.git
synced 2025-09-05 11:06:22 +00:00
⚡️ 提升了config
读取写入重置的速度
This commit is contained in:
2
.gitignore
vendored
2
.gitignore
vendored
@ -1,5 +1,5 @@
|
|||||||
# Custom
|
# Custom
|
||||||
*.backup.json
|
*_backup.json
|
||||||
|
|
||||||
# Byte-compiled / optimized / DLL files
|
# Byte-compiled / optimized / DLL files
|
||||||
__pycache__/
|
__pycache__/
|
||||||
|
@ -6,7 +6,15 @@ py-logiliteal's config settings, used to set py-logiliteal's global config
|
|||||||
# encoding = utf-8
|
# encoding = utf-8
|
||||||
# python 3.13.5
|
# python 3.13.5
|
||||||
|
|
||||||
DEFAULT_CONFIG_PATH = "config.json"
|
import json
|
||||||
|
from os import remove
|
||||||
|
import shutil
|
||||||
|
import time
|
||||||
|
from pathlib import Path
|
||||||
|
from typing import Union, Optional, Tuple
|
||||||
|
from logging import error
|
||||||
|
|
||||||
|
DEFAULT_CONFIG_PATH = "logger_config.json"
|
||||||
DEFAULT_CONFIG = {
|
DEFAULT_CONFIG = {
|
||||||
"file_level": "DEBUG",
|
"file_level": "DEBUG",
|
||||||
"file_name": "latest.log",
|
"file_name": "latest.log",
|
||||||
@ -25,9 +33,47 @@ DEFAULT_CONFIG = {
|
|||||||
"level_color": {"DEBUG": "#c1d5ff", "INFO": "#c1ffff", "WARN": "#fff600", "ERRO": "#ffa000", "CRIT": "#ff8181"},
|
"level_color": {"DEBUG": "#c1d5ff", "INFO": "#c1ffff", "WARN": "#fff600", "ERRO": "#ffa000", "CRIT": "#ff8181"},
|
||||||
}
|
}
|
||||||
|
|
||||||
from typing import Union, Optional
|
g_config_cache = None
|
||||||
from logging import error, info
|
g_config_mtime = 0
|
||||||
import json
|
|
||||||
|
def create_backup(config_path: Path) -> Tuple[bool, str]:
|
||||||
|
"""
|
||||||
|
创建配置文件备份
|
||||||
|
:param config_path: 配置文件路径
|
||||||
|
:param backup_prefix: 备份文件前缀
|
||||||
|
:return: (是否成功, 备份路径或错误信息)
|
||||||
|
"""
|
||||||
|
try:
|
||||||
|
if not config_path.exists():
|
||||||
|
return False, f"配置文件不存在: {config_path}"
|
||||||
|
backup_path = config_path.parent / f"logger_config_backup.json"
|
||||||
|
|
||||||
|
if backup_path.exists():
|
||||||
|
remove(backup_path)
|
||||||
|
shutil.copy2(config_path, backup_path)
|
||||||
|
return True, str(backup_path)
|
||||||
|
except PermissionError:
|
||||||
|
return False, f"权限不足,无法创建备份: {config_path}"
|
||||||
|
except Exception as e:
|
||||||
|
return False, f"备份失败: {str(e)}"
|
||||||
|
|
||||||
|
def handle_config_exceptions(func):
|
||||||
|
"""
|
||||||
|
配置操作异常处理装饰器
|
||||||
|
"""
|
||||||
|
def wrapper(*args, **kwargs):
|
||||||
|
try:
|
||||||
|
return func(*args, **kwargs)
|
||||||
|
except json.JSONDecodeError as e:
|
||||||
|
error(f"配置文件格式错误: {e}")
|
||||||
|
return False, f"配置文件格式错误: {str(e)}"
|
||||||
|
except PermissionError:
|
||||||
|
error(f"没有权限操作配置文件: {DEFAULT_CONFIG_PATH}")
|
||||||
|
return False, "没有权限操作配置文件"
|
||||||
|
except Exception as e:
|
||||||
|
error(f"配置操作失败: {e}")
|
||||||
|
return False, f"配置操作失败: {str(e)}"
|
||||||
|
return wrapper
|
||||||
|
|
||||||
def get_config(select: str = None) -> Union[dict, str, bool, int, None]:
|
def get_config(select: str = None) -> Union[dict, str, bool, int, None]:
|
||||||
"""
|
"""
|
||||||
@ -35,27 +81,24 @@ def get_config(select: str = None) -> Union[dict, str, bool, int, None]:
|
|||||||
:param select: 配置项名称 Config item name
|
:param select: 配置项名称 Config item name
|
||||||
:return: 配置项值 Config item value
|
:return: 配置项值 Config item value
|
||||||
"""
|
"""
|
||||||
try:
|
global g_config_cache, g_config_mtime
|
||||||
with open(DEFAULT_CONFIG_PATH, "r", encoding="utf-8") as f:
|
config_path = Path(DEFAULT_CONFIG_PATH)
|
||||||
config = json.load(f)
|
|
||||||
if select:
|
|
||||||
return config[select]
|
|
||||||
return config
|
|
||||||
except (FileNotFoundError, json.JSONDecodeError):
|
|
||||||
with open(DEFAULT_CONFIG_PATH, "w", encoding="utf-8") as f:
|
|
||||||
json.dump(DEFAULT_CONFIG, f, indent=4)
|
|
||||||
if select:
|
|
||||||
return DEFAULT_CONFIG[select]
|
|
||||||
return DEFAULT_CONFIG
|
|
||||||
except KeyError as e:
|
|
||||||
error(f"配置项 '{select}' 不存在")
|
|
||||||
return None
|
|
||||||
except Exception as e:
|
|
||||||
error(f"读取配置文件失败: {e}")
|
|
||||||
return None
|
|
||||||
|
|
||||||
import shutil
|
if config_path.exists():
|
||||||
from pathlib import Path
|
current_mtime = config_path.stat().st_mtime
|
||||||
|
if current_mtime != g_config_mtime or g_config_cache is None:
|
||||||
|
with open(config_path, "r", encoding="utf-8") as f:
|
||||||
|
g_config_cache = json.load(f)
|
||||||
|
g_config_mtime = current_mtime
|
||||||
|
else:
|
||||||
|
with open(config_path, "w", encoding="utf-8") as f:
|
||||||
|
json.dump(DEFAULT_CONFIG, f, indent=4)
|
||||||
|
g_config_cache = DEFAULT_CONFIG
|
||||||
|
g_config_mtime = config_path.stat().st_mtime
|
||||||
|
|
||||||
|
if select:
|
||||||
|
return g_config_cache.get(select)
|
||||||
|
return g_config_cache
|
||||||
|
|
||||||
def set_config(select: str, value: Union[dict, str, bool, int, None]) -> tuple[bool, Optional[str]]:
|
def set_config(select: str, value: Union[dict, str, bool, int, None]) -> tuple[bool, Optional[str]]:
|
||||||
"""
|
"""
|
||||||
@ -64,75 +107,49 @@ def set_config(select: str, value: Union[dict, str, bool, int, None]) -> tuple[b
|
|||||||
:param value: 配置项值 Config item value
|
:param value: 配置项值 Config item value
|
||||||
:return: (设置是否成功, 消息) (Set success or not, message)
|
:return: (设置是否成功, 消息) (Set success or not, message)
|
||||||
"""
|
"""
|
||||||
try:
|
config_path = Path(DEFAULT_CONFIG_PATH)
|
||||||
config_path = Path(DEFAULT_CONFIG_PATH)
|
|
||||||
backup_path = Path(__file__).parent / "configs.backup.json"
|
|
||||||
|
|
||||||
if not config_path.exists():
|
|
||||||
with open(config_path, "w", encoding="utf-8") as f:
|
|
||||||
json.dump(DEFAULT_CONFIG, f, indent=4)
|
|
||||||
|
|
||||||
shutil.copy2(config_path, backup_path)
|
|
||||||
|
|
||||||
with open(DEFAULT_CONFIG_PATH, "r+", encoding="utf-8") as f:
|
|
||||||
config = json.load(f)
|
|
||||||
config[select] = value
|
|
||||||
f.seek(0)
|
|
||||||
json.dump(config, f, indent=4)
|
|
||||||
f.truncate()
|
|
||||||
|
|
||||||
with open(DEFAULT_CONFIG_PATH, "r", encoding="utf-8") as f:
|
|
||||||
verify_config = json.load(f)
|
|
||||||
if verify_config.get(select) != value:
|
|
||||||
error(f"配置项 '{select}' 设置失败,值为 {value}")
|
|
||||||
shutil.move(backup_path, DEFAULT_CONFIG_PATH)
|
|
||||||
return False, f"配置项 '{select}' 设置失败,已恢复备份"
|
|
||||||
|
|
||||||
Path(backup_path).unlink(missing_ok=True)
|
|
||||||
return True, f"配置项 '{select}' 设置成功"
|
|
||||||
|
|
||||||
except json.JSONDecodeError as e:
|
if not config_path.exists():
|
||||||
error(f"配置文件格式错误: {e}")
|
with open(config_path, "w", encoding="utf-8") as f:
|
||||||
return False, f"配置文件格式错误: {str(e)}"
|
json.dump(DEFAULT_CONFIG, f, indent=4)
|
||||||
except PermissionError:
|
|
||||||
error(f"没有权限操作配置文件: {DEFAULT_CONFIG_PATH}")
|
|
||||||
return False, "没有权限操作配置文件"
|
|
||||||
except Exception as e:
|
|
||||||
error(f"设置配置文件失败: {e}")
|
|
||||||
|
|
||||||
if Path(backup_path).exists():
|
backup_success, backup_info = create_backup(config_path)
|
||||||
shutil.move(backup_path, DEFAULT_CONFIG_PATH)
|
if not backup_success:
|
||||||
return False, f"设置失败,已恢复备份: {str(e)}"
|
return False, f"备份失败: {backup_info}"
|
||||||
return False, f"设置配置文件失败: {str(e)}"
|
|
||||||
|
|
||||||
@staticmethod
|
with open(config_path, "r+", encoding="utf-8") as f:
|
||||||
|
config = json.load(f)
|
||||||
|
config[select] = value
|
||||||
|
f.seek(0)
|
||||||
|
json.dump(config, f, indent=4)
|
||||||
|
f.truncate()
|
||||||
|
|
||||||
|
with open(config_path, "r", encoding="utf-8") as f:
|
||||||
|
verify_config = json.load(f)
|
||||||
|
if verify_config.get(select) != value:
|
||||||
|
shutil.move(backup_info, config_path)
|
||||||
|
return False, f"配置项 '{select}' 设置失败,已恢复备份"
|
||||||
|
|
||||||
|
global g_config_cache
|
||||||
|
g_config_cache = None
|
||||||
|
return True, f"配置项 '{select}' 设置成功"
|
||||||
|
|
||||||
|
@handle_config_exceptions
|
||||||
def reset_config() -> tuple[bool, Optional[str]]:
|
def reset_config() -> tuple[bool, Optional[str]]:
|
||||||
"""
|
"""
|
||||||
重置配置信息 Reset config info
|
重置配置信息 Reset config info
|
||||||
:return: (重置是否成功, 消息) (Reset success or not, message)
|
:return: (重置是否成功, 消息) (Reset success or not, message)
|
||||||
"""
|
"""
|
||||||
try:
|
config_path = Path(DEFAULT_CONFIG_PATH)
|
||||||
config_path = Path(DEFAULT_CONFIG_PATH)
|
|
||||||
|
|
||||||
if not config_path.exists():
|
|
||||||
with open(config_path, "w", encoding="utf-8") as f:
|
|
||||||
json.dump(DEFAULT_CONFIG, f, indent=4)
|
|
||||||
return True, "配置文件不存在,已创建默认配置"
|
|
||||||
|
|
||||||
from .time import get_asctime
|
if config_path.exists():
|
||||||
timestamp = get_asctime()
|
backup_success, backup_info = create_backup(config_path)
|
||||||
backup_path = f"{DEFAULT_CONFIG_PATH}_{timestamp}.backup.json"
|
if not backup_success:
|
||||||
backup_path = backup_path.replace(":", "-")
|
return False, f"备份失败: {backup_info}"
|
||||||
shutil.copy2(DEFAULT_CONFIG_PATH, backup_path)
|
|
||||||
with open(DEFAULT_CONFIG_PATH, "w", encoding="utf-8") as f:
|
with open(config_path, "w", encoding="utf-8") as f:
|
||||||
json.dump(DEFAULT_CONFIG, f, indent=4)
|
json.dump(DEFAULT_CONFIG, f, indent=4)
|
||||||
return True
|
|
||||||
except PermissionError:
|
global g_config_cache
|
||||||
error(f"权限不足,无法操作配置文件: {DEFAULT_CONFIG_PATH}")
|
g_config_cache = None
|
||||||
return False, "权限不足,无法重置配置"
|
return True, "配置已重置为默认值"
|
||||||
except json.JSONDecodeError:
|
|
||||||
error("配置文件格式错误,无法解析")
|
|
||||||
return False, "配置文件格式错误,无法重置"
|
|
||||||
except Exception as e:
|
|
||||||
error(f"重置配置文件失败: {e}")
|
|
||||||
return False, f"重置配置失败: {str(e)}"
|
|
||||||
|
Reference in New Issue
Block a user