Initial commit
This commit is contained in:
AO-IT Net
@@ -0,0 +1,192 @@
|
||||
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] %(levelname)s: %(message)s', level=logging.INFO)
|
||||
|
||||
class MatrixBot:
|
||||
__AUTHOR__ = 'anima'
|
||||
__VERSION__ = '0.6.1'
|
||||
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)
|
||||
self._activate_skills()
|
||||
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,
|
||||
'user_id': self.user_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.user_id = session["user_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(session, LoginResponse):
|
||||
logging.info(f'login successfull with account {self.username} on server {self.homeserver}')
|
||||
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}")
|
||||
|
||||
def _activate_skills(self):
|
||||
"""activate all skills which defined in config file"""
|
||||
for skill in self._config['skills'].keys():
|
||||
self.load_skill(skill_name=skill, config=self._config['skills'][skill])
|
||||
|
||||
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)
|
||||
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.client.sync_forever(timeout=30000)
|
||||
|
||||
|
||||
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())
|
||||
Reference in New Issue
Block a user