matrix-bot: Code cleanup

pkgs/matrix-bot/bin/bot

@@ -10,3 +10,4 @@ from matrix_bot import main  # NOQA
 
 if __name__ == "__main__":
     main()
+

pkgs/matrix-bot/matrix_bot/bot.py

@@ -0,0 +1,140 @@
+import logging
+import os
+from pathlib import Path
+
+log = logging.getLogger(__name__)
+
+from dataclasses import dataclass
+
+import aiohttp
+from nio import (
+    AsyncClient,
+    JoinResponse,
+    MatrixRoom,
+    ProfileGetAvatarResponse,
+    ProfileSetAvatarResponse,
+    RoomMessageText,
+    RoomSendResponse,
+    UploadResponse,
+)
+
+
+async def fetch_repo_labels(
+    url: str,
+    owner: str,
+    repo: str,
+    session: aiohttp.ClientSession,
+    access_token: str | None = None,
+) -> list[dict]:
+    """
+    Asynchronously fetch labels from a GitHub repository.
+
+    Args:
+        url (str): Base URL of the API (e.g., "https://api.github.com").
+        owner (str): Repository owner.
+        repo (str): Repository name.
+        access_token (str): GitHub access token for authentication (optional).
+
+    Returns:
+        list: List of labels in the repository.
+    """
+    endpoint = f"{url}/repos/{owner}/{repo}/labels"
+    headers = {"Accept": "application/vnd.github.v3+json"}
+    if access_token:
+        headers["Authorization"] = f"token {access_token}"
+
+    async with session.get(endpoint, headers=headers) as response:
+        if response.status == 200:
+            labels = await response.json()
+            return labels
+        else:
+            # You may want to handle different statuses differently
+            raise Exception(
+                f"Failed to fetch labels: {response.status}, {await response.text()}"
+            )
+
+
+async def upload_image(client: AsyncClient, image_path: str) -> str:
+    with open(image_path, "rb") as image_file:
+        response: UploadResponse
+        response, _ = await client.upload(image_file, content_type="image/png")
+        if not response.transport_response.ok:
+            raise Exception(f"Failed to upload image {response}")
+        return response.content_uri  # This is the MXC URL
+
+
+async def set_avatar(client: AsyncClient, mxc_url: str) -> None:
+    response: ProfileSetAvatarResponse
+    response = await client.set_avatar(mxc_url)
+    if not response.transport_response.ok:
+        raise Exception(f"Failed to set avatar {response}")
+
+
+@dataclass
+class MatrixBotData:
+    server: str
+    user: str
+    avatar: Path
+    repo_owner: str
+    repo_name: str
+    matrix_room: str
+    password: str
+
+
+async def bot_main(
+    m: MatrixBotData,
+) -> None:
+    log.info(f"Connecting to {m.server} as {m.user}")
+    client = AsyncClient(m.server, m.user)
+    client.add_event_callback(message_callback, RoomMessageText)
+
+    password = os.getenv("MATRIX_PASSWORD")
+    if not password:
+        log.error("No password provided set the MATRIX_PASSWORD environment variable")
+
+    log.info(await client.login(m.password))
+
+    avatar: ProfileGetAvatarResponse = await client.get_avatar()
+    if not avatar.avatar_url:
+        mxc_url = await upload_image(client, m.avatar)
+        log.info(f"Uploaded avatar to {mxc_url}")
+        await set_avatar(client, mxc_url)
+    else:
+        log.debug(f"Avatar already set to {avatar.avatar_url}")
+
+    try:
+        async with aiohttp.ClientSession() as session:
+            await bot_loop(client, session, m.matrix_room)
+    except Exception as e:
+        log.exception(e)
+    finally:
+        await client.close()
+
+
+async def message_callback(room: MatrixRoom, event: RoomMessageText) -> None:
+    print(
+        f"Message received in room {room.display_name}\n"
+        f"{room.user_name(event.sender)} | {event.body}"
+    )
+
+
+async def bot_loop(
+    client: AsyncClient, http: aiohttp.ClientSession, matrix_room: str
+) -> None:
+    # If you made a new room and haven't joined as that user, you can use
+    room: JoinResponse = await client.join(matrix_room)
+
+    log.info("Sending message...")
+
+    res: RoomSendResponse = await client.room_send(
+        # Watch out! If you join an old room you'll see lots of old messages
+        room_id=room.room_id,
+        message_type="m.room.message",
+        content={"msgtype": "m.text", "body": "Hello world!"},
+    )
+
+    if not res.transport_response.ok:
+        await client.close()
+        raise Exception(f"Failed to send message {res}")
+
+    await client.sync_forever(timeout=30000)  # milliseconds

pkgs/matrix-bot/matrix_bot/main.py

@@ -1,12 +1,17 @@
 import argparse
 import asyncio
 import logging
+import os
+import sys
+from pathlib import Path
 
-from matrix_bot.bot import bot_main
+from matrix_bot.bot import MatrixBotData, bot_main
 from matrix_bot.custom_logger import setup_logging
 
 log = logging.getLogger(__name__)
 
+curr_dir = Path(__file__).parent
+
 
 def create_parser(prog: str | None = None) -> argparse.ArgumentParser:
     parser = argparse.ArgumentParser(
@@ -26,14 +31,35 @@ def create_parser(prog: str | None = None) -> argparse.ArgumentParser:
         "--server",
         help="The matrix server to connect to",
         default="https://matrix.clan.lol",
-        required=True,
     )
 
     parser.add_argument(
         "--user",
         help="The matrix user to connect as",
         default="@clan-bot:clan.lol",
-        required=True,
+    )
+
+    parser.add_argument(
+        "--avatar",
+        help="The path to the image to use as the avatar",
+        default=curr_dir / "avatar.png",
+    )
+
+    parser.add_argument(
+        "--repo-owner",
+        help="The owner of gitea the repository",
+        default="clan",
+    )
+    parser.add_argument(
+        "--repo-name",
+        help="The name of the repository",
+        default="clan-core",
+    )
+
+    parser.add_argument(
+        "--matrix-room",
+        help="The matrix room to join",
+        default="#bot-test:gchq.icu",
     )
 
     return parser
@@ -49,4 +75,21 @@ def main() -> None:
     else:
         setup_logging(logging.INFO, root_log_name=__name__.split(".")[0])
 
-    asyncio.run(bot_main(args.server, args.user))
+    password = os.getenv("MATRIX_PASSWORD")
+    if not password:
+        log.error("No password provided set the MATRIX_PASSWORD environment variable")
+
+    data = MatrixBotData(
+        args.server,
+        args.user,
+        args.avatar,
+        args.repo_owner,
+        args.repo_name,
+        args.matrix_room,
+        password,
+    )
+
+    try:
+        asyncio.run(bot_main(data))
+    except KeyboardInterrupt:
+        print("User Interrupt", file=sys.stderr)

pkgs/matrix-bot/pyproject.toml

@@ -6,7 +6,7 @@ build-backend = "setuptools.build_meta"
 name = "matrix-bot"
 description = "matrix bot for release messages from git commits"
 dynamic = ["version"]
-scripts = { clan = "matrix_bot:main" }
+scripts = { mbot = "matrix_bot:main" }
 license = {text = "MIT"}
 
 [project.urls]

pkgs/matrix-bot/shell.nix

@@ -16,10 +16,7 @@ let
     ]);
 in
 mkShell {
-  buildInputs = [
-
-    ruff
-  ] ++ devshellTestDeps;
+  buildInputs = [ ruff ] ++ devshellTestDeps;
 
   PYTHONBREAKPOINT = "ipdb.set_trace";