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']

async def main():
    try:
        matrix_loop = None

        matrix_loop = MatrixLoop(MATRIX_HOMESERVER_URL, MATRIX_FULL_USER_ID,
                                 MATRIX_PASSWORD)
        await matrix_loop.prepare()

        telegram_loop = TelegramLoop(TELEGRAM_BOT_TOKEN)

        await asyncio.gather(
            matrix_loop.run(),
            telegram_loop.run(),
        )
    finally:
        if matrix_loop:
            await 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__':
    asyncio.run(main())