251 lines
9.9 KiB
Python
251 lines
9.9 KiB
Python
import logging
|
|
import random
|
|
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 utils.excel_reader import ExcelHelper
|
|
|
|
RDV_URL = "https://rendezvousparis.hermes.com/client/register"
|
|
#
|
|
# RDV_URL = "file:///Users/lpan/Downloads/test_appointment.html"
|
|
# RDV_URL = "https://api.ipify.org"
|
|
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"
|
|
|
|
def get_random_wait_time()-> float:
|
|
wait_time = random.randint(0,10)/10.0 * 2
|
|
return wait_time
|
|
|
|
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
|
|
# reset otp_value to None
|
|
self.otp_value = None
|
|
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:
|
|
first_page = None
|
|
while first_page is None:
|
|
first_page = self.start_brower(proxy, pwright, user_agent)
|
|
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._setPhoneCountryAndStore()
|
|
time.sleep(get_random_wait_time())
|
|
self.page.mouse.wheel(0, 50)
|
|
self._setPhoneNumber(contact.phone)
|
|
time.sleep(get_random_wait_time())
|
|
self._setName(contact.last_name, contact.first_name)
|
|
time.sleep(get_random_wait_time())
|
|
self._set_email(contact.mail)
|
|
time.sleep(get_random_wait_time())
|
|
self.setIdNumber(contact.passport)
|
|
self.page.mouse.wheel(0, random.randint(200,600))
|
|
self._checkCgu()
|
|
# wait for sms_code field
|
|
self.page.mouse.wheel(0, random.randint(200,600))
|
|
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(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(contact, PublishType.SUCCESS.value, self.page.url)
|
|
else:
|
|
self.logger.info("timeout")
|
|
self.reset_air_plan_mode()
|
|
# check and send successful event
|
|
|
|
def start_brower(self, proxy, pwright, user_agent) -> Union[str, None]:
|
|
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)
|
|
return self.page.content()
|
|
except Exception as error:
|
|
self.logger.exception(error)
|
|
self.browser.close()
|
|
return None
|
|
|
|
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):
|
|
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):
|
|
self.page.evaluate("""(phoneNumber)=>document.getElementById("phone_number").value =phoneNumber""",
|
|
phoneNumber)
|
|
|
|
def _setName(self, lastName, firstName):
|
|
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):
|
|
self.page.evaluate("""(email)=>{
|
|
document.getElementById("email").focus();
|
|
document.getElementById("email").value = email;}""", email)
|
|
|
|
def setIdNumber(self, id):
|
|
self.page.evaluate(""" (id) =>{
|
|
document.getElementById("passport_id").focus();
|
|
document.getElementById("passport_id").value = id}""", id)
|
|
|
|
def _checkCgu(self):
|
|
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.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_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-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, 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)
|
|
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)
|