matrix-bot/matrix_bot/cli.py

import logging
import sys

import click

from .bot import Bot
from .client import send_message as client_send_message
from .config import Config
from .exceptions import NoRoomException
from .message import MarkdownMessage, TextMessage
from .utils import run_async, setup_logger

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()
    if client:
        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))


@cli.group()
@click.pass_context
def client(ctx: click.Context) -> None:
    pass


@client.command('send')
@click.pass_context
@click.argument('room_id')
@click.argument('message', nargs=-1, required=True)
def client_send(ctx: click.Context, room_id: str, message: list) -> None:
    run_async(client_send_message(room_id, ' '.join(message)))


if __name__ == '__main__':
    cli(obj={})