From cafa864e4c4a0244436dfc63e28e876bf8402cbc Mon Sep 17 00:00:00 2001 From: jingrow Date: Wed, 29 Oct 2025 22:22:10 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E5=A4=8Dapp=E5=AE=89=E8=A3=85?= =?UTF-8?q?=E5=8D=B8=E8=BD=BD=E5=90=8E=E8=B7=AF=E7=94=B1=E6=B2=A1=E6=9C=89?= =?UTF-8?q?=E5=8D=B3=E6=97=B6=E7=94=9F=E6=95=88=E6=88=96=E5=A4=B1=E6=95=88?= =?UTF-8?q?=E7=9A=84=E9=97=AE=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- apps/jingrow/jingrow/utils/app_manager.py | 178 ++++++++++++---------- 1 file changed, 94 insertions(+), 84 deletions(-) diff --git a/apps/jingrow/jingrow/utils/app_manager.py b/apps/jingrow/jingrow/utils/app_manager.py index 31dca5a..1992bb1 100644 --- a/apps/jingrow/jingrow/utils/app_manager.py +++ b/apps/jingrow/jingrow/utils/app_manager.py @@ -13,13 +13,12 @@ from typing import Set logger = logging.getLogger(__name__) -# 确保各 app 根目录在 sys.path 中（仅初始化一次） -_apps_path_initialized = False +# 已加载的 app 路径集合（避免重复添加） +_loaded_app_paths: Set[str] = set() def get_project_root() -> Path: """获取项目根目录路径""" - # 从当前文件位置计算：apps/jingrow/jingrow/utils/app_manager.py -> 向上4级 return Path(__file__).resolve().parents[4] @@ -34,115 +33,126 @@ def get_apps_txt_path() -> Path: def ensure_apps_on_sys_path(): - """ - 确保各 app 根目录在 sys.path 中，支持跨 app 导入 - 在路由处理时自动调用，确保新安装的 app 路由可以生效 - """ - global _apps_path_initialized - if _apps_path_initialized: + """确保 apps.txt 中所有 app 的路径都在 sys.path 中""" + apps_txt = get_apps_txt_path() + if not apps_txt.exists(): return - try: - apps_dir = get_apps_dir() - apps_txt = get_apps_txt_path() - - # 读取 apps.txt，添加各 app 的根目录（apps/） - if apps_txt.exists(): - for app_name in apps_txt.read_text(encoding='utf-8').splitlines(): - app_name = app_name.strip() - if app_name: - app_root_dir = apps_dir / app_name - if app_root_dir.exists() and str(app_root_dir) not in sys.path: - sys.path.insert(0, str(app_root_dir)) - logger.debug(f"已添加 app 路径到 sys.path: {app_root_dir}") - except Exception as e: - logger.warning(f"初始化 app 路径失败: {e}") - finally: - _apps_path_initialized = True - - -def reset_apps_path_cache(): - """ - 重置 apps path 缓存，强制下次请求时重新初始化 - 在安装/卸载 app 后调用，确保路由立即生效 - """ - global _apps_path_initialized - _apps_path_initialized = False - logger.debug("已重置 apps path 缓存") - - -def get_apps_from_txt() -> Set[str]: - """ - 从 apps.txt 读取已注册的 app 列表 - - Returns: - Set[str]: app 名称集合 - """ - apps_txt = get_apps_txt_path() - apps = set() - - if apps_txt.exists(): - for line in apps_txt.read_text(encoding='utf-8').splitlines(): - line = line.strip() - if line: - apps.add(line) - - return apps + apps_dir = get_apps_dir() + for app_name in apps_txt.read_text(encoding='utf-8').splitlines(): + app_name = app_name.strip() + if app_name: + app_path = str(apps_dir / app_name) + if app_path not in _loaded_app_paths and (apps_dir / app_name).exists(): + if app_path not in sys.path: + sys.path.insert(0, app_path) + logger.debug(f"已添加 app 路径: {app_path}") + _loaded_app_paths.add(app_path) def update_apps_txt(app_name: str, add: bool = True) -> bool: """ - 更新 apps.txt 文件（保持原有顺序；新增仅在末尾追加） + 更新 apps.txt 文件（末尾追加，保持顺序） Args: app_name: 应用名称 - add: True 为添加，False 为删除 + add: True 添加，False 删除 Returns: - bool: 是否成功更新 + bool: 是否成功 """ try: apps_txt = get_apps_txt_path() apps_txt.parent.mkdir(parents=True, exist_ok=True) - - # 读取现有的 app 列表（保持顺序） + + # 读取现有列表（保持顺序） + lines = [] if apps_txt.exists(): - lines = [ln.strip() for ln in apps_txt.read_text(encoding='utf-8').splitlines()] - lines = [ln for ln in lines if ln] - else: - lines = [] - + lines = [ln.strip() for ln in apps_txt.read_text(encoding='utf-8').splitlines() if ln.strip()] + + # 添加或删除 if add: if app_name not in lines: lines.append(app_name) logger.info(f"添加应用到 apps.txt: {app_name}") + + # 立即添加路径并清除缓存 + _add_app_to_sys_path(app_name) else: - new_lines = [ln for ln in lines if ln != app_name] - if len(new_lines) != len(lines): + if app_name in lines: + lines.remove(app_name) logger.info(f"从 apps.txt 移除应用: {app_name}") - lines = new_lines - - # 写回文件（保持顺序，统一以单个换行分隔，并以换行结尾） + + # 立即移除路径并清除缓存 + _remove_app_from_sys_path(app_name) + + # 写回文件 apps_txt.write_text('\n'.join(lines) + ('\n' if lines else ''), encoding='utf-8') - - # 重置缓存，确保下次请求时重新加载 - reset_apps_path_cache() - return True except Exception as e: logger.error(f"更新 apps.txt 失败: {e}", exc_info=True) return False -def is_app_registered(app_name: str) -> bool: - """ - 检查 app 是否已注册到 apps.txt +def _add_app_to_sys_path(app_name: str): + """添加 app 路径到 sys.path 并清除相关缓存""" + apps_dir = get_apps_dir() + app_path = str(apps_dir / app_name) - Args: - app_name: 应用名称 - - Returns: - bool: 是否已注册 - """ - return app_name in get_apps_from_txt() + if (apps_dir / app_name).exists(): + if app_path not in sys.path: + sys.path.insert(0, app_path) + logger.info(f"已添加 app 路径到 sys.path: {app_path}") + _loaded_app_paths.add(app_path) + + # 清除该 app 的模块缓存，确保能重新导入 + _clear_app_modules(app_name) + +def _remove_app_from_sys_path(app_name: str): + """从 sys.path 移除 app 路径并清除相关缓存""" + apps_dir = get_apps_dir() + app_path = str(apps_dir / app_name) + + # 从 sys.path 移除 + if app_path in sys.path: + sys.path.remove(app_path) + logger.info(f"已从 sys.path 移除: {app_path}") + + # 从已加载集合移除 + _loaded_app_paths.discard(app_path) + + # 清除模块缓存，确保卸载后无法访问 + _clear_app_modules(app_name) + + +def _clear_app_modules(app_name: str): + """清除 app 相关的所有模块缓存""" + modules_to_remove = [ + name for name in list(sys.modules.keys()) + if name == app_name or name.startswith(f"{app_name}.") + ] + + for module_name in modules_to_remove: + try: + del sys.modules[module_name] + logger.debug(f"已清除模块缓存: {module_name}") + except KeyError: + pass + + if modules_to_remove: + logger.info(f"已清除 {len(modules_to_remove)} 个 {app_name} 相关模块缓存") + + +def get_apps_from_txt() -> Set[str]: + """从 apps.txt 读取已注册的 app 列表""" + apps_txt = get_apps_txt_path() + if not apps_txt.exists(): + return set() + + return {ln.strip() for ln in apps_txt.read_text(encoding='utf-8').splitlines() if ln.strip()} + + +def is_app_registered(app_name: str) -> bool: + """检查 app 是否已注册到 apps.txt""" + return app_name in get_apps_from_txt()