diff --git a/.env.dev b/.env.dev deleted file mode 100644 index 015a950c..00000000 --- a/.env.dev +++ /dev/null @@ -1,89 +0,0 @@ -SUPERUSERS=[""] - -COMMAND_START=[""] - -SESSION_RUNNING_EXPRESSION="别急呀,小真寻要宕机了!QAQ" - -NICKNAME=["真寻", "小真寻", "绪山真寻", "小寻子"] - -SESSION_EXPIRE_TIMEOUT=00:00:30 - -ALCONNA_USE_COMMAND_START=True - -# 全局图片统一使用bytes发送，当真寻与协议端不在同一服务器上时为True -IMAGE_TO_BYTES = True - -# 回复消息时自称 -SELF_NICKNAME="小真寻" - -# 官bot appid:bot账号 -QBOT_ID_DATA = '{ - -}' - -# 数据库配置 -# 示例: "postgres://user:password@127.0.0.1:5432/database" -# 示例: "mysql://user:password@127.0.0.1:3306/database" -# 示例: "sqlite:data/db/zhenxun.db" 在data目录下建立db文件夹 -DB_URL = "" - -# NONE: 不使用缓存, MEMORY: 使用内存缓存, REDIS: 使用Redis缓存 -CACHE_MODE = NONE -# REDIS配置，使用REDIS替换Cache内存缓存 -# REDIS地址 -# REDIS_HOST = "127.0.0.1" -# REDIS端口 -# REDIS_PORT = 6379 -# REDIS密码 -# REDIS_PASSWORD = "" -# REDIS过期时间 -# REDIS_EXPIRE = 600 - -# 系统代理 -# SYSTEM_PROXY = "http://127.0.0.1:7890" - -PLATFORM_SUPERUSERS = ' - { - "qq": [""], - "dodo": [""] - } -' - -DRIVER=~fastapi+~httpx+~websockets - - -# LOG_LEVEL = DEBUG -# 服务器和端口 -HOST = 127.0.0.1 -PORT = 8080 - -# kook adapter toekn -# kaiheila_bots =[{"token": ""}] - -# # discode adapter -# DISCORD_BOTS=' -# [ -# { -# "token": "", -# "intent": { -# "guild_messages": true, -# "direct_messages": true -# }, -# "application_commands": {"*": ["*"]} -# } -# ] -# ' -# DISCORD_PROXY='' - -# # dodo adapter -# DODO_BOTS=' -# [ -# { -# "client_id": "", -# "token": "" -# } -# ] -# ' - -# application_commands的{"*": ["*"]}代表将全部应用命令注册为全局应用命令 -# {"admin": ["123", "456"]}则代表将admin命令注册为id是123、456服务器的局部命令，其余命令不注册 \ No newline at end of file diff --git a/zhenxun/builtin_plugins/__init__.py b/zhenxun/builtin_plugins/__init__.py index a5aa7a4b..825e23b1 100644 --- a/zhenxun/builtin_plugins/__init__.py +++ b/zhenxun/builtin_plugins/__init__.py @@ -17,7 +17,7 @@ from zhenxun.models.user_console import UserConsole from zhenxun.services.log import logger from zhenxun.utils.decorator.shop import shop_register from zhenxun.utils.manager.priority_manager import PriorityLifecycle -from zhenxun.utils.manager.resource_manager import ResourceManager +from zhenxun.utils.manager.zhenxun_repo_manager import ZhenxunRepoManager from zhenxun.utils.platform import PlatformUtils driver: Driver = nonebot.get_driver() @@ -85,7 +85,8 @@ from bag_users t1 @PriorityLifecycle.on_startup(priority=5) async def _(): - await ResourceManager.init_resources() + if not ZhenxunRepoManager.check_resources_exists(): + await ZhenxunRepoManager.resources_update(branch="test") """签到与用户的数据迁移""" if goods_list := await GoodsInfo.filter(uuid__isnull=True).all(): for goods in goods_list: diff --git a/zhenxun/builtin_plugins/auto_update/__init__.py b/zhenxun/builtin_plugins/auto_update/__init__.py index 0cee97ba..e29efa90 100644 --- a/zhenxun/builtin_plugins/auto_update/__init__.py +++ b/zhenxun/builtin_plugins/auto_update/__init__.py @@ -16,10 +16,6 @@ from nonebot_plugin_uninfo import Uninfo from zhenxun.configs.utils import PluginExtraData from zhenxun.services.log import logger from zhenxun.utils.enum import PluginType -from zhenxun.utils.manager.resource_manager import ( - DownloadResourceException, - ResourceManager, -) from zhenxun.utils.message import MessageUtils from ._data_source import UpdateManager @@ -68,7 +64,6 @@ _matcher = on_alconna( Option("-f|--force", action=store_true, help_text="强制更新"), Option("-s", Args["source?", ["git", "ali"]], help_text="更新源"), Option("-z|--zip", action=store_true, help_text="下载zip文件"), - Option("-t", Args["update_type?", ["git", "download"]], help_text="更新方式"), ), priority=1, block=True, @@ -86,39 +81,52 @@ async def _( force: Query[bool] = Query("force", False), source: Query[str] = Query("source", "ali"), zip: Query[bool] = Query("zip", False), - update_type: Query[str] = Query("update_type", "git"), ): result = "" await MessageUtils.build_message("正在进行检查更新...").send(reply_to=True) - if ver_type.result in {"main", "release"}: + ver_type_str = ver_type.result + source_str = source.result + if ver_type_str in {"main", "release"}: if not ver_type.available: - result = await UpdateManager.check_version() + result += await UpdateManager.check_version() logger.info("查看当前版本...", "检查更新", session=session) await MessageUtils.build_message(result).finish() try: - result = await UpdateManager.update( + result += await UpdateManager.update_zhenxun( bot, session.user.id, - ver_type.result, + ver_type_str, # type: ignore force.result, - source.result, + source_str, # type: ignore zip.result, - update_type.result, ) + await MessageUtils.build_message(result).finish(reply_to=True) except Exception as e: logger.error("版本更新失败...", "检查更新", session=session, e=e) await MessageUtils.build_message(f"更新版本失败...e: {e}").finish() elif ver_type.result == "webui": - result = await UpdateManager.update_webui(zip.result, source.result) + if zip.result: + source_str = None + try: + result += await UpdateManager.update_webui( + source_str, # type: ignore + "dist", + ) + except Exception as e: + logger.error("WebUI更新失败...", "检查更新", session=session, e=e) + result += "\nWebUI更新错误..." if resource.result or ver_type.result == "resource": try: - await ResourceManager.init_resources(True, zip.result, source.result) - result += "\n资源文件更新成功！" - except DownloadResourceException: - result += "\n资源更新下载失败..." + if zip.result: + source_str = None + result += await UpdateManager.update_resources( + source_str, # type: ignore + "main", + force.result, + ) except Exception as e: logger.error("资源更新下载失败...", "检查更新", session=session, e=e) - result += "\n资源更新未知错误..." + result += "\n资源更新错误..." if result: await MessageUtils.build_message(result.strip()).finish() await MessageUtils.build_message("更新版本失败...").finish() diff --git a/zhenxun/builtin_plugins/auto_update/_data_source.py b/zhenxun/builtin_plugins/auto_update/_data_source.py index eca520d3..62d2edab 100644 --- a/zhenxun/builtin_plugins/auto_update/_data_source.py +++ b/zhenxun/builtin_plugins/auto_update/_data_source.py @@ -1,167 +1,16 @@ -import os -from pathlib import Path -import shutil -import tarfile -import zipfile +from typing import Literal from nonebot.adapters import Bot -from nonebot.utils import run_sync -from zhenxun.configs.path_config import DATA_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.manager.zhenxun_repo_manager import ZhenxunRepoManager from zhenxun.utils.platform import PlatformUtils -from zhenxun.utils.repo_utils import AliyunRepoManager, GithubRepoManager -from .config import ( - BACKUP_PATH, - BASE_PATH, - BASE_PATH_STRING, - COMMAND, - DEFAULT_GITHUB_URL, - DOWNLOAD_GZ_FILE, - DOWNLOAD_ZIP_FILE, - GIT_GITHUB_URL, - GIT_WEBUI_UI_URL, - PYPROJECT_FILE, - PYPROJECT_FILE_STRING, - PYPROJECT_LOCK_FILE, - PYPROJECT_LOCK_FILE_STRING, - RELEASE_URL, - REPLACE_FOLDERS, - REQ_TXT_FILE, - REQ_TXT_FILE_STRING, - TMP_PATH, - VERSION_FILE, -) - - -@run_sync -def _file_handle(latest_version: str | None): - """文件移动操作 - - 参数: - latest_version: 版本号 - """ - BACKUP_PATH.mkdir(exist_ok=True, parents=True) - logger.debug("开始解压文件压缩包...", COMMAND) - download_file = DOWNLOAD_GZ_FILE - if DOWNLOAD_GZ_FILE.exists(): - tf = tarfile.open(DOWNLOAD_GZ_FILE) - else: - download_file = DOWNLOAD_ZIP_FILE - tf = zipfile.ZipFile(DOWNLOAD_ZIP_FILE) - tf.extractall(TMP_PATH) - logger.debug("解压文件压缩包完成...", COMMAND) - download_file_path = TMP_PATH / next( - x for x in os.listdir(TMP_PATH) if (TMP_PATH / x).is_dir() - ) - _pyproject = download_file_path / PYPROJECT_FILE_STRING - _lock_file = download_file_path / PYPROJECT_LOCK_FILE_STRING - _req_file = download_file_path / REQ_TXT_FILE_STRING - extract_path = download_file_path / BASE_PATH_STRING - target_path = BASE_PATH - if PYPROJECT_FILE.exists(): - logger.debug(f"移除备份文件: {PYPROJECT_FILE}", COMMAND) - shutil.move(PYPROJECT_FILE, BACKUP_PATH / PYPROJECT_FILE_STRING) - if PYPROJECT_LOCK_FILE.exists(): - logger.debug(f"移除备份文件: {PYPROJECT_LOCK_FILE}", COMMAND) - shutil.move(PYPROJECT_LOCK_FILE, BACKUP_PATH / PYPROJECT_LOCK_FILE_STRING) - if REQ_TXT_FILE.exists(): - logger.debug(f"移除备份文件: {REQ_TXT_FILE}", COMMAND) - shutil.move(REQ_TXT_FILE, BACKUP_PATH / REQ_TXT_FILE_STRING) - if _pyproject.exists(): - logger.debug("移动文件: pyproject.toml", COMMAND) - shutil.move(_pyproject, PYPROJECT_FILE) - if _lock_file.exists(): - logger.debug("移动文件: poetry.lock", COMMAND) - shutil.move(_lock_file, PYPROJECT_LOCK_FILE) - if _req_file.exists(): - logger.debug("移动文件: requirements.txt", COMMAND) - shutil.move(_req_file, REQ_TXT_FILE) - for folder in REPLACE_FOLDERS: - """移动指定文件夹""" - _dir = BASE_PATH / folder - _backup_dir = BACKUP_PATH / folder - if _backup_dir.exists(): - logger.debug(f"删除备份文件夹 {_backup_dir}", COMMAND) - shutil.rmtree(_backup_dir) - if _dir.exists(): - logger.debug(f"移动旧文件夹 {_dir}", COMMAND) - shutil.move(_dir, _backup_dir) - else: - logger.warning(f"文件夹 {_dir} 不存在，跳过删除", COMMAND) - for folder in 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}", COMMAND - ) - shutil.move(src_folder_path, dest_folder_path) - else: - logger.debug(f"源文件夹不存在: {src_folder_path}", COMMAND) - if tf: - tf.close() - if download_file.exists(): - logger.debug(f"删除下载文件: {download_file}", COMMAND) - download_file.unlink() - if extract_path.exists(): - logger.debug(f"删除解压文件夹: {extract_path}", COMMAND) - shutil.rmtree(extract_path) - if TMP_PATH.exists(): - shutil.rmtree(TMP_PATH) - if latest_version: - with open(VERSION_FILE, "w", encoding="utf8") as f: - f.write(f"__version__: {latest_version}") +from .config import LOG_COMMAND, REQUIREMENTS_FILE, VERSION_FILE class UpdateManager: - @classmethod - async def update_webui(cls, is_zip: bool, source: str) -> str: - from zhenxun.builtin_plugins.web_ui.public.data_source import ( - update_webui_assets, - ) - - WEBUI_PATH = DATA_PATH / "web_ui" / "public" - BACKUP_PATH = DATA_PATH / "web_ui" / "backup_public" - GIT_WEBUI_PATH = DATA_PATH / "web_ui" / "git_web_ui" - if WEBUI_PATH.exists(): - if BACKUP_PATH.exists(): - logger.debug(f"删除旧的备份webui文件夹 {BACKUP_PATH}", COMMAND) - shutil.rmtree(BACKUP_PATH) - WEBUI_PATH.rename(BACKUP_PATH) - try: - if is_zip: - await update_webui_assets() - logger.info("更新webui成功...", COMMAND) - else: - if source == "ali": - result = await AliyunRepoManager.update( - GIT_WEBUI_UI_URL, GIT_WEBUI_PATH, "dist", force=True - ) - else: - result = await GithubRepoManager.update( - GIT_WEBUI_UI_URL, GIT_WEBUI_PATH, "dist", force=True - ) - if not result.success: - return f"Webui更新失败...错误: {result.error_message}" - shutil.rmtree(WEBUI_PATH, ignore_errors=True) - shutil.copytree(GIT_WEBUI_PATH / "dist", WEBUI_PATH) - if BACKUP_PATH.exists(): - logger.debug(f"删除旧的webui文件夹 {BACKUP_PATH}", COMMAND) - shutil.rmtree(BACKUP_PATH) - return "Webui更新成功！" - except Exception as e: - logger.error("更新webui失败...", COMMAND, e=e) - if BACKUP_PATH.exists(): - logger.debug(f"恢复旧的webui文件夹 {BACKUP_PATH}", COMMAND) - BACKUP_PATH.rename(WEBUI_PATH) - raise e - @classmethod async def check_version(cls) -> str: """检查更新版本 @@ -170,71 +19,88 @@ class UpdateManager: str: 更新信息 """ cur_version = cls.__get_version() - data = await cls.__get_latest_data() - if not data: + release_data = await ZhenxunRepoManager.zhenxun_get_latest_releases_data() + if not release_data: return "检查更新获取版本失败..." return ( "检测到当前版本更新\n" f"当前版本：{cur_version}\n" - f"最新版本：{data.get('name')}\n" - f"创建日期：{data.get('created_at')}\n" - f"更新内容：\n{data.get('body')}" + f"最新版本：{release_data.get('name')}\n" + f"创建日期：{release_data.get('created_at')}\n" + f"更新内容：\n{release_data.get('body')}" ) @classmethod - async def __zip_update(cls, version_type: str): - logger.info("开始下载真寻最新版文件....", COMMAND) - cur_version = cls.__get_version() - url = None - new_version = None - repo_info = GithubUtils.parse_github_url(DEFAULT_GITHUB_URL) - if version_type in {"main"}: - repo_info.branch = version_type - new_version = await cls.__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 cls.__get_latest_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 TMP_PATH.exists(): - logger.debug(f"删除临时文件夹 {TMP_PATH}", COMMAND) - shutil.rmtree(TMP_PATH) - logger.debug( - f"开始更新版本：{cur_version} -> {new_version} | 下载链接：{url}", - COMMAND, + async def update_webui( + cls, + source: Literal["git", "ali"] | None, + branch: str = "main", + force: bool = False, + ): + """更新WebUI + + 参数: + source: 更新源 + branch: 分支 + force: 是否强制更新 + + 返回: + str: 返回消息 + """ + if not source: + await ZhenxunRepoManager.webui_zip_update() + return "WebUI更新完成!" + result = await ZhenxunRepoManager.webui_git_update( + source, + branch=branch, + force=force, ) - download_file = ( - DOWNLOAD_GZ_FILE if version_type == "release" else DOWNLOAD_ZIP_FILE + if not result.success: + logger.error(f"WebUI更新失败...错误: {result.error_message}", LOG_COMMAND) + return f"WebUI更新失败...错误: {result.error_message}" + return "WebUI更新完成!" + + @classmethod + async def update_resources( + cls, + source: Literal["git", "ali"] | None, + branch: str = "main", + force: bool = False, + ) -> str: + """更新资源 + + 参数: + source: 更新源 + branch: 分支 + force: 是否强制更新 + + 返回: + str: 返回消息 + """ + if not source: + await ZhenxunRepoManager.resources_zip_update() + return "真寻资源更新完成!" + result = await ZhenxunRepoManager.resources_git_update( + source, + branch=branch, + force=force, ) - if await AsyncHttpx.download_file(url, download_file, stream=True): - logger.debug("下载真寻最新版文件完成...", COMMAND) - await _file_handle(new_version) - result = "版本更新完成" - return ( - f"{result}\n" - f"版本: {cur_version} -> {new_version}\n" - "请重新启动真寻以完成更新!" + if not result.success: + logger.error( + f"真寻资源更新失败...错误: {result.error_message}", LOG_COMMAND ) - else: - logger.debug("下载真寻最新版文件失败...", COMMAND) - return "" + return f"真寻资源更新失败...错误: {result.error_message}" + return "真寻资源更新完成!" @classmethod - async def update( + async def update_zhenxun( cls, bot: Bot, user_id: str, - version_type: str, + version_type: Literal["main", "release"], force: bool, - source: str, + source: Literal["git", "ali"], zip: bool, - update_type: str, ) -> str: """更新操作 @@ -257,33 +123,38 @@ class UpdateManager: user_id, ) if zip: - return await cls.__zip_update(version_type) - elif source == "git": - result = await GithubRepoManager.update( - GIT_GITHUB_URL, - Path(), - use_git=update_type == "git", - force=force, + new_version = await ZhenxunRepoManager.zhenxun_zip_update(version_type) + await PlatformUtils.send_superuser( + bot, "真寻更新完成，开始安装依赖...", user_id + ) + await VirtualEnvPackageManager.install_requirement(REQUIREMENTS_FILE) + return ( + f"版本更新完成！\n版本: {cur_version} -> {new_version}\n" + "请重新启动真寻以完成更新!" ) else: - result = await AliyunRepoManager.update( - GIT_GITHUB_URL, - Path(), + result = await ZhenxunRepoManager.zhenxun_git_update( + source, + branch=version_type, force=force, ) - if not result.success: - return f"版本更新失败...错误: {result.error_message}" - await PlatformUtils.send_superuser( - bot, "真寻更新完成，开始安装依赖...", user_id - ) - await VirtualEnvPackageManager.install_requirement(REQ_TXT_FILE) - return ( - f"版本更新完成！\n" - f"版本: {cur_version} -> {result.new_version}\n" - f"变更文件个数: {len(result.changed_files)}" - f"{'' if source == 'git' else '（阿里云更新不支持查看变更文件）'}\n" - "请重新启动真寻以完成更新!" - ) + if not result.success: + logger.error( + f"真寻版本更新失败...错误: {result.error_message}", + LOG_COMMAND, + ) + return f"版本更新失败...错误: {result.error_message}" + await PlatformUtils.send_superuser( + bot, "真寻更新完成，开始安装依赖...", user_id + ) + await VirtualEnvPackageManager.install_requirement(REQUIREMENTS_FILE) + return ( + f"版本更新完成！\n" + f"版本: {cur_version} -> {result.new_version}\n" + f"变更文件个数: {len(result.changed_files)}" + f"{'' if source == 'git' else '（阿里云更新不支持查看变更文件）'}\n" + "请重新启动真寻以完成更新!" + ) @classmethod def __get_version(cls) -> str: @@ -297,40 +168,3 @@ class UpdateManager: if text := VERSION_FILE.open(encoding="utf8").readline(): _version = text.split(":")[-1].strip() return _version - - @classmethod - async def __get_latest_data(cls) -> dict: - """获取最新版本信息 - - 返回: - dict: 最新版本数据 - """ - for _ in range(3): - try: - res = await AsyncHttpx.get(RELEASE_URL) - if res.status_code == 200: - return res.json() - except TimeoutError: - pass - except Exception as e: - logger.error("检查更新真寻获取版本失败", e=e) - return {} - - @classmethod - async def __get_version_from_repo(cls, repo_info: RepoInfo) -> str: - """从指定分支获取版本号 - - 参数: - branch: 分支名称 - - 返回: - 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 "未知版本" diff --git a/zhenxun/builtin_plugins/auto_update/config.py b/zhenxun/builtin_plugins/auto_update/config.py index 17d5483e..85a44c9c 100644 --- a/zhenxun/builtin_plugins/auto_update/config.py +++ b/zhenxun/builtin_plugins/auto_update/config.py @@ -1,42 +1,7 @@ from pathlib import Path -from zhenxun.configs.path_config import TEMP_PATH +LOG_COMMAND = "AutoUpdate" -GIT_GITHUB_URL = "https://github.com/zhenxun-org/zhenxun_bot.git" +VERSION_FILE = Path() / "__version__" -DEFAULT_GITHUB_URL = "https://github.com/HibiKier/zhenxun_bot/tree/main" -RELEASE_URL = "https://api.github.com/repos/HibiKier/zhenxun_bot/releases/latest" - -GIT_WEBUI_UI_URL = "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 - -TMP_PATH = TEMP_PATH / "auto_update" - -BACKUP_PATH = Path() / "backup" - -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 - -REPLACE_FOLDERS = [ - "builtin_plugins", - "services", - "utils", - "models", - "configs", -] - -COMMAND = "检查更新" +REQUIREMENTS_FILE = Path() / "requirements.txt" diff --git a/zhenxun/utils/manager/resource_manager.py b/zhenxun/utils/manager/resource_manager.py deleted file mode 100644 index 01b849f9..00000000 --- a/zhenxun/utils/manager/resource_manager.py +++ /dev/null @@ -1,101 +0,0 @@ -import os -from pathlib import Path -import shutil -import zipfile - -from zhenxun.configs.path_config import FONT_PATH, TEMP_PATH -from zhenxun.services.log import logger -from zhenxun.utils.github_utils import GithubUtils -from zhenxun.utils.http_utils import AsyncHttpx -from zhenxun.utils.repo_utils import AliyunRepoManager, GithubRepoManager -from zhenxun.utils.repo_utils.utils import clean_git - -LOG_COMMAND = "ResourceManager" - - -class DownloadResourceException(Exception): - pass - - -class ResourceManager: - GITHUB_URL = "https://github.com/zhenxun-org/zhenxun-bot-resources/tree/main" - - RESOURCE_PATH = Path() / "resources" - - TMP_PATH = TEMP_PATH / "_resource_tmp" - - ZIP_FILE = TMP_PATH / "resources.zip" - - UNZIP_PATH = None - - @classmethod - async def init_resources( - cls, force: bool = False, is_zip: bool = False, git_source: str = "ali" - ): - if (FONT_PATH.exists() and os.listdir(FONT_PATH)) and not force: - return - 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() - cls.file_handle() - except Exception as e: - logger.error("获取resources资源包失败", LOG_COMMAND, e=e) - else: - if git_source == "ali": - await AliyunRepoManager.update(cls.GITHUB_URL, cls.RESOURCE_PATH) - else: - await GithubRepoManager.update(cls.GITHUB_URL, cls.RESOURCE_PATH) - cls.UNZIP_PATH = cls.TMP_PATH / "resources" - cls.file_handle() - if cls.TMP_PATH.exists(): - logger.debug("移除resources临时文件夹", LOG_COMMAND) - await clean_git(cls.TMP_PATH) - shutil.rmtree(cls.TMP_PATH) - - @classmethod - def file_handle(cls): - if not cls.UNZIP_PATH: - return - cls.__recursive_folder(cls.UNZIP_PATH, ".") - - @classmethod - def __recursive_folder(cls, dir: Path, parent_path: str): - for file in dir.iterdir(): - if file.is_dir(): - cls.__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) - - @classmethod - async def __download_resources(cls): - """获取resources文件夹""" - repo_info = GithubUtils.parse_github_url(cls.GITHUB_URL) - url = await repo_info.get_archive_download_urls() - logger.debug("开始下载resources资源包...", LOG_COMMAND) - if not await AsyncHttpx.download_file(url, cls.ZIP_FILE, stream=True): - logger.error( - "下载resources资源包失败，请尝试重启重新下载或前往 " - "https://github.com/zhenxun-org/zhenxun-bot-resources 手动下载..." - ) - raise DownloadResourceException("下载resources资源包失败...") - logger.debug("下载resources资源文件压缩包完成...", LOG_COMMAND) - tf = zipfile.ZipFile(cls.ZIP_FILE) - tf.extractall(cls.TMP_PATH) - logger.debug("解压文件压缩包完成...", LOG_COMMAND) - download_file_path = cls.TMP_PATH / next( - x for x in os.listdir(cls.TMP_PATH) if (cls.TMP_PATH / x).is_dir() - ) - cls.UNZIP_PATH = download_file_path / "resources" - if tf: - tf.close() diff --git a/zhenxun/utils/manager/zhenxun_repo_manager.py b/zhenxun/utils/manager/zhenxun_repo_manager.py index 2b63eb6e..3db37540 100644 --- a/zhenxun/utils/manager/zhenxun_repo_manager.py +++ b/zhenxun/utils/manager/zhenxun_repo_manager.py @@ -6,28 +6,24 @@ import os from pathlib import Path import shutil -import tarfile -from typing import ClassVar +from typing import ClassVar, Literal import zipfile -from zhenxun.configs.path_config import DATA_PATH, FONT_PATH, TEMP_PATH +import aiofiles + +from zhenxun.configs.path_config import DATA_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 +from zhenxun.utils.repo_utils.models import RepoUpdateResult +from zhenxun.utils.repo_utils.utils import check_git LOG_COMMAND = "ZhenxunRepoManager" -class DownloadException(Exception): +class ZhenxunUpdateException(Exception): """资源下载异常""" pass @@ -36,79 +32,73 @@ class DownloadException(Exception): class ZhenxunRepoConfig: """真寻仓库配置""" - # GitHub 仓库 URL + # Zhenxun Bot 相关配置 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" + ZHENXUN_BOT_DOWNLOAD_FILE_STRING = "zhenxun_bot.zip" + ZHENXUN_BOT_DOWNLOAD_FILE = TEMP_PATH / ZHENXUN_BOT_DOWNLOAD_FILE_STRING + ZHENXUN_BOT_UNZIP_PATH = TEMP_PATH / "zhenxun_bot" + ZHENXUN_BOT_CODE_PATH = Path() / "zhenxun" + ZHENXUN_BOT_RELEASES_API_URL = ( + "https://api.github.com/repos/HibiKier/zhenxun_bot/releases/latest" ) - - # 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 - + ZHENXUN_BOT_BACKUP_PATH = Path() / "backup" # 需要替换的文件夹 - REPLACE_FOLDERS: ClassVar[list[str]] = [ + ZHENXUN_BOT_UPDATE_FOLDERS: ClassVar[list[str]] = [ "builtin_plugins", "services", "utils", "models", "configs", ] + ZHENXUN_BOT_VERSION_FILE_STRING = "__version__" + ZHENXUN_BOT_VERSION_FILE = Path() / ZHENXUN_BOT_VERSION_FILE_STRING - # 日志标识 - COMMAND = "真寻仓库管理" + # 备份杂项 + BACKUP_FILES: ClassVar[list[str]] = [ + "pyproject.toml", + "poetry.lock", + "requirements.txt", + ".env.dev", + ".env.example", + ] + + # WEB UI 相关配置 + WEBUI_GIT = "https://github.com/HibiKier/zhenxun_bot_webui.git" + WEBUI_DIST_GITHUB_URL = "https://github.com/HibiKier/zhenxun_bot_webui/tree/dist" + WEBUI_DOWNLOAD_FILE_STRING = "webui_assets.zip" + WEBUI_DOWNLOAD_FILE = TEMP_PATH / WEBUI_DOWNLOAD_FILE_STRING + WEBUI_UNZIP_PATH = TEMP_PATH / "web_ui" + WEBUI_PATH = DATA_PATH / "web_ui" / "public" + WEBUI_BACKUP_PATH = DATA_PATH / "web_ui" / "backup_public" + + # 资源管理相关配置 + RESOURCE_GIT = "https://github.com/zhenxun-org/zhenxun-bot-resources.git" + RESOURCE_GITHUB_URL = ( + "https://github.com/zhenxun-org/zhenxun-bot-resources/tree/main" + ) + RESOURCE_ZIP_FILE_STRING = "resources.zip" + RESOURCE_ZIP_FILE = TEMP_PATH / RESOURCE_ZIP_FILE_STRING + RESOURCE_UNZIP_PATH = TEMP_PATH / "resources" + RESOURCE_PATH = Path() / "resources" + + REQUIREMENTS_FILE_STRING = "requirements.txt" + REQUIREMENTS_FILE = Path() / REQUIREMENTS_FILE_STRING -class ZhenxunRepoManager: +class ZhenxunRepoManagerClass: """真寻仓库管理器""" def __init__(self): self.config = ZhenxunRepoConfig() - # 初始化子模块管理器 - self.submodule_manager = SubmoduleManager(GithubRepoManager) def __clear_folder(self, folder_path: Path): + """ + 清空文件夹 + + 参数: + folder_path: 文件夹路径 + """ for filename in os.listdir(folder_path): file_path = folder_path / filename try: @@ -119,548 +109,347 @@ class ZhenxunRepoManager: except Exception as e: logger.warning(f"无法删除 {file_path}", LOG_COMMAND, e=e) - async def check_version(self) -> str: - """检查真寻更新版本 - - 返回: - str: 更新信息 + def __copy_files(self, src_path: Path, dest_path: Path, incremental: bool = False): """ - 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: 更新结果消息 + src_path: 源文件或文件夹路径 + dest_path: 目标文件或文件夹路径 + incremental: 是否增量复制 """ - cur_version = self._get_current_version() - await PlatformUtils.send_superuser( - bot, - f"检测真寻已更新，当前版本：{cur_version}\n开始更新...", - user_id, - ) + if src_path.is_file(): + shutil.copy(src_path, dest_path) + logger.debug(f"复制文件 {src_path} -> {dest_path}", LOG_COMMAND) + elif src_path.is_dir(): + for filename in os.listdir(src_path): + file_path = src_path / filename + dest_file = dest_path / filename + dest_file.parent.mkdir(exist_ok=True, parents=True) + if file_path.is_file(): + if dest_file.exists(): + dest_file.unlink() + shutil.copy(file_path, dest_file) + logger.debug(f"复制文件 {file_path} -> {dest_file}", LOG_COMMAND) + elif file_path.is_dir(): + if incremental: + self.__copy_files(file_path, dest_file, incremental=True) + else: + if dest_file.exists(): + shutil.rmtree(dest_file, True) + shutil.copytree(file_path, dest_file) + logger.debug( + f"复制文件夹 {file_path} -> {dest_file}", + LOG_COMMAND, + ) - 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, - ) + # ==================== Zhenxun Bot 相关方法 ==================== - 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: + async def zhenxun_get_version_from_repo(self) -> str: """从指定分支获取版本号 - 参数: - repo_info: 仓库信息 返回: str: 版本号 """ - version_url = await repo_info.get_raw_download_urls(path="__version__") + repo_info = GithubUtils.parse_github_url(self.config.ZHENXUN_BOT_GITHUB_URL) + version_url = await repo_info.get_raw_download_urls( + path=self.config.ZHENXUN_BOT_VERSION_FILE_STRING + ) 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) + logger.error(f"获取 {repo_info.branch} 分支版本失败", LOG_COMMAND, e=e) return "未知版本" + async def zhenxun_write_version_file(self, version: str): + """写入版本文件""" + async with aiofiles.open( + self.config.ZHENXUN_BOT_VERSION_FILE, "w", encoding="utf8" + ) as f: + await f.write(f"__version__: {version}") + + def __backup_zhenxun(self): + """备份真寻文件""" + for filename in os.listdir(self.config.ZHENXUN_BOT_CODE_PATH): + file_path = self.config.ZHENXUN_BOT_CODE_PATH / filename + if file_path.exists(): + self.__copy_files( + file_path, + self.config.ZHENXUN_BOT_BACKUP_PATH / filename, + True, + ) + for filename in self.config.BACKUP_FILES: + file_path = Path() / filename + if file_path.exists(): + self.__copy_files( + file_path, + self.config.ZHENXUN_BOT_BACKUP_PATH / filename, + ) + + async def zhenxun_get_latest_releases_data(self) -> dict: + """获取真寻releases最新版本信息 + + 返回: + dict: 最新版本数据 + """ + try: + res = await AsyncHttpx.get(self.config.ZHENXUN_BOT_RELEASES_API_URL) + if res.status_code == 200: + return res.json() + except Exception as e: + logger.error("检查更新真寻获取版本失败", LOG_COMMAND, e=e) + return {} + + async def zhenxun_download_zip(self, ver_type: Literal["main", "release"]) -> str: + """下载真寻最新版文件 + + 参数: + ver_type: 版本类型，main 为最新版，release 为最新release版 + + 返回: + str: 版本号 + """ + repo_info = GithubUtils.parse_github_url(self.config.ZHENXUN_BOT_GITHUB_URL) + if ver_type == "main": + download_url = await repo_info.get_archive_download_urls() + new_version = await self.zhenxun_get_version_from_repo() + else: + release_data = await self.zhenxun_get_latest_releases_data() + logger.debug(f"获取真寻RELEASES最新版本信息: {release_data}", LOG_COMMAND) + if not release_data: + raise ZhenxunUpdateException("获取真寻RELEASES最新版本失败...") + new_version = release_data.get("name", "") + download_url = await repo_info.get_release_source_download_urls_tgz( + new_version + ) + if not download_url: + raise ZhenxunUpdateException("获取真寻最新版文件下载链接失败...") + if self.config.ZHENXUN_BOT_DOWNLOAD_FILE.exists(): + self.config.ZHENXUN_BOT_DOWNLOAD_FILE.unlink() + if await AsyncHttpx.download_file( + download_url, self.config.ZHENXUN_BOT_DOWNLOAD_FILE, stream=True + ): + logger.debug("下载真寻最新版文件完成...", LOG_COMMAND) + else: + raise ZhenxunUpdateException("下载真寻最新版文件失败...") + return new_version + + async def zhenxun_unzip(self): + """解压真寻最新版文件""" + if not self.config.ZHENXUN_BOT_DOWNLOAD_FILE.exists(): + raise FileNotFoundError("真寻最新版文件不存在") + if self.config.ZHENXUN_BOT_UNZIP_PATH.exists(): + shutil.rmtree(self.config.ZHENXUN_BOT_UNZIP_PATH) + tf = None + try: + tf = zipfile.ZipFile(self.config.ZHENXUN_BOT_DOWNLOAD_FILE) + tf.extractall(self.config.ZHENXUN_BOT_UNZIP_PATH) + logger.debug("解压Zhenxun Bot文件压缩包完成!", LOG_COMMAND) + self.__backup_zhenxun() + for filename in self.config.BACKUP_FILES: + self.__copy_files( + self.config.ZHENXUN_BOT_UNZIP_PATH / filename, + Path() / filename, + ) + for folder in self.config.ZHENXUN_BOT_UPDATE_FOLDERS: + self.__copy_files( + self.config.ZHENXUN_BOT_UNZIP_PATH / folder, + self.config.ZHENXUN_BOT_CODE_PATH / folder, + ) + logger.debug("移动真寻更新文件完成!", LOG_COMMAND) + if self.config.ZHENXUN_BOT_DOWNLOAD_FILE.exists(): + self.config.ZHENXUN_BOT_DOWNLOAD_FILE.unlink() + if self.config.ZHENXUN_BOT_UNZIP_PATH.exists(): + shutil.rmtree(self.config.ZHENXUN_BOT_UNZIP_PATH) + except Exception as e: + logger.error("解压真寻最新版文件失败...", LOG_COMMAND, e=e) + raise + finally: + if tf: + tf.close() + + async def zhenxun_zip_update(self, ver_type: Literal["main", "release"]) -> str: + """使用zip更新真寻 + + 参数: + ver_type: 版本类型，main 为最新版，release 为最新release版 + + 返回: + str: 版本号 + """ + new_version = await self.zhenxun_download_zip(ver_type) + await self.zhenxun_unzip() + await self.zhenxun_write_version_file(new_version) + return new_version + + async def zhenxun_git_update( + self, source: Literal["git", "ali"], branch: str = "main", force: bool = False + ) -> RepoUpdateResult: + """使用git或阿里云更新真寻 + + 参数: + source: 更新源，git 为 git 更新，ali 为阿里云更新 + branch: 分支名称 + force: 是否强制更新 + """ + if source == "git": + return await GithubRepoManager.update_via_git( + self.config.ZHENXUN_BOT_GIT, + Path(), + branch=branch, + force=force, + ) + else: + return await AliyunRepoManager.update_via_git( + self.config.ZHENXUN_BOT_GIT, + Path(), + branch=branch, + force=force, + ) + + async def zhenxun_update( + self, + source: Literal["git", "ali"] = "ali", + branch: str = "main", + force: bool = False, + ver_type: Literal["main", "release"] = "main", + ): + """更新真寻 + + 参数: + source: 更新源，git 为 git 更新，ali 为阿里云更新 + branch: 分支名称 + force: 是否强制更新 + ver_type: 版本类型，main 为最新版，release 为最新release版 + """ + if await check_git(): + await self.zhenxun_git_update(source, branch, force) + logger.debug("使用git更新真寻!", LOG_COMMAND) + else: + await self.zhenxun_zip_update(ver_type) + logger.debug("使用zip更新真寻!", LOG_COMMAND) + + async def install_requirements(self): + """安装真寻依赖""" + await VirtualEnvPackageManager.install_requirement( + self.config.REQUIREMENTS_FILE + ) + # ==================== 资源管理相关方法 ==================== - 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) + def check_resources_exists(self) -> bool: + """检查资源文件是否存在 返回: - str: 操作结果 + bool: 是否存在 """ - if (FONT_PATH.exists() and os.listdir(FONT_PATH)) and not force: - return "资源文件已存在，跳过初始化" + if self.config.RESOURCE_PATH.exists(): + font_path = self.config.RESOURCE_PATH / "fonts" + if font_path.exists() and os.listdir(font_path): + return True + return False - 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): + async def resources_download_zip(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 + download_url = await GithubUtils.parse_github_url( + self.config.RESOURCE_GITHUB_URL + ).get_archive_download_urls() + logger.debug("开始下载resources资源包...", LOG_COMMAND) + if await AsyncHttpx.download_file( + download_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资源文件压缩包成功!", LOG_COMMAND) + else: + raise ZhenxunUpdateException("下载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: 操作结果 - """ + async def resources_unzip(self): + """解压资源文件""" + if not self.config.RESOURCE_ZIP_FILE.exists(): + raise FileNotFoundError("资源文件压缩包不存在") + if self.config.RESOURCE_UNZIP_PATH.exists(): + shutil.rmtree(self.config.RESOURCE_UNZIP_PATH) + tf = None 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 + tf = zipfile.ZipFile(self.config.RESOURCE_ZIP_FILE) + tf.extractall(self.config.RESOURCE_UNZIP_PATH) + logger.debug("解压文件压缩包完成...", LOG_COMMAND) + self.__copy_files( + self.config.RESOURCE_UNZIP_PATH, self.config.RESOURCE_PATH, True ) - - if success: - return "子模块初始化成功！" - else: - return "子模块初始化失败！" - except Exception as e: - logger.error("子模块初始化失败", self.config.COMMAND, e=e) - return f"子模块初始化失败: {e}" + logger.error("解压资源文件失败...", LOG_COMMAND, e=e) + raise + finally: + if tf: + tf.close() - async def update_submodules(self) -> str: - """更新子模块 + async def resources_zip_update(self): + """使用zip更新资源文件""" + await self.resources_download_zip() + await self.resources_unzip() - 返回: - str: 操作结果 + async def resources_git_update( + self, source: Literal["git", "ali"], branch: str = "main", force: bool = False + ) -> RepoUpdateResult: + """使用git或阿里云更新资源文件 + + 参数: + source: 更新源，git 为 git 更新，ali 为阿里云更新 + branch: 分支名称 + force: 是否强制更新 """ - 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 + if source == "git": + return await GithubRepoManager.update_via_git( + self.config.RESOURCE_GIT, + self.config.RESOURCE_PATH, + branch=branch, + force=force, + ) + else: + return await AliyunRepoManager.update_via_git( + self.config.RESOURCE_GIT, + self.config.RESOURCE_PATH, + branch=branch, + force=force, ) - success_count = sum(1 for result in results if result.success) - total_count = len(results) + async def resources_update( + self, + source: Literal["git", "ali"] = "ali", + branch: str = "main", + force: bool = False, + ): + """更新资源文件 - 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: 子模块信息 + 参数: + source: 更新源，git 为 git 更新，ali 为阿里云更新 + branch: 分支名称 + force: 是否强制更新 """ - 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}" + if await check_git(): + await self.resources_git_update(source, branch, force) + logger.debug("使用git更新资源文件!", LOG_COMMAND) + else: + await self.resources_zip_update() + logger.debug("使用zip更新资源文件!", LOG_COMMAND) # ==================== Web UI 管理相关方法 ==================== - async def webui_download_zip(self) -> str: + async def webui_download_zip(self): """下载 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 + download_url, self.config.WEBUI_DOWNLOAD_FILE, follow_redirects=True ): logger.info("下载 WEBUI_ASSETS 成功!", LOG_COMMAND) - raise DownloadException("下载 WEBUI_ASSETS 失败", LOG_COMMAND) + else: + raise ZhenxunUpdateException("下载 WEBUI_ASSETS 失败", LOG_COMMAND) def __backup_webui(self): """备份 WEBUI_ASSERT 资源""" @@ -673,15 +462,82 @@ class ZhenxunRepoManager: 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 + async def webui_unzip(self): + """解压 WEBUI_ASSETS 资源 - # 参数: - # is_zip: 是否下载 ZIP 文件 - # source: 更新源 (git/ali) + 返回: + str: 更新结果 + """ + if not self.config.WEBUI_DOWNLOAD_FILE.exists(): + raise FileNotFoundError("备份webui文件夹不存在") + tf = None + try: + self.__backup_webui() + self.__clear_folder(self.config.WEBUI_PATH) + tf = zipfile.ZipFile(self.config.WEBUI_DOWNLOAD_FILE) + tf.extractall(self.config.WEBUI_UNZIP_PATH) + logger.debug("解压文件压缩包完成...", LOG_COMMAND) + self.__copy_files(self.config.WEBUI_UNZIP_PATH, self.config.WEBUI_PATH) + logger.debug("复制 WEBUI_ASSETS 成功!", LOG_COMMAND) + shutil.rmtree(self.config.WEBUI_UNZIP_PATH, ignore_errors=True) + except Exception as e: + if self.config.WEBUI_BACKUP_PATH.exists(): + self.__copy_files(self.config.WEBUI_BACKUP_PATH, self.config.WEBUI_PATH) + logger.debug("恢复备份 WEBUI_ASSETS 成功!", LOG_COMMAND) + shutil.rmtree(self.config.WEBUI_BACKUP_PATH, ignore_errors=True) + logger.error("Web UI 更新失败", LOG_COMMAND, e=e) + raise + finally: + if tf: + tf.close() - # 返回: - # str: 更新结果 - # """ - # self.__backup_webui() - # self.__clear_folder(self.config.WEBUI_PATH) + async def webui_zip_update(self): + """使用zip更新 Web UI""" + await self.webui_download_zip() + await self.webui_unzip() + + async def webui_git_update( + self, source: Literal["git", "ali"], branch: str = "main", force: bool = False + ) -> RepoUpdateResult: + """使用git或阿里云更新 Web UI + + 参数: + source: 更新源，git 为 git 更新，ali 为阿里云更新 + branch: 分支名称 + force: 是否强制更新 + """ + if source == "git": + return await GithubRepoManager.update_via_git( + self.config.WEBUI_GIT, + self.config.WEBUI_PATH, + branch=branch, + force=force, + ) + else: + return await AliyunRepoManager.update_via_git( + self.config.WEBUI_GIT, + self.config.WEBUI_PATH, + branch=branch, + force=force, + ) + + async def webui_update( + self, + source: Literal["git", "ali"] = "ali", + branch: str = "main", + force: bool = False, + ): + """更新 Web UI + + 参数: + source: 更新源，git 为 git 更新，ali 为阿里云更新 + """ + if await check_git(): + await self.webui_git_update(source, branch, force) + logger.debug("使用git更新Web UI!", LOG_COMMAND) + else: + await self.webui_zip_update() + logger.debug("使用zip更新Web UI!", LOG_COMMAND) + + +ZhenxunRepoManager = ZhenxunRepoManagerClass() diff --git a/zhenxun/utils/repo_utils/__init__.py b/zhenxun/utils/repo_utils/__init__.py index 39769d1d..f37ccd26 100644 --- a/zhenxun/utils/repo_utils/__init__.py +++ b/zhenxun/utils/repo_utils/__init__.py @@ -24,9 +24,6 @@ from .models import ( RepoFileInfo, RepoType, RepoUpdateResult, - SubmoduleConfig, - SubmoduleInfo, - SubmoduleUpdateResult, ) from .utils import check_git, filter_files, glob_to_regex, run_git_command @@ -56,9 +53,6 @@ __all__ = [ "RepoType", "RepoUpdateError", "RepoUpdateResult", - "SubmoduleConfig", - "SubmoduleInfo", - "SubmoduleUpdateResult", "check_git", "filter_files", "glob_to_regex", diff --git a/zhenxun/utils/repo_utils/base_manager.py b/zhenxun/utils/repo_utils/base_manager.py index c3b77803..aa69dff7 100644 --- a/zhenxun/utils/repo_utils/base_manager.py +++ b/zhenxun/utils/repo_utils/base_manager.py @@ -283,10 +283,11 @@ class BaseRepoManager(ABC): return result # 如果目录存在，检查是否是Git仓库 - success, _, _ = await run_git_command( - "rev-parse --is-inside-work-tree", cwd=local_path - ) - if not success: + # 首先检查目录本身是否有.git文件夹 + git_dir = local_path / ".git" + is_git_repo = git_dir.exists() and git_dir.is_dir() + + if not is_git_repo: # 如果不是Git仓库，尝试初始化它 logger.info(f"目录 {local_path} 不是Git仓库，尝试初始化", LOG_COMMAND) init_success, _, init_stderr = await run_git_command( @@ -338,7 +339,7 @@ class BaseRepoManager(ABC): ) # 获取远程更新 - logger.info("获取远程更新", LOG_COMMAND) + logger.info(f"获取远程更新: {repo_url}", LOG_COMMAND) success, _, stderr = await run_git_command("fetch origin", cwd=local_path) if not success: return RepoUpdateResult( @@ -373,7 +374,7 @@ class BaseRepoManager(ABC): ) # 拉取最新代码 - logger.info("拉取最新代码", LOG_COMMAND) + logger.info(f"拉取最新代码: {repo_url}", LOG_COMMAND) pull_cmd = f"pull origin {branch}" if force: pull_cmd = f"fetch --all && git reset --hard origin/{branch}" diff --git a/zhenxun/utils/repo_utils/github_manager.py b/zhenxun/utils/repo_utils/github_manager.py index 01c13790..462c2723 100644 --- a/zhenxun/utils/repo_utils/github_manager.py +++ b/zhenxun/utils/repo_utils/github_manager.py @@ -29,11 +29,7 @@ from .models import ( RepoFileInfo, RepoType, RepoUpdateResult, - SubmoduleConfig, - SubmoduleInfo, - SubmoduleUpdateResult, ) -from .submodule_manager import SubmoduleManager class GithubManager(BaseRepoManager): @@ -47,7 +43,6 @@ class GithubManager(BaseRepoManager): config: 配置，如果为None则使用默认配置 """ super().__init__(config) - self.submodule_manager = SubmoduleManager(self) async def update_repo( self, @@ -529,158 +524,3 @@ 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 d3ec393f..170e60f3 100644 --- a/zhenxun/utils/repo_utils/models.py +++ b/zhenxun/utils/repo_utils/models.py @@ -15,62 +15,6 @@ 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: """仓库文件信息""" @@ -123,8 +67,6 @@ 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 deleted file mode 100644 index c9927e92..00000000 --- a/zhenxun/utils/repo_utils/submodule_manager.py +++ /dev/null @@ -1,408 +0,0 @@ -""" -子模块管理工具 -""" - -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 deleted file mode 100644 index 22bef33e..00000000 --- a/zhenxun/utils/repo_utils/submodule_setup.py +++ /dev/null @@ -1,210 +0,0 @@ -#!/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()) diff --git a/zhenxun/utils/repo_utils/utils.py b/zhenxun/utils/repo_utils/utils.py index 194e021e..7aceb231 100644 --- a/zhenxun/utils/repo_utils/utils.py +++ b/zhenxun/utils/repo_utils/utils.py @@ -57,11 +57,13 @@ async def run_git_command( """ try: full_command = f"git {command}" + # 将Path对象转换为字符串 + cwd_str = str(cwd) if cwd else None process = await asyncio.create_subprocess_shell( full_command, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, - cwd=cwd, + cwd=cwd_str, ) stdout_bytes, stderr_bytes = await process.communicate()