Skip to content

Instantly share code, notes, and snippets.

@TestPrab
Last active June 23, 2021 16:15
Show Gist options
  • Select an option

  • Save TestPrab/3464e751c6a5e25e738b00c2ad784074 to your computer and use it in GitHub Desktop.

Select an option

Save TestPrab/3464e751c6a5e25e738b00c2ad784074 to your computer and use it in GitHub Desktop.
Vaccination Bot , Notifies you about the availability of Vaccine for the particular district on telegram . Reference taken from https://github.com/RAvengineer/vad-bot , by @RAvengineer
from requests import get, post
from time import sleep
from datetime import datetime
from random import randint
API_SETU_URL = "https://cdn-api.co-vin.in/api/v2/"
APPOINTMENTS_AVAILABILITY = "appointment/sessions/public/"
CALENDAR_DISTRICT = "calendarByDistrict"
CALENDAR_PINCODE = "calendarByPin"
API_HEADERS = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/90.0.4430.93 Safari/537.36 Edg/90.0.818.51"
}
PINCODE = 0
DISTRICT_ID = "" ,# Enter Your District Id from "https://github.com/bhattbhavesh91/cowin-vaccination-slot-availability/blob/main/district_mapping.csv"
bot_token = "1892929544:AAHp_ls9b5WeaAqNjONi3IwHmDJMiQqp-J8"
bot_chatID = "XXXX" # enter bot your telegram chat ID
def fetchCalendarByDistrict(district_id: str, date: datetime) -> dict:
params = {"district_id": district_id, "date": date.strftime("%d-%m-%Y")}
resp = get(
url=API_SETU_URL + APPOINTMENTS_AVAILABILITY + CALENDAR_DISTRICT,
params=params,
headers=API_HEADERS,
)
print(f"{str(date)}\tAPI response: {resp.status_code}")
if resp.status_code != 200:
return None
return resp.json()
def getNewCenters(old_centers: list, new_centers: list) -> list:
x: set = set(old_centers)
y: set = set(new_centers)
return list(y.union(x).difference(x))
def fetchCalendarByPinCode(pincode: str, date: datetime) -> dict:
params = {"pincode": pincode, "date": date.strftime("%d-%m-%Y")}
resp = get(
url=API_SETU_URL + APPOINTMENTS_AVAILABILITY + CALENDAR_PINCODE,
params=params,
headers=API_HEADERS,
)
print(f"{str(date)}\tAPI response: {resp.status_code}")
if resp.status_code != 200:
print(f"fetchCalendarByPinCode response: {resp.content}")
return None
return resp.json()
def getAvailableCenters(data: dict) -> list:
available_centers = list()
for center in data["centers"]:
for session in center["sessions"]:
if session["available_capacity"] > 0:
available_centers.append(
f"{center['pincode']} - {center['name']} -- {session['min_age_limit']} -- {session['vaccine']} -- {session['date']} -- D1 --{session['available_capacity_dose1']}--D2 --{session['available_capacity_dose2']} "
)
break
return available_centers
def fetchData() -> dict:
dt = datetime.now()
try:
data = fetchCalendarByDistrict(district_id=DISTRICT_ID, date=dt)
if not data:
raise Exception("API_Error: fetchCalendarByDistrict returned None")
return data
except:
error_title = "Error in fetchData: Invalid SEARCH_BY mode."
error_body = "SEARCH_BY variable in .env should be either DISTRICT or PINCODE"
raise Exception(error_title + "\n" + error_body)
if __name__ == "__main__":
try:
previous_centers = list()
while True:
# Fetch & process data
data = fetchData()
centers = getNewCenters(
old_centers=previous_centers, new_centers=getAvailableCenters(data)
)
# Avoid Repetition of centers
# Current solution repeats centers every alternate cycle
previous_centers = centers
if centers:
message = str(datetime.now) + " Vaccination Info "
message = "\n".join(centers)
send_text = (
"https://api.telegram.org/bot"
+ bot_token
+ "/sendMessage?chat_id="
+ bot_chatID
+ "&parse_mode=Markdown&text="
+ message
)
response = get(send_text)
pass
# APIs are subject to a rate limit of 100 API calls per 5 minutes per IP
delay_duration = randint(20, 40)
print(f"Train will depart after {delay_duration} sec")
sleep(delay_duration)
except Exception as e:
print(str(e))
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment