aoc/main.py

import logging
import argparse
import os
import asyncio
import yaml
import importlib
import pathlib
from nio import AsyncClient, LoginResponse

from skills import *

logging.basicConfig(format='[%(asctime)s] %(process)s {%(filename)s:%(lineno)d} %(levelname)s - %(message)s', level=logging.INFO)

class MatrixBot:
    __AUTHOR__      = 'anima'
    __VERSION__     = '0.6.3'
    def __init__(self, config_file: str = 'bot.conf', session_file: str = 'session.tmp') -> None:
        """
        Args:
            config_file (str, optional): path to config file
            session_file (str, optional): path to session file
        """
        self.botname            = 'Bot'
        self.config_file        = config_file
        self.session_file       = session_file
        self._config             = dict()
        self.silence            = False
        self.skills             = list()
        self._find_skills()
        
        if self._load_config():
            self.client = AsyncClient(self.homeserver, self.username)
            self.helper = BotHelper(self.client)
        else:
            logging.critical('no config file was found')
            exit(1)

    def _load_config(self) -> bool:
        """load given config file"""
        if not os.path.exists(self.config_file):
            return False
        logging.info(f'load config from file {self.config_file}')
        with open(self.config_file, 'r') as f:
            self._config = yaml.safe_load(f)
        self.botname    = self._config['general']['botname']
        self.domain     = self._config['general']['domain']
        self.homeserver = f"https://{self.domain}"
        self.username   = f"@{self._config['general']['user']}:{self._config['general']['domain']}"
        self.__password   = self._config['general']['pass']
        del self._config['general']['pass']
        return True

    def _save_session(self) -> bool:
        """save session data as file

        Returns:
            bool: True if file was saved
        """
        session = {
            'access_token': self.session.access_token,
            'device_id': self.session.device_id,
            'next_batch': self.client.next_batch
        }
        logging.debug('save session file')
        with open(self.session_file, 'w') as f:
            yaml.dump(session, f)
        return True

    def _load_session(self) -> bool:
        """load session date from file

        Returns:
            bool: True if file was found and load
        """
        if not os.path.exists(self.session_file):
            return False
        logging.info(f'restore session from file {self.session_file}')
        with open(self.session_file, 'r') as f:
            session = yaml.safe_load(f)
        self.client.access_token    = session['access_token']
        self.client.device_id       = session['device_id']
        self.client.next_batch      = session['next_batch']
        return True

    async def _login(self) -> bool:
        """restore login with session file or run full login process

        Returns:
            bool: True when bot has a valid login
        """
        if self._load_session():
            self.session = await self.client.sync(timeout=3000)
            if self.session.transport_response.ok:
                logging.debug('restore session successfully')
                return True
        else:
            logging.warning('no session file was found or file is broken')

        logging.debug(f'try login with account {self.username}')
        self.session = await self.client.login(self.__password)
        if isinstance(self.session, LoginResponse):
            logging.info(f'login successfull with account {self.username} on server {self.homeserver}')
            logging.info(self.session)
            self._save_session()
            return True
        return False

    def _find_skills(self):
        """search all modules in skill folder and init these"""
        skills_path = pathlib.Path(__file__).parent / "skills"
        for module in skills_path.iterdir():
            if module.is_dir() and (module / "__init__.py").exists():
                importlib.import_module(f"skills.{module.name}")

    async def _activate_skills(self):
        """activate all skills which defined in config file"""
        for skill in self._config['skills'].keys():
            await self.load_skill(skill_name=skill, config=self._config['skills'][skill])

    async def load_skill(self, skill_name: str, config: dict = None) -> bool:
        """load skill from skill registry

        Args:
            skill_name (str): name of skill (same as classname)
            config (dict, optional): additional config. Defaults to None.

        Returns:
            bool: True is skill is loaded
        """
        if not skill_name in Skill.registry.keys():
            return False
        logging.info(f'load skill {skill_name} version {Skill.registry[skill_name].__VERSION__}')
        skill = Skill.registry[skill_name](client=self.client, config=config)
        await skill.load()
        self.skills.append(skill)
        return True

    def unload_skill(self):
        pass

    async def send_status(self, message: str) -> bool:
        """send a message to a predefined room

        Args:
            message (str): message content

        Returns:
            bool: True if message send
        """
        if self.silence:
            return False
        await self.helper.send_message(f"{self._config['general']['status_room']}:{self.domain}", message)
        return True

    async def run(self):
        if await self._login():
            await self.send_status(f"{self.botname} Online!")
            await self._activate_skills()
            await self.client.sync_forever(timeout=30000, set_presence='online')
            await self.set_displayname(self.conf['general']['botname'])


def parse_args() -> argparse.Namespace: 
    """evaluates given arguments

    Returns:
        argsparse.Namespace: Namespace Object with all arguments insert (use: args.long_name_of_argument)
    """
    argp = argparse.ArgumentParser(description=__doc__)
    # Default args
    argp.add_argument('-v', '--verbose', action='count', default=0,
        help='increase output verbosity (use up to 3 times)')
    argp.add_argument('-c', '--config', default='aoc.conf',
        help='path to config file')
    argp.add_argument('-s', '--session', default='session.tmp',
        help='path to session file')

    args = argp.parse_args()
    return args

async def main():
    args = parse_args()
    if 1 <= args.verbose <= 3:
        logging.getLogger().setLevel(40 - (args.verbose * 10))
    bot = MatrixBot(config_file=args.config, session_file=args.session)

    bot.silence = True
    await bot.run()


if __name__ == '__main__': 
    asyncio.run(main())