feat(notifications): Terminbenachrichtigung ohne Buchung - Remote Buchung über Telegram

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":

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 = {

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: