68 lines
2.2 KiB
Python
68 lines
2.2 KiB
Python
import json
|
||
import os
|
||
import copy
|
||
from threading import Lock
|
||
|
||
CONFIG_PATH = 'config.json'
|
||
|
||
class ConfigManager:
|
||
"""线程安全地处理 config.json 的读取、写入和访问"""
|
||
def __init__(self, path=CONFIG_PATH):
|
||
self.path = path
|
||
self._lock = Lock()
|
||
self.config = self._load_config()
|
||
|
||
def _load_config(self):
|
||
"""加载配置文件，如果不存在则创建一个空的"""
|
||
with self._lock:
|
||
if os.path.exists(self.path):
|
||
try:
|
||
with open(self.path, 'r', encoding='utf-8') as f:
|
||
return json.load(f)
|
||
except (json.JSONDecodeError, IOError):
|
||
return {}
|
||
return {}
|
||
|
||
def get_config(self, sensitive=False):
|
||
"""
|
||
获取配置。
|
||
:param sensitive: False时，将敏感信息替换为 '******'
|
||
"""
|
||
config_copy = copy.deepcopy(self.config)
|
||
if sensitive:
|
||
return config_copy
|
||
|
||
# 隐藏敏感信息
|
||
sensitive_keys = {
|
||
'douban': 'cookie', 'ptskit': 'cookie',
|
||
'emby': 'api_key', 'plex': 'token', 'fnos': 'authorization'
|
||
}
|
||
for section, key in sensitive_keys.items():
|
||
if section in config_copy and key in config_copy[section]:
|
||
config_copy[section][key] = '******' if config_copy[section][key] else ''
|
||
|
||
# 隐藏密码
|
||
for section in ['qbittorrent', 'transmission']:
|
||
if section in config_copy and 'password' in config_copy[section]:
|
||
config_copy[section]['password'] = ''
|
||
|
||
return config_copy
|
||
|
||
def save_config(self, new_config):
|
||
"""保存配置到文件"""
|
||
with self._lock:
|
||
try:
|
||
with open(self.path, 'w', encoding='utf-8') as f:
|
||
json.dump(new_config, f, indent=4, ensure_ascii=False)
|
||
self.config = new_config # 更新内存中的配置
|
||
return True, None
|
||
except Exception as e:
|
||
return False, str(e)
|
||
|
||
def get(self, key, default=None):
|
||
"""获取特定配置项"""
|
||
return self.config.get(key, default)
|
||
|
||
# 创建一个全局单例
|
||
config_manager = ConfigManager()
|