import logging import re from decimal import Decimal, InvalidOperation import requests from bs4 import BeautifulSoup logger = logging.getLogger(__name__) _HEADERS = { 'User-Agent': ( 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) ' 'AppleWebKit/537.36 (KHTML, like Gecko) ' 'Chrome/124.0.0.0 Safari/537.36' ), 'Accept-Language': 'en-AU,en;q=0.9', } def parse_price(text: str) -> Decimal | None: """Extract the first numeric price from a string.""" if not text: return None cleaned = text.strip().replace(',', '') match = re.search(r'\d+\.?\d*', cleaned) if not match: return None try: return Decimal(match.group()) except InvalidOperation: return None def _fetch_with_requests(url: str, css_selector: str) -> tuple[str, str | None]: """Returns (raw_text, error). raw_text is empty string on failure.""" try: resp = requests.get(url, headers=_HEADERS, timeout=15) resp.raise_for_status() soup = BeautifulSoup(resp.content, 'html.parser') el = soup.select_one(css_selector) if el: return el.get_text(strip=True), None return '', f'Selector "{css_selector}" matched no element' except Exception as exc: return '', str(exc) def _fetch_with_playwright(url: str, css_selector: str) -> tuple[str, str | None]: """Fallback for JS-rendered pages.""" try: from playwright.sync_api import sync_playwright with sync_playwright() as p: browser = p.chromium.launch(headless=True) page = browser.new_page(extra_http_headers={'Accept-Language': 'en-AU,en;q=0.9'}) page.goto(url, wait_until='domcontentloaded', timeout=30000) page.wait_for_timeout(2000) el = page.query_selector(css_selector) text = el.inner_text() if el else '' browser.close() if not text: return '', f'Selector "{css_selector}" matched no element (playwright)' return text.strip(), None except Exception as exc: return '', f'Playwright error: {exc}' def fetch_price(watcher) -> tuple[Decimal | None, str, str | None]: """ Returns (price, raw_text, error). Tries requests first; falls back to Playwright if the element isn't found. """ raw_text, error = _fetch_with_requests(watcher.url, watcher.css_selector) if not raw_text and not error: logger.info(f'Falling back to Playwright for watcher {watcher.pk}') raw_text, error = _fetch_with_playwright(watcher.url, watcher.css_selector) price = parse_price(raw_text) if raw_text else None logger.info(f'Watcher {watcher.pk}: raw="{raw_text}" price={price} error={error}') return price, raw_text, error