diff --git a/pkgs/matrix-bot/matrix_bot/__init__.py b/pkgs/matrix-bot/matrix_bot/__init__.py index 40e2b013..d81f1cb4 100644 --- a/pkgs/matrix-bot/matrix_bot/__init__.py +++ b/pkgs/matrix-bot/matrix_bot/__init__.py @@ -1,4 +1,100 @@ -from .main import main +import argparse +import asyncio +import logging +import os +import sys +from pathlib import Path + +from matrix_bot.custom_logger import setup_logging +from matrix_bot.main import bot_main +from matrix_bot.matrix import MatrixBotData + +log = logging.getLogger(__name__) + +curr_dir = Path(__file__).parent + + +def create_parser(prog: str | None = None) -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + prog=prog, + description="A gitea bot for matrix", + formatter_class=argparse.RawTextHelpFormatter, + ) + + parser.add_argument( + "--debug", + help="Enable debug logging", + action="store_true", + default=False, + ) + + parser.add_argument( + "--server", + help="The matrix server to connect to", + default="https://matrix.clan.lol", + ) + + parser.add_argument( + "--user", + help="The matrix user to connect as", + default="@clan-bot:clan.lol", + ) + + 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 + + +def main() -> None: + parser = create_parser() + args = parser.parse_args() + + if args.debug: + setup_logging(logging.DEBUG, root_log_name=__name__.split(".")[0]) + log.debug("Debug log activated") + else: + setup_logging(logging.INFO, root_log_name=__name__.split(".")[0]) + + 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) + if __name__ == "__main__": main() diff --git a/pkgs/matrix-bot/matrix_bot/__main__.py b/pkgs/matrix-bot/matrix_bot/__main__.py index 40e2b013..868d99ef 100644 --- a/pkgs/matrix-bot/matrix_bot/__main__.py +++ b/pkgs/matrix-bot/matrix_bot/__main__.py @@ -1,4 +1,4 @@ -from .main import main +from . import main if __name__ == "__main__": main() diff --git a/pkgs/matrix-bot/matrix_bot/bot.py b/pkgs/matrix-bot/matrix_bot/bot.py index b7a7c032..14b0ff78 100644 --- a/pkgs/matrix-bot/matrix_bot/bot.py +++ b/pkgs/matrix-bot/matrix_bot/bot.py @@ -1,116 +1,18 @@ 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" diff --git a/pkgs/matrix-bot/matrix_bot/main.py b/pkgs/matrix-bot/matrix_bot/main.py index 0f34c02a..bfb0a5e8 100644 --- a/pkgs/matrix-bot/matrix_bot/main.py +++ b/pkgs/matrix-bot/matrix_bot/main.py @@ -1,95 +1,45 @@ -import argparse -import asyncio import logging -import os -import sys from pathlib import Path -from matrix_bot.bot import MatrixBotData, bot_main -from matrix_bot.custom_logger import setup_logging +import aiohttp + +from matrix_bot.matrix import MatrixBotData log = logging.getLogger(__name__) curr_dir = Path(__file__).parent +from nio import ( + AsyncClient, + ProfileGetAvatarResponse, + RoomMessageText, +) -def create_parser(prog: str | None = None) -> argparse.ArgumentParser: - parser = argparse.ArgumentParser( - prog=prog, - description="A gitea bot for matrix", - formatter_class=argparse.RawTextHelpFormatter, - ) - - parser.add_argument( - "--debug", - help="Enable debug logging", - action="store_true", - default=False, - ) - - parser.add_argument( - "--server", - help="The matrix server to connect to", - default="https://matrix.clan.lol", - ) - - parser.add_argument( - "--user", - help="The matrix user to connect as", - default="@clan-bot:clan.lol", - ) - - 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 +from matrix_bot.bot import bot_loop, message_callback +from matrix_bot.matrix import set_avatar, upload_image -def main() -> None: - parser = create_parser() - args = parser.parse_args() +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) - if args.debug: - setup_logging(logging.DEBUG, root_log_name=__name__.split(".")[0]) - log.debug("Debug log activated") + 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: - setup_logging(logging.INFO, root_log_name=__name__.split(".")[0]) - - 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, - ) + log.debug(f"Avatar already set to {avatar.avatar_url}") try: - asyncio.run(bot_main(data)) - except KeyboardInterrupt: - print("User Interrupt", file=sys.stderr) + 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()