import logging import sys import click from .bot import Bot from .utils import run_async, setup_logger from .config import Config from .message import TextMessage, MarkdownMessage from .exceptions import NoRoomException logger = setup_logger(__name__) def get_loglevel(verbosity: int) -> int: level = logging.ERROR if verbosity == 1: level = logging.WARNING elif verbosity == 2: level = logging.INFO elif verbosity >= 3: level = logging.DEBUG return level async def send_message(room_id: str, message: str, markdown: bool) -> None: logger.debug('Sending a message to %s', room_id) bot = Bot() client = await bot.login() await bot.sync() room = await bot.find_room_by_id(room_id) if markdown: message = MarkdownMessage(message) else: message = TextMessage(message) try: await message.send(client, room) except NoRoomException: click.echo(f'No Room with id {room_id} found') finally: await bot.shutdown() async def main_run() -> None: logger.debug('Starting') bot = Bot() await bot.run() async def main_verify() -> None: logger.debug('Starting') bot = Bot() await bot.verify() @click.group(invoke_without_command=True) @click.option('-v', '--verbose', count=True, help='Set the loglevel (Can be used multiple times e.g. -vvv)') @click.option('-b', '--botname', default='default', help='The name of the bot to start') @click.pass_context def cli(ctx: click.Context, verbose: bool, botname: str, help: str = 'matrix-bot - A Bot That listens in matrix chatrooms' ) -> None: ctx.ensure_object(dict) Config.init(botname, get_loglevel(verbose)) if ctx.invoked_subcommand is None: click.echo(ctx.get_help()) sys.exit(1) @cli.command(help='Start the Bot') @click.pass_context @click.option('-v', '--verbose', count=True, help='Set the loglevel (Can be used multiple times e.g. -vvv)') def run(ctx: click.Context, verbose: int) -> None: run_async(main_run()) @cli.command(help='Verify the bot with a matrix homeserver') @click.pass_context @click.option('-v', '--verbose', count=True, help='Set the loglevel (Can be used multiple times e.g. -vvv)') def verify(ctx: click.Context, verbose: int) -> None: run_async(main_verify()) @cli.command(help='Send a message with the bot') @click.pass_context @click.argument('room_id') @click.argument('message', nargs=-1, required=True) @click.option('-m', '--markdown/--no-markdown', default=False) @click.option('-v', '--verbose', count=True, help='Set the loglevel (Can be used multiple times e.g. -vvv)') def send(ctx: click.Context, room_id: str, message: list, markdown: bool, verbose: int) -> None: run_async(send_message(room_id, ' '.join(message), markdown)) if __name__ == '__main__': cli(obj={})