🐛 fix: 修复插件更新检查逻辑

Author: Kaguya233qwq

pyproject.toml

@@ -27,6 +27,7 @@ dependencies = [
     "nonebot-adapter-onebot>=2.4.6",
     "nonebot-plugin-apscheduler>=0.5.0",
     "nonebot2>=2.4.2",
+    "packaging>=25.0",
     "setuptools>=80.9.0",
 ]
 

src/nonebot_plugin_sky/tools/update.py

@@ -1,7 +1,6 @@
 import asyncio
-from pathlib import Path
 import sys
-from typing import Union
+from pathlib import Path
 
 import httpx
 import logging
@@ -10,72 +9,62 @@
 from nonebot import logger
 from nonebot.matcher import Matcher
 
-logging.captureWarnings(True)  # 去掉建议使用SSL验证的显示
+from packaging.version import parse as parse_version, Version
+
+logging.captureWarnings(True)
 
 try:
     __version__ = metadata.version("nonebot-plugin-sky")
 except metadata.PackageNotFoundError:
     __version__ = "0.0.0-dev"
 
+CURRENT_PLUGIN_VERSION: Version = parse_version(__version__)
 
-async def get_datapack_ver() -> str | None:
-    """
-    检查本地数据包版本
-    """
 
-    ver = Path("SkyDataPack/version")
-    if not ver.exists():
-        return
-    with open(ver, "r", encoding="utf-8") as f:
-        return f.read()
+async def get_datapack_ver() -> str | None:
+    """检查本地数据包版本"""
+    ver_path = Path("SkyDataPack/version")
+    if not ver_path.exists():
+        return None
+    try:
+        import aiofiles
 
+        async with aiofiles.open(ver_path, "r", encoding="utf-8") as f:
+            return (await f.read()).strip()
+    except (ImportError, IOError):
+        with open(ver_path, "r", encoding="utf-8") as f:
+            return f.read().strip()
 
-async def check_plugin_latest() -> str:
-    """
-    通过 PyPI API 检查插件的最新版本
-    """
 
+async def check_plugin_latest() -> Version | None:
+    """通过 PyPI API 检查插件的最新版本，返回 Version 对象"""
     pypi_url = "https://pypi.org/pypi/nonebot-plugin-sky/json"
-
+    pypi_mirror_url = "https://pypi.tuna.tsinghua.edu.cn/pypi/nonebot-plugin-sky/json"
     headers = {"User-Agent": "nonebot_plugin_sky_update_checker/1.0"}
 
-    try:
-        async with httpx.AsyncClient(timeout=10) as client:
-            # 优先尝试官方源，如果失败则尝试镜像源
+    async with httpx.AsyncClient(timeout=10) as client:
+        for url in [pypi_url, pypi_mirror_url]:
             try:
-                res = await client.get(url=pypi_url, headers=headers)
-                res.raise_for_status()
-            except httpx.RequestError:
-                logger.warning("访问 PyPI 官方源失败，正在尝试镜像源...")
-                pypi_mirror_url = (
-                    "https://pypi.tuna.tsinghua.edu.cn/pypi/nonebot-plugin-sky/json"
-                )
-                res = await client.get(url=pypi_mirror_url, headers=headers)
+                res = await client.get(url=url, headers=headers)
                 res.raise_for_status()
-                return "网络错误"
-            data = res.json()
-            latest_version = data.get("info", {}).get("version")
-
-            if latest_version:
-                return str(latest_version)
-            else:
-                return ""
-
-    except httpx.HTTPStatusError:
-        return ""
-    except Exception:
-        return ""
-
-
-async def check_datapack_latest() -> str:
-    """
-    通过 Gitee API 检查最新发布的数据包版本，稳定且快速。
-    """
-
+                data = res.json()
+                latest_version_str = data.get("info", {}).get("version")
+                if latest_version_str:
+                    return parse_version(latest_version_str)
+                return None
+            except httpx.RequestError as e:
+                logger.warning(f"访问 PyPI 源 {url} 失败: {e}")
+            except (KeyError, TypeError, Exception) as e:
+                logger.error(f"解析 PyPI 响应失败: {e}")
+                return None
+    return None
+
+
+async def check_datapack_latest() -> Version | None:
+    """通过 Gitee API 检查最新数据包版本，返回 Version 对象"""
     gitee_api_url = (
         "https://gitee.com/api/v5/repos/Kaguyaaa/nonebot_plugin_sky/releases/latest"
     )
-
     headers = {"User-Agent": "nonebot_plugin_sky_datapack_checker/1.0"}
 
     try:
@@ -85,94 +74,79 @@ async def check_datapack_latest() -> str:
             data = res.json()
             tag_name = data.get("tag_name")
             if not tag_name:
-                return ""
+                return None
+
             version_prefix = "SkyDataPack-v"
             if tag_name.startswith(version_prefix):
-                version = tag_name.replace(version_prefix, "", 1)
-                return version
-            else:
-                logger.warning(f"Gitee 最新 Release 的 tag 格式不符合预期: {tag_name}")
-                return ""
+                version_str = tag_name.replace(version_prefix, "", 1)
+                return parse_version(version_str)
 
-    except httpx.HTTPStatusError as e:
-        if e.response.status_code == 404:
-            logger.info("Gitee 仓库中没有找到任何 Release。")
-        else:
-            logger.error(f"访问 Gitee API 时发生HTTP错误: {e}")
-        return ""
+            logger.warning(f"Gitee 最新 Release 的 tag 格式不符合预期: {tag_name}")
+            return None
     except Exception as e:
         logger.error(f"获取数据包更新信息失败: {e}")
-        return ""
+        return None
 
 
-async def upgrade_plugin() -> Union[bool, None]:
-    latest = await check_plugin_latest()
-    if latest:
-        if latest == __version__:
-            logger.info("已是最新版本，无需更新")
-            return False
-        else:
-            logger.info("发现新版本，开始下载更新")
+async def upgrade_plugin() -> bool:
+    """检查插件是否有新版本可供升级"""
+    latest_version = await check_plugin_latest()
+    if latest_version:
+        if latest_version > CURRENT_PLUGIN_VERSION:
+            logger.info(f"发现新版本: {CURRENT_PLUGIN_VERSION} -> {latest_version}")
             return True
-    else:
-        return None
+    return False
 
 
 async def check_handle(matcher: Matcher):
-    await matcher.send("正在检查更新，请稍候")
+    await matcher.send("正在检查更新，请稍候...")
     results = ""
-    plugin_latest = await check_plugin_latest()
 
+    # 检查插件更新
+    plugin_latest = await check_plugin_latest()
     if plugin_latest:
-        if plugin_latest == __version__:
-            results += "-当前插件已是最新版本(v%s)\n" % __version__
-            logger.info("当前插件已是最新版本(v%s)" % __version__)
+        if plugin_latest > CURRENT_PLUGIN_VERSION:
+            results += f"-[New] 发现新版本插件：v{CURRENT_PLUGIN_VERSION} -> v{plugin_latest}\n"
+        elif plugin_latest < CURRENT_PLUGIN_VERSION:
+            results += f"-[Info] 当前插件版本 v{CURRENT_PLUGIN_VERSION} 高于 PyPI 最新版 v{plugin_latest}\n"
         else:
-            results += f"-[New]发现新版本插件：v{__version__} -> v{plugin_latest}\n"
-            logger.info("发现新版本插件：v%s" % plugin_latest)
+            results += f"- 当前插件已是最新版本 (v{CURRENT_PLUGIN_VERSION})\n"
     else:
-        results += "-[Error]插件检查更新失败，请稍后重试"
+        results += "-[Error] 插件检查更新失败，请稍后重试\n"
 
+    # 检查数据包更新
     datapack_latest = await check_datapack_latest()
-    datapack_current = await get_datapack_ver()
-    datapack_current = "1.0.0"
-    if not datapack_current:
-        results += "-数据包未安装"
-        logger.info("数据包未安装")
+    datapack_current_str = await get_datapack_ver()
+
+    if not datapack_current_str:
+        results += "- 数据包未安装"
     else:
+        datapack_current = parse_version(datapack_current_str)
         if datapack_latest:
-            if datapack_latest == datapack_current:
-                results += "-本地数据包已是最新版本(v%s)" % datapack_current
-                logger.info("本地数据包已是最新版本(v%s)" % datapack_current)
+            if datapack_latest > datapack_current:
+                results += f"-[New] 发现新版本数据包：v{datapack_current} -> v{datapack_latest}"
+            elif datapack_latest < datapack_current:
+                results += f"-[Info] 当前数据包版本 v{datapack_current} 高于 Gitee 最新版 v{datapack_latest}"
             else:
-                results += (
-                    f"-[New]发现新版本数据包：v{datapack_current} -> v{datapack_latest}"
-                )
-                logger.warning("发现新版本数据包：v%s" % datapack_latest)
+                results += f"- 本地数据包已是最新版本 (v{datapack_current})"
         else:
-            results += "-[Error]数据包检查更新失败，请稍后重试"
+            results += "-[Error] 数据包检查更新失败，请稍后重试"
 
-    await matcher.send(results)
+    await matcher.send(results.strip())
 
 
 async def upgrade_handle(matcher: Matcher):
     await matcher.send("正在检查插件更新...")
 
     has_update = await upgrade_plugin()
 
-    if has_update is None:
-        await matcher.send("检查更新失败，请检查网络或稍后再试。")
-        logger.error("检查更新失败")
-        return
-
     if not has_update:
-        await matcher.send("插件已是最新版本，无需更新。")
+        await matcher.send("插件已是最新版本或检查更新失败，无需更新。")
         return
 
     await matcher.send("发现新版本，正在尝试更新，请稍候...")
 
     python_executable = sys.executable
-
     command = [
         python_executable,
         "-m",
@@ -191,16 +165,11 @@ async def upgrade_handle(matcher: Matcher):
         logger.info("插件更新成功。")
         log_output = stdout.decode("utf-8", errors="ignore")
         logger.debug(f"Pip output:\n{log_output}")
-        await matcher.send("插件更新成功！\n请手动重启 NoneBot 以使新版本生效。")
+        await matcher.send("✅ 插件更新成功！\n请手动重启 NoneBot 以使新版本生效。")
     else:
         error_message = stderr.decode("utf-8", errors="ignore")
         logger.error(f"插件更新失败，返回码: {process.returncode}")
         logger.error(f"Pip 错误信息:\n{error_message}")
-
-        # 将关键错误信息发送给用户
         await matcher.send(
-            f"插件更新失败！\n错误信息: `{error_message[-200:]}`\n请检查日志获取详细信息。"
+            f"❌ 插件更新失败！\n错误信息: `{error_message[-200:]}`\n请检查日志获取详细信息。"
         )
-
-
-__all__ = ("check_handle", "upgrade_handle", "check_datapack_latest")