| import requests |
| from bs4 import BeautifulSoup |
| import re |
| import time |
| from datetime import datetime |
| from fake_useragent import UserAgent |
|
|
| |
| SYDNEY_STORE_ID = "3108" |
| SYDNEY_POSTAL = "B1P7H9" |
| SYDNEY_API_COORDS = (46.1368, -60.1942) |
| CORRECT_STORE_ADDRESS = "80 Sydney Port Access Rd, Sydney, NS B1P 7H9" |
|
|
| def get_sydney_session(): |
| """Create a session locked to Sydney NS Walmart""" |
| session = requests.Session() |
| headers = { |
| 'User-Agent': UserAgent().random, |
| 'Accept-Language': 'en-CA,en-US;q=0.9,en;q=0.8', |
| 'Referer': 'https://www.walmart.ca/en/store/3108', |
| 'X-Requested-With': 'XMLHttpRequest' |
| } |
| session.headers.update(headers) |
| |
| |
| session.cookies.update({ |
| 'preferredStore': SYDNEY_STORE_ID, |
| 'storeSelected': 'true', |
| 'regionSelected': 'NS', |
| 'bm_sv': '1' |
| }) |
| |
| return session |
|
|
| def scrape_walmart_product(url, postal_code="B1P 7H9"): |
| """Main scraping function that works with your app.py""" |
| try: |
| |
| clean_postal = re.sub(r'[^A-Za-z0-9]', '', postal_code).upper() |
| is_sydney = clean_postal.startswith('B1P') |
| |
| session = get_sydney_session() |
| |
| |
| time.sleep(2) |
| store_url = f"https://www.walmart.ca/en/store/{SYDNEY_STORE_ID}" if is_sydney else "https://www.walmart.ca/en" |
| session.get(store_url, timeout=10) |
| |
| |
| time.sleep(2) |
| response = session.get(url, timeout=15) |
| |
| if "verify-your-identity" in response.url.lower(): |
| raise Exception("CAPTCHA triggered - try again later") |
| |
| soup = BeautifulSoup(response.text, 'html.parser') |
| |
| |
| product_data = { |
| 'product_name': soup.find('h1', {'itemprop': 'name'}).get_text(strip=True) if soup.find('h1', {'itemprop': 'name'}) else None, |
| 'online_price': extract_price(soup), |
| 'quantity': extract_quantity(soup), |
| 'url': url, |
| 'requested_postal_code': postal_code, |
| 'target_store': CORRECT_STORE_ADDRESS if is_sydney else f"Nearest to {postal_code}" |
| } |
| |
| |
| if is_sydney: |
| store_data = get_sydney_store_price(url) |
| if store_data: |
| product_data.update(store_data) |
| |
| return product_data |
| |
| except Exception as e: |
| return { |
| 'error': str(e), |
| 'solution': 'Try again with different network', |
| 'store_info': CORRECT_STORE_ADDRESS |
| } |
|
|
| def get_sydney_store_price(product_url): |
| """Get price EXCLUSIVELY for Sydney NS Walmart""" |
| try: |
| session = get_sydney_session() |
| |
| |
| product_id = re.search(r'/ip/([^/?]+)', product_url) |
| if not product_id: |
| return None |
| |
| |
| api_url = "https://www.walmart.ca/api/product-page/find-in-store" |
| params = { |
| 'lat': str(SYDNEY_API_COORDS[0]), |
| 'lng': str(SYDNEY_API_COORDS[1]), |
| 'locale': 'en', |
| 'postalCode': SYDNEY_POSTAL[:3] + " " + SYDNEY_POSTAL[3:], |
| 'upc': product_id.group(1), |
| 'storeId': SYDNEY_STORE_ID |
| } |
| |
| headers = { |
| 'Accept': 'application/json', |
| 'Referer': product_url |
| } |
| |
| time.sleep(3) |
| |
| |
| response = session.get(api_url, params=params, headers=headers, timeout=20) |
| data = response.json() |
| |
| |
| if data.get('info'): |
| for store in data['info']: |
| if str(store.get('id')) == SYDNEY_STORE_ID: |
| return { |
| 'store_price': store.get('sellPrice'), |
| 'store_availability': store.get('availabilityStatus'), |
| 'confirmed_location': CORRECT_STORE_ADDRESS, |
| 'store_id': SYDNEY_STORE_ID, |
| 'last_checked': datetime.now().strftime("%Y-%m-%d %H:%M:%S") |
| } |
| |
| return {"error": "Sydney store not found in API response"} |
| |
| except Exception as e: |
| return { |
| "api_error": str(e), |
| "solution": "Try again in 1 hour", |
| "store_info": CORRECT_STORE_ADDRESS |
| } |
|
|
| def extract_price(soup): |
| """Extract product price""" |
| price_selectors = [ |
| ('span', {'itemprop': 'price'}), |
| ('div', {'data-automation': 'price-amount'}), |
| ('span', {'class': 'price-characteristic'}) |
| ] |
| for tag, attrs in price_selectors: |
| element = soup.find(tag, attrs) |
| if element: |
| price = element.get('content') or element.get_text(strip=True) |
| if price: |
| cleaned = re.sub(r'[^\d.,]', '', price) |
| if '.' in cleaned: |
| return f"{float(cleaned):.2f}" |
| return cleaned |
| return None |
|
|
| def extract_quantity(soup): |
| """Extract product quantity""" |
| name_element = soup.find('h1', {'itemprop': 'name'}) |
| if name_element: |
| name = name_element.get_text(strip=True) |
| match = re.search(r'(\d+\s*(g|kg|ml|l|oz|lb|pack|ct|count|ea)\b)', name, re.IGNORECASE) |
| if match: |
| return match.group(1) |
| |
| quantity_selectors = [ |
| ('span', {'data-automation': 'product-quantity'}), |
| ('div', {'class': 'package-size'}) |
| ] |
| for tag, attrs in quantity_selectors: |
| element = soup.find(tag, attrs) |
| if element: |
| return element.get_text(strip=True) |
| return None |
