97 lines
2.8 KiB
Python
97 lines
2.8 KiB
Python
import re
|
|
from typing import Optional, Tuple
|
|
|
|
import httpx
|
|
|
|
HEADERS = {
|
|
"User-Agent": "Mozilla/5.0 (X11; Linux x86_64; rv:121.0) Gecko/20100101 Firefox/121.0",
|
|
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,*/*;q=0.8",
|
|
"Accept-Language": "en-US,en;q=0.5",
|
|
"Accept-Encoding": "gzip, deflate, br",
|
|
"DNT": "1",
|
|
"Sec-GPC": "1",
|
|
"Connection": "keep-alive",
|
|
"Upgrade-Insecure-Requests": "1",
|
|
"Sec-Fetch-Dest": "document",
|
|
"Sec-Fetch-Mode": "navigate",
|
|
"Sec-Fetch-Site": "none",
|
|
"Sec-Fetch-User": "?1",
|
|
"Pragma": "no-cache",
|
|
"Cache-Control": "no-cache",
|
|
}
|
|
|
|
|
|
async def _get_cookies(client):
|
|
# Make a request to https://www.woolworths.com.au/ as if we were a normal browser, then return the cookies
|
|
response = await client.get(
|
|
"https://www.woolworths.com.au/", headers=HEADERS, follow_redirects=True
|
|
)
|
|
return dict(response.cookies)
|
|
|
|
|
|
def _get_package_size(data: dict) -> Tuple[int, str]:
|
|
size = data["Product"]["PackageSize"]
|
|
if size:
|
|
match = re.match(r"(\d+)(.*)", size)
|
|
if match:
|
|
return int(match.group(1)), match.group(2)
|
|
|
|
return 1, "items"
|
|
|
|
|
|
def _get_client() -> httpx.AsyncClient:
|
|
return httpx.AsyncClient()
|
|
|
|
|
|
cached_cookies = None
|
|
|
|
|
|
async def _request_url(url: str) -> Optional[dict]:
|
|
global cached_cookies
|
|
|
|
async with _get_client() as client:
|
|
if not cached_cookies:
|
|
cached_cookies = await _get_cookies(client)
|
|
|
|
cookies = cached_cookies
|
|
try:
|
|
response = await client.get(
|
|
url, headers=HEADERS, follow_redirects=True, cookies=cookies
|
|
)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
except httpx.HTTPError as ne:
|
|
print(ne)
|
|
cached_cookies = None
|
|
return None
|
|
|
|
|
|
def _get_product_details_url(product_id: str) -> str:
|
|
return f"https://www.woolworths.com.au/apis/ui/product/detail/{product_id}"
|
|
|
|
|
|
def get_product_id(url: str) -> Optional[str]:
|
|
woolies_regex = r"https://www.woolworths.com.au/shop/productdetails/(\d+)/?.*"
|
|
match = re.match(woolies_regex, url)
|
|
if match:
|
|
return match.group(1)
|
|
return None
|
|
|
|
|
|
async def scrape(product_id: str) -> Tuple[dict, dict]:
|
|
details_url = _get_product_details_url(product_id)
|
|
raw_data = await _request_url(details_url)
|
|
if raw_data is None:
|
|
# Return a minimal structure; callers treat this as raw payload for logging
|
|
raw_data = {}
|
|
quantity, unit = _get_package_size(raw_data)
|
|
|
|
product_data = {
|
|
"name": raw_data["Product"]["Name"],
|
|
"quantity": quantity,
|
|
"unit": unit,
|
|
"img_small": raw_data["Product"]["SmallImageFile"],
|
|
"img_large": raw_data["Product"]["LargeImageFile"],
|
|
}
|
|
|
|
return product_data, raw_data
|