Prullenbakvaccin bot for Telegram

prullenbakvaccin.py

"""
Copyright (C) 2021 Bas Stottelaar <[email protected]>

License: Beerware

Disclaimer: use at your own risk

Usage:

- Install dependencies
  - Python 3.9 (might work with 3.8 if you adapt type hints)
  - Install modules with `pip install -U requests bs4`
- Create a Telegram bot
- Find the chat identifier
  - Send a message to your bot
  - Check `https://api.telegram.org/bot<TOKEN>/getUpdates` and find the chat identifier
- Configure the details below
- Run the script and leave it running

Known issues:

- It does not work anymore, because the website changed
- No retries if it fails
- Does not de-duplicate results
- Keeps sending notifications if there are vaccins available
"""

from datetime import datetime

import requests
import random
import time
import bs4
import sys
import os
import re


# Configure the details below.
POSTAL_CODES = []

TELEGRAM_BOT_TOKEN = ""
TELEGRAM_CHAT_ID = ""


def create_session() -> requests.Session:
    """
    Create a browser-like session instance.
    """

    session = requests.Session()

    session.headers.update({
        "User-Agent": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/90.0.4430.93 Safari/537.36"
    })
    session.headers.update({
        "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.9"
    })
    session.headers.update({
        "Accept-Language": "nl-NL,nl;q=0.9,en-US;q=0.8,en;q=0.7"
    })

    return session


def get_token(session: requests.Session) -> str:
    """
    Retrieve the CSRF(?) token from the main page.
    """

    response = session.get("https://www.prullenbakvaccin.nl/")
    response.raise_for_status()

    soup = bs4.BeautifulSoup(response.content, features="html.parser")

    if "niet beschikbaar" in soup.text:
        raise requests.exceptions.RequestException("Service not available.")

    return soup.select_one("input[name=_token]")["value"]


def get_vaccines(
    session: requests.Session,
    token: str,
    postal_code: str
) -> list[str]:
    """
    Retrieve the location information page.
    """

    response = session.post(
        "https://www.prullenbakvaccin.nl/", data={
            "_token": token,
            "location": postal_code,
        }
    )
    response.raise_for_status()

    soup = bs4.BeautifulSoup(response.content, features="html.parser")

    cards = soup.select("div[class=card-body]")

    if not cards:
        return []

    results = []

    for card in cards:
        text = card.text.strip().replace("\n", " ")
        text = re.sub(r"\s\s+", " ", text)
        
        if "Heeft geen vaccins" not in text:
            results.append(text)

    return results


def send_message(message: str) -> None:
    """
    Send a telegram message.
    """

    response = requests.get(
        f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendMessage",
        params={
            "chat_id": TELEGRAM_CHAT_ID,
            "parse_mode": "Markdown",
            "text": message
        }
    )
    response.raise_for_status()


def main(argv: list[str]) -> int:
    """
    Application entry point.
    """

    send_message("Bot started 💉.")

    while True:
        results = {}

        postal_codes = POSTAL_CODES.copy()
        
        random.shuffle(postal_codes)

        for postal_code in postal_codes:
            try:
                with create_session() as session:
                    print(f"Requesting for {postal_code}.")

                    token = get_token(session)
                    vaccines = get_vaccines(session, token, postal_code)

                    if vaccines:
                        results[postal_code] = vaccines

                time.sleep(random.randrange(1.0, 5.0))
            except requests.exceptions.RequestException as e:
                send_message(f"Unable to fetch: {e}")
                break

        # Send message if needed.
        if results:
            print("Found %d results." % len(results))

            for postal_code, vaccines in results.items():
                lines = []

                for vaccine in vaccines:
                    lines.append(f"- {vaccine}")

                message.append(
                    f"Postcode {postal_code}:\n" + "\n".join(lines)
                )

            send_message("\n\n".join(message))

        # Sleep some time.
        now = datetime.now()

        if now.hour > 8 and now.hour < 22:
            sleep = random.randrange(60.0 * 1, 60.0 * 3)
        else:
            sleep = 60.0 * 60

        print("Sleeping for %s seconds." % sleep)

        time.sleep(sleep)

    return 0


if __name__ == "__main__":
    main(sys.argv)