feat: 商店存放 json5 格式的数据

避免合并冲突

Author: he0119

CHANGELOG.md

@@ -10,6 +10,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/lang/zh-CN/
 ### Added
 
 - 通过名称加主页避免机器人重复
+- 商店存放 json5 格式的数据
 
 ### Fixed
 

src/plugins/github/plugins/publish/utils.py

@@ -10,8 +10,9 @@
 from src.plugins.github.models import IssueHandler
 from src.plugins.github.models.github import GithubHandler
 from src.plugins.github.utils import commit_message as _commit_message
-from src.plugins.github.utils import dump_json, load_json, run_shell_command
+from src.plugins.github.utils import run_shell_command
 from src.providers.models import RegistryUpdatePayload, to_store
+from src.providers.utils import dump_json5, load_json5_from_file
 from src.providers.validation import PublishType, ValidationDict
 
 from .constants import (
@@ -149,9 +150,9 @@ def update_file(result: ValidationDict) -> None:
 
     logger.info(f"正在更新文件: {path}")
 
-    data = load_json(path)
+    data = load_json5_from_file(path)
     data.append(new_data)
-    dump_json(path, data, 2)
+    dump_json5(path, data)
 
     logger.info("文件更新完成")
 

src/plugins/github/plugins/remove/utils.py

@@ -9,11 +9,8 @@
     get_type_by_labels,
 )
 from src.plugins.github.models import GithubHandler, IssueHandler
-from src.plugins.github.utils import (
-    commit_message,
-    dump_json,
-    run_shell_command,
-)
+from src.plugins.github.utils import commit_message, run_shell_command
+from src.providers.utils import dump_json5
 from src.providers.validation.models import PublishType
 
 from .constants import COMMIT_MESSAGE_PREFIX, REMOVE_LABEL
@@ -40,7 +37,7 @@ def update_file(type: PublishType, key: str):
     data = load_publish_data(type)
     # 删除对应的数据项
     data.pop(key)
-    dump_json(path, list(data.values()))
+    dump_json5(path, list(data.values()))
     logger.info(f"已更新 {path.name} 文件")
 
 

src/plugins/github/plugins/remove/validation.py

@@ -4,8 +4,9 @@
 
 from src.plugins.github import plugin_config
 from src.plugins.github.models import AuthorInfo
-from src.plugins.github.utils import extract_issue_info_from_issue, load_json
+from src.plugins.github.utils import extract_issue_info_from_issue
 from src.providers.constants import BOT_KEY_TEMPLATE, PYPI_KEY_TEMPLATE
+from src.providers.utils import load_json5_from_file
 from src.providers.validation.models import PublishType
 
 from .constants import (
@@ -25,23 +26,27 @@ def load_publish_data(publish_type: PublishType):
                     project_link=adapter["project_link"],
                     module_name=adapter["module_name"],
                 ): adapter
-                for adapter in load_json(plugin_config.input_config.adapter_path)
+                for adapter in load_json5_from_file(
+                    plugin_config.input_config.adapter_path
+                )
             }
         case PublishType.BOT:
             return {
                 BOT_KEY_TEMPLATE.format(
                     name=bot["name"],
                     homepage=bot["homepage"],
                 ): bot
-                for bot in load_json(plugin_config.input_config.bot_path)
+                for bot in load_json5_from_file(plugin_config.input_config.bot_path)
             }
         case PublishType.PLUGIN:
             return {
                 PYPI_KEY_TEMPLATE.format(
                     project_link=plugin["project_link"],
                     module_name=plugin["module_name"],
                 ): plugin
-                for plugin in load_json(plugin_config.input_config.plugin_path)
+                for plugin in load_json5_from_file(
+                    plugin_config.input_config.plugin_path
+                )
             }
         case PublishType.DRIVER:
             raise ValueError("不支持的删除类型")

src/plugins/github/utils.py

@@ -1,11 +1,7 @@
-import json
 import subprocess
-from pathlib import Path
 from re import Pattern
-from typing import Any
 
 from nonebot import logger
-from pydantic_core import to_jsonable_python
 
 
 def run_shell_command(command: list[str]):
@@ -30,20 +26,6 @@ def commit_message(prefix: str, message: str, issue_number: int):
     return f"{prefix} {message} (#{issue_number})"
 
 
-def load_json(path: Path) -> list[dict[str, str]]:
-    """加载 JSON 文件"""
-    with path.open("r", encoding="utf-8") as f:
-        return json.load(f)
-
-
-def dump_json(path: Path, data: Any, indent: int = 4):
-    """保存 JSON 文件"""
-    with path.open("w", encoding="utf-8") as f:
-        # 结尾加上换行符，不然会被 pre-commit fix
-        json.dump(to_jsonable_python(data), f, ensure_ascii=False, indent=indent)
-        f.write("\n")
-
-
 def extract_issue_info_from_issue(
     patterns: dict[str, Pattern[str]], body: str
 ) -> dict[str, str | None]:

src/providers/store_test/store.py

@@ -26,6 +26,7 @@
     StorePlugin,
     StoreTestResult,
 )
+from src.providers.utils import dump_json, load_json5_from_web, load_json_from_web
 from src.providers.validation.utils import get_author_name
 
 from .constants import (
@@ -36,7 +37,7 @@
     PLUGINS_PATH,
     RESULTS_PATH,
 )
-from .utils import dump_json, get_latest_version, load_json
+from .utils import get_latest_version
 from .validation import validate_plugin
 
 print = click.echo
@@ -52,63 +53,65 @@ def __init__(self) -> None:
                 project_link=adapter["project_link"],
                 module_name=adapter["module_name"],
             ): StoreAdapter(**adapter)
-            for adapter in load_json(STORE_ADAPTERS_URL)
+            for adapter in load_json5_from_web(STORE_ADAPTERS_URL)
         }
         self._store_bots: dict[str, StoreBot] = {
             BOT_KEY_TEMPLATE.format(
                 name=bot["name"],
                 homepage=bot["homepage"],
             ): StoreBot(**bot)
-            for bot in load_json(STORE_BOTS_URL)
+            for bot in load_json5_from_web(STORE_BOTS_URL)
         }
         self._store_drivers: dict[str, StoreDriver] = {
             PYPI_KEY_TEMPLATE.format(
                 project_link=driver["project_link"],
                 module_name=driver["module_name"],
             ): StoreDriver(**driver)
-            for driver in load_json(STORE_DRIVERS_URL)
+            for driver in load_json5_from_web(STORE_DRIVERS_URL)
         }
         self._store_plugins: dict[str, StorePlugin] = {
             PYPI_KEY_TEMPLATE.format(
                 project_link=plugin["project_link"],
                 module_name=plugin["module_name"],
             ): StorePlugin(**plugin)
-            for plugin in load_json(STORE_PLUGINS_URL)
+            for plugin in load_json5_from_web(STORE_PLUGINS_URL)
         }
         # 上次测试的结果
         self._previous_results: dict[str, StoreTestResult] = {
             key: StoreTestResult(**value)
-            for key, value in load_json(REGISTRY_RESULTS_URL).items()
+            for key, value in load_json_from_web(REGISTRY_RESULTS_URL).items()
         }
         self._previous_adapters: dict[str, RegistryAdapter] = {
             PYPI_KEY_TEMPLATE.format(
                 project_link=adapter["project_link"],
                 module_name=adapter["module_name"],
             ): RegistryAdapter(**adapter)
-            for adapter in load_json(REGISTRY_ADAPTERS_URL)
+            for adapter in load_json_from_web(REGISTRY_ADAPTERS_URL)
         }
         self._previous_bots: dict[str, RegistryBot] = {
             BOT_KEY_TEMPLATE.format(
                 name=bot["name"],
                 homepage=bot["homepage"],
             ): RegistryBot(**bot)
-            for bot in load_json(url=REGISTRY_BOTS_URL)
+            for bot in load_json_from_web(url=REGISTRY_BOTS_URL)
         }
         self._previous_drivers: dict[str, RegistryDriver] = {
             PYPI_KEY_TEMPLATE.format(
                 project_link=driver["project_link"],
                 module_name=driver["module_name"],
             ): RegistryDriver(**driver)
-            for driver in load_json(REGISTRY_DRIVERS_URL)
+            for driver in load_json_from_web(REGISTRY_DRIVERS_URL)
         }
         self._previous_plugins: dict[str, RegistryPlugin] = {
             PYPI_KEY_TEMPLATE.format(
                 project_link=plugin["project_link"], module_name=plugin["module_name"]
             ): RegistryPlugin(**plugin)
-            for plugin in load_json(REGISTRY_PLUGINS_URL)
+            for plugin in load_json_from_web(REGISTRY_PLUGINS_URL)
         }
         # 插件配置文件
-        self._plugin_configs: dict[str, str] = load_json(REGISTRY_PLUGIN_CONFIG_URL)
+        self._plugin_configs: dict[str, str] = load_json_from_web(
+            REGISTRY_PLUGIN_CONFIG_URL
+        )
 
     def should_skip(self, key: str, force: bool = False) -> bool:
         """是否跳过测试"""

src/providers/store_test/utils.py

@@ -1,31 +1,9 @@
-import json
 from functools import cache
-from pathlib import Path
 from typing import Any
 
 import httpx
-from pydantic_core import to_jsonable_python
 
-
-def load_json(url: str) -> Any:
-    """从网络加载 JSON 文件"""
-    r = httpx.get(url)
-    if r.status_code != 200:
-        raise ValueError(f"下载文件失败：{r.text}")
-    return r.json()
-
-
-def dump_json(path: Path, data: Any, minify: bool = True) -> None:
-    """保存 JSON 文件
-
-    为减少文件大小，还需手动设置 separators
-    """
-    data = to_jsonable_python(data)
-    with open(path, "w", encoding="utf-8") as f:
-        if minify:
-            json.dump(data, f, ensure_ascii=False, separators=(",", ":"))
-        else:
-            json.dump(data, f, ensure_ascii=False, indent=2)
+from src.providers.utils import load_json_from_web
 
 
 @cache
@@ -55,5 +33,5 @@ def get_upload_time(project_link: str) -> str:
 
 def get_user_id(name: str) -> int:
     """获取用户信息"""
-    data = load_json(f"https://api.github.com/users/{name}")
+    data = load_json_from_web(f"https://api.github.com/users/{name}")
     return data["id"]

src/providers/utils.py

@@ -0,0 +1,64 @@
+import json
+from pathlib import Path
+from typing import Any
+
+import httpx
+import json5
+from pydantic_core import to_jsonable_python
+
+
+def load_json_from_file(file_path: Path):
+    """从文件加载 JSON 文件"""
+    with open(file_path, encoding="utf-8") as file:
+        return json.load(file)
+
+
+def load_json_from_web(url: str):
+    """从网络加载 JSON 文件"""
+    r = httpx.get(url)
+    if r.status_code != 200:
+        raise ValueError(f"下载文件失败：{r.text}")
+    return r.json()
+
+
+def dump_json(path: Path, data: Any, minify: bool = True) -> None:
+    """保存 JSON 文件
+
+    为减少文件大小，还需手动设置 separators
+    """
+    data = to_jsonable_python(data)
+    with open(path, "w", encoding="utf-8") as f:
+        if minify:
+            json.dump(data, f, ensure_ascii=False, separators=(",", ":"))
+        else:
+            json.dump(data, f, ensure_ascii=False, indent=2)
+
+
+def load_json5_from_file(file_path: Path):
+    """从文件加载 JSON5 文件"""
+    with open(file_path, encoding="utf-8") as file:
+        return json5.load(file)
+
+
+def load_json5_from_web(url: str):
+    """从网络加载 JSON5 文件"""
+    r = httpx.get(url)
+    if r.status_code != 200:
+        raise ValueError(f"下载文件失败：{r.text}")
+    return json5.loads(r.text)
+
+
+def dump_json5(path: Path, data: Any) -> None:
+    """保存 JSON5 文件
+
+    手动添加末尾的逗号和换行符
+    """
+    data = to_jsonable_python(data)
+
+    content = json.dumps(data, ensure_ascii=False, indent=2)
+    # 手动添加末尾的逗号和换行符
+    # 避免合并时出现冲突
+    content = content.replace("}\n]", "},\n]\n")
+
+    with open(path, "w", encoding="utf-8") as f:
+        f.write(content)

tests/conftest.py

@@ -1,8 +1,8 @@
-import json
 from pathlib import Path
 from typing import TYPE_CHECKING
 
 import httpx
+import json5
 import nonebot
 import pytest
 from nonebot.adapters.github import Adapter
@@ -49,7 +49,7 @@ async def app(app: App, tmp_path: Path, mocker: MockerFixture):
 
     adapter_path = tmp_path / "adapters.json"
     with adapter_path.open("w") as f:
-        json.dump(
+        json5.dump(
             [
                 {
                     "module_name": "module_name1",
@@ -66,7 +66,7 @@ async def app(app: App, tmp_path: Path, mocker: MockerFixture):
         )
     bot_path = tmp_path / "bots.json"
     with bot_path.open("w") as f:
-        json.dump(
+        json5.dump(
             [
                 {
                     "name": "name",
@@ -81,7 +81,7 @@ async def app(app: App, tmp_path: Path, mocker: MockerFixture):
         )
     plugin_path = tmp_path / "plugins.json"
     with plugin_path.open("w") as f:
-        json.dump(
+        json5.dump(
             [
                 {
                     "module_name": "module_name1",

tests/github/publish/utils/test_publish_resolve_conflict_pull_requests.py

@@ -1,19 +1,19 @@
 import json
 from pathlib import Path
-from typing import Any
 
 import pytest
 from inline_snapshot import snapshot
 from nonebug import App
 from pytest_mock import MockerFixture
 from respx import MockRouter
 
-from tests.github.utils import MockBody, MockIssue, MockUser, get_github_bot
-
-
-def check_json_data(file: Path, data: Any) -> None:
-    with open(file, encoding="utf-8") as f:
-        assert json.load(f) == data
+from tests.github.utils import (
+    MockBody,
+    MockIssue,
+    MockUser,
+    check_json_data,
+    get_github_bot,
+)
 
 
 @pytest.fixture