From a3142ad0657aedb199560f271e1534ce030b9300 Mon Sep 17 00:00:00 2001 From: HibiKier <775757368@qq.com> Date: Sun, 3 Aug 2025 23:57:13 +0800 Subject: [PATCH] =?UTF-8?q?=E2=9C=A8=20feat(submodule):=20=E6=B7=BB?= =?UTF-8?q?=E5=8A=A0=E5=AD=90=E6=A8=A1=E5=9D=97=E7=AE=A1=E7=90=86=E5=8A=9F?= =?UTF-8?q?=E8=83=BD=EF=BC=8C=E6=94=AF=E6=8C=81=E5=AD=90=E6=A8=A1=E5=9D=97?= =?UTF-8?q?=E7=9A=84=E5=88=9D=E5=A7=8B=E5=8C=96=E3=80=81=E6=9B=B4=E6=96=B0?= =?UTF-8?q?=E5=92=8C=E4=BF=A1=E6=81=AF=E8=8E=B7=E5=8F=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- zhenxun/utils/manager/resource_manager.py | 18 +- zhenxun/utils/manager/zhenxun_repo_manager.py | 687 ++++++++++++++++++ zhenxun/utils/repo_utils/__init__.py | 6 + zhenxun/utils/repo_utils/github_manager.py | 160 ++++ zhenxun/utils/repo_utils/models.py | 58 ++ zhenxun/utils/repo_utils/submodule_manager.py | 408 +++++++++++ zhenxun/utils/repo_utils/submodule_setup.py | 210 ++++++ 7 files changed, 1538 insertions(+), 9 deletions(-) create mode 100644 zhenxun/utils/manager/zhenxun_repo_manager.py create mode 100644 zhenxun/utils/repo_utils/submodule_manager.py create mode 100644 zhenxun/utils/repo_utils/submodule_setup.py diff --git a/zhenxun/utils/manager/resource_manager.py b/zhenxun/utils/manager/resource_manager.py index 38e90654..01b849f9 100644 --- a/zhenxun/utils/manager/resource_manager.py +++ b/zhenxun/utils/manager/resource_manager.py @@ -34,13 +34,13 @@ class ResourceManager: ): if (FONT_PATH.exists() and os.listdir(FONT_PATH)) and not force: return - if cls.TMP_PATH.exists(): - logger.debug( - "resources临时文件夹已存在，移除resources临时文件夹", LOG_COMMAND - ) - await clean_git(cls.TMP_PATH) - shutil.rmtree(cls.TMP_PATH, ignore_errors=True) if is_zip: + if cls.TMP_PATH.exists(): + logger.debug( + "resources临时文件夹已存在，移除resources临时文件夹", LOG_COMMAND + ) + await clean_git(cls.TMP_PATH) + shutil.rmtree(cls.TMP_PATH, ignore_errors=True) cls.TMP_PATH.mkdir(parents=True, exist_ok=True) try: await cls.__download_resources() @@ -49,9 +49,9 @@ class ResourceManager: logger.error("获取resources资源包失败", LOG_COMMAND, e=e) else: if git_source == "ali": - await AliyunRepoManager.update(cls.GITHUB_URL, cls.TMP_PATH) + await AliyunRepoManager.update(cls.GITHUB_URL, cls.RESOURCE_PATH) else: - await GithubRepoManager.update(cls.GITHUB_URL, cls.TMP_PATH) + await GithubRepoManager.update(cls.GITHUB_URL, cls.RESOURCE_PATH) cls.UNZIP_PATH = cls.TMP_PATH / "resources" cls.file_handle() if cls.TMP_PATH.exists(): @@ -63,7 +63,7 @@ class ResourceManager: def file_handle(cls): if not cls.UNZIP_PATH: return - cls.__recursive_folder(cls.UNZIP_PATH, "resources") + cls.__recursive_folder(cls.UNZIP_PATH, ".") @classmethod def __recursive_folder(cls, dir: Path, parent_path: str): diff --git a/zhenxun/utils/manager/zhenxun_repo_manager.py b/zhenxun/utils/manager/zhenxun_repo_manager.py new file mode 100644 index 00000000..2b63eb6e --- /dev/null +++ b/zhenxun/utils/manager/zhenxun_repo_manager.py @@ -0,0 +1,687 @@ +""" +真寻仓库管理器 +负责真寻主仓库的更新、版本检查、文件处理等功能 +""" + +import os +from pathlib import Path +import shutil +import tarfile +from typing import ClassVar +import zipfile + +from zhenxun.configs.path_config import DATA_PATH, FONT_PATH, TEMP_PATH +from zhenxun.services.log import logger +from zhenxun.utils.github_utils import GithubUtils +from zhenxun.utils.github_utils.models import RepoInfo +from zhenxun.utils.http_utils import AsyncHttpx +from zhenxun.utils.manager.virtual_env_package_manager import VirtualEnvPackageManager +from zhenxun.utils.platform import PlatformUtils +from zhenxun.utils.repo_utils import AliyunRepoManager, GithubRepoManager +from zhenxun.utils.repo_utils.models import ( + SubmoduleConfig, +) +from zhenxun.utils.repo_utils.submodule_manager import SubmoduleManager +from zhenxun.utils.repo_utils.utils import clean_git + +LOG_COMMAND = "ZhenxunRepoManager" + + +class DownloadException(Exception): + """资源下载异常""" + + pass + + +class ZhenxunRepoConfig: + """真寻仓库配置""" + + # GitHub 仓库 URL + ZHENXUN_BOT_GIT = "https://github.com/zhenxun-org/zhenxun_bot.git" + ZHENXUN_BOT_GITHUB_URL = "https://github.com/HibiKier/zhenxun_bot/tree/main" + DEFAULT_GITHUB_URL = "https://github.com/HibiKier/zhenxun_bot/tree/main" + RELEASE_URL = "https://api.github.com/repos/HibiKier/zhenxun_bot/releases/latest" + + # 资源仓库 URL + RESOURCE_GITHUB_URL = ( + "https://github.com/zhenxun-org/zhenxun-bot-resources/tree/main" + ) + + # Web UI 仓库 URL + WEBUI_GIT = "https://github.com/HibiKier/zhenxun_bot_webui.git" + + # 文件路径配置 + VERSION_FILE_STRING = "__version__" + VERSION_FILE = Path() / VERSION_FILE_STRING + + PYPROJECT_FILE_STRING = "pyproject.toml" + PYPROJECT_FILE = Path() / PYPROJECT_FILE_STRING + PYPROJECT_LOCK_FILE_STRING = "poetry.lock" + PYPROJECT_LOCK_FILE = Path() / PYPROJECT_LOCK_FILE_STRING + REQ_TXT_FILE_STRING = "requirements.txt" + REQ_TXT_FILE = Path() / REQ_TXT_FILE_STRING + + BASE_PATH_STRING = "zhenxun" + BASE_PATH = Path() / BASE_PATH_STRING + + # 资源路径配置 + RESOURCE_PATH = Path() / "resources" + + # Web UI 路径配置 + WEBUI_PATH = DATA_PATH / "web_ui" / "public" + WEBUI_BACKUP_PATH = DATA_PATH / "web_ui" / "backup_public" + WEBUI_GIT_PATH = DATA_PATH / "web_ui" / "git_web_ui" + + # 临时文件路径 + TMP_PATH = TEMP_PATH / "zhenxun_update" + BACKUP_PATH = Path() / "backup" + RESOURCE_TMP_PATH = TEMP_PATH / "_resource_tmp" + + # 下载文件配置 + DOWNLOAD_GZ_FILE_STRING = "download_latest_file.tar.gz" + DOWNLOAD_ZIP_FILE_STRING = "download_latest_file.zip" + DOWNLOAD_GZ_FILE = TMP_PATH / DOWNLOAD_GZ_FILE_STRING + DOWNLOAD_ZIP_FILE = TMP_PATH / DOWNLOAD_ZIP_FILE_STRING + + # 资源文件配置 + RESOURCE_ZIP_FILE = RESOURCE_TMP_PATH / "resources.zip" + UNZIP_PATH: Path | None = None + + # 需要替换的文件夹 + REPLACE_FOLDERS: ClassVar[list[str]] = [ + "builtin_plugins", + "services", + "utils", + "models", + "configs", + ] + + # 日志标识 + COMMAND = "真寻仓库管理" + + +class ZhenxunRepoManager: + """真寻仓库管理器""" + + def __init__(self): + self.config = ZhenxunRepoConfig() + # 初始化子模块管理器 + self.submodule_manager = SubmoduleManager(GithubRepoManager) + + def __clear_folder(self, folder_path: Path): + for filename in os.listdir(folder_path): + file_path = folder_path / filename + try: + if file_path.is_file(): + os.unlink(file_path) + elif file_path.is_dir() and not filename.startswith("."): + shutil.rmtree(file_path) + except Exception as e: + logger.warning(f"无法删除 {file_path}", LOG_COMMAND, e=e) + + async def check_version(self) -> str: + """检查真寻更新版本 + + 返回: + str: 更新信息 + """ + cur_version = self._get_current_version() + data = await self._get_latest_release_data() + if not data: + return "检查更新获取版本失败..." + return ( + "检测到当前版本更新\n" + f"当前版本：{cur_version}\n" + f"最新版本：{data.get('name')}\n" + f"创建日期：{data.get('created_at')}\n" + f"更新内容：\n{data.get('body')}" + ) + + async def update_repository( + self, + bot, + user_id: str, + version_type: str, + force: bool, + source: str, + zip_update: bool, + update_type: str, + ) -> str: + """更新真寻仓库 + + 参数: + bot: Bot实例 + user_id: 用户ID + version_type: 更新版本类型 (main/release) + force: 是否强制更新 + source: 更新源 (git/ali) + zip_update: 是否下载zip文件 + update_type: 更新方式 (git/download) + + 返回: + str: 更新结果消息 + """ + cur_version = self._get_current_version() + await PlatformUtils.send_superuser( + bot, + f"检测真寻已更新，当前版本：{cur_version}\n开始更新...", + user_id, + ) + + if zip_update: + return await self._zip_update(version_type) + elif source == "git": + result = await GithubRepoManager.update( + self.config.ZHENXUN_BOT_GIT, + Path(), + use_git=update_type == "git", + force=force, + ) + else: + result = await AliyunRepoManager.update( + self.config.ZHENXUN_BOT_GIT, + Path(), + force=force, + ) + + if not result.success: + return f"版本更新失败...错误: {result.error_message}" + + await PlatformUtils.send_superuser( + bot, "真寻更新完成，开始安装依赖...", user_id + ) + await VirtualEnvPackageManager.install_requirement(self.config.REQ_TXT_FILE) + + return ( + f"版本更新完成！\n" + f"版本: {cur_version} -> {result.new_version}\n" + f"变更文件个数: {len(result.changed_files)}" + f"{'' if source == 'git' else '（阿里云更新不支持查看变更文件）'}\n" + "请重新启动真寻以完成更新!" + ) + + async def _zip_update(self, version_type: str) -> str: + """ZIP文件更新 + + 参数: + version_type: 版本类型 (main/release) + + 返回: + str: 更新结果 + """ + logger.info("开始下载真寻最新版文件....", self.config.COMMAND) + cur_version = self._get_current_version() + url = None + new_version = None + repo_info = GithubUtils.parse_github_url(self.config.DEFAULT_GITHUB_URL) + + if version_type in {"main"}: + repo_info.branch = version_type + new_version = await self._get_version_from_repo(repo_info) + if new_version: + new_version = new_version.split(":")[-1].strip() + url = await repo_info.get_archive_download_urls() + elif version_type == "release": + data = await self._get_latest_release_data() + if not data: + return "获取更新版本失败..." + new_version = data.get("name", "") + url = await repo_info.get_release_source_download_urls_tgz(new_version) + + if not url: + return "获取版本下载链接失败..." + + if self.config.TMP_PATH.exists(): + logger.debug(f"删除临时文件夹 {self.config.TMP_PATH}", self.config.COMMAND) + shutil.rmtree(self.config.TMP_PATH) + + logger.debug( + f"开始更新版本：{cur_version} -> {new_version} | 下载链接：{url}", + self.config.COMMAND, + ) + + download_file = ( + self.config.DOWNLOAD_GZ_FILE + if version_type == "release" + else self.config.DOWNLOAD_ZIP_FILE + ) + + if await AsyncHttpx.download_file(url, download_file, stream=True): + logger.debug("下载真寻最新版文件完成...", self.config.COMMAND) + self._handle_downloaded_files(new_version) + result = "版本更新完成" + return ( + f"{result}\n" + f"版本: {cur_version} -> {new_version}\n" + "请重新启动真寻以完成更新!" + ) + else: + logger.debug("下载真寻最新版文件失败...", self.config.COMMAND) + return "" + + def _handle_downloaded_files(self, latest_version: str | None): + """处理下载的文件 + + 参数: + latest_version: 最新版本号 + """ + self.config.BACKUP_PATH.mkdir(exist_ok=True, parents=True) + logger.debug("开始解压文件压缩包...", self.config.COMMAND) + + download_file = self.config.DOWNLOAD_GZ_FILE + if self.config.DOWNLOAD_GZ_FILE.exists(): + tf = tarfile.open(self.config.DOWNLOAD_GZ_FILE) + else: + download_file = self.config.DOWNLOAD_ZIP_FILE + tf = zipfile.ZipFile(self.config.DOWNLOAD_ZIP_FILE) + + tf.extractall(self.config.TMP_PATH) + logger.debug("解压文件压缩包完成...", self.config.COMMAND) + + download_file_path = self.config.TMP_PATH / next( + x + for x in os.listdir(self.config.TMP_PATH) + if (self.config.TMP_PATH / x).is_dir() + ) + + _pyproject = download_file_path / self.config.PYPROJECT_FILE_STRING + _lock_file = download_file_path / self.config.PYPROJECT_LOCK_FILE_STRING + _req_file = download_file_path / self.config.REQ_TXT_FILE_STRING + extract_path = download_file_path / self.config.BASE_PATH_STRING + target_path = self.config.BASE_PATH + + # 备份现有文件 + if self.config.PYPROJECT_FILE.exists(): + logger.debug( + f"移除备份文件: {self.config.PYPROJECT_FILE}", self.config.COMMAND + ) + shutil.move( + self.config.PYPROJECT_FILE, + self.config.BACKUP_PATH / self.config.PYPROJECT_FILE_STRING, + ) + if self.config.PYPROJECT_LOCK_FILE.exists(): + logger.debug( + f"移除备份文件: {self.config.PYPROJECT_LOCK_FILE}", self.config.COMMAND + ) + shutil.move( + self.config.PYPROJECT_LOCK_FILE, + self.config.BACKUP_PATH / self.config.PYPROJECT_LOCK_FILE_STRING, + ) + if self.config.REQ_TXT_FILE.exists(): + logger.debug( + f"移除备份文件: {self.config.REQ_TXT_FILE}", self.config.COMMAND + ) + shutil.move( + self.config.REQ_TXT_FILE, + self.config.BACKUP_PATH / self.config.REQ_TXT_FILE_STRING, + ) + + # 移动新文件 + if _pyproject.exists(): + logger.debug("移动文件: pyproject.toml", self.config.COMMAND) + shutil.move(_pyproject, self.config.PYPROJECT_FILE) + if _lock_file.exists(): + logger.debug("移动文件: poetry.lock", self.config.COMMAND) + shutil.move(_lock_file, self.config.PYPROJECT_LOCK_FILE) + if _req_file.exists(): + logger.debug("移动文件: requirements.txt", self.config.COMMAND) + shutil.move(_req_file, self.config.REQ_TXT_FILE) + + # 处理文件夹 + for folder in self.config.REPLACE_FOLDERS: + _dir = self.config.BASE_PATH / folder + _backup_dir = self.config.BACKUP_PATH / folder + if _backup_dir.exists(): + logger.debug(f"删除备份文件夹 {_backup_dir}", self.config.COMMAND) + shutil.rmtree(_backup_dir) + if _dir.exists(): + logger.debug(f"移动旧文件夹 {_dir}", self.config.COMMAND) + shutil.move(_dir, _backup_dir) + else: + logger.warning(f"文件夹 {_dir} 不存在，跳过删除", self.config.COMMAND) + + for folder in self.config.REPLACE_FOLDERS: + src_folder_path = extract_path / folder + dest_folder_path = target_path / folder + if src_folder_path.exists(): + logger.debug( + f"移动文件夹: {src_folder_path} -> {dest_folder_path}", + self.config.COMMAND, + ) + shutil.move(src_folder_path, dest_folder_path) + else: + logger.debug(f"源文件夹不存在: {src_folder_path}", self.config.COMMAND) + + # 清理临时文件 + if tf: + tf.close() + if download_file.exists(): + logger.debug(f"删除下载文件: {download_file}", self.config.COMMAND) + download_file.unlink() + if extract_path.exists(): + logger.debug(f"删除解压文件夹: {extract_path}", self.config.COMMAND) + shutil.rmtree(extract_path) + if self.config.TMP_PATH.exists(): + shutil.rmtree(self.config.TMP_PATH) + + # 更新版本文件 + if latest_version: + with open(self.config.VERSION_FILE, "w", encoding="utf8") as f: + f.write(f"__version__: {latest_version}") + + def _get_current_version(self) -> str: + """获取当前版本 + + 返回: + str: 当前版本号 + """ + _version = "v0.0.0" + if self.config.VERSION_FILE.exists(): + if text := self.config.VERSION_FILE.open(encoding="utf8").readline(): + _version = text.split(":")[-1].strip() + return _version + + async def _get_latest_release_data(self) -> dict: + """获取最新版本信息 + + 返回: + dict: 最新版本数据 + """ + for _ in range(3): + try: + res = await AsyncHttpx.get(self.config.RELEASE_URL) + if res.status_code == 200: + return res.json() + except TimeoutError: + pass + except Exception as e: + logger.error("检查更新真寻获取版本失败", e=e) + return {} + + async def _get_version_from_repo(self, repo_info: RepoInfo) -> str: + """从指定分支获取版本号 + + 参数: + repo_info: 仓库信息 + + 返回: + str: 版本号 + """ + version_url = await repo_info.get_raw_download_urls(path="__version__") + try: + res = await AsyncHttpx.get(version_url) + if res.status_code == 200: + return res.text.strip() + except Exception as e: + logger.error(f"获取 {repo_info.branch} 分支版本失败", e=e) + return "未知版本" + + # ==================== 资源管理相关方法 ==================== + + async def init_resources( + self, force: bool = False, is_zip: bool = False, git_source: str = "ali" + ) -> str: + """初始化资源文件 + + 参数: + force: 是否强制更新 + is_zip: 是否下载zip文件 + git_source: 更新源 (ali/git) + + 返回: + str: 操作结果 + """ + if (FONT_PATH.exists() and os.listdir(FONT_PATH)) and not force: + return "资源文件已存在，跳过初始化" + + try: + if is_zip: + if self.config.RESOURCE_TMP_PATH.exists(): + logger.debug( + "resources临时文件夹已存在，移除resources临时文件夹", + self.config.COMMAND, + ) + await clean_git(self.config.RESOURCE_TMP_PATH) + shutil.rmtree(self.config.RESOURCE_TMP_PATH, ignore_errors=True) + self.config.RESOURCE_TMP_PATH.mkdir(parents=True, exist_ok=True) + await self._download_resources() + self._handle_resource_files() + else: + if git_source == "ali": + result = await AliyunRepoManager.update( + self.config.RESOURCE_GITHUB_URL, self.config.RESOURCE_PATH + ) + else: + result = await GithubRepoManager.update( + self.config.RESOURCE_GITHUB_URL, self.config.RESOURCE_PATH + ) + if not result.success: + return f"资源更新失败...错误: {result.error_message}" + self.config.UNZIP_PATH = self.config.RESOURCE_TMP_PATH / "resources" + self._handle_resource_files() + + if self.config.RESOURCE_TMP_PATH.exists(): + logger.debug("移除resources临时文件夹", self.config.COMMAND) + await clean_git(self.config.RESOURCE_TMP_PATH) + shutil.rmtree(self.config.RESOURCE_TMP_PATH) + + return "资源文件初始化成功！" + except Exception as e: + logger.error("资源文件初始化失败", self.config.COMMAND, e=e) + return f"资源文件初始化失败: {e}" + + def _handle_resource_files(self): + """处理资源文件""" + if not hasattr(self.config, "UNZIP_PATH") or not self.config.UNZIP_PATH: + return + self._recursive_folder(self.config.UNZIP_PATH, ".") + + def _recursive_folder(self, dir: Path, parent_path: str): + """递归处理文件夹 + + 参数: + dir: 目录路径 + parent_path: 父路径 + """ + for file in dir.iterdir(): + if file.is_dir(): + self._recursive_folder(file, f"{parent_path}/{file.name}") + else: + res_file = Path(parent_path) / file.name + if res_file.exists(): + res_file.unlink() + res_file.parent.mkdir(parents=True, exist_ok=True) + file.rename(res_file) + + async def _download_resources(self): + """下载资源文件""" + repo_info = GithubUtils.parse_github_url(self.config.RESOURCE_GITHUB_URL) + url = await repo_info.get_archive_download_urls() + logger.debug("开始下载resources资源包...", self.config.COMMAND) + + if not await AsyncHttpx.download_file( + url, self.config.RESOURCE_ZIP_FILE, stream=True + ): + logger.error( + "下载resources资源包失败，请尝试重启重新下载或前往 " + "https://github.com/zhenxun-org/zhenxun-bot-resources 手动下载..." + ) + raise DownloadException("下载resources资源包失败...") + + logger.debug("下载resources资源文件压缩包完成...", self.config.COMMAND) + tf = zipfile.ZipFile(self.config.RESOURCE_ZIP_FILE) + tf.extractall(self.config.RESOURCE_TMP_PATH) + logger.debug("解压文件压缩包完成...", self.config.COMMAND) + + download_file_path = self.config.RESOURCE_TMP_PATH / next( + x + for x in os.listdir(self.config.RESOURCE_TMP_PATH) + if (self.config.RESOURCE_TMP_PATH / x).is_dir() + ) + self.config.UNZIP_PATH = download_file_path / "resources" + + if tf: + tf.close() + + # ==================== 子模块管理相关方法 ==================== + + async def init_submodules(self) -> str: + """初始化子模块 + + 返回: + str: 操作结果 + """ + try: + # 定义子模块配置 + submodule_configs = [ + SubmoduleConfig( + name="resources", + path="resources", + repo_url=self.config.RESOURCE_GITHUB_URL, + branch="main", + enabled=True, + ), + SubmoduleConfig( + name="web_ui", + path="data/web_ui/public", + repo_url=self.config.WEBUI_GIT, + branch="main", + enabled=True, + ), + ] + + # 初始化子模块 + success = await self.submodule_manager.init_submodules( + Path(), submodule_configs + ) + + if success: + return "子模块初始化成功！" + else: + return "子模块初始化失败！" + + except Exception as e: + logger.error("子模块初始化失败", self.config.COMMAND, e=e) + return f"子模块初始化失败: {e}" + + async def update_submodules(self) -> str: + """更新子模块 + + 返回: + str: 操作结果 + """ + try: + # 定义子模块配置 + submodule_configs = [ + SubmoduleConfig( + name="resources", + path="resources", + repo_url=self.config.RESOURCE_GITHUB_URL, + branch="main", + enabled=True, + ), + SubmoduleConfig( + name="web_ui", + path="data/web_ui/public", + repo_url=self.config.WEBUI_GIT, + branch="main", + enabled=True, + ), + ] + + # 更新子模块 + results = await self.submodule_manager.update_submodules( + Path(), submodule_configs + ) + + success_count = sum(1 for result in results if result.success) + total_count = len(results) + + return f"子模块更新完成！成功: {success_count}/{total_count}" + + except Exception as e: + logger.error("子模块更新失败", self.config.COMMAND, e=e) + return f"子模块更新失败: {e}" + + async def get_submodule_info(self) -> str: + """获取子模块信息 + + 返回: + str: 子模块信息 + """ + try: + # 定义子模块配置 + submodule_configs = [ + SubmoduleConfig( + name="resources", + path="resources", + repo_url=self.config.RESOURCE_GITHUB_URL, + branch="main", + enabled=True, + ), + SubmoduleConfig( + name="web_ui", + path="data/web_ui/public", + repo_url=self.config.WEBUI_GIT, + branch="main", + enabled=True, + ), + ] + + # 获取子模块信息 + submodule_infos = await self.submodule_manager.get_submodule_info( + Path(), submodule_configs + ) + + info_text = "子模块信息:\n" + for info in submodule_infos: + info_text += f"- {info.config.name}:\n" + info_text += f" 路径: {info.config.path}\n" + info_text += f" 当前版本: {info.current_version}\n" + info_text += f" 最新版本: {info.latest_version}\n" + info_text += f" 状态: {info.update_status}\n" + + return info_text + + except Exception as e: + logger.error("获取子模块信息失败", self.config.COMMAND, e=e) + return f"获取子模块信息失败: {e}" + + # ==================== Web UI 管理相关方法 ==================== + + async def webui_download_zip(self) -> str: + """下载 WEBUI_ASSETS 资源""" + webui_assets_path = TEMP_PATH / "webui_assets.zip" + download_url = await GithubUtils.parse_github_url( + self.config.WEBUI_GIT + ).get_archive_download_urls() + logger.info("开始下载 WEBUI_ASSETS 资源...", LOG_COMMAND) + if await AsyncHttpx.download_file( + download_url, webui_assets_path, follow_redirects=True + ): + logger.info("下载 WEBUI_ASSETS 成功!", LOG_COMMAND) + raise DownloadException("下载 WEBUI_ASSETS 失败", LOG_COMMAND) + + def __backup_webui(self): + """备份 WEBUI_ASSERT 资源""" + if self.config.WEBUI_PATH.exists(): + if self.config.WEBUI_BACKUP_PATH.exists(): + logger.debug( + f"删除旧的备份webui文件夹 {self.config.WEBUI_BACKUP_PATH}", + LOG_COMMAND, + ) + shutil.rmtree(self.config.WEBUI_BACKUP_PATH) + shutil.copytree(self.config.WEBUI_PATH, self.config.WEBUI_BACKUP_PATH) + + # async def webui_unzip(self) -> str: + # """使用zip更新 Web UI + + # 参数: + # is_zip: 是否下载 ZIP 文件 + # source: 更新源 (git/ali) + + # 返回: + # str: 更新结果 + # """ + # self.__backup_webui() + # self.__clear_folder(self.config.WEBUI_PATH) diff --git a/zhenxun/utils/repo_utils/__init__.py b/zhenxun/utils/repo_utils/__init__.py index f37ccd26..39769d1d 100644 --- a/zhenxun/utils/repo_utils/__init__.py +++ b/zhenxun/utils/repo_utils/__init__.py @@ -24,6 +24,9 @@ from .models import ( RepoFileInfo, RepoType, RepoUpdateResult, + SubmoduleConfig, + SubmoduleInfo, + SubmoduleUpdateResult, ) from .utils import check_git, filter_files, glob_to_regex, run_git_command @@ -53,6 +56,9 @@ __all__ = [ "RepoType", "RepoUpdateError", "RepoUpdateResult", + "SubmoduleConfig", + "SubmoduleInfo", + "SubmoduleUpdateResult", "check_git", "filter_files", "glob_to_regex", diff --git a/zhenxun/utils/repo_utils/github_manager.py b/zhenxun/utils/repo_utils/github_manager.py index 462c2723..01c13790 100644 --- a/zhenxun/utils/repo_utils/github_manager.py +++ b/zhenxun/utils/repo_utils/github_manager.py @@ -29,7 +29,11 @@ from .models import ( RepoFileInfo, RepoType, RepoUpdateResult, + SubmoduleConfig, + SubmoduleInfo, + SubmoduleUpdateResult, ) +from .submodule_manager import SubmoduleManager class GithubManager(BaseRepoManager): @@ -43,6 +47,7 @@ class GithubManager(BaseRepoManager): config: 配置，如果为None则使用默认配置 """ super().__init__(config) + self.submodule_manager = SubmoduleManager(self) async def update_repo( self, @@ -524,3 +529,158 @@ class GithubManager(BaseRepoManager): raise RepoDownloadError("下载文件失败") raise RepoDownloadError("下载文件失败: 超过最大重试次数") + + # 子模块相关方法 + async def init_submodules( + self, + main_repo_path: Path, + submodule_configs: list[SubmoduleConfig], + ) -> bool: + """ + 初始化子模块 + + 参数: + main_repo_path: 主仓库路径 + submodule_configs: 子模块配置列表 + + 返回: + bool: 是否成功 + """ + return await self.submodule_manager.init_submodules( + main_repo_path, submodule_configs + ) + + async def update_submodules( + self, + main_repo_path: Path, + submodule_configs: list[SubmoduleConfig], + ) -> list[SubmoduleUpdateResult]: + """ + 更新子模块 + + 参数: + main_repo_path: 主仓库路径 + submodule_configs: 子模块配置列表 + + 返回: + list[SubmoduleUpdateResult]: 更新结果列表 + """ + return await self.submodule_manager.update_submodules( + main_repo_path, submodule_configs + ) + + async def get_submodule_info( + self, + main_repo_path: Path, + submodule_configs: list[SubmoduleConfig], + ) -> list[SubmoduleInfo]: + """ + 获取子模块信息 + + 参数: + main_repo_path: 主仓库路径 + submodule_configs: 子模块配置列表 + + 返回: + list[SubmoduleInfo]: 子模块信息列表 + """ + return await self.submodule_manager.get_submodule_info( + main_repo_path, submodule_configs + ) + + def save_submodule_configs( + self, + main_repo_path: Path, + submodule_configs: list[SubmoduleConfig], + ) -> bool: + """ + 保存子模块配置到文件 + + 参数: + main_repo_path: 主仓库路径 + submodule_configs: 子模块配置列表 + + 返回: + bool: 是否成功 + """ + return self.submodule_manager.save_submodule_configs( + main_repo_path, submodule_configs + ) + + async def load_submodule_configs( + self, main_repo_path: Path + ) -> list[SubmoduleConfig]: + """ + 从文件加载子模块配置 + + 参数: + main_repo_path: 主仓库路径 + + 返回: + list[SubmoduleConfig]: 子模块配置列表 + """ + return await self.submodule_manager.load_submodule_configs(main_repo_path) + + async def update_with_submodules( + self, + repo_url: str, + local_path: Path, + branch: str = "main", + submodule_configs: list[SubmoduleConfig] | None = None, + use_git: bool = True, + force: bool = False, + include_patterns: list[str] | None = None, + exclude_patterns: list[str] | None = None, + ) -> RepoUpdateResult: + """ + 更新仓库并处理子模块 + + 参数: + repo_url: 仓库URL，格式为 https://github.com/owner/repo + local_path: 本地保存路径 + branch: 分支名称 + submodule_configs: 子模块配置列表 + use_git: 是否使用Git命令更新 + force: 是否强制更新 + include_patterns: 包含的文件模式列表 + exclude_patterns: 排除的文件模式列表 + + 返回: + RepoUpdateResult: 更新结果 + """ + # 更新主仓库 + result = await self.update( + repo_url, + local_path, + branch, + use_git, + force, + include_patterns, + exclude_patterns, + ) + + # 如果没有子模块配置，直接返回结果 + if not submodule_configs: + return result + + # 处理子模块 + try: + submodule_results = await self.update_submodules( + local_path, submodule_configs + ) + result.submodule_results = submodule_results + + # 检查子模块更新是否成功 + failed_submodules = [r for r in submodule_results if not r.success] + if failed_submodules: + logger.warning( + "部分子模块更新失败:" + f" {[r.submodule_name for r in failed_submodules]}", + LOG_COMMAND, + ) + + except Exception as e: + logger.error(f"处理子模块时发生错误: {e}", LOG_COMMAND) + result.error_message += f"; 子模块处理失败: {e}" + + return result diff --git a/zhenxun/utils/repo_utils/models.py b/zhenxun/utils/repo_utils/models.py index 170e60f3..d3ec393f 100644 --- a/zhenxun/utils/repo_utils/models.py +++ b/zhenxun/utils/repo_utils/models.py @@ -15,6 +15,62 @@ class RepoType(str, Enum): ALIYUN = "aliyun" +@dataclass +class SubmoduleConfig: + """子模块配置""" + + # 子模块名称 + name: str + # 子模块路径（相对于主仓库） + path: str + # 子模块仓库URL + repo_url: str + # 分支名称 + branch: str = "main" + # 是否启用 + enabled: bool = True + # 包含的文件模式列表 + include_patterns: list[str] | None = None + # 排除的文件模式列表 + exclude_patterns: list[str] | None = None + + +@dataclass +class SubmoduleInfo: + """子模块信息""" + + # 子模块配置 + config: SubmoduleConfig + # 当前版本 + current_version: str = "" + # 最新版本 + latest_version: str = "" + # 最后更新时间 + last_update: datetime | None = None + # 更新状态 + update_status: str = "unknown" # unknown, up_to_date, outdated, error + + +@dataclass +class SubmoduleUpdateResult: + """子模块更新结果""" + + # 子模块名称 + submodule_name: str + # 子模块路径 + submodule_path: str + # 旧版本 + old_version: str + # 新版本 + new_version: str + # 是否成功 + success: bool = False + # 错误消息 + error_message: str = "" + # 变更的文件列表 + changed_files: list[str] = field(default_factory=list) + + @dataclass class RepoFileInfo: """仓库文件信息""" @@ -67,6 +123,8 @@ class RepoUpdateResult: error_message: str = "" # 变更的文件列表 changed_files: list[str] = field(default_factory=list) + # 子模块更新结果 + submodule_results: list[SubmoduleUpdateResult] = field(default_factory=list) @dataclass diff --git a/zhenxun/utils/repo_utils/submodule_manager.py b/zhenxun/utils/repo_utils/submodule_manager.py new file mode 100644 index 00000000..c9927e92 --- /dev/null +++ b/zhenxun/utils/repo_utils/submodule_manager.py @@ -0,0 +1,408 @@ +""" +子模块管理工具 +""" + +import json +from pathlib import Path + +from zhenxun.services.log import logger + +from .config import LOG_COMMAND +from .github_manager import GithubManager +from .models import SubmoduleConfig, SubmoduleInfo, SubmoduleUpdateResult +from .utils import run_git_command + + +class SubmoduleManager: + """子模块管理器""" + + def __init__(self, github_manager: GithubManager): + """ + 初始化子模块管理器 + + 参数: + github_manager: GitHub管理器实例 + """ + self.github_manager = github_manager + + async def init_submodules( + self, main_repo_path: Path, submodule_configs: list[SubmoduleConfig] + ) -> bool: + """ + 初始化子模块 + + 参数: + main_repo_path: 主仓库路径 + submodule_configs: 子模块配置列表 + + 返回: + bool: 是否成功 + """ + try: + # 检查是否在Git仓库中 + success, stdout, stderr = await run_git_command("status", main_repo_path) + if not success: + logger.error(f"路径 {main_repo_path} 不是有效的Git仓库", LOG_COMMAND) + return False + + # 初始化每个子模块 + for config in submodule_configs: + if not config.enabled: + continue + + await self._init_single_submodule(main_repo_path, config) + + # 更新子模块 + await self._update_submodules(main_repo_path) + + return True + + except Exception as e: + logger.error(f"初始化子模块失败: {e}", LOG_COMMAND) + return False + + async def _init_single_submodule( + self, main_repo_path: Path, config: SubmoduleConfig + ) -> bool: + """ + 初始化单个子模块 + + 参数: + main_repo_path: 主仓库路径 + config: 子模块配置 + + 返回: + bool: 是否成功 + """ + try: + submodule_path = main_repo_path / config.path + + # 检查子模块是否已存在 + if submodule_path.exists() and (submodule_path / ".git").exists(): + logger.info(f"子模块 {config.name} 已存在，跳过初始化", LOG_COMMAND) + return True + + # 添加子模块 + success, stdout, stderr = await run_git_command( + f"submodule add -b {config.branch} {config.repo_url} {config.path}", + main_repo_path, + ) + + if not success: + logger.error(f"添加子模块 {config.name} 失败: {stderr}", LOG_COMMAND) + return False + + logger.info(f"成功添加子模块 {config.name}", LOG_COMMAND) + return True + + except Exception as e: + logger.error(f"初始化子模块 {config.name} 失败: {e}", LOG_COMMAND) + return False + + async def _update_submodules(self, main_repo_path: Path) -> bool: + """ + 更新所有子模块 + + 参数: + main_repo_path: 主仓库路径 + + 返回: + bool: 是否成功 + """ + try: + # 更新子模块 + success, stdout, stderr = await run_git_command( + "submodule update --init --recursive", main_repo_path + ) + + if not success: + logger.error(f"更新子模块失败: {stderr}", LOG_COMMAND) + return False + + logger.info("成功更新所有子模块", LOG_COMMAND) + return True + + except Exception as e: + logger.error(f"更新子模块失败: {e}", LOG_COMMAND) + return False + + async def update_submodules( + self, main_repo_path: Path, submodule_configs: list[SubmoduleConfig] + ) -> list[SubmoduleUpdateResult]: + """ + 更新子模块 + + 参数: + main_repo_path: 主仓库路径 + submodule_configs: 子模块配置列表 + + 返回: + List[SubmoduleUpdateResult]: 更新结果列表 + """ + results = [] + + for config in submodule_configs: + if not config.enabled: + continue + + result = await self._update_single_submodule(main_repo_path, config) + results.append(result) + + return results + + async def _update_single_submodule( + self, main_repo_path: Path, config: SubmoduleConfig + ) -> SubmoduleUpdateResult: + """ + 更新单个子模块 + + 参数: + main_repo_path: 主仓库路径 + config: 子模块配置 + + 返回: + SubmoduleUpdateResult: 更新结果 + """ + result = SubmoduleUpdateResult( + submodule_name=config.name, + submodule_path=config.path, + old_version="", + new_version="", + ) + + try: + submodule_path = main_repo_path / config.path + + # 检查子模块是否存在 + if not submodule_path.exists(): + result.error_message = f"子模块路径不存在: {submodule_path}" + return result + + # 获取当前版本 + success, stdout, stderr = await run_git_command( + "rev-parse HEAD", submodule_path + ) + + if not success: + result.error_message = f"获取当前版本失败: {stderr}" + return result + + old_version = stdout.strip() + result.old_version = old_version + + # 获取远程最新版本 + success, stdout, stderr = await run_git_command( + f"ls-remote origin {config.branch}", submodule_path + ) + + if not success: + result.error_message = f"获取远程版本失败: {stderr}" + return result + + # 解析最新版本 + lines = stdout.strip().split("\n") + if not lines or not lines[0]: + result.error_message = "无法获取远程版本信息" + return result + + latest_version = lines[0].split("\t")[0] + result.new_version = latest_version + + # 检查是否需要更新 + if old_version == latest_version: + result.success = True + logger.info(f"子模块 {config.name} 已是最新版本", LOG_COMMAND) + return result + + # 更新子模块 + success, stdout, stderr = await run_git_command( + f"pull origin {config.branch}", submodule_path + ) + + if not success: + result.error_message = f"更新子模块失败: {stderr}" + return result + + # 更新主仓库中的子模块引用 + success, stdout, stderr = await run_git_command( + f"add {config.path}", main_repo_path + ) + + if not success: + result.error_message = f"更新主仓库引用失败: {stderr}" + return result + + result.success = True + logger.info( + f"成功更新子模块 {config.name}: {old_version} -> {latest_version}", + LOG_COMMAND, + ) + + except Exception as e: + result.error_message = f"更新子模块时发生错误: {e}" + logger.error(f"更新子模块 {config.name} 失败: {e}", LOG_COMMAND) + + return result + + async def get_submodule_info( + self, main_repo_path: Path, submodule_configs: list[SubmoduleConfig] + ) -> list[SubmoduleInfo]: + """ + 获取子模块信息 + + 参数: + main_repo_path: 主仓库路径 + submodule_configs: 子模块配置列表 + + 返回: + List[SubmoduleInfo]: 子模块信息列表 + """ + submodule_infos = [] + + for config in submodule_configs: + if not config.enabled: + continue + + info = await self._get_single_submodule_info(main_repo_path, config) + submodule_infos.append(info) + + return submodule_infos + + async def _get_single_submodule_info( + self, main_repo_path: Path, config: SubmoduleConfig + ) -> SubmoduleInfo: + """ + 获取单个子模块信息 + + 参数: + main_repo_path: 主仓库路径 + config: 子模块配置 + + 返回: + SubmoduleInfo: 子模块信息 + """ + info = SubmoduleInfo(config=config) + + try: + submodule_path = main_repo_path / config.path + + if not submodule_path.exists(): + info.update_status = "error" + return info + + # 获取当前版本 + success, stdout, stderr = await run_git_command( + "rev-parse HEAD", submodule_path + ) + + if success: + info.current_version = stdout.strip() + + # 获取远程最新版本 + success, stdout, stderr = await run_git_command( + f"ls-remote origin {config.branch}", submodule_path + ) + + if success and stdout.strip(): + lines = stdout.strip().split("\n") + if lines and lines[0]: + info.latest_version = lines[0].split("\t")[0] + + # 确定更新状态 + if info.current_version and info.latest_version: + if info.current_version == info.latest_version: + info.update_status = "up_to_date" + else: + info.update_status = "outdated" + else: + info.update_status = "unknown" + + except Exception as e: + info.update_status = "error" + logger.error(f"获取子模块 {config.name} 信息失败: {e}", LOG_COMMAND) + + return info + + def save_submodule_configs( + self, main_repo_path: Path, submodule_configs: list[SubmoduleConfig] + ) -> bool: + """ + 保存子模块配置到文件 + + 参数: + main_repo_path: 主仓库路径 + submodule_configs: 子模块配置列表 + + 返回: + bool: 是否成功 + """ + try: + config_file = main_repo_path / ".submodules.json" + + # 转换为字典格式 + configs_dict = [] + for config in submodule_configs: + config_dict = { + "name": config.name, + "path": config.path, + "repo_url": config.repo_url, + "branch": config.branch, + "enabled": config.enabled, + "include_patterns": config.include_patterns, + "exclude_patterns": config.exclude_patterns, + } + configs_dict.append(config_dict) + + # 保存到文件 + with open(config_file, "w", encoding="utf-8") as f: + json.dump(configs_dict, f, indent=2, ensure_ascii=False) + + logger.info(f"子模块配置已保存到 {config_file}", LOG_COMMAND) + return True + + except Exception as e: + logger.error(f"保存子模块配置失败: {e}", LOG_COMMAND) + return False + + def load_submodule_configs(self, main_repo_path: Path) -> list[SubmoduleConfig]: + """ + 从文件加载子模块配置 + + 参数: + main_repo_path: 主仓库路径 + + 返回: + List[SubmoduleConfig]: 子模块配置列表 + """ + try: + config_file = main_repo_path / ".submodules.json" + + if not config_file.exists(): + logger.warning(f"子模块配置文件不存在: {config_file}", LOG_COMMAND) + return [] + + with open(config_file, encoding="utf-8") as f: + configs_dict = json.load(f) + + # 转换为SubmoduleConfig对象 + configs = [] + for config_dict in configs_dict: + config = SubmoduleConfig( + name=config_dict["name"], + path=config_dict["path"], + repo_url=config_dict["repo_url"], + branch=config_dict.get("branch", "main"), + enabled=config_dict.get("enabled", True), + include_patterns=config_dict.get("include_patterns"), + exclude_patterns=config_dict.get("exclude_patterns"), + ) + configs.append(config) + + logger.info( + f"从 {config_file} 加载了 {len(configs)} 个子模块配置", LOG_COMMAND + ) + return configs + + except Exception as e: + logger.error(f"加载子模块配置失败: {e}", LOG_COMMAND) + return [] diff --git a/zhenxun/utils/repo_utils/submodule_setup.py b/zhenxun/utils/repo_utils/submodule_setup.py new file mode 100644 index 00000000..22bef33e --- /dev/null +++ b/zhenxun/utils/repo_utils/submodule_setup.py @@ -0,0 +1,210 @@ +#!/usr/bin/env python3 +""" +GitHub子模块快速设置脚本 +""" + +import asyncio +from pathlib import Path +import sys + +from zhenxun.services.log import logger +from zhenxun.utils.repo_utils import ( + GithubRepoManager, + SubmoduleConfig, +) + + +def create_sample_configs(): + """创建示例子模块配置""" + return [ + SubmoduleConfig( + name="frontend-ui", + path="frontend/ui", + repo_url="https://github.com/your-org/frontend-ui", + branch="main", + enabled=True, + include_patterns=["*.js", "*.css", "*.html", "*.vue", "*.ts"], + exclude_patterns=["node_modules/*", "*.log", "dist/*", "coverage/*"], + ), + SubmoduleConfig( + name="backend-api", + path="backend/api", + repo_url="https://github.com/your-org/backend-api", + branch="develop", + enabled=True, + include_patterns=["*.py", "*.json", "requirements.txt", "*.yml"], + exclude_patterns=["__pycache__/*", "*.pyc", "venv/*", ".pytest_cache/*"], + ), + SubmoduleConfig( + name="shared-lib", + path="libs/shared", + repo_url="https://github.com/your-org/shared-lib", + branch="main", + enabled=True, + include_patterns=["*.py", "*.js", "*.ts", "*.json"], + exclude_patterns=["tests/*", "docs/*", "examples/*"], + ), + ] + + +async def setup_submodules(project_path: str, configs: list[SubmoduleConfig]): + """设置子模块""" + main_repo_path = Path(project_path) + + logger.info(f"正在为项目 {project_path} 设置子模块...") + + # 检查路径是否存在 + if not main_repo_path.exists(): + logger.info(f"错误: 项目路径 {project_path} 不存在") + return False + + # 检查是否是Git仓库 + git_dir = main_repo_path / ".git" + if not git_dir.exists(): + logger.info(f"错误: {project_path} 不是Git仓库") + logger.info("请先执行: git init") + return False + + # 初始化子模块 + logger.info("正在初始化子模块...") + success = await GithubRepoManager.init_submodules(main_repo_path, configs) + + if not success: + logger.info("子模块初始化失败！") + return False + + # 保存配置 + logger.info("正在保存子模块配置...") + await GithubRepoManager.save_submodule_configs(main_repo_path, configs) + + logger.info("✓ 子模块设置完成！") + logger.info(f"配置文件已保存到: {main_repo_path / '.submodules.json'}") + + return True + + +async def update_submodules(project_path: str): + """更新子模块""" + main_repo_path = Path(project_path) + + logger.info(f"正在更新项目 {project_path} 的子模块...") + + # 加载配置 + configs = await GithubRepoManager.load_submodule_configs(main_repo_path) + + if not configs: + logger.info("未找到子模块配置") + return False + + logger.info(f"找到 {len(configs)} 个子模块配置") + + # 获取子模块信息 + infos = await GithubRepoManager.get_submodule_info(main_repo_path, configs) + + logger.info("\n子模块状态:") + for info in infos: + status_icon = ( + "✓" + if info.update_status == "up_to_date" + else "⚠" + if info.update_status == "outdated" + else "✗" + ) + logger.info( + f"{status_icon} {info.config.name}" + f"({info.config.path}) - {info.update_status}" + ) + + # 更新子模块 + logger.info("\n正在更新子模块...") + results = await GithubRepoManager.update_submodules(main_repo_path, configs) + + success_count = 0 + for result in results: + if result.success: + success_count += 1 + if result.old_version != result.new_version: + logger.info(f" ✓ {result.submodule_name} 已更新") + else: + logger.info(f" ✓ {result.submodule_name} 已是最新版本") + else: + logger.info(f" ✗ {result.submodule_name} 更新失败: {result.error_message}") + + logger.info(f"\n更新完成: {success_count}/{len(results)} 个子模块更新成功") + return success_count == len(results) + + +async def show_submodule_info(project_path: str): + """显示子模块信息""" + main_repo_path = Path(project_path) + + logger.info(f"项目 {project_path} 的子模块信息:") + + # 加载配置 + configs = await GithubRepoManager.load_submodule_configs(main_repo_path) + + if not configs: + logger.info("未找到子模块配置") + return + + # 获取详细信息 + infos = await GithubRepoManager.get_submodule_info(main_repo_path, configs) + + for info in infos: + logger.info(f"\n子模块: {info.config.name}") + logger.info(f" 路径: {info.config.path}") + logger.info(f" 仓库: {info.config.repo_url}") + logger.info(f" 分支: {info.config.branch}") + logger.info(f" 状态: {info.update_status}") + logger.info(f" 启用: {info.config.enabled}") + + if info.current_version: + logger.info(f" 当前版本: {info.current_version[:8]}") + if info.latest_version: + logger.info(f" 最新版本: {info.latest_version[:8]}") + + if info.config.include_patterns: + logger.info(f" 包含文件: {', '.join(info.config.include_patterns)}") + if info.config.exclude_patterns: + logger.info(f" 排除文件: {', '.join(info.config.exclude_patterns)}") + + +def print_info_usage(): + """打印使用说明""" + logger.info("GitHub子模块管理工具") + logger.info("用法:") + logger.info(" python submodule_setup.py setup <项目路径>") + logger.info(" python submodule_setup.py update <项目路径>") + logger.info(" python submodule_setup.py info <项目路径>") + logger.info("示例:") + logger.info(" python submodule_setup.py setup ./my_project") + logger.info(" python submodule_setup.py update ./my_project") + logger.info(" python submodule_setup.py info ./my_project") + + +async def main(): + """主函数""" + if len(sys.argv) < 3: + print_info_usage() + return + + command = sys.argv[1] + project_path = sys.argv[2] + + if command == "setup": + configs = create_sample_configs() + await setup_submodules(project_path, configs) + + elif command == "update": + await update_submodules(project_path) + + elif command == "info": + await show_submodule_info(project_path) + + else: + logger.info(f"未知命令: {command}") + print_info_usage() + + +if __name__ == "__main__": + asyncio.run(main())