create bot class

This commit is contained in:
agatha 2024-06-06 13:59:53 -04:00
parent 53020ae0f2
commit b6c6788edf
2 changed files with 47 additions and 45 deletions

46
main.py
View File

@ -1,52 +1,8 @@
"""Matrix Bot Framework""" """Matrix Bot Framework"""
import asyncio import asyncio
import json import json
from loguru import logger
from nio import AsyncClient, LoginResponse
from matrix import MatrixBot
class MatrixBot:
def __init__(self, config: dict):
# TODO: Test configuration for required settings
self.config = config
self.client = AsyncClient(
homeserver=self.config['homeserver'],
user=self.config['username']
)
async def send_message(self, message: str):
if not self.client.access_token:
logged_in = await self.login()
if not logged_in:
return
await self.client.room_send(
room_id=self.config['room_id'],
message_type="m.room.message",
content={
"msgtype": "m.text",
"body": message
}
)
logger.info("Message sent")
async def login(self):
response = await self.client.login(
password=self.config['password']
)
if isinstance(response, LoginResponse):
logger.info(f"Logged in as {self.config['username']}")
else:
logger.error(f"Failed to login as {self.config['username']}: {response}")
return False
return True
async def logout(self):
await self.client.logout()
await self.client.close()
logger.info(f"Logged out from {self.config['homeserver']}")
def load_config(path: str) -> dict: def load_config(path: str) -> dict:

46
matrix.py Normal file
View File

@ -0,0 +1,46 @@
from loguru import logger
from nio import AsyncClient, LoginResponse
class MatrixBot:
def __init__(self, config: dict):
# TODO: Test configuration for required settings
self.config = config
self.client = AsyncClient(
homeserver=self.config['homeserver'],
user=self.config['username']
)
async def send_message(self, message: str):
if not self.client.access_token:
logged_in = await self.login()
if not logged_in:
return
await self.client.room_send(
room_id=self.config['room_id'],
message_type="m.room.message",
content={
"msgtype": "m.text",
"body": message
}
)
logger.info("Message sent")
async def login(self):
response = await self.client.login(
password=self.config['password']
)
if isinstance(response, LoginResponse):
logger.info(f"Logged in as {self.config['username']}")
else:
logger.error(f"Failed to login as {self.config['username']}: {response}")
return False
return True
async def logout(self):
await self.client.logout()
await self.client.close()
logger.info(f"Logged out from {self.config['homeserver']}")