修复app安装卸载后路由没有即时生效或失效的问题
This commit is contained in:
parent
33dee5c329
commit
cafa864e4c
@ -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/<app>）
|
||||
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()
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user