1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
|
import argparse
import sys
import platform
from importlib.metadata import version
from iamai import Plugin
import os
import pickle
import threading
# 创建全局 ArgumentParser 对象
global_parser = argparse.ArgumentParser(description='hydroroll[水系] 全局命令参数')
# 定义全局配置类
class GlobalConfig:
_name = "hydroroll"
_version = "0.1.0"
_svn = "2"
_author = "简律纯"
_iamai_version = version('iamai')
_python_ver = sys.version
_python_ver_raw= '.'.join(map(str, platform.python_version_tuple()[:3]))
current_path = os.path.dirname(os.path.abspath('__file__'))
# 定义系统组件
class HydroSystem:
def __init__(self):
self.parser = argparse.ArgumentParser(description='hydroroll[水系].system 系统命令参数')
self.subparsers = self.parser.add_subparsers()
self.status_parser = self.subparsers.add_parser('status', aliases=['s'], help='系统状态')
self.reload_parser = self.subparsers.add_parser('reload', aliases=['rld'], help='重新加载系统')
self.restart_parser = self.subparsers.add_parser('restart', aliases=['rst'], help='重启系统')
self.help = '\n'.join(self.parser.format_help().replace('\r\n', '\n').replace('\r', '').split('\n')[2:-3])
class HydroBot:
def __init__(self) -> None:
self.parser = argparse.ArgumentParser(description="Bot命令")
class ConfigManager:
def __init__(self, directory):
self.directory = directory
self.configs = {}
self.locks = {}
def get(self, filename, section, key, default=None):
if not self._check_file_exists(filename):
return default
config = self._get_config(filename)
if section not in config:
return default
return config[section].get(key, default)
def set(self, filename, section, key, value):
config = self._get_config(filename)
if section not in config:
config[section] = {}
config[section][key] = value
self._save_config(filename, config)
def delete(self, filename, section, key):
config = self._get_config(filename)
if section not in config:
return
if key not in config[section]:
return
del config[section][key]
self._save_config(filename, config)
def sections(self, filename):
if not self._check_file_exists(filename):
return []
config = self._get_config(filename)
return list(config.keys())
def section_items(self, filename, section):
if not self._check_file_exists(filename):
return []
config = self._get_config(filename)
if section not in config:
return []
return list(config[section].items())
def files(self):
return [filename for filename in os.listdir(self.directory) if filename.endswith('.dat')]
def _get_lock(self, filename):
if filename not in self.locks:
self.locks[filename] = threading.Lock()
return self.locks[filename]
def _get_config(self, filename):
with self._get_lock(filename):
if filename not in self.configs:
filepath = os.path.join(self.directory, filename)
if os.path.exists(filepath):
try:
with open(filepath, 'rb') as f:
self.configs[filename] = pickle.load(f)
except:
pass
if filename not in self.configs:
self.configs[filename] = {}
return self.configs[filename]
def _save_config(self, filename, config):
with self._get_lock(filename):
try:
filepath = os.path.join(self.directory, filename)
backuppath = os.path.join(self.directory, filename + '.bak')
lockpath = os.path.join(self.directory, filename + '.lock')
with open(lockpath, 'wb') as f:
pass
if os.path.exists(filepath):
os.replace(filepath, backuppath)
with open(filepath, 'wb') as f:
pickle.dump(config, f)
os.remove(backuppath)
os.remove(lockpath)
except:
pass
def _check_file_exists(self, filename):
filepath = os.path.join(self.directory, filename)
return os.path.exists(filepath) and filename.endswith('.dat')
|