From a40ac95689e1aa6948158a8cba62a824a3fbb93b Mon Sep 17 00:00:00 2001 From: Marco Rombach Date: Wed, 16 Jun 2021 16:57:02 +0200 Subject: [PATCH 1/2] bugfix notifications not working in GUI --- gui.py | 2 ++ tools/gui/qtterminsuche.py | 15 +++++++++------ 2 files changed, 11 insertions(+), 6 deletions(-) diff --git a/gui.py b/gui.py index 5080e2d1..ddbdc22d 100644 --- a/gui.py +++ b/gui.py @@ -261,8 +261,10 @@ def __start_terminsuche(self, kontaktdaten: dict, zeitrahmen: dict): check_delay = self.i_interval.value() codes = kontaktdaten["codes"] + notifications = kontaktdaten.get("notifications", {}) terminsuche_prozess = multiprocessing.Process(target=QtTerminsuche.start_suche, name=f"{codes[0]}-{self.prozesse_counter}", daemon=True, kwargs={ "kontaktdaten": kontaktdaten, + "notifications": notifications, "zeitrahmen": zeitrahmen, "ROOT_PATH": PATH, "check_delay": check_delay}) diff --git a/tools/gui/qtterminsuche.py b/tools/gui/qtterminsuche.py index 9c028b21..9812ed07 100644 --- a/tools/gui/qtterminsuche.py +++ b/tools/gui/qtterminsuche.py @@ -37,7 +37,7 @@ class Worker(QObject): fertig = pyqtSignal() fehlschlag = pyqtSignal(Exception) - def __init__(self, kontaktdaten: dict, zeitrahmen: dict, ROOT_PATH: str, check_delay: int): + def __init__(self, kontaktdaten: dict, notifications: dict, zeitrahmen: dict, ROOT_PATH: str, check_delay: int): """ Args: kontaktdaten (dict): kontakdaten aus kontaktdaten.json @@ -47,6 +47,7 @@ def __init__(self, kontaktdaten: dict, zeitrahmen: dict, ROOT_PATH: str, check_d super().__init__() self.kontaktdaten = kontaktdaten + self.notifications = notifications self.zeitrahmen = zeitrahmen self.ROOT_PATH = ROOT_PATH self.check_delay = check_delay @@ -62,7 +63,8 @@ def suchen(self): try: ImpfterminService.terminsuche(codes=codes, plz_impfzentren=plz_impfzentren, kontakt=kontakt, - PATH=self.ROOT_PATH, check_delay=self.check_delay, zeitrahmen=self.zeitrahmen) + notifications=self.notifications, PATH=self.ROOT_PATH, + check_delay=self.check_delay, zeitrahmen=self.zeitrahmen) self.fertig.emit() @@ -87,7 +89,7 @@ class QtTerminsuche(QtWidgets.QMainWindow): ### QTextEdit (readonly) ### # console_text_edit - def __init__(self, kontaktdaten: dict, zeitrahmen: dict, ROOT_PATH: str, check_delay: int, pfad_fenster_layout=os.path.join(PATH, "terminsuche.ui")): + def __init__(self, kontaktdaten: dict, notifications: dict, zeitrahmen: dict, ROOT_PATH: str, check_delay: int, pfad_fenster_layout=os.path.join(PATH, "terminsuche.ui")): super().__init__() @@ -99,6 +101,7 @@ def __init__(self, kontaktdaten: dict, zeitrahmen: dict, ROOT_PATH: str, check_d # Attribute erstellen self.kontaktdaten = kontaktdaten + self.notifications = notifications self.zeitrahmen = zeitrahmen self.ROOT_PATH = ROOT_PATH self.check_delay = check_delay @@ -120,7 +123,7 @@ def __init__(self, kontaktdaten: dict, zeitrahmen: dict, ROOT_PATH: str, check_d self.thread.start() @staticmethod - def start_suche(kontaktdaten: dict, zeitrahmen: dict, ROOT_PATH: str, check_delay: int): + def start_suche(kontaktdaten: dict, notifications: dict, zeitrahmen: dict, ROOT_PATH: str, check_delay: int): """ Startet die Suche in einem eigenen Fenster mit Umlenkung der Konsolenausgabe in das Fenster @@ -132,7 +135,7 @@ def start_suche(kontaktdaten: dict, zeitrahmen: dict, ROOT_PATH: str, check_dela """ app = QtWidgets.QApplication(list()) app.setAttribute(QtCore.Qt.AA_X11InitThreads) - window = QtTerminsuche(kontaktdaten, zeitrahmen, ROOT_PATH, check_delay) + window = QtTerminsuche(kontaktdaten, notifications, zeitrahmen, ROOT_PATH, check_delay) app.exec_() def setup_infos(self): @@ -150,7 +153,7 @@ def setup_thread(self): """ self.thread = QThread(parent=self) - self.worker = Worker(self.kontaktdaten, self.zeitrahmen, self.ROOT_PATH, self.check_delay) + self.worker = Worker(self.kontaktdaten, self.notifications, self.zeitrahmen, self.ROOT_PATH, self.check_delay) # Worker und Thread verbinden self.worker.moveToThread(self.thread) From 065384e9ca9aed7ab63072715e0d7b502fb9d4b2 Mon Sep 17 00:00:00 2001 From: Marco Rombach Date: Mon, 21 Jun 2021 18:55:32 +0200 Subject: [PATCH 2/2] added possibility to not automatically book appointments added remote booking via telegram --- main.py | 22 +++++-- tools/its.py | 172 +++++++++++++++++++++++++++++++++++++++++++++++-- tools/utils.py | 32 +++++++++ 3 files changed, 216 insertions(+), 10 deletions(-) diff --git a/main.py b/main.py index bddcae8e..e5f2e7ca 100755 --- a/main.py +++ b/main.py @@ -240,7 +240,7 @@ def input_kontaktdaten_key( print(f"\n{str(exc)}\n") -def run_search_interactive(kontaktdaten_path, configure_notifications, check_delay): +def run_search_interactive(kontaktdaten_path, configure_notifications, check_delay, booking=True): """ Interaktives Setup für die Terminsuche: 1. Ggf. zuerst Eingabe, ob Kontaktdaten aus kontaktdaten.json geladen @@ -269,10 +269,10 @@ def run_search_interactive(kontaktdaten_path, configure_notifications, check_del print() kontaktdaten = update_kontaktdaten_interactive( kontaktdaten, "search", configure_notifications, kontaktdaten_path) - return run_search(kontaktdaten, check_delay) + return run_search(kontaktdaten, check_delay, booking) -def run_search(kontaktdaten, check_delay): +def run_search(kontaktdaten, check_delay, booking=True): """ Nicht-interaktive Terminsuche @@ -311,7 +311,8 @@ def run_search(kontaktdaten, check_delay): notifications=notifications, zeitrahmen=zeitrahmen, check_delay=check_delay, - PATH=PATH) + PATH=PATH, + booking=booking) def gen_code_interactive(kontaktdaten_path): @@ -401,6 +402,16 @@ def subcommand_search(args): run_search_interactive(args.file, args.configure_notifications, check_delay=args.retry_sec) +def subcommand_search_no_booking(args): + if args.configure_only: + update_kontaktdaten_interactive( + get_kontaktdaten(args.file), "search", args.configure_notifications, args.file) + elif args.read_only: + run_search(get_kontaktdaten(args.file), check_delay=args.retry_sec, booking=False) + else: + run_search_interactive(args.file, args.configure_notifications, check_delay=args.retry_sec, booking=False) + + def subcommand_code(args): if args.configure_only: update_kontaktdaten_interactive( @@ -514,6 +525,7 @@ def main(): "[1] Termin suchen\n" "[2] Vermittlungscode generieren\n" "[3] Eigene Chromium Instanz im Vaccipy Ordner installieren\n" + "[4] Termin suchen ohne zu buchen (nur Benachrichtigen)\n" f"[x] Erweiterte Einstellungen {'verbergen' if extended_settings else 'anzeigen'}\n") if extended_settings: @@ -533,6 +545,8 @@ def main(): subcommand_code(args) elif option == "3": subcommand_install_chromium() + elif option == "4": + subcommand_search_no_booking(args) elif option == "x": extended_settings = not extended_settings elif extended_settings and option == "c": diff --git a/tools/its.py b/tools/its.py index aeec1d4e..696ea93c 100644 --- a/tools/its.py +++ b/tools/its.py @@ -32,7 +32,7 @@ from tools.kontaktdaten import decode_wochentag, validate_codes, validate_kontakt, \ validate_zeitrahmen from tools.mousemover import move_mouse_to_coordinates, move_mouse_to_element -from tools.utils import fire_notifications, unique +from tools.utils import fire_notifications, read_telegram, unique try: import beepy @@ -43,7 +43,7 @@ class ImpfterminService(): - def __init__(self, codes: list, kontakt: dict, PATH: str, notifications=None): + def __init__(self, codes: list, kontakt: dict, PATH: str, notifications=None, booking=True): if notifications is None: notifications = dict() self.PATH = PATH @@ -52,6 +52,8 @@ def __init__(self, codes: list, kontakt: dict, PATH: str, notifications=None): self.notifications = notifications + self.booking = booking + # Logging einstellen self.log = CLogger("impfterminservice") @@ -693,8 +695,13 @@ def reservierung_finden(self, zeitrahmen: dict, plz: str) -> Optional[Dict]: code = codepoint["code"] try: - reservierung = self.reservierung_finden_mit_code( - zeitrahmen, plz, code) + if self.booking: + reservierung = self.reservierung_finden_mit_code( + zeitrahmen, plz, code) + else: + reservierung = self.reservierung_finden_benachrichtigung( + zeitrahmen, plz, code) + if reservierung is not None: return reservierung except UnmatchingCodeError: @@ -847,6 +854,159 @@ def reservierung_finden_mit_code( "terminpaar": tp_angenommen, } + def reservierung_finden_benachrichtigung( + self, zeitrahmen: Dict, plz: str, code: str) -> Optional[Dict]: + """ + Es wird überprüft, ob im Impfzentrum in der gegebenen PLZ ein oder + mehrere Terminpaare (oder Einzeltermine) verfügbar sind, die dem + Zeitrahmen entsprechen. + Falls ja, wird eine Benachrichtigung gesendet und + 10min auf eine Antwort gewartet. Fall die Antwort kommt wird ein + Dict mit der reservierung zurückgegeben, falls nicht wird None + zurückgegeben. + Zum Format der Rückgabe, siehe Beispiel. + + Beispiel: + zeitrahmen = { + 'einhalten_bei': '1', + 'von_datum': '29.03.2021' + } + + self.reservierung_finden_mit_code( + zeitrahmen, '68163', 'XXXX-XXXX-XXXX') + { + 'code': 'XXXX-XXXX-XXXX', + 'impfzentrum': { + 'Zentrumsname': 'Maimarkthalle', + 'PLZ': '68163', + 'Ort': 'Mannheim', + 'URL': 'https://001-iz.impfterminservice.de/', + }, + 'terminpaar': [ + { + 'slotId': 'slot-56817da7-3f46-4f97-9868-30a6ddabcdef', + 'begin': 1616999901000, + 'bsnr': '005221080' + }, + { + 'slotId': 'slot-d29f5c22-384c-4928-922a-30a6ddabcdef', + 'begin': 1623999901000, + 'bsnr': '005221080' + } + ] + } + + :param zeitrahmen: Zeitrahmen, dem das Terminpaar entsprechen muss + :param plz: PLZ des Impfzentrums, in dem geprüft wird + :param code: Vermittlungscode, für den eventuell gefundene Terminpaare + reserviert werden. + :return: Reservierungs-Objekt (siehe obiges Beispiel), falls ein + passender Termin gefunden wurde, sonst None. + :raise RuntimeError: Termine können nicht geladen werden + """ + + impfzentrum = self.impfzentrum_in_plz(plz) + url = impfzentrum["URL"] + location = f"{url}rest/suche/impfterminsuche?plz={plz}" + + try: + self.s.cookies.clear() + res = self.s.get(location, headers=get_headers(code), timeout=5) + except RequestException as exc: + raise RuntimeError( + f"Termine in {plz} können nicht geladen werden: {str(exc)}") + if res.status_code == 401: + raise UnmatchingCodeError( + f"Termine in {plz} können nicht geladen werden: " + f"Vermittlungscode nicht gültig für diese PLZ") + if not res.ok: + raise RuntimeError( + f"Termine in {plz} können nicht geladen werden: " + f"{res.status_code} {res.text}") + + if 'Virtueller Warteraum des Impfterminservice' in res.text: + return None + + try: + terminpaare = res.json().get("termine") + except JSONDecodeError as exc: + raise RuntimeError( + f"Termine in {plz} können nicht geladen werden: " + f"JSONDecodeError: {str(exc)}") + if not terminpaare: + self.log.info(f"Keine Termine verfügbar in {plz}") + return None + + if ENABLE_BEEPY: + beepy.beep('coin') + else: + print("\a") + + terminpaare_angenommen = [ + tp for tp in terminpaare + if terminpaar_im_zeitrahmen(tp, zeitrahmen) + ] + terminpaare_abgelehnt = [ + tp for tp in terminpaare + if tp not in terminpaare_angenommen + ] + + zentrumsname = impfzentrum.get('Zentrumsname').strip() + ort = impfzentrum.get('Ort') + + for tp_abgelehnt in terminpaare_abgelehnt: + self.log.warn( + "Termin gefunden - jedoch nicht im entsprechenden Zeitraum:") + self.log.info('-' * 50) + self.log.warn(f"'{zentrumsname}' in {plz} {ort}") + for num, termin in enumerate(tp_abgelehnt, 1): + ts = datetime.fromtimestamp(termin["begin"] / 1000).strftime( + '%d.%m.%Y um %H:%M Uhr') + self.log.warn(f"{num}. Termin: {ts}") + self.log.warn(f"Link: {url}impftermine/suche/{code}/{plz}") + self.log.info('-' * 50) + + if not terminpaare_angenommen: + raise TimeframeMissed() + + msg = f"'{zentrumsname}' in {plz} {ort}\n" + self.log.success(f"Termin gefunden!") + self.log.success(f"'{zentrumsname}' in {plz} {ort}") + + termine_num = {} + + for n, tp in enumerate(terminpaare_angenommen, 1): + termine_num[n] = tp + msg += f"\nTerminpaar {n} - Buchen mit #{n} (Telegram)\n" + for num, termin in enumerate(tp, 1): + ts = datetime.fromtimestamp(termin["begin"] / 1000).strftime( + '%d.%m.%Y um %H:%M Uhr') + self.log.success(f"{num}. Termin: {ts}") + msg += f"{num}. Termin: {ts}\n" + + msg += f"Link: {url}impftermine/suche/{code}/{plz}" + + self.notify("Termin gefunden!", msg) + self.log.info("Warte auf Antwort...") + t = 0 + while True: + t += 1 + answer = read_telegram(self.notifications["telegram"]) + if answer is not None: + self.log.info(f"Antwort über Telegram: {answer}") + if int(answer) in termine_num.keys(): + self.log.info(f"Termin {answer} ausgewählt.") + return { + "code": code, + "impfzentrum": impfzentrum, + "terminpaar": termine_num[int(answer)], + } + if t >= 120: + break + time.sleep(5) + + return None + def termin_buchen(self, reservierung): """Termin wird gebucht für die Kontaktdaten, die beim Starten des Programms eingetragen oder aus der JSON-Datei importiert wurden. @@ -1219,7 +1379,7 @@ def notify(self, title: str, msg: str): @staticmethod def terminsuche(codes: list, plz_impfzentren: list, kontakt: dict, PATH: str, notifications: Dict = None, zeitrahmen: Dict = None, - check_delay: int = 30): + check_delay: int = 30, booking=True): """ Sucht mit mehreren Vermittlungscodes bei einer Liste von Impfzentren nach Terminen und bucht den erstbesten, der dem Zeitrahmen entspricht, @@ -1253,7 +1413,7 @@ def terminsuche(codes: list, plz_impfzentren: list, kontakt: dict, if len(plz_impfzentren) == 0: raise ValueError("Kein Impfzentrum ausgewählt") - its = ImpfterminService(codes, kontakt, PATH, notifications) + its = ImpfterminService(codes, kontakt, PATH, notifications, booking) # Prüfen, ob in allen angegebenen PLZs ein Impfzentrum verfügbar ist izs_by_plz = { diff --git a/tools/utils.py b/tools/utils.py index 554d926f..cea67d74 100644 --- a/tools/utils.py +++ b/tools/utils.py @@ -1,4 +1,5 @@ import os +import re import time import traceback import random @@ -7,6 +8,7 @@ from json import JSONDecodeError from pathlib import Path from threading import Thread +from time import time import requests from plyer import notification @@ -98,6 +100,8 @@ def desktop_notification(operating_system: str, title: str, message: str): if 'windows' not in operating_system: return + message = message[:256] + try: Thread(target=notification.notify( app_name="Impfterminservice", @@ -230,6 +234,34 @@ def telegram_validation(notifications: dict): return validation_code +def read_telegram(notifications: dict): + if 'api_token' not in notifications or 'chat_id' not in notifications: + return + + headers = { + 'Accept': 'application/json', + 'User-Agent': 'vaccipy' + } + + url = f'https://api.telegram.org/bot{notifications["api_token"]}/getUpdates' + params = { + 'chat_id': notifications["chat_id"], + 'offset': -1 + } + + r = requests.get(url, params=params, headers=headers) + + if r.status_code != 200: + raise TelegramNotificationError(r.status_code, r.text) + for message_object in r.json().get('result'): + message = message_object.get('message') + if not message: + return + + if message.get('text') and re.search(r"#\d", message.get('text')) and time() - message.get('date') <= 10: + return message.get('text').strip('#') + + def fire_notifications(notifications: dict, operating_system: str, title: str, message: str): desktop_notification(operating_system, title, message) if 'pushover' in notifications: