#!/usr/bin/python3 # Based on https://github.com/mh-g/python-signal-cli (c)2017 mh-g # (c) 2021 FA, PM import json import logging import random import time from datetime import date, datetime, timedelta from threading import Thread import paho.mqtt.client as mqtt # EPL V1.0 import requests # Apache 2.0 import schedule # MIT from gi.repository import GLib # LGPL v2.1+ from pydbus import SystemBus # LGPL v2+ import deepl # MIT from mod_birthdayreminder import ModuleBirthdayReminder from mod_challenge import ModuleChallenge from mod_freegames import ModuleFreeGames from mod_genderneutral import ModuleGenderneutral from mod_commands import ModuleCommands from mod_quotes import ModuleQuotes from mod_tex import ModuleTex from mod_eventreminder import ModuleEventReminder from mod_today import ModuleToday from mod_spotify import ModuleSpotify # @TODO Einfacher Wrapper zum senden von Nachrichten per shell # @TODO Config Programm bot-config.py mit dem allgemeine Einstellungen wie Profilname oder Profilbild angepasst werden können # create logger log = logging.getLogger("signalbot") log.setLevel(logging.DEBUG) # create console handler and set level to debug mdch = logging.StreamHandler() fh = logging.FileHandler("signalbot_debug.log") fh.setLevel(logging.DEBUG) # create formatter formatter = logging.Formatter("%(asctime)s - %(levelname)s - %(message)s") # add formatters to handlers fh.setFormatter(formatter) # add handlers to logger log.addHandler(fh) # load modules gn = None quotes = None games = None challenges = None birthday_reminder = None tex = None commands = None event_reminder = None translator = None today = None def load_config(filename): global CONFIG try: CONFIG = {} with open(filename, "r") as file: CONFIG = json.load(file) except FileNotFoundError: log.error(f"File:'{filename}' not accessible. Running in reduced mode.") print("File not accessible") print("Running in reduced mode") def handle_action(timestamp, source, groupID, message, attachments): if ( not message.startswith(".") or len(message) <= 2 or message[0:2] == ".." or message[0:3] == "._." ): return print(timestamp) print(source) print(groupID) print(message) print(attachments) msgDict = { "timestamp": timestamp, "source": source, "groupID": groupID, "attachments": attachments, "receiver": source, } msgDict["groupInfo"] = None if len(groupID) > 0: msgDict["receiver"] = groupID # wenn gruppe, dann berechtigungsobjekt suchen um die rechte zu beschneiden for group in GROUPS: if group["ID"] == groupID: msgDict["groupInfo"] = group break try: message = message[1:].strip() msg_splitted = message.split(" ") if len(msg_splitted) == 1: msg_splitted = message.replace("\n", " ", 1).split(" ") msgDict["action"] = msg_splitted[0].lower() msgDict["content"] = " ".join(str(x) for x in msg_splitted[1:]).strip() if len(msgDict["action"]) == 0: return globals()[commands.get_function(msgDict)](msgDict) except Exception as ex: log.critical( f'Unknown error:"{ex}", timestamp="{timestamp}", source="{source}", groupID="{groupID}", ' f'message="{message}", attachments="{attachments}"' ) def send(message, receiver): if not receiver: return try: if type(receiver) is str: signal.sendMessage(message, [], receiver) elif type(receiver) is list: signal.sendGroupMessage(message, [], receiver) except Exception as ex: err_message = f'Send Message Exception:"{ex}", message="{message}" , receiver="{receiver}" ' log.error(err_message) if "f43e6540-0274-481e-838d-feefbb0686df" in err_message: return testung = next((group for group in GROUPS if group["NAME"] == "Testung"), None) if testung is not None: send(err_message, testung["ID"]) def send_a(message, receiver, attachment): if not receiver: return try: if type(receiver) is str: signal.sendMessage(message, [attachment], receiver) elif type(receiver) is list: signal.sendGroupMessage(message, [attachment], receiver) except Exception as ex: err_message = f'Send Message with attachment Exception:"{ex}"", message="{message}", receiver="{receiver}", attachments="{attachment}"' log.error(err_message) if "f43e6540-0274-481e-838d-feefbb0686df" in err_message: return testung = next((group for group in GROUPS if group["NAME"] == "Testung"), None) if testung is not None: send(err_message, testung["ID"]) def echo(msgDict): send(FUN_DICT[msgDict["action"]], msgDict["receiver"]) def help_func(msgDict): answer = "" if len(msgDict["content"]) > 0: answer = commands.get_help_description(msgDict["content"], msgDict["groupInfo"]) else: answer = commands.get_all_commands(msgDict["groupInfo"]) send(answer, msgDict["receiver"]) def remus_feinste(msgDict): message = ( "- 1,5l Weißwein\n" "- 2,5l Rotwein\n" "- Obst nach Geschmack (bspw.: grüne Äpfel und Orangen)\n" "- N Batzen Zucker\n" "- guter Schuss Obstschnaps\n" "- Gran Manier (Prangenlikör) is beste\n" "- Fanta / O-Saft\n" "Originalrezept von Onkel Remus\n" ) send(message, msgDict["receiver"]) def true_american(msgDict): message = ( "True American Regeln: So funktioniert das Spiel aus New Girl\n\n" "- Sie nun folgende Utensilien: Mindestens zwei weitere Mitspieler, drei Dosen Bier pro Spieler, eine Flasche Schnaps, einen Mülleimer, sowie einen Raum mit Möbeln und einem großen Tisch, den Sie am nächsten Tag mit Sicherheit putzen müssen. \n\n" '- Stellen Sie die Schnapsflasche, den "König", in die Mitte des Tisches und die Dosenbiere (die "Soldaten") um den König herum, um ihn zu beschützen. Platzieren Sie die Möbel im Raum in vier Zonen kreisförmig um den Tisch herum, sodass Sie sich bewegen können, ohne den Boden zu berühren, denn: Der Boden ist Lava! \n\n' "- Nun wählen Sie Teams: Halten sie sich mit Ihren Fingern eine Zahl von eins bis fünf vor die Stirn, die Person mit einem Finger weniger vor der Stirn ist Ihr Teamkollege. Falls Sie niemanden finden, sind Sie auf sich alleine gestellt. \n\n" "JFK! Jetzt wird gespielt.\n\n" '- Zu Beginn ruft einer der Mitspieler Laut "JFK", alle Anderen antworten "FDR", rennen zur Burg, nehmen sich ein Bier und sichern sich den Platz auf einem der Möbelstücke in der äußersten Zone. Nun wird im Uhrzeigersinn gespielt. Wer eine Aufgabe erfolgreich erledigt, trinkt einen Schluck und darf einen Platz weiter. Dabei gibt es drei verschiedene Aufgabentypen. \n\n' "- Erstens: Der Spieler, der an der Reihe ist, nennt zwei Gegenstände oder Personen, die Anderen schreien eine Gemeinsamkeit der Objekte in die Runde. Zweitens: Der Spieler beginnt ein Sprichwort, die anderen Mitspieler vollenden das Sprichwort. Drittens: Alle halten sich Zahlen (mit einer Hand) an die Stirn, Spieler mit gleichen Zahlen vollenden die Aufgabe. \n\n" '- Ist eine Aufgabe erfolgreich absolviert, trinken Sie einen Schluck und ziehen ein Möbelstück weiter. Dabei dürfen Sie nie den Boden berühren oder ein leeres Bier in der Hand halten. Schreit ein beliebiger Mitspieler "JFK", antworten alle Anderen mit "FDR" und müssen ihr Bier austrinken. \n\n' '- Wenn alle Biere vom Tisch getrunken wurden, ist der König, also die Schnapsflache, "verwundbar". Der Spieler, der zuerst die innerste Zone erreicht, kann dann einen Schluck aus der Schnapsflasche trinken und hat damit - zusammen mit seinem Teamkollegen - gewonnen. \n\n' "- Verloren hat der Spieler, der während des Spieles mit einem leeren Bier erwischt wird oder der den Boden, also die Lava berührt. Zurück ins Spiel kommen Sie, indem Sie ein ganzes Bier austrinken. Auf welchen Platz Sie dann gesetzt werden, entscheiden die Mitspielenden. \n\n" "Willkürliche Zusatzregeln\n" "Als wäre das Spiel noch nicht kompliziert und verwirrend genug, erklären wir Ihnen in unserem letzten Kapitel noch einige - zugegeben ziemlich willkürliche - Zusatzregeln.\n\n" '- Spielen Sie mit der "Clinton-Zusatzregel", so muss jeder Spieler, der eine Aufgabe nicht vollenden konnte, ein Kleidungsstück ablegen. Zusätzlich müssen sich bei Aufgabentyp 3 Spieler mit der gleichen Zahl "hinter dem eisernen Vorhang" (zum Beispiel in einem anderen Raum) küssen. \n\n' '- Ruft ein Spieler "All trash belongs..." so antworten die Anderen "in the junkyard!" und werfen eine leere Dose in den Mülleimer am anderen Ende des Raumes. Wenn jemand "Donald Trump" ruft, so müssen alle mit "Build the wall!" antworten und trinken. \n\n' "-Die wichtigste Regeln bei True American: Alles was bei True American gesagt wird, ist angeblich eine Lüge und es geht nicht um das Gewinnen, sondern darum, ein richtiger Amerikaner zu sein. \n\n\n" "#DankeFocus: https://praxistipps.focus.de/true-american-regeln-so-funktioniert-das-spiel-aus-new-girl_126279\n\n" "https://www.trueamericanrules.com/working-rules/" ) send_a(message, msgDict["receiver"], "/home/pi/signalbot/true_american_layout.png") def yes_no_wtf(msgDict): response = str(requests.get("https://yesno.wtf/api").content) result = response[2:-1] json_response = json.loads(result) img_data = requests.get(json_response["image"]).content with open("yn.gif", "wb") as handler: handler.write(img_data) send_a(json_response["answer"], msgDict["receiver"], "/home/pi/signalbot/yn.gif") def dadjoke(msgDict): response = requests.get( f"https://icanhazdadjoke.com", headers={"Accept": "application/json"} ) result = response.json() send(result["joke"], msgDict["receiver"]) def links(msgDict): links = {} if msgDict["groupInfo"] is not None and "LINKS" in msgDict["groupInfo"]: links = msgDict["groupInfo"]["LINKS"] if len(links) == 0: message = "Keine Links hinterlegt!" else: message = "" for key, value in links.items(): if type(value) is str: message += key + ":\n" # Gruppenname / Linkbezeichnung message += value + " \n" else: message += "## " + key + ": ##\n" # Gruppenname / Linkbezeichnung for key2, value2 in value.items(): message += key2 + ":\n" # Linkbezeichnung if type(value2) is str: message += value2 + " \n" message += "\n" message += "\n" message += "\n" send(message, msgDict["receiver"]) def save_quote(msgDict): send( quotes.save_quote(msgDict["content"], msgDict["groupInfo"]), msgDict["receiver"] ) def random_quote(msgDict): send(quotes.get_random_quote(msgDict["groupInfo"]), msgDict["receiver"]) def credits(msgDict): message = ( "Made with ☕ and ❤️\n" "by Frederic Aust & Philip Maas\n" "Git: https://gitlab.cvh-server.de/faust/signalbot \n" "Licence: AGPL v3" ) send(message, msgDict["receiver"]) def free_games(msgDict): # games.freegames(msgDict["receiver"]) send("Sry freegamesyo.com has been discontinued... ☹️", msgDict["receiver"]) def genderneutral(msgDict): global gn message = gn.search_in_db(msgDict["content"]) send(message, msgDict["receiver"]) def sips(msgDict): ex = random.randrange(0, 33) amount_sips = random.randrange(1, 10) message = "Trinke " backfire = random.randrange(0, 50) if backfire == 49: message = "Jokes on You!\n" + message if ex == 32: message = "Trink dein Getränk auf EX!!1" else: if amount_sips == 1: message += "einen Schluck!" else: message += f"{amount_sips} Schlücke!" send(message, msgDict["receiver"]) def roll(msgDict): message = "" range = msgDict["content"] if not range: range = "1-100" try: splitted = range.strip().split("-") if len(splitted) != 2: message = "Rangesyntax invalid! exp.: 1-100" elif int(splitted[0]) >= int(splitted[1]): message = "Rangesyntax invalid! exp.: 1-100" else: rolled = random.randrange(int(splitted[0]), int(splitted[1])) message = f"{rolled} it is!" except Exception: message = "Rangesyntax invalid! exp.: 1-100" send(message, msgDict["receiver"]) def request(msgDict): answer = "" request_group = next( (group for group in GROUPS if group["NAME"] == "Requests"), None ) if not msgDict["content"]: answer = "Was soll denn hinzugefügt werden?" else: answer = "Besten Dank! Die Anfrage wurde weitergeleitet." if request_group is not None: send(answer, msgDict["receiver"]) send(msgDict["content"], request_group["ID"]) else: send("Der Befehl Requests funktioniert gerade nicht :(", msgDict["receiver"]) def get_random_challenge(msgDict): global challenges message = "Keine Challenge verfügbar - bitte melde dich bei uns per .request !" if challenges: message = challenges.get_random_challenge() send(message, msgDict["receiver"]) def save_birthday(msgDict): send( birthday_reminder.save_birthday(msgDict["content"], msgDict["groupInfo"]), msgDict["receiver"], ) def get_all_birthdays(msgDict): send(birthday_reminder.get_birthdaylist(msgDict["groupInfo"]), msgDict["receiver"]) def get_next_birthday(msgDict): send(birthday_reminder.get_next_birthday(msgDict["groupInfo"]), msgDict["receiver"]) def save_event(msgDict): send( event_reminder.save_event(msgDict["content"], msgDict["groupInfo"]), msgDict["receiver"], ) def get_all_upcomming_events(msgDict): send(event_reminder.get_eventlist(msgDict["groupInfo"]), msgDict["receiver"]) def get_all_events(msgDict): send( event_reminder.get_complete_eventlist(msgDict["groupInfo"]), msgDict["receiver"] ) def get_next_event(msgDict): send(event_reminder.get_next_event(msgDict["groupInfo"]), msgDict["receiver"]) def unknown_command(msgDict): log.debug( f'Unknown action: {msgDict["action"]} {msgDict["content"]} from: {msgDict["receiver"]}' ) send( f'Das habe ich nicht verstanden: "{msgDict["action"]} {msgDict["content"]}" ', msgDict["receiver"], ) def get_tex_formula(msgDict): send_a( msgDict["content"], msgDict["receiver"], tex.create_formula(msgDict["content"]) ) def get_deepl_translation(msgDict): if translator is None: send("There is no translator available!", msgDict["receiver"]) return splitted = msgDict["content"].strip().split(",") lang = splitted[0].upper() text = ",".join(str(x) for x in splitted[1:]).strip() try: result = translator.translate_text(text, target_lang=lang) except Exception as ex: send(f"{ex}", msgDict["receiver"]) return answer = "Computer says no" if result: answer = str(result.detected_source_lang) + " → " + str(lang) + "\n" answer += text + "\n\n" answer += result.text send(answer, msgDict["receiver"]) def add_song_to_playlist(msgDict): spotify_config = CONFIG.get("Spotify", {}) if ( spotify is None or spotify_config.get("client_id", None) is None or spotify_config.get("client_secret", None) is None or spotify_config.get("redirect_uri", None) is None ): # TODO bereits in der INit berücksichtigen, wenn die Secrets nicht am start sind send("There is no Spotify!", msgDict["receiver"]) return try: send(spotify.add_song_to_playlist(msgDict["content"]), msgDict["receiver"]) except Exception as ex: send(f"{ex}", msgDict["receiver"]) return def get_kings_cup(msgDict): send_a( "Don't drink and derive!", msgDict["receiver"], "/home/pi/signalbot/kings_cup_regeln.png", ) def get_today(msgDict): send(today.getTodayMessage(), msgDict["receiver"]) def on_mqtt_message(client, userdata, msg): print(msg.topic + " " + str(msg.payload)) send(msg.topic + " " + str(msg.payload), CONFIG["Admins"]["Fred"]) def mqtt_client(): mqtt_local = CONFIG.get("MQTT", {}).get("mqtt.local", {}) client2 = mqtt.Client() client2.on_message = on_mqtt_message client2.username_pw_set( mqtt_local.get("MQTT_USER", ""), mqtt_local.get("MQTT_PWD", "") ) client2.connect( mqtt_local.get("MQTT_IP", ""), mqtt_local.get("MQTT_PORT", None), mqtt_local.get("MQTT_TIMEOUT", None), ) # TODO Topics und receiver in config auslagern, um dynamischer zu subscriben client2.subscribe("Sensoren/Arbeitszimmer/temperature") client2.loop_forever() def send_heartbeat(): testung = next((group for group in GROUPS if group["NAME"] == "Testung"), None) if testung is not None: send("Still alive - there will be 🎂", testung["ID"]) def init_schedule_jobs(): schedule.every().day.at("19:00").do(run_threaded, send_heartbeat) schedule.every().monday.at("06:00").do(run_threaded, gn.update_db) # schedule.every().hour.do(run_threaded, games.auto_newsletter) schedule.every().day.at("00:01").do( run_threaded, birthday_reminder.check_for_birthdays ) schedule.every().day.at("09:00").do(run_threaded, event_reminder.check_for_events) schedule.every().tuesday.at("10:00").do(run_threaded, gg_remind_schichten) schedule.every().thursday.at("11:00").do(run_threaded, gg_remind_stundenzettel) schedule.every().day.at("13:00").do(run_threaded, today.update) # schedule.every().thursday.at("10:00").do(run_threaded, remind_pflanzen) # schedule.every(10).seconds.do(run_threaded, birthday_reminder.check_for_birthdays) def remind_pflanzen(): tierwg = next( (group for group in GROUPS if group["NAME"] == "TierWG"), None ) # Hier eine neue Gruppe angeben, da es die TierWG nicht mehr gibt if tierwg is not None: send(f"Bitte gebt den armen Pflanzen etwas Lebenselixier!", tierwg["ID"]) def gg_remind_schichten(): gg_group = next((group for group in GROUPS if group["NAME"] == "GGOffiziell"), None) if gg_group is not None: send( f'Bitte tragt euch in die Planung ein:\n {gg_group["LINKS"]["Planung"]}', gg_group["ID"], ) def gg_remind_stundenzettel(): # Get the current date and time now = datetime.datetime.now() # Calculate the last day of the current month # If the current month is December (12), the next month would be January of the next year # Otherwise, it's simply the next month of the current year # We then subtract one day to get the last day of the current month if now.month == 12: last_day = datetime.datetime(now.year + 1, 1, 1) - datetime.timedelta(days=1) else: last_day = datetime.datetime(now.year, now.month + 1, 1) - datetime.timedelta( days=1 ) # Determine the last Tuesday of the month # Here, we enter a loop that continues subtracting one day from the last day of the month # until it hits a Tuesday. The weekday() function of a datetime object returns a number # where Monday is 0 and Sunday is 6. Therefore, Tuesday is 1. We keep subtracting one day # until last_day.weekday() equals 1, which signifies it's a Tuesday. while last_day.weekday() != 1: last_day -= datetime.timedelta(days=1) last_day -= datetime.timedelta(days=5) gg_group = next((group for group in GROUPS if group["NAME"] == "GGOffiziell"), None) if last_day == date.today().day and gg_group is not None: send( """Bitte schickt eure ausgefüllten Stundenzettel! • ihr müsst eure Abrechnungen jeden Monat sowohl an Sarah, als auch an Ben senden. • sie müssen signiert und mit dem Kreuzchen im oberen Kästchen versehen sein • eure Emails müssen einen Anhang haben (duh) • der Betreff eurer Email sollte "Abrechnung Nachname Monat Jahr" beinhalten, das macht es leichter Sie zu finden. • in die Zeile "Tätigkeit" muss bitte "aktivierende Arbeit im offenen Bereich" eingetragen werden. Nichts anderes!!1 • sollten eure Abrechnungen mal falsch/ zu spät sein ist das kein Weltuntergang, dann gibt's das Geld einen Monat später - es verfällt nicht. Sollten sich dadurch persönliche Probleme ergeben sprecht uns bitte an, es gibt immer Lösungen.""", gg_group["ID"], ) def start_schedule(): while True: schedule.run_pending() time.sleep(1) def run_threaded(job_func): job_thread = Thread(target=job_func) job_thread.start() if __name__ == "__main__": load_config("/home/pi/signalbot/config.json") # global GROUPS GROUPS = CONFIG.get("GROUPS", {}) spotify_config = CONFIG.get("Spotify", {}) FUN_DICT = CONFIG.get("FUN_DICT", {}) gn = ModuleGenderneutral("gn.json", log) quotes = ModuleQuotes(GROUPS, log) games = ModuleFreeGames(CONFIG.get("FREE_GAMES_SUBSCRIBER", {}), send_a, log) challenges = ModuleChallenge("/home/pi/signalbot/challenge.json", log) birthday_reminder = ModuleBirthdayReminder(GROUPS, send, log) tex = ModuleTex(log) commands = ModuleCommands( "/home/pi/signalbot/commands.json", CONFIG.get("FUN_DICT", {}), log ) event_reminder = ModuleEventReminder(GROUPS, send, log) spotify = ModuleSpotify( send, spotify_config.get("client_id", ""), spotify_config.get("client_secret", ""), spotify_config.get("redirect_uri", ""), log, ) today = ModuleToday(log) loop = GLib.MainLoop() bus = SystemBus() signal = bus.get("org.asamk.Signal") signal.onMessageReceived = handle_action # mqtt_thread = Thread(target=mqtt_client) # mqtt_thread.start() # print("mqtt_thread started") init_schedule_jobs() schedule_thread = Thread(target=start_schedule) schedule_thread.start() print("schedule_thread started") loop.run()