Python库 | BotpySE-0.7.5.tar.gz

# # Bot.py # Botpy # # Created by Ashish Ahuja on 1st August 2017. # # import os import sys import json import logging import jsonpickle as jp import pyRedunda as redunda from .CommandManager import * from .BackgroundTaskManager import * from .BackgroundTask import * from . import ChatRoom from . import ChatUser from . import Utilities from . import Redunda class Bot(ce.client.Client): def __init__(self, bot_name, commands, room_ids, background_tasks=[], host='stackexchange.com', email=None, password=None, send_aggressively=False): super().__init__(host, email, password, send_aggressively=send_aggressively) background_tasks.append(BackgroundTask(self._stop_reason_check, interval=5)) background_tasks.append(BackgroundTask(self._save_users, interval=60)) self.name = bot_name self.aliases = [] self.is_alive = False self.at_standby = False self._ids = room_ids self.commands = commands self._command_manager = CommandManager(commands) self._users = list() self._rooms = list() self._storage_prefix = os.path.expanduser("~") + "/." + self.name.lower() + "/" self._version = None self._location = "unknown location" self._startup_message = self.name + " starting..." self._standby_message = "Switching to standby mode." self._failover_message = "Failover received." background_tasks.append(BackgroundTask(self._command_manager.cleanup_finished_commands, interval=3)) self._background_task_manager = BackgroundTaskManager(background_tasks) # Redunda storage self._redunda = None self._redunda_key = None self._sync_files = list() self._using_redunda = False self._redunda_task_manager = None try: with open(self._storage_prefix + 'location.txt', 'r') as file_handle: content = [x.rstrip('\n') for x in file_handle.readlines()] if len(content) != 2: raise ValueError('Invalid format in "location.txt"') self._location = content[0] + "/" + content[1] logging.info(self._location) except IOError as ioerr: logging.error(str(ioerr)) logging.info('"location.txt" probably does not exist at: ' + self._storage_prefix) except ValueError as value_error: logging.error(str(value_error)) def add_alias(self, alias): """ Adds a new name alias for the bot. """ if not isinstance(alias, str): raise TypeError('Bot.add_alias: "alias" can only be of type str!') self.aliases.append(alias) def set_startup_message(self, message): """ Sets a startup message to be posted when the bot starts. """ if not isinstance(message, str): raise TypeError('Bot.set_startup_message: "message" should be of type str!') self._startup_message = message def set_standby_message(self, message): """ Sets a message posted across all rooms when the bot switches to standby. """ if not isinstance(message, str): raise TypeError('Bot.set_standby_message: "message" should be of type str!') self._standby_message = message def set_failover_message(self, message): """ Sets a message posted across all rooms when the bot exits standby (failover). """ if not isinstance(message, str): raise TypeError('Bot.set_failover_message: "message" should be of type str!') self._failover_message = message def set_room_owner_privs_max(self, ids=[]): """ Sets room owner privileges to maximum in all rooms. Keyword arguments: ids -- Sets maximum privs for only these rooms if specified. """ room_ids = list() if len(ids) > 0: if not all(isinstance(x, int) for x in ids): raise TypeError('Bot.set_room_owner_privs_max: Argument "ids" should only consist Integers.') room_ids = ids else: room_ids = self._ids for id in room_ids: room = self.get_room(id) max_privs = room.get_max_privileges() for user in room.owners: if max_privs is not None: user.change_privilege_level(max_privs) def set_storage_prefix(self, new_prefix): """ Sets a new path for the bot's data files to be stored. """ self._storage_prefix = new_prefix try: with open(self._storage_prefix + 'location.txt', 'r') as file_handle: content = [x.rstrip('\n') for x in file_handle.readlines()] if len(content) != 2: raise ValueError('Invalid format in "location.txt"') self._location = content[0] + "/" + content[1] except IOError as ioerr: logging.error(str(ioerr)) logging.info('"location.txt" probably does not exist at: ' + self._storage_prefix) except ValueError as value_error: logging.error(str(value_error)) def set_bot_version(self, bot_version): """ Sets a new bot version. Argument should be of type str. """ if not isinstance(bot_version, str): raise TypeError('Bot.set_botversion: "bot_version" argument must be of type str!') self._version = bot_version def set_bot_location(self, new_location): """ Sets a new location. """ self._location = new_location def set_redunda_key(self, key): """ Sets Redunda key necessary to use Redunda. """ if not isinstance(key, str): raise TypeError('Bot.set_redundakey: "key" is not of type str!') self._redunda_key = key def add_file_to_sync(self, new_file_dict): """ Adds a new file to sync; must be in sync file format supported by pyRedunda. """ if not isinstance(new_file_dict, dict): raise TypeError('Bot.add_file_to_sync: "new_file_dict" must be of type dict') self._sync_files.append(new_file_dict) def redunda_init(self, file_sync=True, bot_version=None): """ Initializes the self._redunda object for further use. Keyword arguments: file_sync -- specifies whether data files should be synced with Redunda (default: True). bot_version -- specifies the version to be sent to Redunda (default: self._version) Returns True on success. """ if bot_version is None: bot_version = self._version if not isinstance(bot_version, str): raise TypeError('Bot.redunda_init: "bot_version" is not of type str!') if not isinstance(self._redunda_key, str): raise TypeError('Bot.redunda_init: "self._redunda_key" is not of type str!') if (self._redunda_key is None) or (bot_version is None): return False if file_sync: if not isinstance(self._sync_files, list): raise TypeError('Bot.redunda_init: "self._sync_files" is not a list!') for id in self._ids: self._sync_files.append({"name": self._convert_to_save_filename(id), "ispickle": False, "at_home": False}) self._redunda = Redunda.RedundaManager(redunda.Redunda(self._redunda_key, self._sync_files, bot_version)) self._redunda_task_manager = BackgroundTaskManager([BackgroundTask(self._redunda.update, interval=60)]) return True def set_redunda_status(self, status): """ Changes redunda status. If True, Redunda will now be actively used. Keyword arguments: status -- type Bool. Returns True on success. """ if type(status) is not bool: raise TypeError('Bot.set_redunda_status: "status" is not of type bool!') i