mirror of
https://github.com/wahyd4/links.git
synced 2026-08-09 05:06:16 +10:00
81 lines
2.7 KiB
Python
81 lines
2.7 KiB
Python
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
|