Last active
May 19, 2021 08:46
-
-
Save basilfx/6d18c642be3e8a2619859c787b063106 to your computer and use it in GitHub Desktop.
Prullenbakvaccin bot for Telegram
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| """ | |
| Copyright (C) 2021 Bas Stottelaar <basstottelaar@gmail.com> | |
| 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) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment