MirrorTea/mirrortea/__main__.py

import asyncio
import os
import sys

import aiogram as telegram
import nio as matrix

MATRIX_HOMESERVER_URL = os.environ['MATRIX_HOMESERVER_URL']
MATRIX_FULL_USER_ID   = os.environ['MATRIX_FULL_USER_ID']
MATRIX_PASSWORD       = os.environ['MATRIX_PASSWORD']
TELEGRAM_BOT_TOKEN    = os.environ['TELEGRAM_BOT_TOKEN']

def main():
    asyncio.run(Application().run())

class Application:
    def __init__(self):
        self.matrix_loop = MatrixLoop(
            MATRIX_HOMESERVER_URL,
            MATRIX_FULL_USER_ID,
            MATRIX_PASSWORD,
        )
        self.telegram_loop = TelegramLoop(TELEGRAM_BOT_TOKEN)

    async def run(self):
        try:
            await self.matrix_loop.prepare()
            await asyncio.gather(
                self.matrix_loop.run(),
                self.telegram_loop.run(),
            )
        finally:
            if self.matrix_loop:
                await self.matrix_loop.finish()

class MatrixLoop:
    def __init__(self, homeserver_url, full_user_id, password):
        self.password = password
        self.client = matrix.AsyncClient(homeserver_url, full_user_id)
        self.client.add_event_callback(self.on_message, matrix.RoomMessage)

    async def prepare(self):
        await self.client.login(self.password)

    async def finish(self):
        await self.client.close()

    async def run(self):
        await self.client.sync_forever(timeout=30000)

    async def on_message(self, room, event):
        print(room, event, file=sys.stderr)

class TelegramLoop:
    def __init__(self, bot_token):
        self.bot = telegram.Bot(token=bot_token)
        self.dispatcher = telegram.Dispatcher(bot=self.bot)
        self.dispatcher.register_message_handler(self.on_message)

    async def run(self):
        await self.dispatcher.start_polling()

    async def on_message(self, msg):
        print(msg, file=sys.stderr)

if __name__ == '__main__':
    main()