Skip to content

Commit

Permalink
changes to support search by pincode
Browse files Browse the repository at this point in the history
  • Loading branch information
@pallupz
pallupz committed May 5, 2021
1 parent f07ea5e commit 6fc464c
Show file tree
Hide file tree
Showing 3 changed files with 125 additions and 46 deletions.
2 changes: 1 addition & 1 deletion requirements.txt
View file
Original file line number Diff line number Diff line change
@@ -1,3 +1,3 @@
requests
tabulate
inputimeout
inputimeout
26 changes: 19 additions & 7 deletions src/covid-vaccine-slot-booking.py
View file
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
from collections import Counter
import requests, sys, argparse, os
from utils import generate_token_OTP, get_beneficiaries, check_and_book, get_districts, get_min_age, beep, \
from utils import generate_token_OTP, get_beneficiaries, check_and_book, get_districts, get_pincodes, beep, \
BENEFICIARIES_URL, WARNING_BEEP_DURATION


Expand Down Expand Up @@ -37,15 +37,27 @@ def main():
os.system("pause")
sys.exit(1)

# Collect vaccination center preferance
district_dtls = get_districts()
print("================================= Location Info =================================\n")

search_option = input("""Search by Pincode? Or by State/District? \nEnter 1 for Pincode or 2 for State/District. (Default 2) : """)
search_option = int(search_option) if int(search_option) in [1, 2] else 2

print("================================= Additional Info =================================")
if search_option == 2:
# Collect vaccination center preferance
location_dtls = get_districts()

else:
# Collect vaccination center preferance
location_dtls = get_pincodes()

print("================================= Additional Info =================================\n")

# Set filter condition
minimum_slots = int(input(f'Filter out centers with availability less than ? Minimum {len(beneficiary_dtls)} : '))
minimum_slots = minimum_slots if minimum_slots >= len(beneficiary_dtls) else len(beneficiary_dtls)
minimum_slots = input(f'Filter out centers with availability less than ? Minimum {len(beneficiary_dtls)} : ')
if minimum_slots:
minimum_slots = int(minimum_slots) if int(minimum_slots) >= len(beneficiary_dtls) else len(beneficiary_dtls)
else:
minimum_slots = len(beneficiary_dtls)

# Get refresh frequency
refresh_freq = input('How often do you want to refresh the calendar (in seconds)? Default 15. Minimum 5. : ')
Expand All @@ -56,7 +68,7 @@ def main():
request_header = {"Authorization": f"Bearer {token}"}

# call function to check and book slots
token_valid = check_and_book(request_header, beneficiary_dtls, district_dtls,
token_valid = check_and_book(request_header, beneficiary_dtls, location_dtls, search_option,
min_slots=minimum_slots,
ref_freq=refresh_freq)

Expand Down
143 changes: 105 additions & 38 deletions src/utils.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,8 @@

BOOKING_URL = "https://cdn-api.co-vin.in/api/v2/appointment/schedule"
BENEFICIARIES_URL = "https://cdn-api.co-vin.in/api/v2/appointment/beneficiaries"
CALENDAR_URL = "https://cdn-api.co-vin.in/api/v2/appointment/sessions/calendarByDistrict?district_id={0}&date={1}"
CALENDAR_URL_DISTRICT = "https://cdn-api.co-vin.in/api/v2/appointment/sessions/calendarByDistrict?district_id={0}&date={1}"
CALENDAR_URL_PINCODE = "https://cdn-api.co-vin.in/api/v2/appointment/sessions/calendarByPin?pincode={0}&date={1}"
WARNING_BEEP_DURATION = (1000, 2000)

try:
Expand All @@ -22,6 +23,33 @@ def beep(freq, duration):
winsound.Beep(freq, duration)


def viable_options(resp, minimum_slots, min_age_booking):
options = []
if len(resp['centers']) >= 0:
for center in resp['centers']:
for session in center['sessions']:
if (session['available_capacity'] >= minimum_slots) \
and (session['min_age_limit'] <= min_age_booking):
out = {
'name': center['name'],
'district': center['district_name'],
'pincode': center['pincode'],
'center_id': center['center_id'],
'available': session['available_capacity'],
'date': session['date'],
'slots': session['slots'],
'session_id': session['session_id']
}
options.append(out)

else:
pass
else:
pass

return options


def display_table(dict_list):
"""
This function
Expand All @@ -34,7 +62,7 @@ def display_table(dict_list):
print(tabulate.tabulate(rows, header, tablefmt='grid'))


def check_calendar(request_header, vaccine_type, district_dtls, minimum_slots, min_age_booking):
def check_calendar_by_district(request_header, vaccine_type, location_dtls, minimum_slots, min_age_booking):
"""
This function
1. Takes details required to check vaccination calendar
Expand All @@ -47,52 +75,76 @@ def check_calendar(request_header, vaccine_type, district_dtls, minimum_slots, m
today = datetime.datetime.today()
tomorrow = (today + datetime.timedelta(days=1)).strftime("%d-%m-%Y")

base_url = CALENDAR_URL
base_url = CALENDAR_URL_DISTRICT

if vaccine_type:
base_url += f"&vaccine={vaccine_type}"

options = []
for district in district_dtls:
resp = requests.get(base_url.format(district['district_id'], tomorrow), headers=request_header)
for location in location_dtls:
resp = requests.get(base_url.format(location['district_id'], tomorrow), headers=request_header)

if resp.status_code == 401:
print('TOKEN INVALID')
return False

elif resp.status_code == 200:
resp = resp.json()
print(
f"Centers available in {district['district_name']} from {tomorrow} as of {today.strftime('%Y-%m-%d %H:%M:%S')}: {len(resp['centers'])}")

if len(resp['centers']) >= 0:
for center in resp['centers']:
for session in center['sessions']:
if (session['available_capacity'] >= minimum_slots) \
and (session['min_age_limit'] <= min_age_booking):
out = {
'name': center['name'],
'district': center['district_name'],
'center_id': center['center_id'],
'available': session['available_capacity'],
'date': session['date'],
'slots': session['slots'],
'session_id': session['session_id']
}
options.append(out)

else:
pass
else:
pass
print(f"Centers available in {location['district_name']} from {tomorrow} as of {today.strftime('%Y-%m-%d %H:%M:%S')}: {len(resp['centers'])}")
options += viable_options(resp, minimum_slots, min_age_booking)

else:
pass

for dist in set([item['district'] for item in options]):
for district in district_dtls:
if dist == district['district_name']:
for _ in range(2):
# beep twice for each district with a slot
beep(district['district_alert_freq'], 150)
for location in location_dtls:
if location['district_name'] in [option['district'] for option in options]:
for _ in range(2):
beep(location['alert_freq'], 150)
return options

except Exception as e:
print(str(e))
beep(WARNING_BEEP_DURATION[0], WARNING_BEEP_DURATION[1])


def check_calendar_by_pincode(request_header, vaccine_type, location_dtls, minimum_slots, min_age_booking):
"""
This function
1. Takes details required to check vaccination calendar
2. Filters result by minimum number of slots available
3. Returns False if token is invalid
4. Returns list of vaccination centers & slots if available
"""
try:
print('===================================================================================')
today = datetime.datetime.today()
tomorrow = (today + datetime.timedelta(days=1)).strftime("%d-%m-%Y")

base_url = CALENDAR_URL_PINCODE

if vaccine_type:
base_url += f"&vaccine={vaccine_type}"

options = []
for location in location_dtls:
resp = requests.get(base_url.format(location['pincode'], tomorrow), headers=request_header)

if resp.status_code == 401:
print('TOKEN INVALID')
return False

elif resp.status_code == 200:
resp = resp.json()
print(f"Centers available in {location['pincode']} from {tomorrow} as of {today.strftime('%Y-%m-%d %H:%M:%S')}: {len(resp['centers'])}")
options += viable_options(resp, minimum_slots, min_age_booking)

else:
pass

for location in location_dtls:
if int(location['pincode']) in [option['pincode'] for option in options]:
for _ in range(2):
beep(location['alert_freq'], 150)

return options

Expand Down Expand Up @@ -134,7 +186,7 @@ def book_appointment(request_header, details):
beep(WARNING_BEEP_DURATION[0], WARNING_BEEP_DURATION[1])


def check_and_book(request_header, beneficiary_dtls, district_dtls, **kwargs):
def check_and_book(request_header, beneficiary_dtls, location_dtls, search_option, **kwargs):
"""
This function
1. Checks the vaccination calendar for available slots,
Expand All @@ -150,13 +202,16 @@ def check_and_book(request_header, beneficiary_dtls, district_dtls, **kwargs):
minimum_slots = kwargs['min_slots']
refresh_freq = kwargs['ref_freq']

options = check_calendar(request_header, vaccine_type, district_dtls, minimum_slots, min_age_booking)
if search_option == 2:
options = check_calendar_by_district(request_header, vaccine_type, location_dtls, minimum_slots, min_age_booking)
else:
options = check_calendar_by_pincode(request_header, vaccine_type, location_dtls, minimum_slots, min_age_booking)

if isinstance(options, bool):
return False

options = sorted(options,
key=lambda k: (k['district'].lower(),
key=lambda k: (k['district'].lower(), k['pincode'],
k['name'].lower(),
datetime.datetime.strptime(k['date'], "%d-%m-%Y"))
)
Expand Down Expand Up @@ -212,6 +267,18 @@ def check_and_book(request_header, beneficiary_dtls, district_dtls, **kwargs):
pass


def get_pincodes():
locations = []
pincodes = input("Enter all the pincodes you are interested separated by commas: ")
for idx, pincode in enumerate(pincodes.split(',')):
pincode = {
'pincode': pincode,
'alert_freq': 440 + ((2 * idx) * 110)
}
locations.append(pincode)
return locations


def get_districts():
"""
This function
Expand Down Expand Up @@ -256,7 +323,7 @@ def get_districts():
reqd_districts = [{
'district_id': item['district_id'],
'district_name': item['district_name'],
'district_alert_freq': 440 + ((2 * idx) * 110)
'alert_freq': 440 + ((2 * idx) * 110)
} for idx, item in enumerate(districts) if idx in districts_idx]

print(f'Selected districts: ')
Expand Down

0 comments on commit 6fc464c

Please sign in to comment.