visa_rescheduler
visa_rescheduler copied to clipboard
Stuck on Log-in Page
It was working perfectly fine for months but suddenly stop working for the past 2 days. Program not able to continue in the log in page.
I modified the Line 103 to fix that element
href = driver.find_element(By.LINK_TEXT, 'Continue')
Thank you for your help @sirfanmkm but my line 103 is different. I am using the new file from pull request. Appreciate if you can assist me which one I need to change.
import time import json import random import requests import configparser from datetime import datetime
from selenium import webdriver from selenium.webdriver.chrome.service import Service from selenium.webdriver.support import expected_conditions as EC from selenium.webdriver.support.ui import WebDriverWait as Wait from selenium.webdriver.common.by import By from webdriver_manager.chrome import ChromeDriverManager
from sendgrid import SendGridAPIClient from sendgrid.helpers.mail import Mail
from embassy import *
config = configparser.ConfigParser() config.read('config.ini')
Personal Info:
Account and current appointment info from https://ais.usvisa-info.com
USERNAME = config['PERSONAL_INFO']['USERNAME'] PASSWORD = config['PERSONAL_INFO']['PASSWORD']
Find SCHEDULE_ID in re-schedule page link:
https://ais.usvisa-info.com/en-am/niv/schedule/{SCHEDULE_ID}/appointment
SCHEDULE_ID = config['PERSONAL_INFO']['SCHEDULE_ID']
Target Period:
PRIOD_START = config['PERSONAL_INFO']['PRIOD_START'] PRIOD_END = config['PERSONAL_INFO']['PRIOD_END']
Embassy Section:
YOUR_EMBASSY = config['PERSONAL_INFO']['YOUR_EMBASSY'] EMBASSY = Embassies[YOUR_EMBASSY][0] FACILITY_ID = Embassies[YOUR_EMBASSY][1] REGEX_CONTINUE = Embassies[YOUR_EMBASSY][2]
Notification:
Get email notifications via https://sendgrid.com/ (Optional)
SENDGRID_API_KEY = config['NOTIFICATION']['SENDGRID_API_KEY']
Get push notifications via https://pushover.net/ (Optional)
PUSHOVER_TOKEN = config['NOTIFICATION']['PUSHOVER_TOKEN'] PUSHOVER_USER = config['NOTIFICATION']['PUSHOVER_USER']
Get push notifications via PERSONAL WEBSITE http://yoursite.com (Optional)
PERSONAL_SITE_USER = config['NOTIFICATION']['PERSONAL_SITE_USER'] PERSONAL_SITE_PASS = config['NOTIFICATION']['PERSONAL_SITE_PASS'] PUSH_TARGET_EMAIL = config['NOTIFICATION']['PUSH_TARGET_EMAIL'] PERSONAL_PUSHER_URL = config['NOTIFICATION']['PERSONAL_PUSHER_URL']
Time Section:
minute = 60 hour = 60 * minute
Time between steps (interactions with forms)
STEP_TIME = 0.5
Time between retries/checks for available dates (seconds)
RETRY_TIME_L_BOUND = config['TIME'].getfloat('RETRY_TIME_L_BOUND') RETRY_TIME_U_BOUND = config['TIME'].getfloat('RETRY_TIME_U_BOUND')
Cooling down after WORK_LIMIT_TIME hours of work (Avoiding Ban)
WORK_LIMIT_TIME = config['TIME'].getfloat('WORK_LIMIT_TIME') WORK_COOLDOWN_TIME = config['TIME'].getfloat('WORK_COOLDOWN_TIME')
Temporary Banned (empty list): wait COOLDOWN_TIME hours
BAN_COOLDOWN_TIME = config['TIME'].getfloat('BAN_COOLDOWN_TIME')
CHROMEDRIVER
Details for the script to control Chrome
LOCAL_USE = config['CHROMEDRIVER'].getboolean('LOCAL_USE')
Optional: HUB_ADDRESS is mandatory only when LOCAL_USE = False
HUB_ADDRESS = config['CHROMEDRIVER']['HUB_ADDRESS']
FIRST_PAGE_LINK = f"https://ais.usvisa-info.com/{EMBASSY}/niv" DATE_URL = f"https://ais.usvisa-info.com/{EMBASSY}/niv/schedule/{SCHEDULE_ID}/appointment/days/{FACILITY_ID}.json?appointments[expedite]=false" TIME_URL = f"https://ais.usvisa-info.com/{EMBASSY}/niv/schedule/{SCHEDULE_ID}/appointment/times/{FACILITY_ID}.json?date=%s&appointments[expedite]=false" APPOINTMENT_URL = f"https://ais.usvisa-info.com/{EMBASSY}/niv/schedule/{SCHEDULE_ID}/appointment" SIGN_OUT_LINK = f"https://ais.usvisa-info.com/{EMBASSY}/niv/users/sign_out"
def send_notification(title, msg): print(f"Sending notification!") if SENDGRID_API_KEY: message = Mail(from_email=USERNAME, to_emails=USERNAME, subject=msg, html_content=msg) try: sg = SendGridAPIClient(SENDGRID_API_KEY) response = sg.send(message) print(response.status_code) print(response.body) print(response.headers) except Exception as e: print(e.message) if PUSHOVER_TOKEN: url = "https://api.pushover.net/1/messages.json" data = { "token": PUSHOVER_TOKEN, "user": PUSHOVER_USER, "message": msg } requests.post(url, data) if PERSONAL_SITE_USER: url = PERSONAL_PUSHER_URL data = { "title": "VISA - " + str(title), "user": PERSONAL_SITE_USER, "pass": PERSONAL_SITE_PASS, "email": PUSH_TARGET_EMAIL, "msg": msg, } requests.post(url, data)
def auto_action(label, find_by, el_type, action, value, sleep_time=0): print("\t"+ label +":", end="") # Find Element By match find_by.lower(): case 'id': item = driver.find_element(By.ID, el_type) case 'name': item = driver.find_element(By.NAME, el_type) case 'class': item = driver.find_element(By.CLASS_NAME, el_type) case 'xpath': item = driver.find_element(By.XPATH, el_type) case _: return 0 # Do Action: match action.lower(): case 'send': item.send_keys(value) case 'click': item.click() case _: return 0 print("\t\tCheck!") if sleep_time: time.sleep(sleep_time)
def start_process(): # Bypass reCAPTCHA driver.get(FIRST_PAGE_LINK) time.sleep(STEP_TIME)
auto_action("Arrow down bounce", "xpath", '//a[@class="down-arrow bounce"]', "click", "", STEP_TIME)
auto_action("Login start", "xpath",'//*[@id="header"]/nav/div[2]/div[1]/ul/li[3]/a', "click", "", STEP_TIME)
Wait(driver, 60).until(EC.presence_of_element_located((By.NAME, "commit")))
auto_action("Click bounce", "xpath", '//a[@class="down-arrow bounce"]', "click", "", STEP_TIME)
auto_action("Email", "id", "user_email", "send", USERNAME, random.randint(1, 3))
auto_action("Password", "id", "user_password", "send", PASSWORD, random.randint(1, 3))
auto_action("Privacy", "class", "icheckbox", "click", "", random.randint(1, 3))
auto_action("Commit", "name", "commit", "click", "", random.randint(1, 3))
Wait(driver, 60).until(EC.presence_of_element_located((By.XPATH, "//a[contains(text(), '" + REGEX_CONTINUE + "')]")))
print("\n\tlogin successful!")
def reschedule(date): time = get_time(date) driver.get(APPOINTMENT_URL) headers = { "User-Agent": driver.execute_script("return navigator.userAgent;"), "Referer": APPOINTMENT_URL, "Cookie": "_yatri_session=" + driver.get_cookie("_yatri_session")["value"] } data = { "utf8": driver.find_element(by=By.NAME, value='utf8').get_attribute('value'), "authenticity_token": driver.find_element(by=By.NAME, value='authenticity_token').get_attribute('value'), "confirmed_limit_message": driver.find_element(by=By.NAME, value='confirmed_limit_message').get_attribute('value'), "use_consulate_appointment_capacity": driver.find_element(by=By.NAME, value='use_consulate_appointment_capacity').get_attribute('value'), "appointments[consulate_appointment][facility_id]": FACILITY_ID, "appointments[consulate_appointment][date]": date, "appointments[consulate_appointment][time]": time, } r = requests.post(APPOINTMENT_URL, headers=headers, data=data) if(r.text.find('Successfully Scheduled') != -1): title = "SUCCESS" msg = f"Rescheduled Successfully! {date} {time}" else: title = "FAIL" msg = f"Reschedule Failed!!! {date} {time}" return [title, msg]
def get_date(): # Requesting to get the whole available dates driver.get(DATE_URL) if not is_logged_in(): start_process() return get_date() else: content = driver.find_element(By.TAG_NAME, 'pre').text return json.loads(content)
def get_time(date): time_url = TIME_URL % date driver.get(time_url) content = driver.find_element(By.TAG_NAME, 'pre').text data = json.loads(content) time = data.get("available_times")[-1] print(f"Got time successfully! {date} {time}") return time
def is_logged_in(): content = driver.page_source if(content.find("error") != -1): return False return True
def get_available_date(dates): # Evaluation of different available dates def is_in_period(date, PSD, PED): new_date = datetime.strptime(date, "%Y-%m-%d") result = ( PED > new_date and new_date > PSD ) # print(f'{new_date.date()} : {result}', end=", ") return result
PED = datetime.strptime(PRIOD_END, "%Y-%m-%d")
PSD = datetime.strptime(PRIOD_START, "%Y-%m-%d")
for d in dates:
date = d.get('date')
if is_in_period(date, PSD, PED):
return date
print(f"\n\nNo available dates between ({PSD.date()}) and ({PED.date()})!")
def info_logger(file_path, log): # file_path: e.g. "log.txt" with open(file_path, "a") as file: file.write(str(datetime.now().time()) + ":\n" + log + "\n")
if LOCAL_USE: driver = webdriver.Chrome(service=Service(ChromeDriverManager().install())) else: driver = webdriver.Remote(command_executor=HUB_ADDRESS, options=webdriver.ChromeOptions())
if name == "main": first_loop = True while 1: LOG_FILE_NAME = "log_" + str(datetime.now().date()) + ".txt" if first_loop: t0 = time.time() total_time = 0 Req_count = 0 start_process() first_loop = False Req_count += 1 try: msg = "-" * 60 + f"\nRequest count: {Req_count}, Log time: {datetime.today()}\n" print(msg) info_logger(LOG_FILE_NAME, msg) dates = get_date() if not dates: # Ban Situation msg = f"List is empty, Probabely banned!\n\tSleep for {BAN_COOLDOWN_TIME} hours!\n" print(msg) info_logger(LOG_FILE_NAME, msg) send_notification("BAN", msg) driver.get(SIGN_OUT_LINK) time.sleep(BAN_COOLDOWN_TIME * hour) first_loop = True else: # Print Available dates: msg = "" for d in dates: msg = msg + "%s" % (d.get('date')) + ", " msg = "Available dates:\n"+ msg print(msg) info_logger(LOG_FILE_NAME, msg) date = get_available_date(dates) if date: # A good date to schedule for END_MSG_TITLE, msg = reschedule(date) break RETRY_WAIT_TIME = random.randint(RETRY_TIME_L_BOUND, RETRY_TIME_U_BOUND) t1 = time.time() total_time = t1 - t0 msg = "\nWorking Time: ~ {:.2f} minutes".format(total_time/minute) print(msg) info_logger(LOG_FILE_NAME, msg) if total_time > WORK_LIMIT_TIME * hour: # Let program rest a little send_notification("REST", f"Break-time after {WORK_LIMIT_TIME} hours | Repeated {Req_count} times") driver.get(SIGN_OUT_LINK) time.sleep(WORK_COOLDOWN_TIME * hour) first_loop = True else: msg = "Retry Wait Time: "+ str(RETRY_WAIT_TIME)+ " seconds" print(msg) info_logger(LOG_FILE_NAME, msg) time.sleep(RETRY_WAIT_TIME) except: # Exception Occured msg = f"Break the loop after exception!\n" END_MSG_TITLE = "EXCEPTION" break
print(msg) info_logger(LOG_FILE_NAME, msg) send_notification(END_MSG_TITLE, msg) driver.get(SIGN_OUT_LINK) driver.stop_client() driver.quit()
Ok. This is how my visa.py file looks now with changes. I added new case for link_text
and updated auto_action
. Hope this helps.
def auto_action(label, find_by, el_type, action, value, sleep_time=0):
print("\t"+ label +":", end="")
# Find Element By
match find_by.lower():
case 'id':
item = driver.find_element(By.ID, el_type)
case 'name':
item = driver.find_element(By.NAME, el_type)
case 'class':
item = driver.find_element(By.CLASS_NAME, el_type)
case 'xpath':
item = driver.find_element(By.XPATH, el_type)
case 'link_text':
item = driver.find_element(By.LINK_TEXT, el_type)
case _:
return 0
# Do Action:
match action.lower():
case 'send':
item.send_keys(value)
case 'click':
item.click()
case _:
return 0
print("\t\tCheck!")
if sleep_time:
time.sleep(sleep_time)
def start_process():
# Bypass reCAPTCHA
driver.get(FIRST_PAGE_LINK)
time.sleep(STEP_TIME)
auto_action("Arrow down bounce", "xpath", '//a[@class="down-arrow bounce"]', "click", "", STEP_TIME)
# auto_action("Login start", "xpath", '//*[@id="header"]/nav/div[2]/div[1]/ul/li[3]/a', "click", "", STEP_TIME)
auto_action("Login start", "link_text", 'Continue', "click", "", STEP_TIME)
Thank so much. It is working now. Cheers bro!