95 lines
3.5 KiB
Python
95 lines
3.5 KiB
Python
# -------------------------------
|
|
# Difficult Rocket
|
|
# Copyright © 2020-2023 by shenjackyuanjie 3695888@qq.com
|
|
# All rights reserved
|
|
# -------------------------------
|
|
|
|
import sys
|
|
import time
|
|
import logging
|
|
import traceback
|
|
import importlib
|
|
from pathlib import Path
|
|
from typing import List, Dict, Optional
|
|
|
|
from Difficult_Rocket.mod.api import ModInfo
|
|
from Difficult_Rocket.utils.translate import tr
|
|
from Difficult_Rocket.api.types import Options, Version
|
|
|
|
logger = logging.getLogger('mod_manager')
|
|
ONE_FILE_SUFFIX = ('.py', '.pyc', '.pyd')
|
|
PACKAGE_SUFFIX = ('.pyz', '.zip', '.dr_mod')
|
|
|
|
|
|
class ModManager(Options):
|
|
name = 'Mod Manager'
|
|
|
|
mods_path: List[Path] = [Path('./mods')]
|
|
loaded_mod_modules: Dict[str, ModInfo] = {}
|
|
|
|
def dispatch_event(self, event_name: str, *args, **kwargs):
|
|
"""
|
|
分发事件
|
|
:param event_name:
|
|
:param args:
|
|
:param kwargs:
|
|
:return:
|
|
"""
|
|
for mod in self.loaded_mod_modules.values():
|
|
if hasattr(mod, event_name):
|
|
try:
|
|
getattr(mod, event_name)(*args, **kwargs)
|
|
except Exception as e:
|
|
logger.error(tr().mod.event.error().format(mod, event_name, e, traceback.format_exc()))
|
|
|
|
def load_mods(self, extra_path: Optional[List[Path]] = None) -> List[type(ModInfo)]:
|
|
"""
|
|
加载所有 mod (可提供额外的 mod 路径)
|
|
:param extra_path: 额外的 mod 路径
|
|
:return:
|
|
"""
|
|
find_path = self.mods_path + (extra_path if extra_path is not None else [])
|
|
sys.path += [str(path_) for path_ in find_path]
|
|
mods = []
|
|
start_time = time.time()
|
|
logger.info(tr().mod.load.start().format(find_path))
|
|
for path in find_path:
|
|
if not path.exists():
|
|
path.mkdir(parents=True)
|
|
continue
|
|
for mod in path.iterdir():
|
|
try:
|
|
if mod.name == '__pycache__':
|
|
# 忽略 __pycache__ 文件夹 (Python 编译文件)
|
|
continue
|
|
logger.info(tr().mod.load.loading().format(mod))
|
|
if mod.is_dir() or mod.suffix in PACKAGE_SUFFIX or mod.suffix in ONE_FILE_SUFFIX:
|
|
# 文件夹 mod
|
|
loading_mod = importlib.import_module(mod.name)
|
|
if not hasattr(loading_mod, 'mod_class'):
|
|
logger.warning(tr().mod.load.faild.no_mod_class().format(mod))
|
|
continue
|
|
mod_class: type(ModInfo) = loading_mod.mod_class # 获取 mod 类
|
|
mods.append(mod_class)
|
|
except ImportError:
|
|
logger.warning(tr().mod.load.faild.error().format(mod))
|
|
logger.info(tr().mod.load.use_time().format(time.time() - start_time))
|
|
return mods
|
|
|
|
def init_mods(self, mods: List[type(ModInfo)]):
|
|
"""
|
|
加载 mod
|
|
:param mods:
|
|
:return:
|
|
"""
|
|
start_time = time.time()
|
|
for mod_class in mods:
|
|
try:
|
|
init_mod = mod_class()
|
|
self.loaded_mod_modules[init_mod.name] = init_mod
|
|
logger.info(tr().mod.init.success().format(init_mod, init_mod.version))
|
|
except Exception as e:
|
|
logger.error(tr().mod.init.faild().format(mod_class, e, traceback.format_exc()))
|
|
continue
|
|
logger.info(tr().mod.init.use_time().format(time.time() - start_time))
|