up

Author: lgc2333

README.md

@@ -14,7 +14,7 @@
 
 _✨ 新一代的图片帮助插件 ✨_
 
-<img src="https://img.shields.io/badge/python-3.9+-blue.svg" alt="python">
+<img src="https://img.shields.io/badge/python-3.10+-blue.svg" alt="python">
 <a href="https://pdm.fming.dev">
   <img src="https://img.shields.io/badge/pdm-managed-blueviolet" alt="pdm-managed">
 </a>

nonebot_plugin_picmenu_next/__init__.py

@@ -11,7 +11,7 @@
 from .data_source import refresh_infos
 from .templates import load_builtin_templates
 
-__version__ = "0.1.3"
+__version__ = "0.1.4"
 __plugin_meta__ = PluginMetadata(
     name="PicMenu Next",
     description="新一代的图片帮助插件",

nonebot_plugin_picmenu_next/__main__.py

@@ -1,6 +1,6 @@
 from collections.abc import Sequence
 from pathlib import Path
-from typing import Optional, TypeVar
+from typing import TypeVar
 
 from arclet.alconna import Alconna, Arg, Args, CommandMeta, Option, store_true
 from loguru import logger
@@ -81,7 +81,7 @@ def get_name_similarities(
 T = TypeVar("T")
 
 
-def handle_query_index(query: str, infos: Sequence[T]) -> Optional[tuple[int, T]]:
+def handle_query_index(query: str, infos: Sequence[T]) -> tuple[int, T] | None:
     if query.isdigit() and query.strip("0"):
         return (
             ((i := qn - 1), infos[i])
@@ -95,7 +95,7 @@ async def query_plugin(
     infos: list[PMNPluginInfo],
     query: str,
     score_cutoff: float = 60,
-) -> Optional[tuple[int, PMNPluginInfo]]:
+) -> tuple[int, PMNPluginInfo] | None:
     if r := handle_query_index(query, infos):
         return r
 
@@ -121,7 +121,7 @@ async def query_func_detail(
     pm_data: list[PMDataItem],
     query: str,
     score_cutoff: float = 60,
-) -> Optional[tuple[int, PMDataItem]]:
+) -> tuple[int, PMDataItem] | None:
     if r := handle_query_index(query, pm_data):
         return r
 
@@ -147,8 +147,8 @@ async def query_func_detail(
 async def _(
     bot: BaseBot,
     ev: BaseEvent,
-    q_plugin: Query[Optional[str]] = Query("~plugin", None),
-    q_function: Query[Optional[str]] = Query("~function", None),
+    q_plugin: Query[str | None] = Query("~plugin", None),
+    q_function: Query[str | None] = Query("~function", None),
     q_show_hidden: Query[bool] = Query("~show-hidden.value", default=False),
 ):
     show_hidden = q_show_hidden.result
@@ -159,7 +159,7 @@ async def _(
     ):
         await (
             UniMessage.image(raw=TIP_IMG_PATH.read_bytes())
-            .text("别想看咱藏起来的东西！")
+            .text("不是主人不给看")
             .finish(reply_to=True)
         )
 

nonebot_plugin_picmenu_next/config.py

@@ -1,7 +1,19 @@
+from pathlib import Path
+
+from cookit.nonebot.localstore import ensure_localstore_path_config
 from cookit.pyd import model_with_alias_generator
 from nonebot import get_plugin_config
+from nonebot_plugin_localstore import get_plugin_config_dir
 from pydantic import BaseModel
 
+ensure_localstore_path_config()
+
+config_dir = get_plugin_config_dir()
+
+pm_menus_dir = Path.cwd() / "menu_config/menus"
+external_infos_dir = config_dir / "external_infos"
+external_infos_dir.mkdir(parents=True, exist_ok=True)
+
 
 @model_with_alias_generator(lambda x: f"pmn_{x}")
 class ConfigModel(BaseModel):

nonebot_plugin_picmenu_next/data_source/collect.py

@@ -1,19 +1,20 @@
 import asyncio
 import importlib
-from collections.abc import Iterable
+from collections.abc import Generator, Iterable
 from contextlib import suppress
 from functools import lru_cache
 from importlib.metadata import Distribution, PackageNotFoundError, distribution
-from typing import Optional
+from pathlib import Path
 
 from cookit.loguru import warning_suppress
-from cookit.pyd import type_validate_python
+from cookit.pyd import type_validate_json, type_validate_python
 from nonebot import logger
 from nonebot.plugin import Plugin
 
+from ..config import external_infos_dir, pm_menus_dir
 from ..utils import normalize_plugin_name
-from .mixin import chain_mixins, plugin_collect_mixins
-from .models import PMNData, PMNPluginExtra, PMNPluginInfo
+from .mixin import PluginCollectMixinNext, chain_mixins, plugin_collect_mixins
+from .models import ExternalPluginInfo, PMNData, PMNPluginExtra, PMNPluginInfo
 
 
 def normalize_metadata_user(info: str, allow_multi: bool = False) -> str:
@@ -24,7 +25,7 @@ def normalize_metadata_user(info: str, allow_multi: bool = False) -> str:
 
 
 @lru_cache
-def get_dist(module_name: str) -> Optional[Distribution]:
+def get_dist(module_name: str) -> Distribution | None:
     with warning_suppress(f"Unexpected error happened when getting info of package {module_name}"),\
         suppress(PackageNotFoundError):  # fmt: skip
         return distribution(module_name)
@@ -35,7 +36,7 @@ def get_dist(module_name: str) -> Optional[Distribution]:
 
 
 @lru_cache
-def get_version_attr(module_name: str) -> Optional[str]:
+def get_version_attr(module_name: str) -> str | None:
     with warning_suppress(f"Unexpected error happened when importing {module_name}"),\
         suppress(ImportError):  # fmt: skip
         m = importlib.import_module(module_name)
@@ -49,7 +50,7 @@ def get_version_attr(module_name: str) -> Optional[str]:
 
 async def get_info_from_plugin(plugin: Plugin) -> PMNPluginInfo:
     meta = plugin.metadata
-    extra: Optional[PMNPluginExtra] = None
+    extra: PMNPluginExtra | None = None
     if meta:
         with warning_suppress(f"Failed to parse plugin metadata of {plugin.id_}"):
             extra = type_validate_python(PMNPluginExtra, meta.extra)
@@ -68,19 +69,25 @@ async def get_info_from_plugin(plugin: Plugin) -> PMNPluginInfo:
         else None
     )
     if not author and (dist := get_dist(plugin.module_name)):
-        if author := dist.metadata.get("Author") or dist.metadata.get("Maintainer"):
+        if (("Author" in dist.metadata) and (author := dist.metadata["Author"])) or (
+            ("Maintainer" in dist.metadata) and (author := dist.metadata["Maintainer"])
+        ):
             author = normalize_metadata_user(author)
-        elif author := dist.metadata.get("Author-Email") or dist.metadata.get(
-            "Maintainer-Email",
+        elif (
+            ("Author-Email" in dist.metadata)
+            and (author := dist.metadata["Author-Email"])
+        ) or (
+            ("Maintainer-Email" in dist.metadata)
+            and (author := dist.metadata["Maintainer-Email"])
         ):
             author = normalize_metadata_user(author, allow_multi=True)
 
     description = (
         meta.description
         if meta
         else (
-            dist.metadata.get("Summary")
-            if (dist := get_dist(plugin.module_name))
+            dist.metadata["Summary"]
+            if (dist := get_dist(plugin.module_name)) and "Summary" in dist.metadata
             else None
         )
     )
@@ -102,6 +109,107 @@ async def get_info_from_plugin(plugin: Plugin) -> PMNPluginInfo:
     )
 
 
+def scan_path(path: Path, suffixes: Iterable[str] | None = None) -> Generator[Path]:
+    for child in path.iterdir():
+        if child.is_dir():
+            yield from scan_path(child, suffixes)
+        elif suffixes and child.suffix in suffixes:
+            yield child
+
+
+async def collect_menus():
+    yaml = None
+
+    supported_suffixes = {".json", ".yml", ".yaml", ".toml"}
+
+    def _load_file(path: Path) -> ExternalPluginInfo:
+        nonlocal yaml
+
+        if path.suffix == ".json":
+            return type_validate_json(ExternalPluginInfo, path.read_text("u8"))
+
+        if path.suffix in {".yml", ".yaml"}:
+            if not yaml:
+                try:
+                    from ruamel.yaml import YAML
+                except ImportError as e:
+                    raise ImportError(
+                        "Missing dependency for parsing yaml files, please install using"
+                        " `pip install nonebot-plugin-picmenu-next[yaml]`",
+                    ) from e
+                yaml = YAML()
+            return type_validate_python(
+                ExternalPluginInfo,
+                yaml.load(path.read_text("u8")),
+            )
+
+        if path.suffix == ".toml":
+            try:
+                import tomllib
+            except ImportError:
+                try:
+                    import tomli as tomllib
+                except ImportError as e:
+                    raise ImportError(
+                        "Missing dependency for parsing toml files, please install using"
+                        " `pip install nonebot-plugin-picmenu-next[toml]`",
+                    ) from e
+            return type_validate_python(
+                ExternalPluginInfo,
+                tomllib.loads(path.read_text("u8")),
+            )
+
+        raise ValueError("Unsupported file type")
+
+    infos: dict[str, ExternalPluginInfo] = {}
+
+    def _load_to_infos(path: Path):
+        if path.name in infos:
+            logger.warning(
+                f"Find file with duplicated name `{path.name}`! Skip loading {{path}}",
+            )
+            return
+        with warning_suppress(f"Failed to load file {path}"):
+            infos[path.name] = _load_file(path)
+
+    def _load_all(path: Path):
+        for x in scan_path(path, supported_suffixes):
+            _load_to_infos(x)
+
+    if pm_menus_dir.exists():
+        logger.warning(
+            "Old PicMenu menus dir is deprecated"
+            ", recommended to migrate to PicMenu Next config dir",
+        )
+        _load_all(pm_menus_dir)
+
+    _load_all(external_infos_dir)
+
+    return infos
+
+
+@plugin_collect_mixins(priority=1)
+async def load_user_custom_infos_mixin(
+    next_mixin: PluginCollectMixinNext,
+    infos: list[PMNPluginInfo],
+) -> list[PMNPluginInfo]:
+    external_infos = await collect_menus()
+    if not external_infos:
+        return await next_mixin(infos)
+    logger.info(f"Collected {len(external_infos)} external infos")
+
+    infos_map = {x.plugin_id: x for x in infos if x.plugin_id}
+    for k, v in external_infos.items():
+        if k in infos_map:
+            logger.debug(f"Found `{k}` in infos, will merge to original")
+            v.merge_to(infos_map[k], plugin_id=k, copy=False)
+        else:
+            logger.debug(f"Not found `{k}` in infos, will add into")
+            infos.append(v.to_plugin_info(k))
+
+    return await next_mixin(infos)
+
+
 async def collect_plugin_infos(plugins: Iterable[Plugin]):
     async def _get(p: Plugin):
         with warning_suppress(f"Failed to get plugin info of {p.id_}"):

nonebot_plugin_picmenu_next/data_source/mixin.py

@@ -1,8 +1,8 @@
 from collections import defaultdict
-from collections.abc import Coroutine, Sequence
+from collections.abc import Callable, Coroutine, Sequence
 from dataclasses import dataclass
-from typing import Any, Callable, Generic, Optional, TypeVar
-from typing_extensions import Concatenate, ParamSpec, TypeAlias
+from typing import Any, Concatenate, Generic, TypeAlias, TypeVar
+from typing_extensions import ParamSpec
 
 from cookit import DecoListCollector
 from cookit.loguru import warning_suppress
@@ -67,15 +67,15 @@
 class MixinInfo(Generic[T]):
     func: T
     priority: int
-    source: Optional[MatcherSource]
+    source: MatcherSource | None
 
 
 class MixinCollector(DecoListCollector[MixinInfo[T]]):
     def __call__(  # pyright: ignore[reportIncompatibleMethodOverride]
         self,
         priority: int = 5,
         _depth: int = 0,
-        _matcher_source: Optional[MatcherSource] = None,
+        _matcher_source: MatcherSource | None = None,
     ):
         def deco(func: T) -> T:
             self.data.append(
@@ -99,7 +99,7 @@ def __call__(
         self,
         priority: int = 1,
         _depth: int = 0,
-        _matcher_source: Optional[MatcherSource] = None,
+        _matcher_source: MatcherSource | None = None,
     ):
         def deco(f: T) -> T:
             s = _matcher_source or get_matcher_source()