#!/usr/bin/env python3 from collections import namedtuple from modules import STOP_PROCESSING, RestartException from telegram.ext import Updater, CommandHandler, MessageHandler, Filters, InlineQueryHandler, CallbackQueryHandler from telegram import InlineQueryResultArticle, InputTextMessageContent from util import logsetup from typing import Dict, Callable, Deque, List, Set, Any import collections import functools import inspect import logging, coloredlogs import modules.musicleg import modules.tts import modules.match import modules.plusone import modules.admin import modules.bsh import modules.python import modules.haskell import modules.help import modules.misc import modules.poll import modules.scale import modules.sed import modules.spelling import modules.stickers import modules.tagauth as auth import modules.text import modules.wilco import modules.woordenlijst import modules.wiktionary import modules.brief import modules.silence import os import schema import sys import telegram as tg import threading import time import yaml import traceback import hashlib try: import sdnotify except ImportError: sdnotify = None print("sd_notify support missing, not notifying systemd of startup success") LOGLEVEL = logging.INFO FORMAT = "[%(asctime)s] [%(levelname)7s] %(name)7s: %(message)s" logger = logsetup("bea.log", FORMAT, LOGLEVEL) in_error = False Command = Callable[['Bea', tg.Bot, tg.Update], None] class Bea: def __init__(self) -> None: with open('config.yaml') as f: self.config = yaml.load(f) self.commands: List[Command] = [] self.bypass_fn: Dict[int, Command] = {} # A map from @ to user-id. if 'known_users' not in self.config: self.config['known_users'] = {} self.known_users = self.config['known_users'] if 'known_names' not in self.config: self.config['known_names'] = {} self.known_names = self.config['known_names'] if 'known_channels' not in self.config: self.config['known_channels'] = {} self.known_channels = self.config['known_channels'] self.updater = Updater(token=self.config['token'], use_context=True) self.dispatcher = self.updater.dispatcher self.bot = self.updater.bot # callbacks for callback queries. # dictionary of data prefix -> handler # (data: prefix;rest) self.callback_handler = {} self.chataction_event = threading.Event() self.load_modules() self.banned_uids: Set[int] = set() self.message_log: Deque[tg.Message] = collections.deque(maxlen=16) msg = MessageHandler(Filters.all, self.handle_message) # type: ignore callback = CallbackQueryHandler(self.handle_callback) # type: ignore inline = InlineQueryHandler(self.handle_message) # type: ignore self.dispatcher.add_handler(callback) self.dispatcher.add_handler(inline) self.dispatcher.add_handler(msg) self.dispatcher.add_error_handler(self.error_callback) self.is_restarting = False self.try_auth = functools.partial(auth.try_auth, self) self.current_update: tg.Update if (sdnotify): sdnotify.SystemdNotifier().notify("READY=1") def handle_callback(self, update: tg.Update, context: tg.ext.CallbackContext) -> None: user = update.effective_user if user: uid = user.id first_name = user.first_name self.known_names[uid] = first_name prefix = update.callback_query.data.split(';')[0] if (handler := self.callback_handler.get(prefix)): handler(self, self.bot, update) def inline_reply_text(self, title, text=None, parse_mode=None, is_personal=False, cache_time=0) -> bool: text = text or title message_content = InputTextMessageContent(text, parse_mode=parse_mode) results = [InlineQueryResultArticle(id=hashlib.sha256(text.encode('utf-8')).hexdigest(), title=title, input_message_content=message_content)] return self.current_update.inline_query.answer(results=results, is_personal=is_personal, cache_time=cache_time) def reply(self, *args: Any, quote=False, **kwargs: Any) -> tg.Message: if not self.current_update: raise ValueError("no update to reply to") if not hasattr(self.current_update, "message") or self.current_update.message is None: raise NotImplementedError("can't reply to update with no message yet") return self.current_update.message.reply_text(*args, quote=quote, **kwargs) def dump_config(self) -> None: self.config['known_users'] = self.known_users with open('config.yaml', 'w') as f: yaml.dump(self.config, f) def start(self) -> None: try: self.updater.start_polling() self.updater.idle() finally: schema.db.close() self.dump_config() if (self.is_restarting): raise RestartException def try_register_bypass(self, uid: int, fn: Command) -> bool: if uid in self.bypass_fn: return False else: self.bypass_fn[uid] = fn return True def start_uploading_photo(self) -> None: def uploading_thread(event: threading.Event) -> None: while not event.is_set(): if self.current_update.effective_chat: self.bot.send_chat_action(self.current_update.effective_chat.id, tg.ChatAction.UPLOAD_PHOTO) time.sleep(3) event.clear() self.chataction_thread = threading.Thread(target=uploading_thread, args=(self.chataction_event,), daemon=True) self.chataction_thread.start() def stop_uploading_photo(self) -> None: self.chataction_event.set() def set_typing(self) -> None: if self.current_update.effective_chat: self.bot.send_chat_action(self.current_update.effective_chat.id, tg.ChatAction.TYPING) def unregister_bypass(self, uid: int, fn: Command) -> bool: if uid not in self.bypass_fn: return False if self.bypass_fn[uid] != fn: return False del self.bypass_fn[uid] return True def load_modules(self) -> None: extension_modules = {k:v for k, v in sys.modules.items() if k.startswith('modules.')} for name, module in extension_modules.items(): name = name.split('modules.')[1] logger.info(f'Loading module {name}') if name not in self.config: self.config[name] = {} init_func = getattr(module, "init", lambda *args: []) self.commands.extend(init_func(self, self.config[name])) def handle_message(self, update: tg.Update, ctx: tg.ext.CallbackContext) -> None: self.current_update = update if update.effective_chat and update.effective_chat.title: self.known_channels[update.effective_chat.title] = update.effective_chat.id if update.effective_message: self.message_log.append(update.effective_message) if not update.effective_user: return if not update.inline_query: toprint = str(update.effective_message) else: toprint = str(update.inline_query) logger.debug(f"Saw message: {toprint}") msg = update.effective_message if msg and msg.text: logger.info(f"[{update.effective_user.username}] {msg.text}") update.message = update.effective_message uid = update.effective_user.id if isinstance(update.effective_user.username, str): username = update.effective_user.username.lower() self.known_users[username] = uid if isinstance(update.effective_user.first_name, str): self.known_names[uid] = update.effective_user.first_name if (uid in self.banned_uids): return if uid in self.bypass_fn: return self.bypass_fn[uid](self, ctx.bot, update) for command in self.commands: if (command(self, ctx.bot, update) == STOP_PROCESSING): break def error_callback(self, up: tg.Update, ctx: tg.ext.CallbackContext) -> None: global in_error error = ctx.error if not in_error and error: # only try to send this once, otherwise we may get a loop in_error = True ss = traceback.extract_tb(error.__traceback__) full_tb = traceback.format_tb(error.__traceback__) for frame in ss[::-1]: if frame.line and 'site-packages' not in frame.filename and frame.filename.startswith('/home/bea/textfuckery'): break else: logger.error(''.join(full_tb)) if up.message and hasattr(up.message, 'reply_text'): up.message.reply_text(f"Ik deed krak: {error}.") return name = frame.filename.split('textfuckery/')[1] if up.message and hasattr(up.message, 'reply_text'): up.message.reply_text(f"Oepsiewoepsie!! Uwu, de bot is stukkiewukkie! Een beetje kapotjewotje!\n\nLast non-library source line is in {name}, and reads:\n{frame.line}") logger.error(''.join(full_tb)) logger.error(error) in_error = False try: bea = Bea() bea.start() except RestartException: rel_file = inspect.getsourcefile(Bea) if rel_file: abs_file = os.path.abspath(rel_file) os.execl(abs_file, abs_file)