fix: simplify code to fix PR suggestions

This commit is contained in:
Alvaro Santos Andres
2025-08-11 06:57:42 +00:00
parent bb88c8e111
commit 155a9076f1
5 changed files with 50 additions and 67 deletions
+2 -2
View File
@@ -37,9 +37,9 @@ Built with [seleniumbase](https://seleniumbase.io/) and [FastAPI](https://fastap
|----------------------|------------------------|------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
| `HOST` | `0.0.0.0` | Host address to bind the server to. Use `0.0.0.0` to bind to all IPv4 interfaces, `::` for all IPv6 interfaces, or `127.0.0.1`/`localhost` for local access only. |
| `PROXY` | None | Proxy to use in format: `protocol://username:password@host:port`. [SOCKS5 with authentication is not supported by Chrome](https://stackoverflow.com/questions/75602916/connection-to-private-proxy-socks5-with-chrome-webrequest-onauthrequired-and), see `compose.yaml` file for a workaround |
| `USE_HEADLESS` | `False` | Use headless chromium. |
| `USE_XVFB` | `SeleniumBase default` | Use activate the special virtual display. |
| `USE_HEADLESS` | `SeleniumBase default` | Use headless chromium. |
| `USE_XVFB` | `SeleniumBase default` | Use activate the special virtual display. |
| `CAPTCHA_RETRIES` | 5 | Number of times to retry solving a CAPTCHA challenge before giving up. This helps in cases where CAPTCHA challenges fail intermittently and can be retried successfully. |
## Proxy Recommendation
+1 -1
View File
@@ -23,4 +23,4 @@ app.include_router(router=router)
if __name__ == "__main__":
host = os.getenv("HOST", "0.0.0.0")
uvicorn.run("main:app", host=host, port=8191, log_level=LOG_LEVEL, reload=True) # noqa: S104
uvicorn.run(app, host=host, port=8191, log_level=LOG_LEVEL) # noqa: S104
+1 -1
View File
@@ -30,7 +30,7 @@ VERSION = get_version_from_env() or "unknown"
USE_XVFB = os.getenv("USE_XVFB") in ["true", "1"] if os.getenv("USE_XVFB") else None
USE_HEADLESS = (
os.getenv("USE_HEADLESS") in ["true", "1"] if os.getenv("USE_HEADLESS") else False
os.getenv("USE_HEADLESS") in ["true", "1"] if os.getenv("USE_HEADLESS") else None
)
CAPTCHA_RETRIES = int(os.getenv("CAPTCHA_RETRIES", "5"))
+39 -50
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, CAPTCHA_RETRIES
from src.consts import CAPTCHA_RETRIES, CHALLENGE_TITLES
from src.models import (
HealthcheckResponse,
LinkRequest,
@@ -49,61 +49,50 @@ def read_item(request: LinkRequest, sb: SeleniumDep) -> LinkResponse:
"""Handle POST requests."""
start_time = int(time.time() * 1000)
try:
request.url = request.url.replace('"', "").strip()
sb.activate_cdp_mode(request.url)
sb.sleep(1)
request.url = request.url.replace('"', "").strip()
sb.activate_cdp_mode(request.url)
sb.sleep(1)
logger.debug(f"Got webpage: {request.url}")
source_bs = sb.get_beautiful_soup()
title_tag = source_bs.title
source_bs = sb.get_beautiful_soup()
title_tag = source_bs.title
if title_tag and title_tag.string in CHALLENGE_TITLES:
for attempt in range(CAPTCHA_RETRIES):
try:
sb.uc_gui_click_captcha()
sb.sleep(2)
if title_tag and title_tag.string in CHALLENGE_TITLES:
logger.debug("Challenge detected")
for attempt in range(CAPTCHA_RETRIES):
try:
sb.uc_gui_click_captcha()
sb.sleep(2)
if sb.get_title() not in CHALLENGE_TITLES:
break
except Exception as e: # noqa
logger.warning(f"Captcha click attempt {attempt + 1} failed: {e}")
logger.info(f"Clicked captcha (attempt {attempt + 1})")
time.sleep(5)
if sb.get_title() not in CHALLENGE_TITLES:
break
except Exception as e:
logger.warning(f"Captcha click attempt {attempt + 1} failed: {e}")
if sb.get_title() in CHALLENGE_TITLES:
save_screenshot(sb)
time.sleep(5)
raise HTTPException(status_code=500, detail="Could not bypass challenge")
if sb.get_title() in CHALLENGE_TITLES:
save_screenshot(sb)
cookies = sb.get_cookies()
for cookie in cookies:
name = cookie["name"]
value = cookie["value"]
cookie["size"] = len(f"{name}={value}".encode())
raise HTTPException(
status_code=500, detail="Could not bypass challenge"
)
cookie["session"] = False
if "expiry" in cookie:
cookie["expires"] = cookie["expiry"]
cookies = sb.get_cookies()
for cookie in cookies:
name = cookie["name"]
value = cookie["value"]
cookie["size"] = len(f"{name}={value}".encode())
cookie["session"] = False
if "expiry" in cookie:
cookie["expires"] = cookie["expiry"]
return LinkResponse(
message="Success",
solution=Solution(
user_agent=sb.get_user_agent(),
url=sb.get_current_url(),
status=200,
cookies=cookies,
headers={},
response=str(sb.get_beautiful_soup()),
),
start_timestamp=start_time,
)
except Exception as e:
logger.error(f"Error processing request: {e}", exc_info=True)
raise
return LinkResponse(
message="Success",
solution=Solution(
user_agent=sb.get_user_agent(),
url=sb.get_current_url(),
status=200,
cookies=cookies,
headers={},
response=str(sb.get_beautiful_soup()),
),
start_timestamp=start_time,
)
+7 -13
View File
@@ -28,20 +28,14 @@ def get_sb(
detail="SOCKS5 proxy with authentication is not supported. Check README for more info.",
)
kwargs = {
"uc": True,
"test": True,
"headless": USE_HEADLESS,
"xvfb": USE_XVFB,
"locale_code": "en",
"ad_block": True,
"proxy": proxy,
}
logger.info("Creating SeleniumBase instance with parameters: %s", kwargs)
with SB(
**kwargs,
uc=True,
test=True,
headless=USE_HEADLESS,
xvfb=USE_XVFB,
locale_code="en",
ad_block=True,
proxy=proxy,
) as sb:
yield sb