149 lines
4.7 KiB
Python
149 lines
4.7 KiB
Python
import asyncio
|
|
import logging
|
|
import random
|
|
import threading
|
|
import time
|
|
from concurrent.futures import ProcessPoolExecutor
|
|
|
|
from playwright.sync_api import sync_playwright
|
|
|
|
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://www.google.fr"
|
|
|
|
|
|
# RDV_URL = "https://api.ipify.org"
|
|
|
|
global_page = None
|
|
otp_value = None
|
|
user_agent_list = ExcelHelper().read_user_agent_list()
|
|
|
|
|
|
class CommandorPage:
|
|
def __init__(self):
|
|
pass
|
|
|
|
def _run(self, e: threading.Event, proxy, contact):
|
|
with sync_playwright() as pwright:
|
|
self.browser = pwright.chromium.launch(headless=False, timeout=90000, proxy=proxy)
|
|
self.page = self.browser.new_page(
|
|
user_agent=random.choice(user_agent_list))
|
|
global global_page
|
|
global_page = self
|
|
self.page.on("load", self.on_page_loaded)
|
|
self.page.goto(RDV_URL, timeout=90000)
|
|
# content = self.page.content()
|
|
self.setPhoneCountry()
|
|
self.setPhoneNumber(contact.phone)
|
|
self.setName(contact.last_name, contact.first_name)
|
|
self.setEmail(contact.mail)
|
|
self.setIdNumber(contact.passport)
|
|
self.checkCgu()
|
|
self.page.on("domcontentloaded", self.on_document_loaded)
|
|
self.clickOnValidBtn()
|
|
event_is_set = e.wait()
|
|
logging.debug('event set: %s', event_is_set)
|
|
global otp_value
|
|
self.send_otp(otp_value)
|
|
self.clickOnValidBtn()
|
|
self.page.on("load", self.on_page_loaded)
|
|
time.sleep(4000)
|
|
# time.sleep(1000)
|
|
|
|
def start_page(self, proxy, contact: ContactPojo) -> threading.Event:
|
|
# specifying an explicit component name
|
|
e = threading.Event()
|
|
t = threading.Thread(target=self._run, args=(e, proxy, contact))
|
|
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_response(self):
|
|
print("on_response called")
|
|
|
|
def on_document_loaded(self):
|
|
print("on_document_loaded called")
|
|
|
|
def setPhoneCountry(self):
|
|
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(".alert")
|
|
if items:
|
|
print(items.inner_html())
|
|
|
|
def setEmail(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 send_otp(self, otp: str):
|
|
self.page.evaluate("""(otp)=> document.getElementById("sms_code").value = otp""", otp)
|
|
|
|
def reset_air_plan_mode(self):
|
|
pass
|
|
|
|
|
|
def main():
|
|
executor = ProcessPoolExecutor(2)
|
|
loop = asyncio.get_event_loop()
|
|
loop.run_in_executor(executor, wait_for_otp)
|
|
loop.run_in_executor(executor, launch_page)
|
|
|
|
|
|
def launch_page() -> threading.Event:
|
|
proxy = {
|
|
"server": "http://gw.ntnt.io:5959",
|
|
"username": "panleicim-cc-fr-sid-179112",
|
|
"password": "M3PZAXgW5V27"
|
|
}
|
|
contact = ContactPojo(phone_number="+33751177505", passport_number="9933251119", last_name="Chen", first_name="Fan",
|
|
mail="lei_chena09@outlook.com", ccid="", position=0)
|
|
page = CommandorPage()
|
|
return page.start_page(proxy, contact)
|
|
|
|
|
|
def wait_for_otp():
|
|
sec = input("Press Enter otp to continue...")
|
|
asyncio.sleep(1)
|
|
print("input otp is: " + sec)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
event = launch_page()
|
|
sec = input("Press Enter otp to continue...")
|
|
print("input otp is: " + sec)
|
|
otp_value = sec
|
|
event.set()
|