diff --git a/src/workers/commandor_page.py b/src/workers/commandor_page.py index a90b96b..040fc36 100644 --- a/src/workers/commandor_page.py +++ b/src/workers/commandor_page.py @@ -21,11 +21,11 @@ from src.workers.GeoCaptchSolver import GeoCaptchaSolver from src.workers.SolveCaptch import SolveCaptcha 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 = "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]+" BLOCKED_MESSAGE_FR = "Pourquoi ce blocage" BLOCKED_MESSAGE_EN = "You have been blocked" @@ -90,7 +90,7 @@ class CommandorPage: self.logger.info("will close timeout modem") self.termine() - def _run(self, proxy): + def _run(self, proxy, use_proxy=True): self.logger.info("will start browser") self.on_success_listener = on_success # reset otp_value to None @@ -101,10 +101,12 @@ class CommandorPage: # if "iPhone" in key or "iPad" in key: # ios_keys.append(key) # 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 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) # self.thread_event = e otp_input = self.page.locator(OTP_FIELD_ID) @@ -126,16 +128,25 @@ class CommandorPage: self.resolve_captcha() 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: - # self.browser = pwright.webkit.launch(headless=self.headless, timeout=PAGE_TIMEOUT) - self.browser = pwright.webkit.launch(headless=self.headless, timeout=PAGE_TIMEOUT, proxy=proxy) - # userAgent = random.choice(params.firefox_user_agent_list) - simulated_mobile = pwright.devices[device] - userAgent = simulated_mobile['user_agent'] - print("user_agent is " + userAgent) - # context = self.browser.new_context(**simulated_mobile, locale='fr-FR') - context = self.browser.new_context(**simulated_mobile) + default_browser_type = device['default_browser_type'] + if default_browser_type == "webkit": + if use_proxy: + self.browser = pwright.webkit.launch(headless=self.headless, timeout=PAGE_TIMEOUT, proxy=proxy) + else: + self.browser = pwright.webkit.launch(headless=self.headless, timeout=PAGE_TIMEOUT) + userAgent = device['user_agent'] + 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.create_and_config_page(context) return self.page.content() @@ -168,9 +179,8 @@ class CommandorPage: 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(proxy) + def start_page(self, proxy, use_proxy=True): + self._run(proxy, use_proxy) def solve_datadome_captcha(self): print("solve_datadome_captcha") @@ -426,7 +436,7 @@ def launch_page(): first_name="xingzhen", mail="ColbyPatel653@gmail.com", ccid="", position=0) 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):