fix: use of CDP Mode

This commit is contained in:
Alvaro Santos Andres
2025-07-25 16:31:28 +00:00
parent 183f4f7f0f
commit 707a9c7a83
3 changed files with 29 additions and 8 deletions
+2
View File
@@ -33,6 +33,8 @@ USE_HEADLESS = (
os.getenv("USE_HEADLESS") in ["true", "1"] if os.getenv("USE_HEADLESS") else False
)
CAPTCHA_RETRIES = int(os.getenv("CAPTCHA_RETRIES", "5"))
PROXY = os.getenv("PROXY")
CHALLENGE_TITLES = [
+21 -7
View File
@@ -6,7 +6,7 @@ from fastapi import APIRouter, Depends, HTTPException
from fastapi.responses import RedirectResponse
from sbase import BaseCase
from src.consts import CHALLENGE_TITLES
from src.consts import CHALLENGE_TITLES, CAPTCHA_RETRIES
from src.models import (
HealthcheckResponse,
LinkRequest,
@@ -51,7 +51,8 @@ def read_item(request: LinkRequest, sb: SeleniumDep) -> LinkResponse:
try:
request.url = request.url.replace('"', "").strip()
sb.uc_open_with_reconnect(request.url)
sb.activate_cdp_mode(request.url)
sb.sleep(1)
logger.debug(f"Got webpage: {request.url}")
@@ -60,13 +61,26 @@ def read_item(request: LinkRequest, sb: SeleniumDep) -> LinkResponse:
if title_tag and title_tag.string in CHALLENGE_TITLES:
logger.debug("Challenge detected")
sb.uc_gui_click_captcha()
logger.info("Clicked captcha")
for attempt in range(CAPTCHA_RETRIES):
try:
sb.uc_gui_click_captcha()
sb.sleep(2)
if sb.get_title() in CHALLENGE_TITLES:
save_screenshot(sb)
logger.info(f"Clicked captcha (attempt {attempt + 1})")
raise HTTPException(status_code=500, detail="Could not bypass challenge")
if sb.get_title() not in CHALLENGE_TITLES:
break
except Exception as e:
logger.warning(f"Captcha click attempt {attempt + 1} failed: {e}")
time.sleep(5)
if sb.get_title() in CHALLENGE_TITLES:
save_screenshot(sb)
raise HTTPException(
status_code=500, detail="Could not bypass challenge"
)
cookies = sb.get_cookies()
for cookie in cookies:
+6 -1
View File
@@ -1,6 +1,7 @@
import logging
from time import gmtime, strftime
from anyio.streams import file
from fastapi import Header, HTTPException
from httpx import codes
from sbase import SB, BaseCase
@@ -29,6 +30,7 @@ def get_sb(
kwargs = {
"uc": True,
"test": True,
"headless": USE_HEADLESS,
"xvfb": USE_XVFB,
"locale_code": "en",
@@ -46,4 +48,7 @@ def get_sb(
def save_screenshot(sb: BaseCase):
"""Save screenshot on HTTPException."""
sb.save_screenshot(f"screenshots_{strftime('%Y-%m-%d %H:%M:%S', gmtime())}.png")
file_name = f"screenshots_{strftime('%Y-%m-%d_%H:%M:%S', gmtime())}.png"
logger.info(f"Saving screenshot to {file_name}")
sb.save_screenshot(file_name)