Last active
June 23, 2021 16:15
-
-
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
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
| 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