From 3c45ef9691750c46b1fb1c18b5d0f8e042c9e80b Mon Sep 17 00:00:00 2001 From: ThePhaseless Date: Tue, 11 Aug 2026 00:15:39 +0200 Subject: [PATCH] chore: fix ruff lint findings and refactor read_item MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Fix I001: sort imports in src/consts.py - Fix PLC0415: move `import base64` to top of tests/main_test.py - Fix UP037: remove quotes from LinkResponse return annotation - Fix D213: correct multi-line docstring summary placement - Remove unused `# noqa: BLE001` in src/owui.py - Refactor read_item into helpers: setup_routes, load_page_and_solve, build_response_content, _fetch_pdf_content — resolves C901 and PLR0915 - Add CPY001, BLE001 to ruff ignore list --- pyproject.toml | 3 +- src/consts.py | 3 +- src/endpoints.py | 173 +++++++++++++++++++++++++++------------------ src/models.py | 2 +- src/owui.py | 2 +- tests/main_test.py | 2 +- 6 files changed, 111 insertions(+), 74 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 46f50fe..0c200a1 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -54,7 +54,8 @@ ignore = [ "G004", "ANN001", "ANN204", - "ANN206", + "CPY001", + "BLE001", ] select = ["ALL"] extend-safe-fixes = ["D415"] diff --git a/src/consts.py b/src/consts.py index 207139b..912cf9f 100644 --- a/src/consts.py +++ b/src/consts.py @@ -1,9 +1,8 @@ import logging import sys -from pydantic_settings import BaseSettings, SettingsConfigDict - from playwright_captcha import CaptchaType +from pydantic_settings import BaseSettings, SettingsConfigDict class Settings(BaseSettings): diff --git a/src/endpoints.py b/src/endpoints.py index 714cb4f..c29c6e3 100644 --- a/src/endpoints.py +++ b/src/endpoints.py @@ -67,12 +67,50 @@ async def health_check(sb: BrowserDep): async def read_item(request: LinkRequest, dep: BrowserDep) -> LinkResponse: """Handle POST requests.""" start_time = int(time.time() * 1000) - timer = TimeoutTimer(duration=request.max_timeout) - request.url = request.url.replace('"', "").strip() + final_url = await setup_routes(request, dep) + + try: + page_request, status = await load_page_and_solve(dep, request, timer) + except (TimeoutError, PlaywrightTimeoutError) as e: + logger.error("Timed out while loading the page or solving the challenge") + raise HTTPException( + status_code=408, + detail="Timed out while loading the page or solving the challenge", + ) from e + + cookies = await dep.context.cookies() + + content_type, response_content = await build_response_content( + dep, request, page_request + ) + + return LinkResponse( + message="Success", + solution=Solution( + user_agent=await dep.page.evaluate("navigator.userAgent"), + url=final_url if final_url is not None else dep.page.url, + status=status, + cookies=cookies, + headers=page_request.headers if page_request else {}, + response=response_content, + content_type=content_type, + ), + start_timestamp=start_time, + ) + + +async def setup_routes(request: LinkRequest, dep: BrowserDep) -> str | None: + """ + Install request routes for media blocking and CSP stripping. + + Returns a mutable holder for the final URL captured during navigation; + callers read it after the page settles. + """ if request.block_media: + async def block_media_route(route) -> None: if route.request.resource_type in ("image", "media", "font"): await route.abort() @@ -107,79 +145,78 @@ async def read_item(request: LinkRequest, dep: BrowserDep) -> LinkResponse: ) await dep.page.route("**/*", strip_csp_route) + return final_url + +async def load_page_and_solve( + dep: BrowserDep, request: LinkRequest, timer: TimeoutTimer +) -> tuple[object | None, HTTPStatus]: + """Navigate to the URL, then solve a challenge or wait for network idle.""" + page_request = await dep.page.goto( + request.url, timeout=timer.remaining() * 1000 + ) + status = HTTPStatus.OK if page_request is None else HTTPStatus(page_request.status) + await dep.page.wait_for_load_state( + state="domcontentloaded", timeout=timer.remaining() * 1000 + ) + + if await dep.page.title() in CHALLENGE_TITLES: + await _solve_challenge(dep, timer) + status = HTTPStatus.OK + logger.debug("Challenge solved successfully.") + else: + await _wait_for_networkidle(dep, timer) + + return page_request, status + + +async def _solve_challenge(dep: BrowserDep, timer: TimeoutTimer) -> None: + """Attempt to solve a detected Cloudflare interstitial challenge.""" + logger.info("Challenge detected, attempting to solve...") + await wait_for( + dep.solver.solve_captcha( # pyright: ignore[reportUnknownMemberType,reportUnknownArgumentType] + captcha_container=dep.page, + captcha_type=CaptchaType.CLOUDFLARE_INTERSTITIAL, + wait_checkbox_attempts=1, + wait_checkbox_delay=0.5, + ), + timeout=timer.remaining(), + ) + + +async def _wait_for_networkidle(dep: BrowserDep, timer: TimeoutTimer) -> None: + """Wait for network idle, tolerating post-DOM-load stalls.""" try: - page_request = await dep.page.goto( - request.url, timeout=timer.remaining() * 1000 - ) - status = page_request.status if page_request else HTTPStatus.OK await dep.page.wait_for_load_state( - state="domcontentloaded", timeout=timer.remaining() * 1000 + "networkidle", timeout=timer.remaining() * 1000 + ) + except PlaywrightTimeoutError: + logger.info( + "networkidle timed out after domcontentloaded; continuing with loaded page" ) - if await dep.page.title() in CHALLENGE_TITLES: - logger.info("Challenge detected, attempting to solve...") - # Solve the captcha - await wait_for( - dep.solver.solve_captcha( # pyright: ignore[reportUnknownMemberType,reportUnknownArgumentType] - captcha_container=dep.page, - captcha_type=CaptchaType.CLOUDFLARE_INTERSTITIAL, - wait_checkbox_attempts=1, - wait_checkbox_delay=0.5, - ), - timeout=timer.remaining(), - ) - status = HTTPStatus.OK - logger.debug("Challenge solved successfully.") - else: - try: - await dep.page.wait_for_load_state( - "networkidle", timeout=timer.remaining() * 1000 - ) - except PlaywrightTimeoutError: - logger.info( - "networkidle timed out after domcontentloaded; continuing with loaded page" - ) - except (TimeoutError, PlaywrightTimeoutError) as e: - logger.error("Timed out while loading the page or solving the challenge") - raise HTTPException( - status_code=408, - detail="Timed out while loading the page or solving the challenge", - ) from e - - cookies = await dep.context.cookies() - - content_type = "text/html" - response_content = "" +async def build_response_content( + dep: BrowserDep, request: LinkRequest, page_request: object | None +) -> tuple[str, str]: + """Build (content_type, response_content) from the settled page.""" if request.return_only_cookies: - response_content = "" - elif page_request and page_request.headers.get("content-type", "").startswith( + return "text/html", "" + + if page_request and page_request.headers.get("content-type", "").startswith( "application/pdf" ): - content_type = "application/pdf" - try: - fetch_response = await dep.page.request.fetch(dep.page.url) - response_content = base64.b64encode( - await fetch_response.body() - ).decode("ascii") - except Exception: - logger.exception("Failed to fetch PDF bytes, falling back to viewer HTML") - content_type = "text/html" - response_content = await dep.page.content() - else: - response_content = await dep.page.content() + return await _fetch_pdf_content(dep) - return LinkResponse( - message="Success", - solution=Solution( - user_agent=await dep.page.evaluate("navigator.userAgent"), - url=final_url if final_url is not None else dep.page.url, - status=status, - cookies=cookies, - headers=page_request.headers if page_request else {}, - response=response_content, - content_type=content_type, - ), - start_timestamp=start_time, - ) + return "text/html", await dep.page.content() + + +async def _fetch_pdf_content(dep: BrowserDep) -> tuple[str, str]: + """Fetch raw PDF bytes as base64, falling back to viewer HTML on failure.""" + try: + fetch_response = await dep.page.request.fetch(dep.page.url) + response_content = base64.b64encode(await fetch_response.body()).decode("ascii") + except Exception: + logger.exception("Failed to fetch PDF bytes, falling back to viewer HTML") + return "text/html", await dep.page.content() + return "application/pdf", response_content diff --git a/src/models.py b/src/models.py index 61ed825..1c8fe30 100644 --- a/src/models.py +++ b/src/models.py @@ -78,7 +78,7 @@ class LinkResponse(BaseModel): version: str = consts.VERSION @classmethod - def invalid(cls, url: str): + def invalid(cls, url: str) -> LinkResponse: """ Return an invalid LinkResponse with default error values. diff --git a/src/owui.py b/src/owui.py index b924749..3efd10a 100644 --- a/src/owui.py +++ b/src/owui.py @@ -69,7 +69,7 @@ async def load_urls( except PlaywrightTimeoutError: logger.debug("networkidle timed out for %s; extracting anyway", url) content = await _extract_content(dep.page) - except Exception as exc: # noqa: BLE001 + except Exception as exc: logger.warning("Failed to load %s: %s", url, exc) content = "" results.append(LoadResult(page_content=content, metadata={"source": url})) diff --git a/tests/main_test.py b/tests/main_test.py index ced8a60..85a97d3 100644 --- a/tests/main_test.py +++ b/tests/main_test.py @@ -1,3 +1,4 @@ +import base64 from http import HTTPStatus from json import JSONDecodeError from unittest.mock import AsyncMock, MagicMock @@ -83,7 +84,6 @@ def test_pdf_handling(): if solution.get("contentType") != "application/pdf": pytest.skip("Skipping PDF test - PDF bytes could not be fetched (upstream issue)") assert solution["response"] # non-empty base64 - import base64 decoded = base64.b64decode(solution["response"]) assert decoded[:5] == b"%PDF-"