Added some room filters

src/mab/filters/room.py

@@ -0,0 +1,107 @@
+from .base import BaseEventFilter
+
+from nio import AsyncClient
+from nio import MatrixRoom, Event
+
+class RoomIdContainsFilter(BaseEventFilter):
+    """
+        This filter returns True if the `room.room_id` contains `needle` (or
+        any of needles from the list). The check will be case insensetive if
+        `any_case` is True.
+    """
+    def __init__(self, needle: str | list[str], *, any_case: bool = True):
+        super().__init__()
+        if type(needle) is str:
+            needle = [needle]
+        self._any_case = any_case
+        if self._any_case:
+            self._needle = [s.lower() for s in needle]
+        else:
+            self._needle = list(needle)
+
+    def __repr__(self):
+        return f"RoomIdContainsFilter({repr(self._needle)}, any_case={repr(self._any_case)})"
+
+    async def __call__(self, room: MatrixRoom, event: Event, client: AsyncClient) -> bool:
+        try:
+            room_id = room.room_id.lower() if self._any_case else room.room_id
+            for s in self._needle:
+                if s in room_id:
+                    return True
+            return False
+        except:
+            return False
+
+class RoomIdStartsWithFilter(BaseEventFilter):
+    """
+        This filter returns True if the `room.room_id` starts with `substring`
+        (or any of substrings from the list). The check will be case insensetive
+        if `any_case` is True.
+    """
+    def __init__(self, substring: str | list[str], *, any_case: bool = True):
+        super().__init__()
+        if type(substring) is str:
+            substring = [substring]
+        self._any_case = any_case
+        if self._any_case:
+            self._substring = [s.lower() for s in substring]
+        else:
+            self._substring = list(substring)
+
+    def __repr__(self):
+        return f"RoomIdStartsWithFilter({repr(self._substring)}, any_case={repr(self._any_case)})"
+
+    async def __call__(self, room: MatrixRoom, event: Event, client: AsyncClient) -> bool:
+        try:
+            room_id = room.room_id.lower() if self._any_case else room.room_id
+            for s in self._substring:
+                if room_id.startswith(s):
+                    return True
+            return False
+        except:
+            return False
+
+class RoomIdEndsWithFilter(BaseEventFilter):
+    """
+        This filter returns True if the `room.room_id` ends with `substring` (or
+        any of substrings from the list). The check will be case insensetive
+        if `any_case` is True.
+    """
+    def __init__(self, substring: str | list[str], *, any_case: bool = True):
+        super().__init__()
+        if type(substring) is str:
+            substring = [substring]
+        self._any_case = any_case
+        if self._any_case:
+            self._substring = [s.lower() for s in substring]
+        else:
+            self._substring = list(substring)
+
+    def __repr__(self):
+        return f"RoomIdEndsWithFilter({repr(self._substring)}, any_case={repr(self._any_case)})"
+
+    async def __call__(self, room: MatrixRoom, event: Event, client: AsyncClient) -> bool:
+        try:
+            room_id = room.room_id.lower() if self._any_case else room.room_id
+            for s in self._substring:
+                if room_id.endswith(s):
+                    return True
+            return False
+        except:
+            return False
+
+class RoomEncryptedFilter(BaseEventFilter):
+    """
+    This filter returns True if the room is encrypted.
+    """
+    def __init__(self):
+        super().__init__()
+
+    def __repr__(self):
+        return f"RoomEncryptedFilter()"
+
+    async def __call__(self, room: MatrixRoom, event: Event, client: AsyncClient) -> bool:
+        try:
+            return room.encrypted
+        except:
+            return False