feat: 适配器和驱动器添加版本号和时间

CHANGELOG.md

@@ -10,6 +10,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/lang/zh-CN/
 ### Added
 
 - 插件测试中使用 uv 来控制 Python 版本并与本体共享依赖
+- 适配器和驱动器添加版本号和时间
 
 ## [4.1.4] - 2024-12-08
 

src/plugins/github/plugins/publish/validation.py

@@ -9,7 +9,7 @@
 from src.plugins.github.models import AuthorInfo
 from src.plugins.github.utils import extract_issue_info_from_issue
 from src.providers.docker_test import DockerPluginTest, Metadata
-from src.providers.utils import get_latest_version, load_json_from_file
+from src.providers.utils import get_pypi_version, load_json_from_file
 from src.providers.validation import PublishType, ValidationDict, validate_info
 
 from .constants import (
@@ -101,7 +101,7 @@ async def validate_plugin_info_from_issue(
         raw_data.update(metadata)
 
         # 跳过测试的时候只能从 pypi 获取版本号
-        raw_data["version"] = get_latest_version(project_link)
+        raw_data["version"] = get_pypi_version(project_link)
         raw_data["load"] = False
         raw_data["test_output"] = "插件未进行测试"
         raw_data["metadata"] = bool(metadata)

src/providers/models.py

@@ -8,6 +8,8 @@
 
 from src.providers.constants import BOT_KEY_TEMPLATE, PYPI_KEY_TEMPLATE
 from src.providers.docker_test import Metadata
+from src.providers.utils import get_author_name, get_pypi_upload_time, get_pypi_version
+from src.providers.validation import validate_info
 from src.providers.validation.models import (
     AdapterPublishInfo,
     BotPublishInfo,
@@ -61,6 +63,21 @@ def from_publish_info(cls, publish_info: AdapterPublishInfo) -> Self:
             is_official=publish_info.is_official,
         )
 
+    def to_registry(self) -> "RegistryAdapter":
+        """将仓库数据转换为注册表数据
+
+        将获取 author 信息，重新验证数据
+        """
+        author = get_author_name(self.author_id)
+        result = validate_info(
+            PublishType.ADAPTER,
+            {**self.model_dump(), "author": author},
+            [],
+        )
+        if result.info is None or not isinstance(result.info, AdapterPublishInfo):
+            raise ValueError(f"数据验证失败: {result.errors}")
+        return RegistryAdapter.from_publish_info(result.info)
+
 
 class StoreBot(BaseModel):
     """NoneBot 仓库中的机器人数据"""
@@ -87,6 +104,21 @@ def from_publish_info(cls, publish_info: BotPublishInfo) -> Self:
             is_official=publish_info.is_official,
         )
 
+    def to_registry(self) -> "RegistryBot":
+        """将仓库数据转换为注册表数据
+
+        将获取 author 信息，重新验证数据
+        """
+        author = get_author_name(self.author_id)
+        result = validate_info(
+            PublishType.BOT,
+            {**self.model_dump(), "author": author},
+            [],
+        )
+        if result.info is None or not isinstance(result.info, BotPublishInfo):
+            raise ValueError(f"数据验证失败: {result.errors}")
+        return RegistryBot.from_publish_info(result.info)
+
 
 class StoreDriver(BaseModel):
     """NoneBot 仓库中的驱动数据"""
@@ -119,6 +151,21 @@ def from_publish_info(cls, publish_info: DriverPublishInfo) -> Self:
             is_official=publish_info.is_official,
         )
 
+    def to_registry(self) -> "RegistryDriver":
+        """将仓库数据转换为注册表数据
+
+        将获取 author 信息，重新验证数据
+        """
+        author = get_author_name(self.author_id)
+        result = validate_info(
+            PublishType.DRIVER,
+            {**self.model_dump(), "author": author},
+            [],
+        )
+        if result.info is None or not isinstance(result.info, DriverPublishInfo):
+            raise ValueError(f"数据验证失败: {result.errors}")
+        return RegistryDriver.from_publish_info(result.info)
+
 
 class StorePlugin(BaseModel):
     """NoneBot 仓库中的插件数据"""
@@ -177,6 +224,8 @@ class RegistryAdapter(BaseModel):
     homepage: str
     tags: list[Tag]
     is_official: bool
+    time: str
+    version: str
 
     @property
     def key(self):
@@ -195,8 +244,21 @@ def from_publish_info(cls, publish_info: AdapterPublishInfo) -> Self:
             homepage=publish_info.homepage,
             tags=[Tag(label=tag.label, color=tag.color) for tag in publish_info.tags],
             is_official=publish_info.is_official,
+            time=publish_info.time,
+            version=publish_info.version,
         )
 
+    def update(self, store: StoreAdapter) -> "RegistryAdapter":
+        """根据商店数据更新注册表数据"""
+        version = get_pypi_version(self.project_link)
+        time = get_pypi_upload_time(self.project_link)
+
+        data = self.model_dump()
+        data.update(store.model_dump())
+        data.update(version=version, time=time)
+
+        return RegistryAdapter(**data)
+
 
 class RegistryBot(BaseModel):
     """NoneBot 商店机器人数据"""
@@ -223,6 +285,12 @@ def from_publish_info(cls, publish_info: BotPublishInfo) -> Self:
             is_official=publish_info.is_official,
         )
 
+    def update(self, store: StoreBot) -> "RegistryBot":
+        """根据商店数据更新注册表数据"""
+        data = self.model_dump()
+        data.update(store.model_dump())
+        return RegistryBot(**data)
+
 
 class RegistryDriver(BaseModel):
     """NoneBot 商店驱动数据"""
@@ -235,6 +303,8 @@ class RegistryDriver(BaseModel):
     homepage: str
     tags: list[Tag]
     is_official: bool
+    time: str
+    version: str
 
     @property
     def key(self):
@@ -253,8 +323,26 @@ def from_publish_info(cls, publish_info: DriverPublishInfo) -> Self:
             homepage=publish_info.homepage,
             tags=[Tag(label=tag.label, color=tag.color) for tag in publish_info.tags],
             is_official=publish_info.is_official,
+            time=publish_info.time,
+            version=publish_info.version,
         )
 
+    def update(self, store: StoreDriver) -> "RegistryDriver":
+        """根据商店数据更新注册表数据"""
+        # ~none 和 ~fastapi 驱动器的项目名一个是空字符串，一个是 nonebot2[fastapi]
+        # 上传时间和版本号均以 nonebot2 为准
+        project_link = self.project_link
+        if project_link == "" or project_link.startswith("nonebot2["):
+            project_link = "nonebot2"
+        version = get_pypi_version(project_link)
+        time = get_pypi_upload_time(project_link)
+
+        data = self.model_dump()
+        data.update(store.model_dump())
+        data.update(version=version, time=time)
+
+        return RegistryDriver(**data)
+
 
 class RegistryPlugin(BaseModel):
     """NoneBot 商店插件数据"""
@@ -282,9 +370,6 @@ def key(self):
 
     @classmethod
     def from_publish_info(cls, publish_info: PluginPublishInfo) -> Self:
-        if publish_info.time is None:
-            raise ValueError("上传时间不能为空")
-
         return cls(
             module_name=publish_info.module_name,
             project_link=publish_info.project_link,
@@ -312,6 +397,13 @@ def metadata(self) -> dict[str, Any]:
             "supported_adapters": self.supported_adapters,
         }
 
+    def update(self, store: StorePlugin) -> "RegistryPlugin":
+        """根据商店数据更新注册表数据"""
+        # TODO: 如果 author_id 变化，应该重新获取 author
+        data = self.model_dump()
+        data.update(store.model_dump())
+        return RegistryPlugin(**data)
+
 
 RegistryModels: TypeAlias = (
     RegistryAdapter | RegistryBot | RegistryDriver | RegistryPlugin

src/providers/store_test/store.py

@@ -30,10 +30,9 @@
 from src.providers.utils import (
     add_step_summary,
     dump_json,
-    get_latest_version,
+    get_pypi_version,
     load_json_from_web,
 )
-from src.providers.validation.utils import get_author_name
 
 from .constants import (
     ADAPTERS_PATH,
@@ -134,7 +133,7 @@ def should_skip(self, key: str, force: bool = False) -> bool:
 
         # 如果插件为最新版本，则跳过测试
         try:
-            latest_version = get_latest_version(previous_plugin.project_link)
+            latest_version = get_pypi_version(previous_plugin.project_link)
         except ValueError as e:
             logger.warning(f"插件 {key} 获取最新版本失败：{e}，跳过测试")
             return True
@@ -316,45 +315,55 @@ async def sync_store(self):
         以商店数据为准，更新商店数据到仓库中，如果仓库中不存在则获取用户名后存储
         """
         for key in self._store_adapters:
-            if key not in self._previous_adapters:
-                author = get_author_name(self._store_adapters[key].author_id)
-                self._previous_adapters[key] = RegistryAdapter(
-                    **self._store_adapters[key].model_dump(), author=author
-                )
-            else:
-                self._previous_adapters[key] = RegistryAdapter(
-                    **self._store_adapters[key].model_dump(),
-                    author=self._previous_adapters[key].author,
-                )
+            try:
+                if key in self._previous_adapters:
+                    new_adapter = self._previous_adapters[key].update(
+                        self._store_adapters[key]
+                    )
+                else:
+                    new_adapter = self._store_adapters[key].to_registry()
+            except Exception as e:
+                logger.error(f"适配器 {key} 同步商店数据失败：{e}")
+                continue
+            self._previous_adapters[key] = new_adapter
         for key in self._store_bots:
-            if key not in self._previous_bots:
-                author = get_author_name(self._store_bots[key].author_id)
-                self._previous_bots[key] = RegistryBot(
-                    **self._store_bots[key].model_dump(), author=author
-                )
-            else:
-                self._previous_bots[key] = RegistryBot(
-                    **self._store_bots[key].model_dump(),
-                    author=self._previous_bots[key].author,
-                )
+            try:
+                if key in self._previous_bots:
+                    new_bot = self._previous_bots[key].update(self._store_bots[key])
+                else:
+                    new_bot = self._store_bots[key].to_registry()
+            except Exception as e:
+                logger.error(f"机器人 {key} 同步商店数据失败：{e}")
+                continue
+
+            self._previous_bots[key] = new_bot
         for key in self._store_drivers:
-            if key not in self._previous_drivers:
-                author = get_author_name(self._store_drivers[key].author_id)
-                self._previous_drivers[key] = RegistryDriver(
-                    **self._store_drivers[key].model_dump(), author=author
-                )
-            else:
-                self._previous_drivers[key] = RegistryDriver(
-                    **self._store_drivers[key].model_dump(),
-                    author=self._previous_drivers[key].author,
-                )
+            try:
+                if key in self._previous_drivers:
+                    new_driver = self._previous_drivers[key].update(
+                        self._store_drivers[key]
+                    )
+                else:
+                    new_driver = self._store_drivers[key].to_registry()
+            except Exception as e:
+                logger.error(f"驱动器 {key} 同步商店数据失败：{e}")
+                continue
+
+            self._previous_drivers[key] = new_driver
         for key in self._store_plugins:
-            if key in self._previous_plugins:
-                plugin_data = self._previous_plugins[key].model_dump()
-                # 更新插件数据，假设商店数据的数据没有问题的
-                # TODO: 如果 author_id 变化，应该重新获取 author
-                plugin_data.update(self._store_plugins[key].model_dump())
-                self._previous_plugins[key] = RegistryPlugin(**plugin_data)
+            try:
+                if key in self._previous_plugins:
+                    new_plugin = self._previous_plugins[key].update(
+                        self._store_plugins[key]
+                    )
+                else:
+                    # TODO: 如果插件不存在，尝试重新测试获取相关信息验证
+                    raise NotImplementedError("插件需要重新测试")
+            except Exception as e:
+                logger.error(f"插件 {key} 同步商店数据失败：{e}")
+                continue
+
+            self._previous_plugins[key] = new_plugin
 
     def generate_github_summary(self, results: dict[str, StoreTestResult]):
         """生成 GitHub 摘要"""

src/providers/store_test/validation.py

@@ -5,13 +5,13 @@
 from src.providers.docker_test import DockerPluginTest
 from src.providers.logger import logger
 from src.providers.models import RegistryPlugin, StorePlugin, StoreTestResult
+from src.providers.utils import get_author_name, get_pypi_upload_time
 from src.providers.validation import (
     PluginPublishInfo,
     PublishType,
     ValidationDict,
     validate_info,
 )
-from src.providers.validation.utils import get_author_name, get_upload_time
 
 
 async def validate_plugin(
@@ -32,7 +32,7 @@ async def validate_plugin(
     module_name = store_plugin.module_name
 
     # 从 PyPI 获取信息
-    pypi_time = get_upload_time(project_link)
+    pypi_time = get_pypi_upload_time(project_link)
 
     # 测试插件
     plugin_test_result = await DockerPluginTest(project_link, module_name, config).run(