panleicim/appointment_tool
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'master' into feature/recaptha

Browse Source
This commit is contained in:
PAN Lei
2022-05-21 15:24:27 +02:00
parent 30cd794709 bce39de000
commit aa641d33a0
12 changed files with 108 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files
workers/commandor_page.py
+42 -21
View File
@@ -1,5 +1,6 @@
import logging
import random
import re
import string
import threading
import time
@@ -18,6 +19,7 @@ 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://bot.sannysoft.com/"
REGEX_RDV_URL = "https:\/\/rendezvousparis\.hermes\.com\/client\/register\/[A-Z0-9]+"
otp_value = None
OTP_FIELD_ID = "#sms_code"
MESSAGE_FIELD_CLASS = ".message"
@@ -28,10 +30,11 @@ TOO_MANY_REQUEST_ERROR_MESSAGE = "Due to a large number of requests"
CAPTCHA_ERROR_MESSAGE = "Error verifying captcha, please try again"
TIME_OUT = 400000
OTP_TIMEOUT = 240
PAGE_TIMEOUT = 30000
def get_random_wait_time() -> float:
wait_time = random.randint(0, 10) / 10.0 * 5
wait_time = random.randint(0, 10) / 10.0 * 1
return wait_time
@@ -43,11 +46,12 @@ class Tls(threading.local):
class CommandorPage:
tls = Tls()
def __init__(self, contact: ContactPojo, store_type=0):
def __init__(self, contact: ContactPojo, store_type=0, proxy_type=0):
self.otp_value = None
self.logger = logging.getLogger("CommandorPage")
self.is_finished = False
self.contact = contact
self.proxy_type = proxy_type
# 0: random
# 1: faubourg
# 2: George
@@ -72,7 +76,7 @@ class CommandorPage:
self.termine()
def _run(self, e: threading.Event, proxy):
print("will start browser")
self.logger.info("will start browser")
self.on_success_listener = on_success
# reset otp_value to None
self.otp_value = None
@@ -80,21 +84,14 @@ class CommandorPage:
first_page = None
while first_page is None:
first_page = self.start_browser(proxy, self.tls.playwright, devices)
proxy_username = "panleicim-res-fr-" + params.get_random_id_number_for_proxy()
proxy_username = params.get_proxy_name_prefix(self.proxy_type) + params.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._setPhoneNumber(self.contact.phone)
self._set_email(self.contact.mail)
self.setIdNumber(self.contact.passport)
#
self._checkCgu()
self.click_recapcha_checkbox()
self.fill_fields()
# wait for sms_code field
# self.clickOnValidBtn()
self.thread_event = e
@@ -119,9 +116,18 @@ class CommandorPage:
self.logger.info("timeout")
self.termine()
def fill_fields(self):
self._set_name(self.contact.last_name, self.contact.first_name)
self._setPhoneCountryAndStore()
self._setPhoneNumber(self.contact.phone)
self._set_email(self.contact.mail)
self.setIdNumber(self.contact.passport)
#
self._checkCgu()
def start_browser(self, proxy, pwright, device) -> Union[str, None]:
try:
self.browser = pwright.webkit.launch(headless=False, timeout=90000, proxy=proxy)
self.browser = pwright.webkit.launch(headless=False, timeout=PAGE_TIMEOUT, proxy=proxy)
self.logger.info("模拟设备: " + device)
pixel_2 = pwright.devices[device]
context = self.browser.new_context(**pixel_2, locale='en-GB')
@@ -136,7 +142,8 @@ class CommandorPage:
}
""")
self.page.on("load", self._on_page_loaded)
self.page.goto(RDV_URL, timeout=90000)
self.page.on("response", self.handle_response)
self.page.goto(RDV_URL, timeout=PAGE_TIMEOUT)
return self.page.content()
except Exception as error:
params.oracle_log_sender.send_error(str(error))
@@ -145,6 +152,12 @@ class CommandorPage:
self.browser.close()
return None
def handle_response(self, response):
pattern = re.compile(REGEX_RDV_URL)
if pattern.match(response.url):
self.logger.info("result url found: " + response.url)
self.publish_message_to_queue(self.contact, PublishType.PENDING, response.url)
def start_page(self, proxy):
e = threading.Event()
self._run(e, proxy)
@@ -153,6 +166,8 @@ class CommandorPage:
self.logger.info("page loaded")
# self.logger.info("content is " + self.page.content())
self.logger.info("url is " + self.page.url)
if self.page.url == RDV_URL:
self.fill_fields()
message = self.page.content()
if CONFIRMED_MESSAGE in message:
# publish the successful message
@@ -185,14 +200,18 @@ class CommandorPage:
except Exception as error:
self.logger.error(error)
def _setName(self, lastName, firstName):
def _set_name(self, lastName, firstName):
time.sleep(get_random_wait_time())
try:
self.page.evaluate("""(name)=> {
document.getElementById("surname").focus();
document.getElementById("surname").value = name.lastName;
self.page.evaluate("""(name)=> {
let surname = document.getElementById("surname");
if(surname.value.length == 0){
surname.focus();
surname.value = name.lastName;
document.getElementById("name").focus();
document.getElementById("name").value = name.firstName}""", {'lastName': lastName, 'firstName': firstName})
document.getElementById("name").value = name.firstName
}}
""", {'lastName': lastName, 'firstName': firstName})
except Exception as error:
self.logger.error(error)
@@ -229,8 +248,10 @@ class CommandorPage:
time.sleep(get_random_wait_time())
try:
self.page.evaluate("""(email)=>{
document.getElementById("email").focus();
document.getElementById("email").value = email;}""", email)
let emailElement = document.getElementById("email")
if(emailElement.value.length == 0){
emailElement.focus();
document.getElementById("email").value = email;}}""", email)
except Exception as error:
self.logger.error(error)