truroshan/cowin-termux
Archived
1
0
Fork 0
Code Issues Pull requests Actions Packages Projects Releases Wiki Activity

Update cowin.py

Browse source
This commit is contained in:
roShan 2021-06-02 14:55:47 +05:30 committed by GitHub
parent 742582b5f2
commit a4cb3c239f
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
1 changed files with 124 additions and 70 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

156
cowin.py
View file

@ -1,4 +1,5 @@
from apscheduler.schedulers.blocking import BlockingScheduler
try:
import schedule
from bs4 import BeautifulSoup
from datetime import datetime
import subprocess
@ -11,40 +12,46 @@ import fire
import sys
import re
import os
except ImportError:
print("First run requirements instaling command")
print("pip install -r requirements.txt")
exit()
OTP_SITE_URL = None
OTP_VALID_DURATION_SECONDS = 180
'''
Add Worker Domain here example : https://db.domain.workers.dev
Add Worker Domain here in Double Quore example : "https://db.domain.workers.dev"
Check this : https://github.com/truroshan/CloudflareCoWinDB
'''
scheduler = BlockingScheduler()
# scheduler = BlockingScheduler()
def line_break(): print("-"*25)
def clear_screen(): os.system("clear")
def clear_screen(): os.system('clear' if os.name =='posix' else 'cls')
class CoWinBook():
def init(
self,
mobile_no,
pin = "Not Passed",
age = 18 ,
vaccine = "ANY",
dose = 1,
otp = 'a',
time = 30,
bookToday = 1,
relogin = None ):
mobile_no, # --m
pin = "Not Passed", # --p
age = 18 , # --a
vaccine = "ANY", # --v
dose = 1, # --d
otp = 'a', # --o
time = 30, # --t
bookToday = 1, # --b
fee_type = "BOTH", # --f
relogin = None # --r
):
self.mobile_no = str(mobile_no)
if relogin and os.path.exists(self.mobile_no): os.remove(self.mobile_no)
# Cron Time
global TIME
TIME = time
TIME = 8 if time < 10 else time
# Include today session for Booking Slot
self.bookToday = 0 if bookToday is True else 1
@ -55,6 +62,7 @@ class CoWinBook():
# Vaccination Center id and Session id for Slot Booking
self.vaccine = vaccine.upper()
self.vacc_fee_type = fee_type.upper()
self.vacc_center = None
self.vacc_session = None
self.slot_time = None
@ -96,7 +104,7 @@ class CoWinBook():
bottomBanner = "for Today and Day After 📆 ..." if self.bookToday == 0 else "for Tomorrow and Day After 📆 ..."
print(f" 📍 {self.pin} 💉 {age}+ ⌛️ {TIME} Seconds")
print(f" 📲 xxxx{self.mobile_no[7:]} 💉 {self.vaccine} (Dose :{self.dose})")
print(f" 📲 XXXX{self.mobile_no[7:]} 💉 {self.vaccine} (Dose :{self.dose})")
print(f"CoWin Auto Slot Booking 🔃\n{bottomBanner}")
line_break()
@ -138,12 +146,16 @@ class CoWinBook():
def checkToken(self):
# Pause job of Checking Slot
scheduler.pause_job('slot')
# scheduler.pause_job('slot')
self.getSession()
response = self.fetch_beneficiaries()
try:
response.json()
except json.decoder.JSONDecodeError:
self.login_cowin()
# Resume job of Checking Slot
scheduler.resume_job('slot')
# scheduler.resume_job('slot')
# Login to selfregistration.cowin.gov.in/
def login_cowin(self):
@ -155,11 +167,14 @@ class CoWinBook():
response = self.session.post('https://cdn-api.co-vin.in/api/v2/auth/generateMobileOTP',data=self.get_data())
if self.otp == 's': requests.delete(f"{OTP_SITE_URL}/{self.mobile_no}")
if self.otp == 's' and OTP_SITE_URL is not None: requests.delete(f"{OTP_SITE_URL}/{self.mobile_no}")
otpSha265 = self.get_otp()
try:
txn_id = response.json()['txnId']
except json.decoder.JSONDecodeError:
print("Wrong OTP Entered")
return
self.data = {
"otp":otpSha265,
@ -200,9 +215,9 @@ class CoWinBook():
last_msg_body = last_msg.get("body",'')
print(f'Waiting for OTP {i} sec')
sys.stdout.write("\033[F")
self.set_cursor()
d1 = datetime.strptime(last_msg.get("received"), '%Y-%m-%d %H:%M:%S')
d1 = datetime.strptime(last_msg.get("received","2019-12-01 09:09:09"), '%Y-%m-%d %H:%M:%S')
d2 = datetime.now() # current date and time
diff = (d2 - d1).total_seconds()
if (curr_msg_body != last_msg_body and "cowin" in last_msg_body.lower()) or diff <= OTP_VALID_DURATION_SECONDS:
@ -233,14 +248,15 @@ class CoWinBook():
msg = json.loads(msg)[0]
return msg
except KeyError:
raise Exception("Install Termux:API 0.31 Version for AutoMode ")
# Get OTP using DB hosted on Cloudflare and Attached with https://play.google.com/store/apps/details?id=com.gawk.smsforwarder
raise Exception("Install Termux:API from FDroid for Auto Mode ")
# Get OTP using DB hosted on Cloudflare and Configure with https://play.google.com/store/apps/details?id=com.gawk.smsforwarder
elif self.otp == 's':
if OTP_SITE_URL is None:
raise Exception("First Setup DB on Cloudflare \nhttps://github.com/truroshan/CloudflareCoWinDB ")
res = requests.get(f"{OTP_SITE_URL}/{self.mobile_no}",timeout=3).json()
res = requests.get(f"{OTP_SITE_URL}/{self.mobile_no}").json()
if res.get("status"):
msg['body'] = res.get('data').get("message")
@ -262,12 +278,7 @@ class CoWinBook():
self.requestStatus = response.status_code
if response.status_code == 200:
# CoWIN server may respond back with HTTP 204 so response.ok
# will not be right to depend upon. The content will be empty
# and so it will crash if we try to get the JSON body.
self.check_slot(response.json())
elif response.ok: # We have received ok response but without content
self.request_slot()
elif response.status_code == 401:
print("Re-login Account : " + datetime.now().strftime("%H:%M:%S") + " 🤳")
self.checkToken()
@ -276,7 +287,8 @@ class CoWinBook():
# Check Slot availability
def check_slot(self,response):
for center in response.get('centers',[]):
centers = response.get('centers',[])
for center in centers:
for session in center.get('sessions')[self.bookToday:]:
@ -287,12 +299,13 @@ class CoWinBook():
capacity = session.get(f'available_capacity_dose{self.dose}')
session_date = session.get('date')
vaccine_name = session.get('vaccine') + "ANY"
vaccine_name = session.get('vaccine')
if capacity > 1 and \
self.vaccine in vaccine_name and \
(self.vaccine in vaccine_name or self.vaccine == "ANY") and \
session.get('min_age_limit') == self.age and \
center.get('center_id') in self.center_id:
self.slot_time = session.get('slots')[0]
MSG = f'💉 {capacity} #{vaccine_name} / {session_date} / {center_name} 📍{self.pin}'
@ -302,13 +315,13 @@ class CoWinBook():
BOOKED = self.book_slot()
if BOOKED:
scheduler.shutdown(wait=False)
# scheduler.shutdown(wait=False)
print("Shutting Down CoWin Script 👩‍💻 ")
return
exit()
# When last Checked
print(f"Last Checked ✅ : " + datetime.now().strftime("%H:%M:%S") + " 🕐")
sys.stdout.write("\033[F")
print(f"Last Checked {self.requestStatus} ✅ : " + datetime.now().strftime("%H:%M:%S") + " 🕐")
self.set_cursor()
# Get Solved Captcha in String
def get_captcha(self):
@ -373,6 +386,8 @@ class CoWinBook():
return True
elif status == 409:
print("This vaccination center is completely booked for the selected date 😥")
elif status == 400:
print("Minimum age criteria is 45 years for this center")
elif status == 401:
self.login_cowin()
self.book_slot()
@ -430,31 +445,48 @@ class CoWinBook():
def select_center(self):
response = self.fetch_center()
# CoWIN server may respond back with HTTP 204 so response.ok
# will not be right to depend upon. The content will be empty
# and so it will crash if we try to get the JSON body.
while response.status_code != 200:
print(f'Trying to fetch center detail. Please wait...')
sys.stdout.write("\033[F")
time.sleep(1)
self.set_cursor()
time.sleep(5)
response = self.fetch_center()
clear_screen()
response = response.json()
CENTERS = {}
INDEX_S = []
if not response.get('centers'):
print("No Centers Available at this location")
print("Shutting Down CoWin Script 👩‍💻 ")
exit()
print(f"Select Vaccination Center ({self.pin}) 💉 \n")
counter = 1
for center in response.get('centers',[]):
vaccine_fee_type = center.get("fee_type").upper()
vaccine__fee_emoji = '🆓' if "FREE" in vaccine_fee_type else '💰'
for session in center.get('sessions'):
if session.get('min_age_limit') == self.age:
print(f'{counter} : {center.get("name")}')
vaccine_name = session.get("vaccine")
if session.get('min_age_limit') == self.age \
and (self.vaccine in vaccine_name or self.vaccine == 'ANY') \
and (self.vacc_fee_type in vaccine_fee_type or self.vacc_fee_type == "BOTH" ):
print(f'{counter} : {center.get("name")} {vaccine__fee_emoji}')
CENTERS[counter] = center.get('center_id')
INDEX_S.append(counter)
counter += 1
break
if counter == 1:
# clear_screen()
print(f"No Center available Vaccine : ( 💉{self.vaccine}) Fee Type : ({self.vacc_fee_type}).")
self.shutting_down()
exit()
print()
line_break()
@ -464,7 +496,10 @@ class CoWinBook():
* Select Mutiple with Space
input : 1 2 3 4
* Select All Center
Hit Enter without Input\n""")
Hit Enter without Input
**pass --f paid or free
**pass --v vacc_name \n""")
line_break()
@ -490,7 +525,7 @@ class CoWinBook():
response = self.fetch_beneficiaries()
while response.status_code != 200:
print(f'Please wait...')
sys.stdout.write("\033[F")
self.set_cursor()
time.sleep(5)
response = self.fetch_beneficiaries()
@ -502,8 +537,8 @@ class CoWinBook():
print(f"Select User for Vaccination 👩‍👦‍👦 \n")
if not response.get('beneficiaries',[]):
print("No user added in beneficiaries")
return
print("No beneficiaries added in Account")
exit()
counter = 1
for user in response.get('beneficiaries'):
@ -513,6 +548,13 @@ class CoWinBook():
INDEX_S.append(counter)
counter += 1
if counter == 1:
# clear_screen()
print(f"No beneficiaries available for Dose {self.dose}.")
self.shutting_down()
exit()
print()
line_break()
print("""
@ -539,6 +581,12 @@ class CoWinBook():
self.user_id = USER_ID
def shutting_down(self):
print("Shutting Down CoWin Script 👩‍💻 ")
def set_cursor(self):
sys.stdout.write("\033[F")
if __name__ == '__main__':
@ -549,9 +597,15 @@ if __name__ == '__main__':
fire.Fire(cowin.init)
# Check for Slot
scheduler.add_job(cowin.book_now, 'interval',id = "slot",seconds = TIME, misfire_grace_time=2)
# scheduler.add_job(cowin.book_now, 'interval',id = "slot",seconds = TIME, misfire_grace_time=2,replace_existing=True)
schedule.every(TIME).seconds.do(cowin.book_now)
# Check Token every 3 min
scheduler.add_job(cowin.checkToken, 'cron',id ="login", minute = f'*/3',misfire_grace_time= 30)
# scheduler.add_job(cowin.checkToken, 'cron',id ="login", minute = f'*/3',misfire_grace_time= 30)
schedule.every(3).minutes.do(cowin.checkToken)
scheduler.start()
# scheduler.start()
while True:
schedule.run_pending()
time.sleep(1)