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

dynamically choose browser

Browse Source
This commit is contained in:
PAN Lei
2022-09-03 09:51:05 +02:00
parent 8e408473fb
commit b96ebf44ac
1 changed files with 28 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/workers/commandor_page.py
+28 -18
View File
@@ -21,11 +21,11 @@ from src.workers.GeoCaptchSolver import GeoCaptchaSolver
from src.workers.SolveCaptch import SolveCaptcha from src.workers.SolveCaptch import SolveCaptcha
from src.workers.TlsPlaywright import TlsPlaywright from src.workers.TlsPlaywright import TlsPlaywright
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 = "file:///Users/lpan/Downloads/test_appointment.html"
# RDV_URL = "https://api.ipify.org" # RDV_URL = "https://api.ipify.org"
# RDV_URL ="https://bot.sannysoft.com/" RDV_URL = "https://bot.sannysoft.com/"
REGEX_RDV_URL = "https:\/\/rendezvousparis\.hermes\.com\/client\/register\/[A-Z0-9]+" REGEX_RDV_URL = "https:\/\/rendezvousparis\.hermes\.com\/client\/register\/[A-Z0-9]+"
BLOCKED_MESSAGE_FR = "Pourquoi ce blocage" BLOCKED_MESSAGE_FR = "Pourquoi ce blocage"
BLOCKED_MESSAGE_EN = "You have been blocked" BLOCKED_MESSAGE_EN = "You have been blocked"
@@ -90,7 +90,7 @@ class CommandorPage:
self.logger.info("will close timeout modem") self.logger.info("will close timeout modem")
self.termine() self.termine()
def _run(self, proxy): def _run(self, proxy, use_proxy=True):
self.logger.info("will start browser") self.logger.info("will start browser")
self.on_success_listener = on_success self.on_success_listener = on_success
# reset otp_value to None # reset otp_value to None
@@ -101,10 +101,12 @@ class CommandorPage:
# if "iPhone" in key or "iPad" in key: # if "iPhone" in key or "iPad" in key:
# ios_keys.append(key) # ios_keys.append(key)
# print(ios_keys) # print(ios_keys)
devices = random.choice(params.DEVICES) # devices = random.choice(params.DEVICES)
device_key = random.sample(list(self.tls.playwright.devices), 1)[0]
device = self.tls.playwright.devices[device_key]
first_page = None first_page = None
while first_page is None: while first_page is None:
first_page = self.start_browser(proxy, self.tls.playwright, devices) first_page = self.start_browser(proxy, self.tls.playwright, device, use_proxy=use_proxy)
proxy = params.get_proxy(self.proxy_type) proxy = params.get_proxy(self.proxy_type)
# self.thread_event = e # self.thread_event = e
otp_input = self.page.locator(OTP_FIELD_ID) otp_input = self.page.locator(OTP_FIELD_ID)
@@ -126,16 +128,25 @@ class CommandorPage:
self.resolve_captcha() self.resolve_captcha()
self.is_filling_fields = False self.is_filling_fields = False
def start_browser(self, proxy, pwright, device) -> Union[str, None]: def start_browser(self, proxy, pwright, device, use_proxy=True) -> Union[str, None]:
try: try:
# self.browser = pwright.webkit.launch(headless=self.headless, timeout=PAGE_TIMEOUT) default_browser_type = device['default_browser_type']
self.browser = pwright.webkit.launch(headless=self.headless, timeout=PAGE_TIMEOUT, proxy=proxy) if default_browser_type == "webkit":
# userAgent = random.choice(params.firefox_user_agent_list) if use_proxy:
simulated_mobile = pwright.devices[device] self.browser = pwright.webkit.launch(headless=self.headless, timeout=PAGE_TIMEOUT, proxy=proxy)
userAgent = simulated_mobile['user_agent'] else:
print("user_agent is " + userAgent) self.browser = pwright.webkit.launch(headless=self.headless, timeout=PAGE_TIMEOUT)
# context = self.browser.new_context(**simulated_mobile, locale='fr-FR') userAgent = device['user_agent']
context = self.browser.new_context(**simulated_mobile) print("user_agent is " + userAgent)
# context = self.browser.new_context(**simulated_mobile, locale='fr-FR')
context = self.browser.new_context(**device)
else:
if use_proxy:
self.browser = pwright.firefox.launch(headless=self.headless, timeout=PAGE_TIMEOUT, proxy=proxy)
else:
self.browser = pwright.firefox.launch(headless=self.headless, timeout=PAGE_TIMEOUT)
userAgent = device['user_agent']
context = self.browser.new_context(user_agent=userAgent)
self.current_context = context self.current_context = context
self.create_and_config_page(context) self.create_and_config_page(context)
return self.page.content() return self.page.content()
@@ -168,9 +179,8 @@ class CommandorPage:
self.logger.info("result url found: " + response.url) self.logger.info("result url found: " + response.url)
# self.publish_message_to_queue(self.contact, PublishType.PENDING, response.url) # self.publish_message_to_queue(self.contact, PublishType.PENDING, response.url)
def start_page(self, proxy): def start_page(self, proxy, use_proxy=True):
e = threading.Event() self._run(proxy, use_proxy)
self._run(proxy)
def solve_datadome_captcha(self): def solve_datadome_captcha(self):
print("solve_datadome_captcha") print("solve_datadome_captcha")
@@ -426,7 +436,7 @@ def launch_page():
first_name="xingzhen", first_name="xingzhen",
mail="ColbyPatel653@gmail.com", ccid="", position=0) mail="ColbyPatel653@gmail.com", ccid="", position=0)
page = CommandorPage(contact, store_type=1) page = CommandorPage(contact, store_type=1)
return page.start_page(params.get_proxy(ProxyType.RESIDENTIAL)) return page.start_page(params.get_proxy(ProxyType.RESIDENTIAL), use_proxy=False)
def wait_for_otp(event: threading.Event, commandor: CommandorPage): def wait_for_otp(event: threading.Event, commandor: CommandorPage):