Merge pull request #7 from muffix/add-notification

Add support for notificaitons through ntfy.sh

CHANGELOG.md

@@ -1,24 +1,28 @@
 # Changelog
 
-## v1.1.0
+## v1.1.1 - 2022-10-04
+
+- Added support for sending a notification through [ntfy.sh](https://ntfy.sh) on success
+
+## v1.1.0 - 2022-10-04
 
 - Added support for the updated BayIMCO API
 - Added filter to find vaccination appointments against a specific variant, such as BA.1 or BA.4/5
 - Added support for the `--debug` option to additionally print debug output
 
-## v1.0.3
+## v1.0.3 - 2021-11-30
 
 - Increased the frequency of authentication token refreshes
 - Added pre-commit hooks config
 
-## v1.0.2
+## v1.0.2 - 2021-11-30
 
 - Added support for the `--latest-day` option
 
-## v1.0.1
+## v1.0.1 - 2021-11-30
 
 - Fixed a bug where the `--earliest-day` option wasn't optional, but should have been
 
-## v1.0.0
+## v1.0.0 - 2021-11-30
 
 - Initial release

README.md

@@ -67,6 +67,9 @@ options:
   --first-vaccine-id FIRST_VACCINE_ID
                         The ID of the vaccine that was used for the first jab. Only required for the second vaccination.
   --variant VARIANT     Variants to find vaccines for; either ba1 or ba45. Leave blank for any.
+  --ntfy-topic NTFY_TOPIC
+                        ntfy.sh topic to send a message to on success. See https://ntfy.sh for details.
+
 ```
 
 #### Vaccine IDs
@@ -87,6 +90,11 @@ first jab. In addition to `--dose=second`, add `--first-vaccine-id` according to
 | Spikevax 0 (Zero)/O (Omicron) (Moderna)             | 010 |
 | Comirnaty Original/Omicron BA.4-5 (BioNTech/Pfizer) | 011 |
 
+#### Notifications
+
+The checker can send a notification to an [ntfy.sh](https://ntfy.sh) topic. Use the `--ntfy-topic` option to specify the
+topic name and subscribe to the notifications on your phone or on the web app.
+
 #### Checking until successful
 
 You can use the optional `--interval` option to tell the script to keep trying until it succeeds in finding (or booking

byimpf/client.py

@@ -58,6 +58,7 @@ class ImpfChecker:
     APPOINTMENTS_URL_FORMAT = (
         "https://impfzentren.bayern/api/v1/citizens/{}/appointments"
     )
+    NTFY_SH_URL = "https://ntfy.sh/"
 
     def auth_token(self):
         if self._auth_token is None:
@@ -68,11 +69,18 @@ def auth_token(self):
                 exit(1)
         return self._auth_token
 
-    def __init__(self, username: str, password: str, citizen_id: str):
+    def __init__(
+        self,
+        username: str,
+        password: str,
+        citizen_id: str,
+        ntfy_topic: Optional[str] = None,
+    ):
         self._user = username
         self._password = password
         self.citizen_id = citizen_id
         self.session = requests.Session()
+        self._ntfy_topic = ntfy_topic
         self._auth_token: Optional[str] = None
         self._auth_token_expiry: Optional[datetime.datetime] = None
         self._refresh_token: Optional[str] = None
@@ -353,6 +361,10 @@ def find(
 
         if book:
             return self._book(appt)
+        else:
+            self.notify(
+                f"An appointment is available on {appt['vaccinationDate']} at {appt['vaccinationTime']}."
+            )
 
         return True
 
@@ -377,6 +389,14 @@ def _book(self, payload: Dict[str, Union[str, Dict[str, bool]]]) -> bool:
             return False
 
         logging.info("Appointment booked.")
+
+        self.notify(
+            (
+                f"Appointment booked for {payload['vaccinationDate']} at {payload['vaccinationTime']}."
+                "Please check your e-mails."
+            )
+        )
+
         return True
 
     def print_appointments(self):
@@ -413,3 +433,26 @@ def print_appointments(self):
                 appt["slotId"]["date"],
                 appt["slotId"]["time"],
             )
+
+    def notify(self, title: str):
+        if not self._ntfy_topic:
+            return
+
+        requests.post(
+            self.NTFY_SH_URL,
+            json={
+                "topic": self._ntfy_topic,
+                "message": title,
+                "title": "",
+                "tags": ["syringe"],
+                "priority": 4,
+                "actions": [
+                    {
+                        "action": "view",
+                        "label": "BayIMCO portal",
+                        "url": f"https://impfzentren.bayern/citizen/overview/{self.citizen_id}",
+                        "clear": True,
+                    }
+                ],
+            },
+        )

impf.py

@@ -103,6 +103,13 @@ def main():
         required=False,
         help="Variants to find vaccines for; either ba1 or ba45. Leave blank for any.",
     )
+    parser.add_argument(
+        "--ntfy-topic",
+        type=str,
+        default=None,
+        required=False,
+        help="ntfy.sh topic to send a message to on success. See https://ntfy.sh for details.",
+    )
     args = parser.parse_args()
 
     if args.dose == VaccinationType.SECOND and args.first_vaccine is None:
@@ -117,7 +124,10 @@ def main():
     )
 
     checker = ImpfChecker(
-        username=args.email, password=args.password, citizen_id=args.citizen_id
+        username=args.email,
+        password=args.password,
+        citizen_id=args.citizen_id,
+        ntfy_topic=args.ntfy_topic,
     )
 
     if args.interval is not None: