Get pushover notification about earliest appoitment for selected doctor/place in znanylekarz.pl

znanylekarzPushover.py

"""
[pushover]
user_key = "..."
api_token = "..."

[api]
doctor_id = 123456
place_id = 234567
start_date = "2025-02-20"
end_date = "2030-11-12"
"""

import tomllib
import requests
import cloudscraper
import re
import time
import logging
from datetime import datetime

logging.basicConfig(
    format="%(asctime)s: %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S",
    level=logging.INFO
)

with open("config.toml", "rb") as f:
    config = tomllib.load(f)

PUSHOVER_USER_KEY = config["pushover"]["user_key"]
PUSHOVER_API_TOKEN = config["pushover"]["api_token"]
DOCTOR_ID = config["api"]["doctor_id"]
PLACE_ID = config["api"]["place_id"]
START_DATE = config["api"]["start_date"]
END_DATE = config["api"]["end_date"]

api_url = f"https://www.znanylekarz.pl/api/v3/doctors/{DOCTOR_ID}/addresses/{PLACE_ID}/slots?start={START_DATE}T00%3A00%3A00%2B01%3A00&end={END_DATE}T00%3A00%3A00%2B01%3A00&with%5B%5D=address.nearest_slot_after_end"

def send_pushover_notification(message):
    data = {
        "token": PUSHOVER_API_TOKEN,
        "user": PUSHOVER_USER_KEY,
        "message": message,
        "title": "Znanylekarz.pl"
    }
    requests.post("https://api.pushover.net/1/messages.json", data=data)

session = cloudscraper.create_scraper()

headers = {
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:131.0) Gecko/20100101 Firefox/131.0",
    "Accept": "application/json, text/plain, */*",
    "Accept-Language": "en-US,en;q=0.5",
    "Accept-Encoding": "gzip, deflate, br, zstd",
    "DNT": "1",
    "Connection": "keep-alive",
    "Pragma": "no-cache",
    "Cache-Control": "no-cache"
}

url = "https://www.znanylekarz.pl"

def get_access_token():
    resp = session.get(url, headers=headers)
    token_match = re.search(r"'ACCESS_TOKEN': '([a-zA-Z0-9]+)'", resp.text)
    if token_match:
        return token_match.group(1)
    return None

lowest_date = None

while True:
    try:
        access_token = get_access_token()
        if not access_token:
            retry_seconds = 60
            logging.info(f"No access token. Retrying in {retry_seconds} seconds...")
            time.sleep(retry_seconds)
            continue

        headers["Authorization"] = f"Bearer {access_token}"
        headers["Referer"] = url

        resp = session.get(api_url, headers=headers)

        if resp.status_code == 200:
            data = resp.json()
            slots = data.get("_items", [])
            if slots:
                earliest_slot = min(slots, key=lambda x: x["start"])["start"]
                earliest_date = datetime.fromisoformat(earliest_slot)

                if lowest_date is None or earliest_date < lowest_date:
                    lowest_date = earliest_date
                    booking_url = next(s["booking_url"] for s in slots if s["start"] == earliest_slot)
                    logging.info(f"Earliest appointment {lowest_date}")
                    send_pushover_notification(f"New appointment\n{lowest_date}\n\n{booking_url}")
                elif earliest_date > lowest_date:
                    logging.info(f"Missed: {lowest_date}, New lowest: {earliest_date}")
                    send_pushover_notification(f"Missed\n{lowest_date}\n\nNew appointment\n{earliest_date}\n\n{booking_url}")
                    lowest_date = earliest_date
                else:
                    logging.info("Nothing fancy.. sleeping...")
            else:
                logging.info("No available slots found.")
        else:
            logging.error(f"Error: {resp.status_code}")
    except Exception as e:
        logging.error(f"An error occurred: {e}")

    time.sleep(120)