Added Member Class Information retrival

examples/member.py

@@ -0,0 +1,28 @@
+from matrix import Bot
+
+bot = Bot()
+
+
+@bot.command()
+async def profile(ctx):
+    member = ctx.member
+
+    name = await member.get_display_name()
+    avatar = await member.get_avatar_url()
+    level = await member.get_room_power_level(ctx.room)
+    presence = await member.get_presence()
+    has_permission = await member.has_room_permission(ctx.room, "ban")
+    has_event_permission = await member.has_event_permission(ctx.room, "m.room.message")
+
+    await ctx.reply(
+        f"Welcome {member.mention()}!\n"
+        f"Display name: {name}\n"
+        f"Avatar URL: {avatar}\n"
+        f"Power level: {level}\n"
+        f"Presence: {presence}\n"
+        f"Has permission to ban users: {has_permission}\n"
+        f"Has permission to send messages: {has_event_permission}"
+    )
+
+
+bot.start(config="config.yaml")

matrix/context.py

@@ -7,6 +7,7 @@
 from .message import Message
 from .room import Room
 from .types import File, Image
+from .member import Member
 
 if TYPE_CHECKING:
     from .bot import Bot  # pragma: no cover
@@ -23,6 +24,7 @@ def __init__(self, bot: "Bot", room: Room, event: Event):
         self.bot = bot
         self.room = room
         self.event = event
+        self.member = Member(event.sender, bot.client)
 
         self.body: str = getattr(event, "body", "")
         self.sender: str = event.sender

matrix/member.py

@@ -0,0 +1,170 @@
+from matrix.room import Room
+from nio import AsyncClient
+from matrix.api import matrix_call
+from dataclasses import dataclass
+from typing import Any
+
+
+@dataclass
+class MemberProfile:
+    displayname: str | None
+    avatar_url: str | None
+    other_info: dict[Any, Any]
+
+
+class Member:
+    def __init__(self, user_id: str, client: AsyncClient) -> None:
+        self._user_id: str = user_id
+        self._client: AsyncClient = client
+
+    def __str__(self) -> str:
+        return self.mention()
+
+    @property
+    def user_id(self) -> str:
+        return self._user_id
+
+    def mention(self) -> str:
+        """Get a Markdown-formatted mention link (matrix.to pill) for this member.
+
+        ## Example
+
+        ```python
+        await ctx.reply(f"Welcome {member.mention()}!")
+        ```
+        """
+        return f"[{self._user_id}](https://matrix.to/#/{self._user_id})"
+
+    async def get_profile(self) -> MemberProfile | None:
+        """Get the profile information for this member.
+
+        ## Example
+
+        ```python
+        profile = await member.get_profile()
+        for key, value in profile.other_info.items():
+            print(f"{key}: {value}")
+        ```
+        """
+        profile = await matrix_call(
+            self._client.get_profile(self._user_id),
+            error_message=f"Failed to get profile for user {self._user_id}",
+        )
+        if profile:
+            return MemberProfile(
+                displayname=profile.displayname,
+                avatar_url=profile.avatar_url,
+                other_info=profile.other_info,
+            )
+        return None
+
+    async def get_display_name(self) -> str | None:
+        """Get the display name for this member.
+
+        ## Example
+
+        ```python
+        display_name = await member.get_display_name()
+        print(f"Display name: {display_name}")
+        ```
+        """
+        display_name = await matrix_call(
+            self._client.get_displayname(self._user_id),
+            error_message=f"Failed to get display name for user {self._user_id}",
+        )
+        return display_name.displayname if display_name else None
+
+    async def get_avatar_url(self) -> str | None:
+        """Get the avatar URL for this member.
+
+        ## Example
+
+        ```python
+        avatar_url = await member.get_avatar_url()
+        print(f"Avatar URL: {avatar_url}")
+        ```
+        """
+        avatar = await matrix_call(
+            self._client.get_avatar(self._user_id),
+            error_message=f"Failed to get avatar for user {self._user_id}",
+        )
+        return await self._client.mxc_to_http(avatar.avatar_url) if avatar else None
+
+    async def get_presence(self) -> str | None:
+        """Get the presence status for this member.
+
+        ## Example
+
+        ```python
+        presence = await member.get_presence()
+        print(f"Presence status: {presence}")
+        ```
+        """
+        presence = await matrix_call(
+            self._client.get_presence(self._user_id),
+            error_message=f"Failed to get presence for user {self._user_id}",
+        )
+        return presence.presence if presence else None
+
+    async def get_room_power_level(self, room: Room) -> int:
+        """Get the power level for this member in a specific room.
+
+        ## Example
+
+        ```python
+        level = await member.get_room_power_level(ctx.room)
+        print(f"Power level in room {room.room_id}: {level}")
+        ```
+        """
+        power_level = await room.get_state_event("m.room.power_levels", "")
+        if power_level is None:
+            return 0
+
+        content = power_level.content
+        users = content.get("users", {})
+        default = content.get("users_default", 0)
+
+        return int(users.get(self._user_id, default))
+
+    async def has_room_permission(self, room: Room, permission: str) -> bool:
+        """Check if this member has a specific permission in a room.
+
+        ## Example
+
+        ```python
+        has_permission = await member.has_room_permission(ctx.room, "ban")
+        print(f"Has ban permission: {has_permission}")
+        ```
+        """
+        power_level = await self.get_room_power_level(room)
+        power_levels_event = await room.get_state_event("m.room.power_levels", "")
+        if power_levels_event is None:
+            return False
+
+        content = power_levels_event.content
+        required_level = content.get(permission)
+        if required_level is None:
+            return False  # Permission not defined in the room's power levels
+
+        return bool(power_level >= required_level)
+
+    async def has_event_permission(self, room: Room, event_type: str) -> bool:
+        """Check if this member has permission to send a specific event type in a room.
+
+        ## Example
+
+        ```python
+        can_send_message = await member.has_event_permission(ctx.room, "m.room.message")
+        print(f"Can send messages: {can_send_message}")
+        ```
+        """
+        power_level = await self.get_room_power_level(room)
+        power_levels_event = await room.get_state_event("m.room.power_levels", "")
+        if power_levels_event is None:
+            return False
+
+        content = power_levels_event.content
+        events = content.get("events", {})
+        required_level = events.get(event_type, 0)
+
+        return bool(power_level >= required_level)

matrix/room.py

@@ -1,6 +1,12 @@
-from typing import Any
-
-from nio import AsyncClient, MatrixRoom, Event
+from typing import Any, Union
+
+from nio import (
+    AsyncClient,
+    MatrixRoom,
+    Event,
+    RoomGetStateEventResponse,
+    RoomGetStateEventError,
+)
 
 from matrix.api import matrix_call
 from matrix.message import Message
@@ -97,6 +103,14 @@ def encrypted(self) -> bool:
         """Whether the room is encrypted."""
         return self._matrix_room.encrypted  # type: ignore[no-any-return]
 
+    async def get_state_event(
+        self, event_type: str, state_key: str = ""
+    ) -> Union[RoomGetStateEventResponse, RoomGetStateEventError]:
+        return await matrix_call(
+            self._client.room_get_state_event(self.room_id, event_type, state_key),
+            error_message=f"Failed to get state event for room {self._matrix_room.room_id}",
+        )
+
     async def send(
         self,
         content: str | None = None,