✨ support schema generate for configs in $files

Author: RF-Tar-Railt

arclet/entari/config/file.py

@@ -120,7 +120,7 @@ class EntariConfig:
     plugin_extra_files: list[str] = field(default_factory=list, init=False)
     save_flag: bool = field(default=False)
     _origin_data: dict[str, Any] = field(init=False)
-    _env_replaced: dict[int, str] = field(default_factory=dict, init=False)
+    _env_replaced: dict[str, dict[int, str]] = field(default_factory=dict, init=False)
 
     instance: ClassVar["EntariConfig"]
 
@@ -136,7 +136,7 @@ def loader(self, path: Path):
             for i, line in enumerate(lines):
 
                 def handle(m: re.Match[str]):
-                    self._env_replaced[i] = line
+                    self._env_replaced.setdefault(path.as_posix(), {})[i] = line
                     expr = m.group("expr")
                     return safe_eval(expr, ctx)
 
@@ -160,9 +160,9 @@ def dumper(self, path: Path, save_path: Path, data: dict, indent: int, apply_sch
             schema_file = f"{save_path.stem}.schema.json"
         if end in _dumpers:
             ans, applied = _dumpers[end](origin, indent, schema_file)
-            if self._env_replaced:
+            if path.as_posix() in self._env_replaced:
                 lines = ans.splitlines(keepends=True)
-                for i, line in self._env_replaced.items():
+                for i, line in self._env_replaced[path.as_posix()].items():
                     lines[i + applied] = line
                 ans = "".join(lines)
             with save_path.open("w", encoding="utf-8") as f:
@@ -222,10 +222,10 @@ def reload(self):
                 raise FileNotFoundError(file)
             if path.is_dir():
                 for _path in path.iterdir():
-                    if not _path.is_file():
+                    if not _path.is_file() or _path.name.endswith(".schema.json"):
                         continue
                     self.plugin[_path.stem] = self.loader(_path)
-            else:
+            elif path.name.endswith(".schema.json"):
                 self.plugin[path.stem] = self.loader(path)
         return True
 
@@ -235,7 +235,7 @@ def _clean(value: T_M) -> T_M:
         value.pop("$static", None)
         return value
 
-    def dump(self, indent: int = 2):
+    def dump(self, indent: int = 2, apply_schema: bool = False) -> dict[str, Any]:
         basic = self._origin_data.get("basic", {})
         if "log" not in basic and ("log_level" in basic or "log_ignores" in basic):
             basic["log"] = {}
@@ -247,12 +247,12 @@ def dump(self, indent: int = 2):
         if self.plugin_extra_files:
             for file in self.plugin_extra_files:
                 path = Path(file)
-                if path.is_file():
-                    self.dumper(path, path, self._clean(self.plugin.pop(path.stem)), indent, False)
+                if path.is_file() and not path.name.endswith(".schema.json"):
+                    self.dumper(path, path, self._clean(self.plugin.pop(path.stem)), indent, apply_schema)
                 else:
                     for _path in path.iterdir():
-                        if _path.is_file():
-                            self.dumper(_path, _path, self._clean(self.plugin.pop(_path.stem)), indent, False)
+                        if _path.is_file() and not _path.name.endswith(".schema.json"):
+                            self.dumper(_path, _path, self._clean(self.plugin.pop(_path.stem)), indent, apply_schema)
         for key in list(self.plugin.keys()):
             if key.startswith("$"):
                 continue
@@ -268,7 +268,7 @@ def dump(self, indent: int = 2):
 
     def save(self, path: str | os.PathLike[str] | None = None, indent: int = 2, apply_schema: bool = False):
         self.save_flag = True
-        self.dumper(self.path, Path(path or self.path), self.dump(indent), indent, apply_schema)
+        self.dumper(self.path, Path(path or self.path), self.dump(indent, apply_schema), indent, apply_schema)
 
     @classmethod
     def load(cls, path: str | os.PathLike[str] | None = None) -> "EntariConfig":
@@ -322,7 +322,10 @@ def generate_schema(self, plugins: list["Plugin"]):
         plugins_properties = {}
         # fmt: off
         plugin_meta_properties = {"$disable": {"type": "string", "description": "Expression for whether disable this plugin"}, "$prefix": {"type": "string", "description": "Plugin name prefix"}, "$priority": {"type": "integer", "description": "Plugin loading priority, lower value means higher priority (default: 16)"}, "$filter": {"type": "string", "description": "Plugin filter expression, which will be evaluated in the context of the plugin"}}  # noqa: E501
+        # Build a mapping from plugin config key to plugin object for $files schema generation
+        plugin_map: dict[str, "Plugin"] = {}  # noqa: UP037
         for plug in plugins:
+            plugin_map[plug._config_key] = plug
             if plug.metadata is not None:
                 if plug.metadata.config:
                     schema = config_model_schema(plug.metadata.config, ref_root=f"/properties/plugins/properties/{plug._config_key}/")  # noqa: E501
@@ -337,7 +340,64 @@ def generate_schema(self, plugins: list["Plugin"]):
         }
         with open(f"{self.path.stem}.schema.json", "w", encoding="utf-8") as f:
             json.dump({"$schema": "https://json-schema.org/draft/2020-12/schema", "type": "object", "properties": schemas, "additionalProperties": False, "required": ["basic"]}, f, indent=2, ensure_ascii=False)  # noqa: E501
+
+        # Generate schema for each file in $files
+        for file in self.plugin_extra_files:
+            path = Path(file)
+            if path.is_file() and not path.name.endswith(".schema.json"):
+                self._generate_extra_file_schema(path, plugin_map, plugin_meta_properties)
+            elif path.is_dir():
+                for _path in path.iterdir():
+                    if _path.is_file() and not _path.name.endswith(".schema.json"):
+                        self._generate_extra_file_schema(_path, plugin_map, plugin_meta_properties)
         # fmt: on
+        plugin_map.clear()
+
+    def _generate_extra_file_schema(self, path: Path, plugin_map: dict[str, "Plugin"], plugin_meta_properties: dict):
+        """Generate schema for an extra config file from $files."""
+        plugin_key = path.stem
+        schema_file = path.with_suffix(".schema.json")
+        plugin_meta_properties = {
+            **plugin_meta_properties,
+            "$optional": {"type": "boolean", "description": "Whether this plugin is optional"},
+        }  # noqa: E501
+
+        # Check if we have a matching plugin with config
+        if plugin_key in plugin_map:
+            plug = plugin_map[plugin_key]
+            if plug.metadata is not None and plug.metadata.config:
+                plugin_schema = config_model_schema(plug.metadata.config, ref_root="/")
+                plugin_schema["properties"].update(plugin_meta_properties)
+            elif plug.metadata is not None:
+                plugin_schema = {
+                    "type": "object",
+                    "description": f"{plug.metadata.description or plug.metadata.name}; no configuration required",
+                    "additionalProperties": True,
+                    "properties": plugin_meta_properties,
+                }  # noqa: E501
+            else:
+                plugin_schema = {
+                    "type": "object",
+                    "description": "No configuration required",
+                    "additionalProperties": True,
+                    "properties": plugin_meta_properties,
+                }  # noqa: E501
+        else:
+            # Plugin not found, generate a generic schema
+            plugin_schema = {
+                "type": "object",
+                "description": f"Configuration for {plugin_key}",
+                "additionalProperties": True,
+                "properties": plugin_meta_properties,
+            }  # noqa: E501
+
+        with open(schema_file, "w", encoding="utf-8") as f:
+            json.dump(
+                {"$schema": "https://json-schema.org/draft/2020-12/schema", **plugin_schema},
+                f,
+                indent=2,
+                ensure_ascii=False,
+            )  # noqa: E501
 
 
 load_config = EntariConfig.load

arclet/entari/core.py

@@ -111,80 +111,71 @@ async def __call__(self, context: Contexts):
         if ITEM_OPERATOR in context:
             return context[ITEM_OPERATOR]
         if ITEM_ORIGIN_EVENT not in context:
-            return
-        return context[ITEM_ORIGIN_EVENT].operator
+            return context[ITEM_ORIGIN_EVENT].operator
 
 
 class UserProvider(Provider[User]):
     async def __call__(self, context: Contexts):
         if ITEM_USER in context:
             return context[ITEM_USER]
         if ITEM_ORIGIN_EVENT not in context:
-            return
-        return context[ITEM_ORIGIN_EVENT].user
+            return context[ITEM_ORIGIN_EVENT].user
 
 
 class MessageProvider(Provider[MessageObject]):
     async def __call__(self, context: Contexts):
         if ITEM_MESSAGE_ORIGIN in context:
             return context[ITEM_MESSAGE_ORIGIN]
         if ITEM_ORIGIN_EVENT not in context:
-            return
-        return context[ITEM_ORIGIN_EVENT].message
+            return context[ITEM_ORIGIN_EVENT].message
 
 
 class ChannelProvider(Provider[Channel]):
     async def __call__(self, context: Contexts):
         if ITEM_CHANNEL in context:
             return context[ITEM_CHANNEL]
         if ITEM_ORIGIN_EVENT not in context:
-            return
-        return context[ITEM_ORIGIN_EVENT].channel
+            return context[ITEM_ORIGIN_EVENT].channel
 
 
 class GuildProvider(Provider[Guild]):
     async def __call__(self, context: Contexts):
         if ITEM_GUILD in context:
             return context[ITEM_GUILD]
         if ITEM_ORIGIN_EVENT not in context:
-            return
-        return context[ITEM_ORIGIN_EVENT].guild
+            return context[ITEM_ORIGIN_EVENT].guild
 
 
 class MemberProvider(Provider[Member]):
     async def __call__(self, context: Contexts):
         if ITEM_MEMBER in context:
             return context[ITEM_MEMBER]
         if ITEM_ORIGIN_EVENT not in context:
-            return
-        return context[ITEM_ORIGIN_EVENT].member
+            return context[ITEM_ORIGIN_EVENT].member
 
 
 class RoleProvider(Provider[Role]):
     async def __call__(self, context: Contexts):
         if ITEM_ROLE in context:
             return context[ITEM_ROLE]
         if ITEM_ORIGIN_EVENT not in context:
-            return
-        return context[ITEM_ORIGIN_EVENT].role
+            return context[ITEM_ORIGIN_EVENT].role
 
 
 class EmojiProvider(Provider[EmojiObject]):
     async def __call__(self, context: Contexts):
         if ITEM_EMOJI_OBJECT in context:
             return context[ITEM_EMOJI_OBJECT]
         if ITEM_ORIGIN_EVENT not in context:
-            return
-        return context[ITEM_ORIGIN_EVENT].emoji
+            return context[ITEM_ORIGIN_EVENT].emoji
 
 
 class LoginProvider(Provider[Login]):
     async def __call__(self, context: Contexts):
         if ITEM_LOGIN in context:
             return context[ITEM_LOGIN]
         if ITEM_ORIGIN_EVENT not in context:
-            return
-        return context[ITEM_ORIGIN_EVENT].login
+            return context[ITEM_ORIGIN_EVENT].login
 
 
 class MessageContentProvider(Provider[MessageChain]):

arclet/entari/plugin/module.py

@@ -2,13 +2,14 @@
 from collections.abc import Sequence
 from importlib import _bootstrap, _bootstrap_external  # type: ignore
 from importlib.abc import MetaPathFinder
-from importlib.machinery import ExtensionFileLoader, PathFinder, SourceFileLoader
+from importlib.machinery import ExtensionFileLoader, ModuleSpec, PathFinder, SourceFileLoader
 from importlib.util import module_from_spec, resolve_name
 from io import BytesIO
 import re
 import sys
 import tokenize
 from types import ModuleType
+from typing import Any
 
 from arclet.letoderea import publish
 from arclet.letoderea.scope import scope_ctx
@@ -241,7 +242,7 @@ def create_module(self, spec) -> ModuleType | None:
             raise ReusablePluginError(f"reusable plugin {self.name!r} cannot be imported directly")
         return super().create_module(spec)
 
-    def exec_module(self, module: ModuleType, config: dict[str, str] | None = None) -> None:
+    def exec_module(self, module: ModuleType, config: dict[str, Any] | None = None) -> None:
         is_sub = False
         if plugin := plugin_service.plugins.get(self.parent_plugin_id) if self.parent_plugin_id else None:
             plugin.subplugins.add(self.plugin_id)
@@ -335,7 +336,7 @@ def _path_find_spec(fullname, path=None, target=None):
     """
     if path is None:
         path = sys.path
-    spec = PathFinder._get_spec(fullname, path, target)  # type: ignore
+    spec: ModuleSpec | None = PathFinder._get_spec(fullname, path, target)  # type: ignore
     if spec is None:
         return None
     elif spec.loader is None:
@@ -418,12 +419,13 @@ def find_spec(
         return
 
 
-def find_spec(id_, package=None):
+def find_spec(id_, package=None) -> ModuleSpec | None:
     uid_index = id_.rfind("@")
     name = id_ if uid_index == -1 else id_[:uid_index]
     fullname = resolve_name(name, package) if name.startswith(".") else name
     parent_name = fullname.rpartition(".")[0]
     if parent_name:
+        parent: ModuleType | None
         parts = parent_name.split(".")
         _current = parts[0]
         if _current in plugin_service.plugins:
@@ -454,13 +456,12 @@ def find_spec(id_, package=None):
                 enter_plugin = False
                 parent = __import__(_current, fromlist=["__path__"])
             _current += "."
-        try:
-            parent_path = parent.__path__
-        except AttributeError as e:
+        if parent is None:
             raise ModuleNotFoundError(
-                f"__path__ attribute not found on {parent_name!r} " f"while trying to find {fullname!r}",
+                f"parent module {parent_name!r} does not have __path__ attribute " f"while trying to find {fullname!r}",
                 name=fullname,
-            ) from e
+            )
+        parent_path = parent.__path__
     else:
         parent_path = None
     if isinstance(parent_path, _bootstrap_external._NamespacePath):  # type: ignore