Get a token from the shared browser unless a fatal browser error occurs.
(
self,
project_id: str,
website_key: str,
action: str = "IMAGE_GENERATION",
token_proxy_url: Optional[str] = None,
token_id: Optional[int] = None,
)
| 2342 | self.note_idle() |
| 2343 | |
| 2344 | async def get_token( |
| 2345 | self, |
| 2346 | project_id: str, |
| 2347 | website_key: str, |
| 2348 | action: str = "IMAGE_GENERATION", |
| 2349 | token_proxy_url: Optional[str] = None, |
| 2350 | token_id: Optional[int] = None, |
| 2351 | ) -> tuple[Optional[str], Optional[str]]: |
| 2352 | """Get a token from the shared browser unless a fatal browser error occurs.""" |
| 2353 | async with self._semaphore: |
| 2354 | self._solve_inflight += 1 |
| 2355 | max_retries = 3 |
| 2356 | |
| 2357 | try: |
| 2358 | for attempt in range(max_retries): |
| 2359 | try: |
| 2360 | start_ts = time.time() |
| 2361 | _, _, context = await self._get_or_create_shared_browser( |
| 2362 | token_proxy_url=token_proxy_url, |
| 2363 | token_id=token_id, |
| 2364 | ) |
| 2365 | |
| 2366 | token = await self._execute_captcha(context, project_id, website_key, action) |
| 2367 | if token: |
| 2368 | self._solve_count += 1 |
| 2369 | self._consecutive_browser_failures = 0 |
| 2370 | debug_logger.log_info( |
| 2371 | f"[BrowserCaptcha] Token-{self.token_id} token acquired ({(time.time()-start_ts)*1000:.0f}ms, launches={self._shared_launch_count}, reuse={self._shared_reuse_count})" |
| 2372 | ) |
| 2373 | return token, None |
| 2374 | |
| 2375 | self._error_count += 1 |
| 2376 | self._consecutive_browser_failures += 1 |
| 2377 | debug_logger.log_warning( |
| 2378 | f"[BrowserCaptcha] Token-{self.token_id} token attempt {attempt + 1}/{max_retries} failed" |
| 2379 | ) |
| 2380 | if self._consecutive_browser_failures >= 2: |
| 2381 | await self.recycle_browser(reason=f"captcha_failed_{attempt + 1}", rotate_profile=False) |
| 2382 | except Exception as e: |
| 2383 | self._error_count += 1 |
| 2384 | self._consecutive_browser_failures += 1 |
| 2385 | error_message = f"{type(e).__name__}: {str(e)}" |
| 2386 | debug_logger.log_error( |
| 2387 | f"[BrowserCaptcha] Token-{self.token_id} browser error: {error_message[:200]}" |
| 2388 | ) |
| 2389 | error_lower = error_message.lower() |
| 2390 | if any(keyword in error_lower for keyword in [ |
| 2391 | "context or browser has been closed", |
| 2392 | "target closed", |
| 2393 | "browser has been closed", |
| 2394 | "connection closed", |
| 2395 | "crash", |
| 2396 | "closed", |
| 2397 | ]): |
| 2398 | await self.recycle_browser(reason="browser_runtime_error", rotate_profile=False) |
| 2399 | |
| 2400 | if attempt < max_retries - 1: |
| 2401 | await asyncio.sleep(1) |
no test coverage detected