import logging import random import string import threading import time from playwright.sync_api import sync_playwright from params import PROXY_SERVER, PROXY_PASSWORD from pojo.ReserveResultPojo import ReserveResultPojo, PublishType from pojo.contact_pojo import ContactPojo from utils.excel_reader import ExcelHelper RDV_URL = "https://rendezvousparis.hermes.com/client/register" # RDV_URL = "file:///Users/lpan/Downloads/test_appointment.html" otp_value = None user_agent_list = ExcelHelper().read_user_agent_list() OTP_FIELD_ID = "#sms_code" MESSAGE_FIELD_CLASS = ".message" CONFIRMED_MESSAGE = "Your request for a Leather Goods appointment has been registered" class CommandorPage: def __init__(self): self.otp_value = None self.logger = logging.getLogger("CommandorPage") def _run(self, e: threading.Event, proxy, contact: ContactPojo, on_ready_for_otp, on_success): self.contact = contact self.on_success_listener = on_success with sync_playwright() as pwright: firefox_user_agents = filter(lambda user_agent: "firefox" in user_agent.lower(), user_agent_list) firefox_user_agents_list = list(firefox_user_agents) user_agent = random.choice(firefox_user_agents_list) # if "chrome" in user_agent.lower(): # self.browser = pwright.chromium.launch(headless=False, timeout=90000, proxy=proxy) # elif "firefox" in user_agent.lower(): # self.browser = pwright.firefox.launch(headless=False, timeout=90000, proxy=proxy) # else: self.start_brower(proxy, pwright, user_agent) self._setPhoneCountryAndStore() self._setPhoneNumber(contact.phone) self._setName(contact.last_name, contact.first_name) self._set_email(contact.mail) self.setIdNumber(contact.passport) self._checkCgu() # wait for sms_code field self.clickOnValidBtn() # self.page.goto("file:///Users/lpan/Downloads/input_otp.html") otp_input = self.page.locator(OTP_FIELD_ID) otp_input.wait_for(state='visible',timeout=90000) on_ready_for_otp(e, self) event_is_set = e.wait() logging.info('event set: %s', event_is_set) if self.otp_value: self.fill_otp(self.otp_value) self.clickOnValidBtn() otp_sent = self.page.locator(MESSAGE_FIELD_CLASS) otp_sent.wait_for(state='visible') message = self.page.content() print("message is:" + message) time.sleep(2) if CONFIRMED_MESSAGE in message: # publish the successful message print("url is " + self.page.url) self.publish_message_to_queue(contact, PublishType.SUCCESS.value, self.page.url) else: print("timeout") self.reset_air_plan_mode() # check and send successful event def start_brower(self, proxy, pwright, user_agent): try: self.browser = pwright.firefox.launch(headless=False, timeout=90000, proxy=proxy) self.logger.info("user_agent is " + user_agent) self.page = self.browser.new_page( user_agent=user_agent) # hide webdriver information self.page.add_init_script("""() => Object.defineProperty(navigator,'webdriver',{get: () => undefined}""") self.page.on("load", self._on_page_loaded) self.page.goto(RDV_URL, timeout=90000) except Exception as error: self.logger.exception(error) def start_page(self, proxy, contact: ContactPojo, on_ready_for_otp, on_sucess) -> threading.Event: e = threading.Event() t = threading.Thread(target=self._run, args=(e, proxy, contact, on_ready_for_otp, on_sucess)) t.start() return e def _on_page_loaded(self): print("page loaded") print("content is " + self.page.content()) print("url is " + self.page.url) self.getErrors() def on_document_loaded(self): print("on_document_loaded called") def _setPhoneCountryAndStore(self): # document.getElementById("prefer").value = \"faubourg\"; self.page.evaluate("""()=>{ document.getElementById("phone_country").value = \"FR\" }""") def _setPhoneNumber(self, phoneNumber): self.page.evaluate("""(phoneNumber)=>document.getElementById("phone_number").value =phoneNumber""", phoneNumber) def _setName(self, lastName, firstName): self.page.evaluate("""(name)=> { document.getElementById("surname").value = name.lastName; document.getElementById("name").value = name.firstName}""", {'lastName': lastName, 'firstName': firstName}) def getErrors(self): items = self.page.query_selector("div.alert") if items: print(items.inner_html()) def _set_email(self, email): self.page.evaluate("""(email)=>document.getElementById("email").value = email""", email) def setIdNumber(self, id): self.page.evaluate(""" (id) => document.getElementById("passport_id").value = id""", id) def _checkCgu(self): self.page.evaluate("""document.getElementById("cgu").checked = true; document.getElementById("processing").checked = true""") def clickOnValidBtn(self): self.page.evaluate("""document.getElementsByClassName("btn")[0].click();""") def clear_app_data(self): pass def fill_otp(self, otp: str): self.page.fill(OTP_FIELD_ID, otp) def reset_air_plan_mode(self): print("will close browser") self.browser.close() def publish_message_to_queue(self, contact: ContactPojo, message: str, url: str): # create the message id = url.split("/")[-1] result = ReserveResultPojo(type=PublishType.SUCCESS, phone=contact.phone, message=message, url=url, firstName=contact.first_name, lastName=contact.last_name, email=contact.mail, passport=contact.passport, ccid=contact.ccid) result.id = id self.on_success_listener(result) def get_random_id_number_for_proxy() -> str: S = 8 # number of characters in the string. ran = ''.join(random.choices(string.digits, k=S)) id_number = str(ran) print("The randomly generated string is : " + str(ran)) # print the random data return id_number def get_random_id_number() -> str: S = 8 # number of characters in the string. ran = ''.join(random.choices(string.digits, k=S)) id_number = "57" + str(ran) print("The randomly generated string is : 94" + str(ran)) # print the random data return id_number def on_success(result: ReserveResultPojo): pass def launch_page(ready_for_otp) -> threading.Event: PROXY_USERNAME = "panleicim-cc-fr-sid-" + get_random_id_number_for_proxy() print("proxy_username is " + PROXY_USERNAME) proxy = { "server": PROXY_SERVER, "username": PROXY_USERNAME, "password": PROXY_PASSWORD } passport_number = get_random_id_number() print("passport_number is " + passport_number) contact = ContactPojo(phone_number="+33758912245", passport_number=passport_number, last_name="XU", first_name="xingzhen", mail="ColbyPatel653@gmail.com", ccid="", position=0) page = CommandorPage() return page.start_page(proxy, contact, ready_for_otp, on_sucess=on_success) def wait_for_otp(event: threading.Event, commandor: CommandorPage): sec = input("Press Enter otp to continue...\n") print("input otp is: " + sec) commandor.otp_value = sec event.set() if __name__ == '__main__': event = launch_page(wait_for_otp)