MirrorTea/mirrortea/__main__.py

import asyncio
import os
import sqlite3
import sys

import aiogram as telegram
import nio as matrix

def main():
    config = Config(
        db_path=os.environ['DB_PATH'],
        matrix_bot_id=os.environ['MATRIX_BOT_ID'],
        matrix_homeserver_url=os.environ['MATRIX_HOMESERVER_URL'],
        matrix_owner_id=os.environ['MATRIX_OWNER_ID'],
        matrix_password=os.environ['MATRIX_PASSWORD'],
        telegram_bot_token=os.environ['TELEGRAM_BOT_TOKEN'],
    )

    asyncio.run(Application(config).run())

class Config:
    def __init__(self, **kwargs):
        self.db_path               = kwargs['db_path']
        self.matrix_bot_id         = kwargs['matrix_bot_id']
        self.matrix_homeserver_url = kwargs['matrix_homeserver_url']
        self.matrix_owner_id       = kwargs['matrix_owner_id']
        self.matrix_password       = kwargs['matrix_password']
        self.telegram_bot_token    = kwargs['telegram_bot_token']

class Application:
    def __init__(self, config):
        self.config = config
        self.sqlite_adapter = SqliteAdapter(self, config.db_path)
        self.matrix_loop = MatrixLoop(
            self,
            config.matrix_bot_id,
            config.matrix_homeserver_url,
            config.matrix_owner_id,
            config.matrix_password,
        )
        self.telegram_loop = TelegramLoop(
            self,
            config.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, app, bot_id, homeserver_url, owner_id, password):
        self.app            = app
        self.bot_id         = bot_id
        self.homeserver_url = homeserver_url
        self.owner_id       = owner_id
        self.password       = password

        self.client = matrix.AsyncClient(homeserver_url, bot_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, app, bot_token):
        self.app = app
        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)

class SqliteAdapter:
    def __init__(self, app, path):
        self.app = app
        self.path = path
        self.conn = sqlite3.connect(path)

if __name__ == '__main__':
    main()