126 lines
4.1 KiB
Python
126 lines
4.1 KiB
Python
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
from app.errors import AppError
|
|
from app.sources.geizhals.taxonomy_parser import is_challenge_html
|
|
|
|
FIXTURE_DIR = Path(__file__).resolve().parents[3] / "fixtures" / "geizhals"
|
|
BROWSER_HEADERS = {
|
|
"User-Agent": (
|
|
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
|
|
"AppleWebKit/537.36 (KHTML, like Gecko) Chrome/128.0.0.0 Safari/537.36"
|
|
),
|
|
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
|
|
"Accept-Language": "de-DE,de;q=0.9,en;q=0.8",
|
|
}
|
|
EXPAND_LABELS = (
|
|
"Erweiterte Filter",
|
|
"Alle Filter ausklappen",
|
|
"Immer alle Filter anzeigen",
|
|
"weitere Filter anzeigen",
|
|
)
|
|
|
|
|
|
def load_result_page(url: str, *, playwright: bool = True) -> tuple[str, str, int, bool]:
|
|
status, html = 0, ""
|
|
try:
|
|
status, html = _fetch_http(url)
|
|
if not _blocked(status, html):
|
|
return html, "live_http", status, False
|
|
except Exception:
|
|
status, html = 0, ""
|
|
if playwright:
|
|
try:
|
|
status, html = _fetch_playwright(url)
|
|
if not _blocked(status, html):
|
|
return html, "live_playwright", status, False
|
|
except Exception:
|
|
pass
|
|
if html and not is_challenge_html(html) and status and status < 400:
|
|
return html, "live_http", status, False
|
|
return html, "blocked", status or 403, True
|
|
|
|
|
|
def load_category_page(url: str, *, category_ref: str, playwright: bool = True) -> tuple[str, str, int]:
|
|
status, html = 0, ""
|
|
if playwright:
|
|
try:
|
|
status, html = _fetch_playwright(url, expand_filters=True)
|
|
if not _blocked(status, html):
|
|
return html, "live_playwright", status
|
|
except Exception:
|
|
pass
|
|
try:
|
|
status, html = _fetch_http(url)
|
|
if not _blocked(status, html):
|
|
return html, "live_http", status
|
|
except Exception:
|
|
status, html = status or 0, html or ""
|
|
recorded = FIXTURE_DIR / f"{category_ref}-category.html"
|
|
if recorded.exists():
|
|
return recorded.read_text(encoding="utf-8"), "recorded_html", 0
|
|
raise AppError(
|
|
423,
|
|
"SOURCE_BLOCKED",
|
|
"Geizhals blockiert den Taxonomie-Abruf (403/Challenge)",
|
|
{"url": url, "http_status": status},
|
|
)
|
|
|
|
|
|
def _fetch_http(url: str) -> tuple[int, str]:
|
|
import httpx
|
|
|
|
response = httpx.get(url, headers=BROWSER_HEADERS, follow_redirects=True, timeout=25.0)
|
|
return response.status_code, response.text
|
|
|
|
|
|
def _fetch_playwright(url: str, *, expand_filters: bool = False) -> tuple[int, str]:
|
|
try:
|
|
from playwright.sync_api import sync_playwright
|
|
except ImportError as exc:
|
|
raise AppError(
|
|
423,
|
|
"SOURCE_BLOCKED",
|
|
"Geizhals verlangt einen Browser-Abruf, Playwright ist nicht installiert",
|
|
) from exc
|
|
with sync_playwright() as playwright:
|
|
browser = playwright.chromium.launch(headless=True)
|
|
page = browser.new_page(locale="de-DE", extra_http_headers=BROWSER_HEADERS)
|
|
response = page.goto(url, wait_until="domcontentloaded", timeout=45000)
|
|
page.wait_for_timeout(4000)
|
|
if expand_filters:
|
|
_expand_filter_sections(page)
|
|
for label in EXPAND_LABELS:
|
|
button = page.get_by_text(label, exact=False)
|
|
if button.count():
|
|
try:
|
|
button.first.click(timeout=4000)
|
|
page.wait_for_timeout(800)
|
|
except Exception:
|
|
continue
|
|
if expand_filters:
|
|
_expand_filter_sections(page)
|
|
html = page.content()
|
|
status = response.status if response is not None else 200
|
|
browser.close()
|
|
return status, html
|
|
|
|
|
|
def _expand_filter_sections(page: object) -> None:
|
|
locator = page.locator('button[aria-label*="aufklappen"]')
|
|
for _ in range(40):
|
|
if locator.count() == 0:
|
|
return
|
|
try:
|
|
locator.first.click(timeout=3000)
|
|
page.wait_for_timeout(700)
|
|
except Exception:
|
|
return
|
|
|
|
|
|
def _blocked(status: int, html: str) -> bool:
|
|
if status in {401, 403, 429}:
|
|
return True
|
|
return is_challenge_html(html)
|