-
Notifications
You must be signed in to change notification settings - Fork 0
/
main.py
177 lines (158 loc) · 6.25 KB
/
main.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
import requests
import json
import hashlib
from datetime import date
import time, traceback
import threading
API_BASE = "https://cdn-api.co-vin.in/api/v2"
BENEF_ID = "<Your_Beneficiary_Id>"
MOBILE_NO = "<Your_Mobile_number>"
DISTRICT_ID = 307 # District Id of Ekm
USER_AGENT_SECRET = "U2FsdGVkX1/bWMpon6LReFzd84D/C+lCPnUr4eAmAFQQwY+CYD/PC3M/kEgD2affS30yycuEXbo/YQYLGE3xmA=="
USER_AGENT = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.77 Safari/537.36"
DOSE = 1
MIN_AGE = 18
DOSE_FILTER = "available_capacity_dose" + str(DOSE)
uaHeaders = {'user-agent': USER_AGENT}
BOOK = False
def every(delay, task):
next_time = time.time() + delay
while True:
time.sleep(max(0, next_time - time.time()))
try:
task()
except Exception:
traceback.print_exc()
# in production code you might want to have this instead of course:
# logger.exception("Problem while executing repetitive task.")
# skip tasks if we are behind schedule:
next_time += (time.time() - next_time) // delay * delay + delay
today = date.today().strftime("%d-%m-%Y")
def encrypt_string(hash_string):
sha_signature = \
hashlib.sha256(hash_string.encode()).hexdigest()
return sha_signature
def hasDose(session):
return session[DOSE_FILTER]!=0
def filterSession(session):
return session["vaccine"] == "COVISHIELD" and session["min_age_limit"] == MIN_AGE and hasDose(session)
headers=uaHeaders
class VaccineBooking:
def __init__(self):
self.token = ""
self.txnId = ""
def getCenters(self, p=False):
url = "".join((API_BASE, "/appointment/sessions/public/calendarByDistrict"))
PARAMS = {"district_id": DISTRICT_ID, "date": today}
# sending get request and saving the response as response object
r = requests.get(url = url, params = PARAMS)
resJson = r.json()
if r.status_code != 200:
print("Error: statusCode={}, resJson={}".format(r.status_code, resJson))
return
centers = resJson["centers"]
filteredCenters = []
for center in centers:
sessions = center["sessions"]
sessions = list(filter(filterSession, sessions))
sessions.sort(key=lambda session:session[DOSE_FILTER], reverse=True)
if sessions:
filteredCenters.append({ **center, "sessions": sessions })
if p:
print(json.dumps(filteredCenters, indent=4))
try:
filteredCenters.sort(key=lambda x: x.sessions[0][DOSE_FILTER], reverse=True) # sort by centers that have highest available doses
except:
pass
return filteredCenters
def sendOTP(self):
# Generate OTP
url = "".join((API_BASE, "/auth/public/generateOTP"))
data = {"mobile": MOBILE_NO, "secret": USER_AGENT_SECRET}
print(data)
r = requests.post(url=url, json=data, headers=headers)
try:
resJson = r.json()
except Exception:
print("Error", r)
traceback.print_exc()
# return
if r.status_code != 200:
print("Error: statusCode={}, resJson={}".format(r.status_code, resJson))
return
self.txnId = resJson["txnId"]
# print("txnId: ", self.txnId)
def confirmOTP(self):
otp = input("Enter OTP: ")
if otp == "x":
return "x" # Dismiss
# Validate OTP
url = "".join((API_BASE, "/auth/public/confirmOTP"))
data = {"otp": encrypt_string(otp), "txnId": self.txnId}
print(data)
r = requests.post(url=url, json=data, headers=headers)
try:
resJson = r.json()
except Exception:
print("Error confirming OTP", r)
traceback.print_exc()
return
if r.status_code != 200:
print("Error: statusCode={}, resJson={}".format(r.status_code, resJson))
return
self.token = resJson["token"]
print("token: ", self.token)
def scheduleAppointment(self, session_id, slot, beneficiaries=[BENEF_ID]):
url = "".join((API_BASE, "/appointment/schedule"))
while not self.token:
print("Unauthorized - Please login")
self.sendOTP()
x = self.confirmOTP()
# x = self.generateAndConfirmOTP()
if x == "x":
return "x"
headers = {"authorization": " ".join(("Bearer", self.token)), **uaHeaders}
data = {"session_id": session_id, "dose": DOSE, "slot": slot, "beneficiaries": beneficiaries}
r = requests.post(url=url, json=data, headers=headers)
try:
resJson = r.json()
except Exception:
print("Error booking appointment", r)
traceback.print_exc()
return
if r.status_code == 401:
self.token = ""
print("401")
self.scheduleAppointment(session_id, slot)
elif r.status_code != 200:
print("Error: statusCode={}, resJson={}".format(r.status_code, resJson))
retry = input("retry? y/n: ")
if retry == "y":
self.token = "" # clear
self.scheduleAppointment(session_id, slot)
return
confirmation = resJson.get("appointment_confirmation_no") or resJson.get("appointment_id")
print(confirmation)
return confirmation
book = VaccineBooking()
def main():
centers = book.getCenters()
if centers:
print(json.dumps(centers, indent=4))
if BOOK:
center = centers[0]
session = center["sessions"][0]
sessionId = session["session_id"]
slot = session["slots"][1]
# print("Chosen center: \n_________________\n", json.dumps(center, indent=4))
confirmation = None
while not confirmation:
confirmation = book.scheduleAppointment(sessionId, slot)
if confirmation != "x":
raise Exception("Completed") # In order to terminate the execution
else:
book.sendOTP() # Just send an OTP (for alert)
else:
print("No centers available!")
threading.Thread(target=lambda: every(5, main)).start()
# target is the callable object to be invoked by the run() method.