refactor: use the library's detector instead of hand-rolled selectors

detect_cloudflare_challenge(page, 'interstitial') matches the challenge on
the page and stops matching once it clears, measured on ext.to (True then
False), and on yggtorrent, nowsecure.nl and google.com, none of which carry
a /cdn-cgi/challenge-platform/ script when cleared. So the custom marker set
was unnecessary.

The turnstile variant is not usable for this: nowsecure.nl embeds turnstile
scripts on its normal page, so it reports a challenge even when cleared.

Also adds PlaywrightTimeoutError to the retryable set. It is a different
class from the builtin TimeoutError -- playwright's derives from its own
Error -- so a Playwright timeout inside the solver escaped the loop and
returned 408 without retrying.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01TDMac4vGGcBhoUB5V6bvFK
This commit is contained in:
ThePhaseless
2026-08-17 11:35:49 +02:00
co-authored by Claude Opus 5
parent fc64fe05d5
commit cb8fb58fa8
2 changed files with 12 additions and 18 deletions
+7 -16
View File
@@ -8,7 +8,6 @@ from typing import Annotated
from fastapi import APIRouter, Depends, HTTPException
from fastapi.responses import RedirectResponse
from playwright.async_api import Page
from playwright.async_api import TimeoutError as PlaywrightTimeoutError
from playwright_captcha import CaptchaType
from playwright_captcha.solvers.click.cloudflare.utils.detection import (
@@ -34,11 +33,6 @@ router = APIRouter()
BrowserDep = Annotated[BrowserDepClass, Depends(get_browser)]
CHALLENGE_MARKERS = (
'script[src*="/cdn-cgi/challenge-platform/"][src*="chl_page"], '
"#challenge-error-text, #challenge-running, #challenge-stage"
)
@router.get("/", include_in_schema=False)
def read_root():
@@ -153,7 +147,12 @@ async def _solve_challenge(dep: BrowserDep, timer: TimeoutTimer) -> None:
"""Attempt to solve a detected Cloudflare interstitial challenge."""
logger.info("Challenge detected, attempting to solve...")
while timer.remaining() > 0:
with suppress(TimeoutError, CaptchaDetectionError, CaptchaSolvingError):
with suppress(
TimeoutError,
PlaywrightTimeoutError,
CaptchaDetectionError,
CaptchaSolvingError,
):
await wait_for(
dep.solver.solve_captcha( # pyright: ignore[reportUnknownMemberType,reportUnknownArgumentType]
captcha_container=dep.page,
@@ -164,7 +163,7 @@ async def _solve_challenge(dep: BrowserDep, timer: TimeoutTimer) -> None:
timeout=min(15, timer.remaining()),
)
if not await _challenge_visible(dep.page):
if not await detect_cloudflare_challenge(dep.page, "interstitial"):
logger.debug("Challenge solved successfully.")
return
@@ -172,14 +171,6 @@ async def _solve_challenge(dep: BrowserDep, timer: TimeoutTimer) -> None:
raise TimeoutError(message)
async def _challenge_visible(page: Page) -> bool:
"""Report whether an unsolved challenge is still on the page."""
try:
return await page.locator(CHALLENGE_MARKERS).count() > 0
except Exception:
return False
async def _wait_for_networkidle(dep: BrowserDep, timer: TimeoutTimer) -> None:
"""Wait for network idle, tolerating post-DOM-load stalls."""
try:
+5 -2
View File
@@ -7,6 +7,9 @@ import httpx2
import pytest
from fastapi import HTTPException
from playwright.async_api import TimeoutError as PlaywrightTimeoutError
from playwright_captcha.solvers.click.cloudflare.utils.detection import (
CF_INTERSTITIAL_INDICATORS_SELECTORS,
)
from playwright_captcha.utils.exceptions import (
CaptchaDetectionError,
CaptchaSolvingError,
@@ -14,7 +17,7 @@ from playwright_captcha.utils.exceptions import (
from starlette.testclient import TestClient
from main import app
from src.endpoints import CHALLENGE_MARKERS, read_item
from src.endpoints import read_item
from src.models import LinkRequest
from src.utils import BrowserDepClass
@@ -165,7 +168,7 @@ def fake_dep(
def count_for(selector: str) -> int:
"""Answer the marker check from the script, else from `challenged`."""
if selector != CHALLENGE_MARKERS or not remaining:
if selector not in CF_INTERSTITIAL_INDICATORS_SELECTORS or not remaining:
return 1 if challenged else 0
return remaining.pop(0) if len(remaining) > 1 else remaining[0]