Last active
May 28, 2021 12:34
-
-
Save Debilski/beacf353ba837efe6e47bd2778923882 to your computer and use it in GitHub Desktop.
Search the Berlin Doctolib site for an available vaccination slot in the next N days. If you run macOS then add `--exec 'say -v Anna {NAME}'` and it will speak to you.
This file contains hidden or 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
import argparse | |
from datetime import date, datetime | |
import subprocess | |
import shlex | |
import sys | |
import time | |
import requests | |
from requests.api import request | |
# These names will be searched in the location data | |
TYPES = ['BIONTECH', 'MODERNA', 'ASTRA'] | |
LOCATIONS = ['VELODROM', 'EISSTADION', 'TEMPELHOF', 'TEGEL', 'ARENA', 'MESSE'] | |
session = requests.Session() | |
session.headers.update({'User-Agent': 'Mozilla/5.0 (Windows NT 6.1; Win64; x64; rv:59.0) Gecko/20100101 Firefox/59.0'}) | |
def run_exec(location): | |
for exec_cmd in args.exec: | |
split_cmd = shlex.split(exec_cmd) | |
cmd = [] | |
for arg in split_cmd: | |
if '{NAME}' in arg: | |
cmd.append(arg.format(NAME=location)) | |
else: | |
cmd.append(arg) | |
subprocess.run(cmd) | |
def RESET_LINE(): | |
_MSWINDOWS = (sys.platform == "win32") | |
if _MSWINDOWS: | |
print('', end='\r', flush=True) | |
else: | |
print('', end='\x1b[1K\r', flush=True) | |
def init_locations(*, start_date, limit, skip_location=None, skip_vaccine=None): | |
if skip_location is None: | |
skip_location = [] | |
if skip_vaccine is None: | |
skip_vaccine = [] | |
print("Loading practice data.") | |
j = session.get('https://www.doctolib.de/booking/ciz-berlin-berlin.json').json() | |
data = j['data'] | |
visit_motives = {} | |
for motive in data['visit_motives']: | |
if not motive['first_shot_motive']: | |
continue | |
for vacc in skip_vaccine: | |
if vacc in motive['name'].upper(): | |
print(f"Found vacc type: {motive['name']}. Ignoring as requested.") | |
break | |
else: | |
print(f"Found vacc type: {motive['name']}.") | |
visit_motives[motive['id']] = motive['name'] | |
print("Finding locations.") | |
locations = {} | |
for location in data['places']: | |
for loc in skip_location: | |
if loc in location['name'].upper(): | |
print(f"Found location {location['name']}. Ignoring as requested.") | |
break | |
else: | |
print(f"Found location {location['name']}.") | |
locations[location['practice_ids'][0]] = location['name'] | |
print("Collecting agendas for each location and vacc type.") | |
pm_agendas = {} | |
for agenda in data['agendas']: | |
if agenda['booking_disabled']: | |
# skip this one | |
continue | |
for practice_id in locations: | |
# need to cast to a string here. little weird … | |
if str(practice_id) in agenda['visit_motive_ids_by_practice_id']: | |
for motive in agenda['visit_motive_ids_by_practice_id'][str(practice_id)]: | |
if not motive in visit_motives: | |
# We have skipped this vaccination type | |
continue | |
if not (practice_id, motive) in pm_agendas: | |
pm_agendas[(practice_id, motive)] = [] | |
pm_agendas[(practice_id, motive)].append(agenda['id']) | |
print('.', end='') | |
print('done') | |
print('Generating URLs.') | |
data = {} | |
for ((practice, motive), agendas) in pm_agendas.items(): | |
agenda_ids = "-".join(map(str, agendas)) | |
url = f'https://www.doctolib.de/availabilities.json?start_date={start_date}&visit_motive_ids={motive}&agenda_ids={agenda_ids}&insurance_sector=public&practice_ids={practice}&destroy_temporary=true&limit={limit}' | |
for loc_code in LOCATIONS: | |
if loc_code in locations[practice].upper(): | |
short_loc = loc_code | |
break | |
else: | |
print(f'Cannot find {locations[practice]} in known locations.') | |
short_loc = 'UNKNWN' | |
short_type = 'UNKNWN' | |
for type_code in TYPES: | |
if type_code in visit_motives[motive].upper(): | |
short_type = type_code | |
break | |
else: | |
print(f'Cannot find {visit_motives[motive]} in known vaccination types.') | |
short_type = 'UNKNWN' | |
short_name = f'{short_loc}_{short_type}' | |
data[short_name] = { | |
'type_short': short_type, | |
'loc_short': short_loc, | |
'url': url, | |
'type': visit_motives[motive], | |
'loc': locations[practice] | |
} | |
return data | |
def check_vaccinations(locations): | |
for loc, data in locations.items(): | |
print(f"[{datetime.now().isoformat(sep=' ', timespec='seconds')}] {loc}: Checking", end='', flush=True) | |
url = data['url'].format(start_date=today, limit=limit) | |
r = session.get(url) | |
try: | |
j = r.json() | |
except ValueError: | |
print("Error decoding the json from", url) | |
continue | |
total = j['total'] | |
next_slot = data.get('next_slot') or "unknown" | |
RESET_LINE() | |
print(f"[{datetime.now().isoformat(sep=' ', timespec='seconds')}] {loc}: Total # of slots {total}. Next date: {next_slot}", end='', flush=True) | |
if total > 0: | |
print("!!!") | |
print("!!!") | |
run_exec(loc) | |
for av in j['availabilities']: | |
if av['slots'] != []: | |
print(f"{av['date']}: {len(av['slots'])}") | |
print("@@@") | |
time.sleep(3) | |
RESET_LINE() | |
if __name__ == '__main__': | |
parser = argparse.ArgumentParser(description='Check for available vaccines') | |
parser.add_argument('--exec', action='append', | |
help='Execute this external command when a match is found.') | |
parser.add_argument('--test-exec', action='store_const', const=True, | |
help='Test exec arcuments') | |
parser.add_argument('--skip-location', action='append', | |
help=f'Skip a location: {" ,".join(LOCATIONS)}') | |
parser.add_argument('--skip-vaccine', action='append', | |
help=f'Skip a vaccine: {" ,".join(TYPES)}') | |
parser.add_argument('--limit', type=int, default=10, | |
help='Search for the next N days.') | |
args = parser.parse_args() | |
if args.test_exec: | |
run_exec("TEST") | |
sys.exit() | |
today = date.today().isoformat() | |
limit = args.limit | |
skip_location = [] | |
for loc in (args.skip_location or []): | |
if not loc in LOCATIONS: | |
print(f"Location code '{loc}' unkown. Ignoring.") | |
continue | |
skip_location.append(loc) | |
skip_vaccine = [] | |
for vacc in (args.skip_vaccine or []): | |
if not vacc in TYPES: | |
print(f"Vaccine code '{vacc}' unkown. Ignoring.") | |
continue | |
skip_vaccine.append(vacc) | |
print(f'Searching vaccination dates starting {today} for the next {limit} days.') | |
locations = init_locations(start_date=today, limit=limit, skip_location=skip_location, skip_vaccine=skip_vaccine) | |
print('Start scraping.') | |
while True: | |
check_vaccinations(locations) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Usage:
To check for an appointment in the Impfzentren in the next 14 days from now, excluding AstraZeneca and speaking with a German voice (
-v Anna
) over the speakers on macOS use: