✨ more filter

arclet/entari/filter.py

@@ -1,7 +1,103 @@
 from typing import Optional
 
 from arclet.letoderea import Interface, JudgeAuxiliary, Scope
-from satori import Channel, ChannelType
+from satori import Channel, ChannelType, Guild, User
+from satori.client import Account
+
+
+class UserFilter(JudgeAuxiliary):
+    def __init__(self, *user_ids: str):
+        self.user_ids = set(user_ids)
+        super().__init__()
+
+    async def __call__(self, scope: Scope, interface: Interface) -> Optional[bool]:
+        if not (user := interface.query(User, "user", force_return=True)):
+            return False
+        return user.id in self.user_ids
+
+    @property
+    def scopes(self) -> set[Scope]:
+        return {Scope.prepare}
+
+    @property
+    def id(self) -> str:
+        return "entari.filter/user"
+
+
+class GuildFilter(JudgeAuxiliary):
+    def __init__(self, *guild_ids: str):
+        self.guild_ids = set(guild_ids)
+        super().__init__()
+
+    async def __call__(self, scope: Scope, interface: Interface) -> Optional[bool]:
+        if not (guild := interface.query(Guild, "guild", force_return=True)):
+            return False
+        return guild.id in self.guild_ids
+
+    @property
+    def scopes(self) -> set[Scope]:
+        return {Scope.prepare}
+
+    @property
+    def id(self) -> str:
+        return "entari.filter/guild"
+
+
+class ChannelFilter(JudgeAuxiliary):
+    def __init__(self, *channel_ids: str):
+        self.channel_ids = set(channel_ids)
+        super().__init__()
+
+    async def __call__(self, scope: Scope, interface: Interface) -> Optional[bool]:
+        if not (channel := interface.query(Channel, "channel", force_return=True)):
+            return False
+        return channel.id in self.channel_ids
+
+    @property
+    def scopes(self) -> set[Scope]:
+        return {Scope.prepare}
+
+    @property
+    def id(self) -> str:
+        return "entari.filter/channel"
+
+
+class SelfFilter(JudgeAuxiliary):
+    def __init__(self, *self_ids: str):
+        self.self_ids = set(self_ids)
+        super().__init__()
+
+    async def __call__(self, scope: Scope, interface: Interface) -> Optional[bool]:
+        if not (account := interface.query(Account, "account", force_return=True)):
+            return False
+        return account.self_id in self.self_ids
+
+    @property
+    def scopes(self) -> set[Scope]:
+        return {Scope.prepare}
+
+    @property
+    def id(self) -> str:
+        return "entari.filter/self"
+
+
+class PlatformFilter(JudgeAuxiliary):
+    def __init__(self, *platforms: str):
+        self.platforms = set(platforms)
+        super().__init__()
+
+    async def __call__(self, scope: Scope, interface: Interface) -> Optional[bool]:
+        if not (account := interface.query(Account, "account", force_return=True)):
+            return False
+        return account.platform in self.platforms
+
+    @property
+    def scopes(self) -> set[Scope]:
+        return {Scope.prepare}
+
+    @property
+    def id(self) -> str:
+        return "entari.filter/platform"
 
 
 class DirectMessageJudger(JudgeAuxiliary):

arclet/entari/plugin/__init__.py

@@ -8,10 +8,11 @@
 
 from ..config import EntariConfig
 from ..logger import log
-from .model import RootlessPlugin as RootlessPlugin
 from .model import Plugin
 from .model import PluginMetadata as PluginMetadata
-from .model import RegisterNotInPluginError, _current_plugin
+from .model import RegisterNotInPluginError
+from .model import RootlessPlugin as RootlessPlugin
+from .model import _current_plugin
 from .model import keeping as keeping
 from .module import import_plugin
 from .module import package as package

arclet/entari/plugin/service.py

@@ -8,7 +8,7 @@
 from ..logger import log
 
 if TYPE_CHECKING:
-    from .model import RootlessPlugin, KeepingVariable, Plugin
+    from .model import KeepingVariable, Plugin, RootlessPlugin
 
 
 class PluginManagerService(Service):