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

optimisation for otp

Browse Source
This commit is contained in:
PAN Lei
2022-03-21 22:39:47 +01:00
parent 1a60ba3a1d
commit ea0aed7caa
2 changed files with 55 additions and 66 deletions
Download Patch File Download Diff File Expand all files Collapse all files
commandor_page.py
+38 -56
View File
@@ -3,86 +3,78 @@ 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 = "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"
RDV_URL = "file:///Users/lpan/Downloads/test_appointment.html"
global_page = None
otp_value = None
user_agent_list = ExcelHelper().read_user_agent_list()
OTP_FIELD_ID = "#sms_code"
class CommandorPage:
def __init__(self):
pass
self.otp_value = None
def _run(self, e: threading.Event, proxy, contact):
def _run(self, e: threading.Event, proxy, contact, on_ready_for_otp):
self.contact = 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.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._setPhoneCountry()
self._setPhoneNumber(contact.phone)
self._setName(contact.last_name, contact.first_name)
self._set_email(contact.mail)
self.setIdNumber(contact.passport)
self.checkCgu()
self.page.on("domcontentloaded", self.on_document_loaded)
# wait for sms_code field
self.clickOnValidBtn()
self.page.goto("file:///Users/lpan/Downloads/input_otp.html")
order_sent = self.page.locator(OTP_FIELD_ID)
order_sent.wait_for(state='visible')
on_ready_for_otp(e, self)
event_is_set = e.wait()
logging.debug('event set: %s', event_is_set)
global otp_value
self.send_otp(otp_value)
self.fill_otp(self.otp_value)
self.clickOnValidBtn()
self.page.on("load", self.on_page_loaded)
time.sleep(4000)
# time.sleep(1000)
time.sleep(4)
# check and send successful event
def start_page(self, proxy, contact: ContactPojo) -> threading.Event:
# specifying an explicit component name
def start_page(self, proxy, contact: ContactPojo, on_ready_for_otp) -> threading.Event:
e = threading.Event()
t = threading.Thread(target=self._run, args=(e, proxy, contact))
t = threading.Thread(target=self._run, args=(e, proxy, contact, on_ready_for_otp))
t.start()
return e
def on_page_loaded(self):
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):
def _setPhoneCountry(self):
self.page.evaluate("""()=>document.getElementById("phone_country").value = \"FR\" """)
def setPhoneNumber(self, phoneNumber):
def _setPhoneNumber(self, phoneNumber):
self.page.evaluate("""(phoneNumber)=>document.getElementById("phone_number").value =phoneNumber""",
phoneNumber)
def setName(self, lastName, firstName):
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})
@@ -92,7 +84,7 @@ class CommandorPage:
if items:
print(items.inner_html())
def setEmail(self, email):
def _set_email(self, email):
self.page.evaluate("""(email)=>document.getElementById("email").value = email""", email)
def setIdNumber(self, id):
@@ -108,41 +100,31 @@ class CommandorPage:
def clear_app_data(self):
pass
def send_otp(self, otp: str):
self.page.evaluate("""(otp)=> document.getElementById("sms_code").value = otp""", otp)
def fill_otp(self, otp: str):
self.page.fill(OTP_FIELD_ID, 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:
def launch_page(ready_for_otp) -> threading.Event:
proxy = {
"server": "http://gw.ntnt.io:5959",
"username": "panleicim-cc-fr-sid-179112",
"username": "panleicim-cc-fr-sid-192012",
"password": "M3PZAXgW5V27"
}
contact = ContactPojo(phone_number="+33758321759", passport_number="9533251119", last_name="Cheng", first_name="Fan",
mail="checn.fan@outlook.com", ccid="", position=0)
contact = ContactPojo(phone_number="+33758735284", passport_number="9537551179", last_name="Hong", first_name="Fan",
mail="hongfan@outlook.com", ccid="", position=0)
page = CommandorPage()
return page.start_page(proxy, contact)
return page.start_page(proxy, contact, ready_for_otp)
def wait_for_otp():
sec = input("Press Enter otp to continue...")
asyncio.sleep(1)
def wait_for_otp(event: threading.Event, commandorPage: CommandorPage):
sec = input("Press Enter otp to continue...\n")
print("input otp is: " + sec)
commandorPage.otp_value = sec
event.set()
if __name__ == '__main__':
event = launch_page()
sec = input("Press Enter otp to continue...")
print("input otp is: " + sec)
otp_value = sec
event.set()
event = launch_page(wait_for_otp)