Genshin-bots/gsuid_core
1
0
Fork 0
mirror of https://github.com/Genshin-bots/gsuid_core.git synced 2025-05-12 06:55:49 +08:00
Code Issues Packages Projects Releases Wiki Activity

✨ 令prefix生效 (#76)

Browse Source
This commit is contained in:
KimigaiiWuyi 2024-10-04 03:56:19 +08:00
parent 8569062749
commit 980fe9d677
4 changed files with 137 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
gsuid_core/config.py
View File

@ -36,7 +36,8 @@ plugins_sample = {
'area': 'SV',
'black_list': [],
'white_list': [],
'prefix': '',
'prefix': [],
'allow_empty_prefix': True,
'sv': {},
}

65
gsuid_core/server.py
View File

@ -24,7 +24,7 @@ auto_update_dep: bool = core_plugins_config.get_config('AutoUpdateDep').data
core_start_def = set()
core_shutdown_def = set()
installed_dependencies = []
installed_dependencies: Dict[str, str] = {}
ignore_dep = ['python', 'fastapi', 'pydantic']
@ -214,8 +214,16 @@ def check_pyproject(pyproject: Path):
def install_dependencies(dependencies: Dict, need_update: bool = False):
global installed_dependencies
start_tool = check_start_tool(True)
to_update = find_dependencies_to_update(
installed_dependencies, dependencies
)
if not to_update:
logger.debug('[安装/更新依赖] 无需更新依赖！')
return
logger.debug(f'[安装/更新依赖] 需更新依赖列表如下：\n{to_update}')
start_tool = check_start_tool(True)
logger.debug(f'[安装/更新依赖] 当前启动工具：{start_tool}')
if start_tool.startswith('pdm') and False:
@ -247,8 +255,8 @@ def install_dependencies(dependencies: Dict, need_update: bool = False):
# 解析依赖项
for (
dependency,
version,
) in dependencies.items():
_version,
) in to_update.items():
if need_update:
condi = dependency not in ignore_dep
else:
@ -262,6 +270,7 @@ def install_dependencies(dependencies: Dict, need_update: bool = False):
)
if condi:
version: str = _version.get('required_version', '')
logger.info(f'[安装/更新依赖] {dependency} 中...')
CMD = f'{start_tool} install "{dependency}{version}" {extra}'
@ -297,7 +306,10 @@ def execute_cmd(CMD: str):
def get_installed_dependencies():
global installed_dependencies
installed_packages = pkg_resources.working_set
installed_dependencies = [package.key for package in installed_packages]
installed_dependencies = {
package.key: package.version for package in installed_packages
}
return installed_dependencies
def parse_dependency(dependency: List):
@ -319,3 +331,46 @@ def parse_dependency_string(dependency_string: str):
dependencies[dependency] = f"{operator}{version}"
return dependencies
def extract_numeric_version(version):
# 提取版本中的数字和小数点部分
numeric_version = re.findall(r'\d+', version)
return tuple(map(int, numeric_version)) if numeric_version else (0,)
def compare_versions(installed_version, required_version):
installed_tuple = extract_numeric_version(installed_version)
required_tuple = extract_numeric_version(
re.sub(r'[<>=]', '', required_version)
)
# 基于符号进行比较
if "<=" in required_version:
return installed_tuple <= required_tuple
elif ">=" in required_version:
return installed_tuple >= required_tuple
elif "==" in required_version:
return installed_tuple == required_tuple
elif "<" in required_version:
return installed_tuple < required_tuple
elif ">" in required_version:
return installed_tuple > required_tuple
return False
def find_dependencies_to_update(
installed_deps, required_deps
) -> Dict[str, Dict[str, str]]:
to_update = {}
for dep, installed_version in installed_deps.items():
if dep in required_deps:
required_version = required_deps[dep]
if not compare_versions(installed_version, required_version):
to_update[dep] = {
"installed_version": installed_version,
"required_version": required_version,
}
return to_update

73
gsuid_core/sv.py
View File

@ -62,7 +62,7 @@ class Plugins:
if name is None:
raise ValueError('Plugins.name is None!')
if name in SL.plugins:
if name in SL.plugins and not kwargs.get('force'):
return SL.plugins[name]
else:
_plugin = super().__new__(cls)
@ -82,9 +82,21 @@ class Plugins:
black_list: List = [],
white_list: List = [],
sv: Dict = {},
prefix: str = '',
prefix: Union[List[str], str] = [],
allow_empty_prefix: Optional[bool] = None,
force: bool = False,
):
if not self.is_initialized:
if isinstance(prefix, str):
prefix = [prefix]
if allow_empty_prefix is None:
if '' in prefix:
prefix.remove('')
if prefix:
allow_empty_prefix = False
else:
allow_empty_prefix = True
self.name = name
self.priority = priority
self.enabled = enabled
@ -94,6 +106,7 @@ class Plugins:
self.white_list = white_list
self.sv = {}
self.prefix = prefix
self.allow_empty_prefix = allow_empty_prefix
self.is_initialized = True
def set(self, **kwargs):
@ -169,8 +182,34 @@ class SV:
core_config.set_config('plugins', config_plugins)
else:
if 'prefix' not in config_plugins[plugins_name]:
config_plugins[plugins_name]['prefix'] = ''
plugins = Plugins(**config_plugins[plugins_name])
if plugins_name in SL.plugins:
config_plugins[plugins_name]['prefix'] = SL.plugins[
plugins_name
].prefix
else:
config_plugins[plugins_name]['prefix'] = []
elif isinstance(config_plugins[plugins_name]['prefix'], str):
if config_plugins[plugins_name]['prefix'] != '':
config_plugins[plugins_name]['prefix'] = [
config_plugins[plugins_name]['prefix']
]
else:
config_plugins[plugins_name]['prefix'] = []
if 'allow_empty_prefix' not in config_plugins[plugins_name]:
if plugins_name in SL.plugins:
config_plugins[plugins_name]['allow_empty_prefix'] = (
SL.plugins[plugins_name].allow_empty_prefix
)
else:
config_plugins[plugins_name][
'allow_empty_prefix'
] = None
plugins = Plugins(
**config_plugins[plugins_name],
force=True,
)
core_config.set_config('plugins', config_plugins)
@ -270,12 +309,22 @@ class SV:
else:
keyword_list = keyword
for _k in keyword_list:
if prefix:
tr = f'{self.plugins.prefix}{_k}'
else:
tr = _k
entry = []
_pp = deepcopy(self.plugins.prefix)
if "" in _pp:
_pp.remove("")
if self.plugins.allow_empty_prefix:
_pp.append("")
for _k in keyword_list:
if prefix and _pp:
for _p in _pp:
entry.append(f'{_p}{_k}')
else:
entry.append(_k)
for tr in entry:
if tr not in self.TL:
logger.trace(f'载入{type}触发器【{tr}】!')
if type not in self.TL:
@ -369,8 +418,12 @@ class SV:
return self._on('message', unique_id, block, to_me, prefix)
def get_plugin_prefix(plugin_name: str) -> str:
def get_plugin_prefixs(plugin_name: str) -> List[str]:
plugin = SL.plugins.get(plugin_name)
if plugin is None:
raise ValueError(f'插件{plugin_name}不存在!')
return plugin.prefix
def get_plugin_prefix(plugin_name: str) -> str:
return get_plugin_prefixs(plugin_name)[0]

13
gsuid_core/utils/download_resource/download_core.py
View File

@ -14,6 +14,7 @@ from gsuid_core.logger import logger
from .download_file import download
global_tag, global_url = '', ''
NOW_SPEED_TEST = False
async def check_url(tag: str, url: str):
@ -62,8 +63,10 @@ async def find_fastest_url(urls: Dict[str, str]):
async def check_speed():
global global_tag
global global_url
global NOW_SPEED_TEST
if not global_tag or not global_url:
if (not global_tag or not global_url) and not NOW_SPEED_TEST:
NOW_SPEED_TEST = True
logger.info('[GsCore资源下载]测速中...')
URL_LIB = {
@ -80,7 +83,15 @@ async def check_speed():
global_tag, global_url = TAG, BASE_URL
logger.info(f"最快资源站: {TAG} {BASE_URL}")
NOW_SPEED_TEST = False
return TAG, BASE_URL
if NOW_SPEED_TEST:
while True:
if not NOW_SPEED_TEST:
return global_tag, global_url
await asyncio.sleep(1)
return global_tag, global_url