import datetime import logging import random import re import string import threading import time from typing import Union from playwright.sync_api import sync_playwright import params from params import PROXY_SERVER, PROXY_PASSWORD from pojo.ReserveResultPojo import ReserveResultPojo, PublishType from pojo.contact_pojo import ContactPojo from pojo.serial_modem import SerialModem from utils.excel_reader import ExcelHelper from utils.operator import Operator, check_operator RDV_URL = "https://rendezvousparis.hermes.com/client/register" # # RDV_URL = "file:///Users/lpan/Downloads/test_appointment.html" # RDV_URL = "https://api.ipify.org" # RDV_URL = "https://duoyi.org/" # RDV_URL ="https://bot.sannysoft.com/" 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" TIME_OUT = 400000 OTP_TIMEOUT = 240 def get_random_wait_time() -> float: wait_time = random.randint(0, 10) / 10.0 * 5 return wait_time class Tls(threading.local): def __init__(self) -> None: self.playwright = sync_playwright().start() print("Create playwright instance in Thread", threading.current_thread().name) class CommandorPage: tls = Tls() def __init__(self, serial_modem: SerialModem, slot_position, sim_position): self.otp_value = None self.logger = logging.getLogger("CommandorPage") self.is_finished = False self.current_gsm_modem = serial_modem self.slot_position = slot_position self.sim_position = sim_position self.contact = serial_modem.contact def on_success(self, result: ReserveResultPojo): self.logger.info("on_success called.") result.sim_position = self.sim_position result.slot_position = self.slot_position self.logger.info(result) params.firebase_store_manager.save(result) params.oracle_log_sender.send_appoint_result(result) self.current_gsm_modem.modem.close() self.is_finished = True def handle_sms(self, sms): self.logger.info( u'== SMS message received ==\nFrom: {0}\nTime: {1}\nMessage:\n{2}\n'.format(sms.number, sms.time, sms.text)) # extract the otp number date = str(sms.time)[0:10] params.oracle_log_sender.send_sms_reception_log(sms.number, sms.text,self.current_gsm_modem.ccid) if date == str(datetime.date.today()): self.logger.info("this sms is for today") if "rendez-vous" in sms.text or "appointment" in sms.text: self.logger.info("try to extract the otp") pattern = r'\d{6,8}' # if re.match(pattern, sms.text): match = re.search(pattern, sms.text) otp = match.group(0) self.logger.info("otp is " + otp) self.otp_value = otp self.logger.info("will set thread event") self.thread_event.set() # wait for the sms for 20 seconds # while not self.is_finished: # time.sleep(2) self.is_finished = True # if self.current_gsm_modem: # self.logger.info("will close used modem") # self.current_gsm_modem.modem.close() else: self.logger.info("The sms is not for RDV") else: self.logger.info("The sms is not for today") def set_up_sms_listener(self): if check_operator(self.current_gsm_modem.ccid) == Operator.LYCAMOBILE: # lycamobile self.current_gsm_modem.modem.deleteMultipleStoredSms(memory="SM") self.current_gsm_modem.modem.smsReceivedCallback = self.handle_sms self.is_finished = False self.current_gsm_modem.smsTextMode = False self.logger.info('Waiting for SMS message, for phone number ' + str(self.current_gsm_modem.phone_number)) listen_at = time.time() while not self.is_finished: time.sleep(2) # check whether timeout now = time.time() if (listen_at + OTP_TIMEOUT) < now: self.logger.info("time out for {}, switch to next contact".format(self.current_gsm_modem.phone_number)) # save the contact in timeout self.timeout_occurred() self.current_gsm_modem.modem.close() return return def timeout_occurred(self): params.firebase_store_manager.save_timeout_contact(self.current_gsm_modem.contact) params.oracle_log_sender.send_timeout_log(self.current_gsm_modem) self.logger.info("will close timeout modem") self.thread_event.set() self.current_gsm_modem.modem.close() self.reset_air_plan_mode() def _run(self, e: threading.Event, proxy): print("will start browser") self.on_success_listener = on_success # reset otp_value to None self.otp_value = None devices = random.choice(params.DEVICES) first_page = None while first_page is None: first_page = self.start_browser(proxy, self.tls.playwright, devices) proxy_username = "panleicim-res-fr-" + get_random_id_number_for_proxy() self.logger.info("proxy_username is " + proxy_username) proxy = { "server": params.PROXY_SERVER, "username": proxy_username, "password": params.PROXY_PASSWORD } self._setName(self.contact.last_name, self.contact.first_name) self._setPhoneCountryAndStore() # self.page.mouse.wheel(0, random.randint(100, 200)) self._setPhoneNumber(self.contact.phone) self._set_email(self.contact.mail) self.setIdNumber(self.contact.passport) # self._checkCgu() # wait for sms_code field # self.clickOnValidBtn() self.thread_event = e self.set_up_sms_listener() otp_input = self.page.locator(OTP_FIELD_ID) otp_input.wait_for(state='visible', timeout=TIME_OUT) event_is_set = e.wait() logging.info('event set: %s', event_is_set) if self.otp_value: self.fill_otp(self.otp_value) time.sleep(get_random_wait_time()) self.clickOnValidBtn() otp_sent = self.page.locator(MESSAGE_FIELD_CLASS) otp_sent.wait_for(state='visible', timeout=TIME_OUT) message = self.page.content() # print("message is:" + message) time.sleep(get_random_wait_time()) if CONFIRMED_MESSAGE in message: # publish the successful message self.logger.info("url is " + self.page.url) self.publish_message_to_queue(self.contact, PublishType.SUCCESS.value, self.page.url) else: self.logger.info("timeout") self.reset_air_plan_mode() def start_browser(self, proxy, pwright, device) -> Union[str, None]: try: self.browser = pwright.webkit.launch(headless=False, timeout=90000, proxy=proxy) self.logger.info("device is " + device) pixel_2 = pwright.devices[device] context = self.browser.new_context(**pixel_2, locale='en-GB') self.page = context.new_page() # hide webdriver information self.page.add_init_script("""() => { Object.defineProperty(navigator,'webdriver',{get: () => undefined}); Object.defineProperty(navigator, 'platform', { get: () => { return "iPhone"; }}); } """) self.page.on("load", self._on_page_loaded) self.page.goto(RDV_URL, timeout=90000) return self.page.content() except Exception as error: params.oracle_log_sender.send_error(str(error)) self.logger.exception(error) self.browser.close() return None def start_page(self, proxy): e = threading.Event() # t = threading.Thread(target=self._run, args=(e, proxy)) # t.start() self._run(e, proxy) # return e def _on_page_loaded(self): self.logger.info("page loaded") # self.logger.info("content is " + self.page.content()) self.logger.info("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").focus(); document.getElementById("phone_country").value = \"FR\" }""") def _setPhoneNumber(self, phoneNumber): time.sleep(get_random_wait_time()) self.page.evaluate("""(phoneNumber)=>document.getElementById("phone_number").value =phoneNumber""", phoneNumber) def _setName(self, lastName, firstName): time.sleep(get_random_wait_time()) self.page.evaluate("""(name)=> { document.getElementById("surname").focus(); document.getElementById("surname").value = name.lastName; document.getElementById("name").focus(); 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): time.sleep(get_random_wait_time()) self.page.evaluate("""(email)=>{ document.getElementById("email").focus(); document.getElementById("email").value = email;}""", email) def setIdNumber(self, id): time.sleep(get_random_wait_time()) self.page.evaluate(""" (id) =>{ document.getElementById("passport_id").focus(); document.getElementById("passport_id").value = id}""", id) def _checkCgu(self): # self.page.mouse.wheel(0, random.randint(200, 600)) self.page.evaluate(""" document.getElementById("cgu").focus(); document.getElementById("cgu").checked = true; document.getElementById("processing").focus(); document.getElementById("processing").checked = true""") def clickOnValidBtn(self): # self.page.mouse.wheel(0, random.randint(400, 500)) time.sleep(get_random_wait_time()) self.page.evaluate("""document.getElementsByClassName("btn")[0].focus();""") time.sleep(get_random_wait_time()) self.page.evaluate(""" document.getElementsByClassName("btn")[0].click();""") def clear_app_data(self): pass def fill_otp(self, otp: str): self.page.focus(OTP_FIELD_ID) time.sleep(get_random_wait_time()) self.page.fill(OTP_FIELD_ID, otp) def reset_air_plan_mode(self): self.logger.info("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(result) time.sleep(2) self.browser.close() 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-res-fr-" + 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) 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) # time = get_random_wait_time() # print(time) # import urllib.request # # proxy = urllib.request.ProxyHandler({'https': 'http://panleicim-res-fr-121:94sY7zwBG13i@gw.ntnt.io:5959'}) # opener = urllib.request.build_opener(proxy) # urllib.request.install_opener(opener) # content = urllib.request.urlopen('https://api.ipify.org').read() # print(content)