matrix-bot: Code cleanup#
This commit is contained in:
parent
00002c787c
commit
5f9ffb4b66
|
|
@ -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()
|
||||
|
|
|
|||
|
|
@ -1,4 +1,4 @@
|
|||
from .main import main
|
||||
from . import main
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
|
|
|
|||
|
|
@ -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"
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
|
|
|
|||
Loading…
Reference in New Issue
Block a user