#!/usr/bin/env python3 """ RW_Site_Scraper-Orders_Page.py ============================== Scrapes Red Wing footwear catalog: - https://order.redwingshoes.com/footwear-rwbr Test mode: - Set ONLY_STYLE at the top of this file to a style number (e.g. "400") to scrape just that single product. Outputs (next to this script): - RW_Orders_Site_Scrape.md - RW_Site_Scraper_v2_checkpoint.json (resume state) - RW_Site_Scraper_v2_errors.txt (links that failed repeatedly) Stability features: - `safe_get()` uses short timeouts + window.stop() so Selenium doesn't hang forever - HTTP fallback (no Selenium) for the few product pages that still time out - Optional salvage pass at the end to retry hard failures (HTTP-first) Cross-platform (Windows + Linux Cinnamon): - Headless Firefox (default) - Geckodriver resolution "like Parts_Auto" (explicit Service path; no Selenium Manager): 1) GECKODRIVER_PATH env var (file or directory) 2) geckodriver(.exe) on PATH 3) auto-download geckodriver (GitHub releases) into a user cache dir 4) (optional) if double-click/no terminal and Tk is available, prompt to pick geckodriver Feature columns are 1/0 (not Yes/No). Includes Brand (string) + brand family flags (Red Wing / Irish Setter / Worx). Dependencies: - Python 3.9+ - Firefox installed - Selenium installed: Linux Mint/Ubuntu: sudo apt install -y python3-selenium Windows: python -m pip install selenium Notes for Linux Mint PEP 668: - Prefer `python3-selenium` from apt (as above). - This script does NOT require webdriver-manager. """ from __future__ import annotations import json import os import platform import re import shutil import stat import sys import tarfile import time import traceback import random import requests import html as html_lib from html.parser import HTMLParser import zipfile from pathlib import Path from typing import Dict, List, Optional, Set, Tuple from urllib.parse import parse_qs, urljoin, urlparse from urllib.request import Request, urlopen # NOTE: Style/Name extraction is now done via BeautifulSoup using the exact # logic requested (shoeguide/printSpacing -> h3 + strong). This intentionally # avoids any other name/title fallbacks. from bs4 import BeautifulSoup from selenium import webdriver from selenium.common.exceptions import TimeoutException, WebDriverException from selenium.webdriver.common.by import By from selenium.webdriver.common.keys import Keys from selenium.webdriver.firefox.options import Options from selenium.webdriver.firefox.service import Service as FirefoxService from selenium.webdriver.support import expected_conditions as EC from selenium.webdriver.support.ui import WebDriverWait # ----------------------------- # Config # ----------------------------- # Catalog pages to scan for product links. CATALOG_URLS = [ "https://order.redwingshoes.com/footwear-rwbr", ] CATALOG_URL_PRIMARY = CATALOG_URLS[0] SITE_HOST = "order.redwingshoes.com" FOOTWEAR_RWB_SLUG = "/footwear-rwbr" # Login credentials (env vars take precedence) RW_SITE_USERNAME = os.environ.get("RW_SITE_USERNAME", "rwss614@redwingshoes.com") RW_SITE_PASSWORD = os.environ.get("RW_SITE_PASSWORD", "WelcomeBack99!") # Default is headless. Set RW_HEADLESS=0 to run with a visible browser. RUN_HEADLESS = os.environ.get("RW_HEADLESS", "1").strip().lower() in {"1", "true", "yes", "y", "on"} # Re-discover catalog links on every run so newly added products are picked up. REFRESH_PRODUCT_LINKS_EACH_RUN = os.environ.get("RW_REFRESH_LINKS", "1").strip().lower() in {"1", "true", "yes", "y", "on"} # Optional: scrape only a single style number (set "" to scrape all discovered styles) ONLY_STYLE: str = "" BASE_DIR = Path(__file__).resolve().parent OUT_MD = BASE_DIR / "RW_Orders_Site_Scrape.md" CHECKPOINT = BASE_DIR / "RW_Site_Scraper_v2_checkpoint.json" ERRORS_TXT = BASE_DIR / "RW_Site_Scraper_v2_errors.txt" MAX_FAILS_PER_LINK = 3 # Auto-repair: if old checkpoint rows have clearly-bad Name/Brand (e.g. "RED WING FOR BUSINESS"), # automatically un-mark those links as done so they get re-scraped on the next run. AUTO_REPAIR_BAD_ROWS = True BAD_NAME_SENTINEL = "RED WING FOR BUSINESS" # Prefer names as shown on the International catalog listing page. # Disabled: listing tiles can include UI strings like "Save to list". PREFER_INTERNATIONAL_LISTING_NAMES = False MD_HEADERS = [ "Style #", "Name", "URL", "Image", "Brand", "Male", "Female", "Red Wing", "Irish Setter", "Worx", "Safety Toe", "Steel Toe", "Non-Metallic Toe", "Aluminum Toe", "Metatarsal Guard", "Soft Toe", "Waterproof", "Insulation", "Slip Resistant", "Electrical Hazard", "Puncture Resistant", "Static Dissipative", "Ankle Protection", "BOA® Lacing System", "Defined Heel", "All Leather Upper", "Resoleable", "Oxford/Athletic", "Chukka", "Hiker", '5"', '6"', '7"', '8"', '9"', '10"', '11"', '12"', "Built in USA", "Made in USA", ] EXPECTED_COLS = len(MD_HEADERS) AUTO_REPAIR_MISSING_MEDIA = True # Navigation tuning (Red Wing pages sometimes never fully 'finish' loading) PAGE_LOAD_TIMEOUT = 45 SCRIPT_TIMEOUT = 30 NAV_SETTLE_SECONDS = 1.25 # Small polite delay between product navigations to reduce transient server hiccups REQUEST_DELAY_RANGE = (0.15, 0.55) # Force private-session behavior so prior cookies/tracking data do not affect scraping. PRIVATE_BROWSER_MODE = True # After the main pass, retry hard-failed links once with a fresh driver + more conservative settings ENABLE_SALVAGE_PASS = True SALVAGE_PAGE_LOAD_TIMEOUT = 90 SALVAGE_MAX_TRIES_PER_LINK = 2 CHECKPOINT_VERSION = 11 # Optional: set this environment variable to explicitly point at geckodriver # Linux: export GECKODRIVER_PATH="/home/you/bin/geckodriver" # Windows: setx GECKODRIVER_PATH "C:\Users\you\bin\geckodriver.exe" GECKODRIVER_PATH = os.environ.get("GECKODRIVER_PATH", "").strip() or None # ----------------------------- # Small helpers # ----------------------------- def b01(v: bool) -> str: return "1" if v else "0" def md_escape_cell(s: str) -> str: s = (s or "").replace("|", r"\|") s = re.sub(r"\s+", " ", s).strip() return s def atomic_write(path: Path, text: str) -> None: tmp = path.with_suffix(path.suffix + ".tmp") tmp.write_text(text, encoding="utf-8") tmp.replace(path) def load_checkpoint() -> Dict: if not CHECKPOINT.exists(): return { "version": CHECKPOINT_VERSION, "catalog_urls": CATALOG_URLS, "product_links": [], "done_links": [], "rows": [], "preferred_names": {}, "media_repair_done": False, "fail_counts": {}, "hard_failed_links": [], "started_at": time.time(), "geckodriver_path": None, } try: data = json.loads(CHECKPOINT.read_text(encoding="utf-8")) old_ver = int(data.get("version", 0)) if old_ver != CHECKPOINT_VERSION: try: CHECKPOINT.replace(CHECKPOINT.with_suffix(f".json.v{old_ver}.bak")) except Exception: pass return { "version": CHECKPOINT_VERSION, "catalog_urls": CATALOG_URLS, "product_links": [], "done_links": [], "rows": [], "fail_counts": {}, "started_at": time.time(), "geckodriver_path": None, } data.setdefault("version", CHECKPOINT_VERSION) data.setdefault("catalog_urls", CATALOG_URLS) data.setdefault("product_links", []) data.setdefault("done_links", []) data.setdefault("rows", []) data.setdefault("fail_counts", {}) data.setdefault("hard_failed_links", []) data.setdefault("started_at", time.time()) data.setdefault("geckodriver_path", None) return data except Exception: try: CHECKPOINT.replace(CHECKPOINT.with_suffix(".json.corrupt")) except Exception: pass return { "version": CHECKPOINT_VERSION, "catalog_urls": CATALOG_URLS, "product_links": [], "done_links": [], "rows": [], "preferred_names": {}, "media_repair_done": False, "fail_counts": {}, "hard_failed_links": [], "started_at": time.time(), "geckodriver_path": None, } def save_checkpoint(data: Dict) -> None: atomic_write(CHECKPOINT, json.dumps(data, indent=2, sort_keys=True)) def write_errors_line(line: str) -> None: with open(ERRORS_TXT, "a", encoding="utf-8") as f: f.write(line.rstrip() + "\n") def _can_use_tk() -> bool: try: import tkinter # noqa: F401 return True except Exception: return False def _tk_pick_file(title: str) -> Optional[str]: # Only use this when not running in a terminal (double-click scenario). if sys.stdout.isatty() or not _can_use_tk(): return None try: import tkinter as tk from tkinter import filedialog, messagebox root = tk.Tk() root.withdraw() root.attributes("-topmost", True) messagebox.showinfo( "RW Site Scraper", "Could not find geckodriver automatically.\n\n" "Please select the geckodriver executable.\n" "Windows: geckodriver.exe\nLinux: geckodriver" ) path = filedialog.askopenfilename(title=title) root.destroy() path = (path or "").strip() return path if path else None except Exception: return None class ProgressReporter: """TTY progress bar, or Tk window if launched by double-click (no TTY).""" def __init__(self, total: int, started_at: float): self.total = max(int(total), 1) self.started_at = started_at self.use_tty = sys.stdout.isatty() self.gui = False self._root = None self._label = None self._pbar = None if (not self.use_tty) and _can_use_tk(): try: import tkinter as tk from tkinter import ttk self._root = tk.Tk() self._root.title("RW Site Scraper") self._root.geometry("620x150") self._root.resizable(False, False) self._label = tk.Label(self._root, text="Starting...", anchor="w") self._label.pack(fill="x", padx=12, pady=(12, 6)) self._pbar = ttk.Progressbar(self._root, maximum=self.total, length=580) self._pbar.pack(padx=12, pady=(0, 10)) self.gui = True self._root.update_idletasks() self._root.update() except Exception: self.gui = False def update(self, current: int, note: str = "") -> None: current = max(0, min(int(current), self.total)) if self.gui and self._root: msg = f"{current}/{self.total} {note}".strip() if self._label: self._label.config(text=msg) if self._pbar: self._pbar["value"] = current self._root.update_idletasks() self._root.update() return width = 32 frac = current / self.total filled = int(round(frac * width)) bar = "#" * filled + "-" * (width - filled) elapsed = max(time.time() - self.started_at, 0.0001) rate = current / elapsed eta = (self.total - current) / rate if rate > 1e-9 else 0.0 msg = f"[{bar}] {current}/{self.total} ({frac*100:5.1f}%) ETA {int(eta)}s" if note: msg += f" {note}" sys.stdout.write("\r" + msg[:240]) sys.stdout.flush() if current == self.total: sys.stdout.write("\n") sys.stdout.flush() def close(self) -> None: if self.gui and self._root: try: self._root.destroy() except Exception: pass # ----------------------------- # Geckodriver resolution / install # ----------------------------- def _cache_dir() -> Path: if os.name == "nt": base = os.environ.get("LOCALAPPDATA") or os.environ.get("APPDATA") or str(Path.home()) d = Path(base) / "RW_Site_Scraper" else: base = os.environ.get("XDG_CACHE_HOME") or str(Path.home() / ".cache") d = Path(base) / "rw_site_scraper" d.mkdir(parents=True, exist_ok=True) return d def _resolve_from_env_or_path() -> Optional[str]: # 1) explicit env var (file or directory) if GECKODRIVER_PATH: p = Path(GECKODRIVER_PATH) if p.is_file(): return str(p) if p.is_dir(): for name in ("geckodriver.exe", "geckodriver"): cand = p / name if cand.is_file(): return str(cand) # 2) alongside this script for name in ("geckodriver.exe", "geckodriver"): cand = BASE_DIR / name if cand.is_file(): return str(cand) # 3) scraper cache dir for name in ("geckodriver.exe", "geckodriver"): cand = _cache_dir() / name if cand.is_file(): return str(cand) # 4) common user bin dirs (linux) if os.name != "nt": for d in (Path.home() / "bin", Path.home() / ".local" / "bin"): cand = d / "geckodriver" if cand.is_file(): return str(cand) # 5) PATH p = shutil.which("geckodriver") or shutil.which("geckodriver.exe") if p: return p return None def _platform_asset_key() -> Tuple[str, str]: """ Returns (asset_contains, archive_type) matching geckodriver releases. """ sysplat = sys.platform.lower() mach = platform.machine().lower() if sysplat.startswith("win"): is_64 = "64" in platform.architecture()[0] return ("win64" if is_64 else "win32", "zip") if sysplat.startswith("linux"): if "aarch64" in mach or "arm64" in mach: return ("linux-aarch64", "tar.gz") return ("linux64", "tar.gz") if sysplat == "darwin": if "arm64" in mach or "aarch64" in mach: return ("macos-aarch64", "tar.gz") return ("macos", "tar.gz") return ("linux64", "tar.gz") def _download_latest_geckodriver(dest_dir: Path) -> Path: """ Downloads and extracts latest geckodriver into dest_dir. Returns path to extracted driver. """ asset_key, archive_type = _platform_asset_key() api = "https://api.github.com/repos/mozilla/geckodriver/releases/latest" req = Request(api, headers={"User-Agent": "RW_Site_Scraper/1.0"}) with urlopen(req, timeout=30) as r: data = json.loads(r.read().decode("utf-8", errors="replace")) assets = data.get("assets", []) dl_url = None dl_name = None for a in assets: name = a.get("name", "") if asset_key in name and name.endswith(archive_type): dl_url = a.get("browser_download_url") dl_name = name break if not dl_url: raise RuntimeError(f"Could not find a geckodriver asset for {asset_key} ({archive_type}).") archive_path = dest_dir / dl_name req2 = Request(dl_url, headers={"User-Agent": "RW_Site_Scraper/1.0"}) with urlopen(req2, timeout=60) as r, open(archive_path, "wb") as f: shutil.copyfileobj(r, f) driver_name = "geckodriver.exe" if os.name == "nt" else "geckodriver" extracted_path = dest_dir / driver_name if archive_type == "zip": with zipfile.ZipFile(archive_path, "r") as z: for member in z.namelist(): if member.endswith(driver_name): z.extract(member, path=dest_dir) p = dest_dir / member p.replace(extracted_path) break else: with tarfile.open(archive_path, "r:gz") as t: for member in t.getmembers(): if member.name.endswith("/" + driver_name) or member.name == driver_name: t.extract(member, path=dest_dir) p = dest_dir / member.name p.replace(extracted_path) break try: archive_path.unlink(missing_ok=True) except Exception: pass if os.name != "nt": try: st = os.stat(extracted_path) os.chmod(extracted_path, st.st_mode | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH) except Exception: pass if not extracted_path.exists(): raise RuntimeError("Download succeeded but geckodriver was not extracted.") return extracted_path def ensure_geckodriver(ck: Dict) -> str: # 1) env/path/script dir p = _resolve_from_env_or_path() if p and Path(p).exists(): return p # 2) saved in checkpoint saved = (ck.get("geckodriver_path") or "").strip() if saved and Path(saved).exists(): return saved # 3) auto-download try: dest = _cache_dir() driver_path = _download_latest_geckodriver(dest) ck["geckodriver_path"] = str(driver_path) save_checkpoint(ck) return str(driver_path) except Exception as e: # 4) optional interactive pick (only when double-click/no TTY) picked = _tk_pick_file("Select geckodriver / geckodriver.exe") if picked and Path(picked).exists(): ck["geckodriver_path"] = picked save_checkpoint(ck) return picked raise RuntimeError( "Unable to locate or install geckodriver.\n\n" "Fix options:\n" " - Put geckodriver on PATH\n" " - OR set GECKODRIVER_PATH to the full driver path\n" " - OR install via package manager (Linux often: sudo apt install firefox-geckodriver)\n\n" f"Underlying error: {e}" ) def create_driver( ck: Dict, headless: bool = True, *, page_load_strategy: str = "eager", block_images: bool = True, page_load_timeout: int = PAGE_LOAD_TIMEOUT, user_agent: Optional[str] = None, ) -> webdriver.Firefox: """Start Firefox using an explicit geckodriver path (no Selenium Manager). Args: page_load_strategy: "eager" returns after DOMContentLoaded. "normal" waits for full load. block_images: If True, blocks images to reduce load stalls/timeouts. page_load_timeout: Seconds for Selenium navigation timeout. """ gecko = ensure_geckodriver(ck) options = Options() if headless: options.add_argument("-headless") if PRIVATE_BROWSER_MODE: # Explicit private mode in addition to private-browsing prefs. options.add_argument("-private") if page_load_strategy: options.set_capability("pageLoadStrategy", page_load_strategy) # Safe stability prefs options.set_preference("dom.webnotifications.enabled", False) options.set_preference("media.volume_scale", "0.0") options.set_preference("browser.privatebrowsing.autostart", True) options.set_preference("network.http.http3.enable", False) options.set_preference("browser.cache.disk.enable", False) options.set_preference("browser.cache.memory.enable", False) options.set_preference("browser.cache.offline.enable", False) options.set_preference("network.http.use-cache", False) options.set_preference("places.history.enabled", False) options.set_preference("signon.rememberSignons", False) options.set_preference("privacy.trackingprotection.enabled", True) options.set_preference("privacy.trackingprotection.pbmode.enabled", True) if block_images: # 2 = block images options.set_preference("permissions.default.image", 2) if user_agent: options.set_preference("general.useragent.override", user_agent) service = FirefoxService(executable_path=gecko) driver = webdriver.Firefox(service=service, options=options) driver.set_page_load_timeout(page_load_timeout) driver.set_script_timeout(SCRIPT_TIMEOUT) return driver def reset_browser_state(driver) -> None: """Clear cookies/storage for a clean private session baseline.""" try: driver.delete_all_cookies() except Exception: pass try: driver.get("about:blank") driver.execute_script( """ try { localStorage.clear(); } catch (e) {} try { sessionStorage.clear(); } catch (e) {} """ ) except Exception: pass # ----------------------------- # Navigation + HTTP fallback helpers # ----------------------------- def _polite_delay() -> None: lo, hi = REQUEST_DELAY_RANGE try: time.sleep(random.uniform(lo, hi)) except Exception: time.sleep(lo) def safe_get( driver, url: str, timeout: int = PAGE_LOAD_TIMEOUT, settle: float = NAV_SETTLE_SECONDS, max_tries: int = 2, ) -> None: """Navigate without getting stuck on pages that never fully finish loading. - Uses page_load_timeout - On timeout, calls window.stop() and continues if the DOM exists - Retries a couple times with light cleanup """ last_exc: Optional[BaseException] = None for attempt in range(1, max_tries + 2): try: driver.set_page_load_timeout(timeout) _polite_delay() # Python-side HTTP timeout to geckodriver (prevents indefinite hangs in driver.get) try: ce = getattr(driver, "command_executor", None) if ce is not None: # Selenium 4: RemoteConnection holds a ClientConfig with a request timeout if hasattr(ce, "_client_config") and hasattr(ce._client_config, "timeout"): ce._client_config.timeout = int(timeout) + 10 # urllib3 PoolManager may also have a timeout attribute if hasattr(ce, "_conn") and hasattr(ce._conn, "timeout"): ce._conn.timeout = int(timeout) + 10 except Exception: pass driver.get(url) # Give the DOM a moment to settle time.sleep(settle) # If we have any DOM, we can proceed. try: driver.find_element(By.TAG_NAME, "body") except Exception: pass return except TimeoutException as e: last_exc = e try: driver.execute_script("window.stop();") except Exception: pass time.sleep(settle) try: driver.find_element(By.TAG_NAME, "body") return except Exception: pass except WebDriverException as e: last_exc = e # Cleanup between retries try: driver.get("about:blank") except Exception: pass time.sleep(0.25 * attempt) if last_exc: raise last_exc def _url_variants(url: str) -> List[str]: """Return a small set of URL variants (www/non-www) to dodge occasional redirects.""" outs: List[str] = [] for u in [url]: if u not in outs: outs.append(u) if "//www." in url: u2 = url.replace("//www.", "//", 1) if u2 not in outs: outs.append(u2) else: u2 = url.replace("//", "//www.", 1) if u2 not in outs: outs.append(u2) return outs def _fetch_html(url: str, timeout: int = 30) -> str: req = Request( url, headers={ "User-Agent": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120 Safari/537.36", "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8", "Accept-Language": "en-US,en;q=0.9", }, ) with urlopen(req, timeout=timeout) as r: data = r.read() # best-effort decode return data.decode("utf-8", errors="replace") class _VisibleTextExtractor(HTMLParser): """Very small HTML->visible text extractor (no external deps).""" _BLOCK_TAGS = { "p", "div", "br", "li", "tr", "td", "th", "h1", "h2", "h3", "h4", "h5", "h6", "section", "article", "header", "footer", } def __init__(self): super().__init__() self.parts: List[str] = [] def handle_starttag(self, tag, attrs): if tag.lower() in self._BLOCK_TAGS: self.parts.append("\n") def handle_endtag(self, tag): if tag.lower() in self._BLOCK_TAGS: self.parts.append("\n") def handle_data(self, data): if data and data.strip(): self.parts.append(data) def _html_to_text(html: str) -> str: p = _VisibleTextExtractor() try: p.feed(html) except Exception: pass raw = "".join(p.parts) raw = html_lib.unescape(raw) raw = re.sub(r"\n{3,}", "\n\n", raw) raw = re.sub(r"[\t\r]+", " ", raw) return raw.strip() def _extract_first_h1(html: str) -> str: m = re.search(r"]*>(.*?)", html, flags=re.IGNORECASE | re.DOTALL) if not m: return "" inner = m.group(1) inner = re.sub(r"<[^>]+>", " ", inner) inner = html_lib.unescape(inner) inner = re.sub(r"\s+", " ", inner).strip() return inner def _extract_first_heading(html: str) -> str: """Extract a likely product title from HTML. Many Red Wing product pages place the actual product name in

, while

can be a site/banner header (e.g., 'RED WING FOR BUSINESS'). We therefore try meaningful

first, then fall back to

. """ BAD = {"RED WING FOR BUSINESS"} STOP = { "FEATURES", "SPECIFICATIONS", "TECHNOLOGY", "DETAILS", "SIZE & FIT", "SIZING", "CARE", "REVIEWS", "RELATED PRODUCTS", } def _clean(inner: str) -> str: inner = re.sub(r"<[^>]+>", " ", inner) inner = html_lib.unescape(inner) inner = re.sub(r"\s+", " ", inner).strip() return inner def _ok(s: str) -> bool: s = (s or "").strip() if not s: return False up = s.upper() if up in BAD or up in STOP: return False if len(up) <= 3: return False if not re.search(r"[A-Z]", up): return False return True # Prefer the 'Name' field from the productInfo table when present mname = re.search( r"]*class=['\"][^'\"]*prTitle[^'\"]*['\"][^>]*>\s*Name\s*\s*" r"]*class=['\"][^'\"]*prValue[^'\"]*['\"][^>]*>(.*?)", html, flags=re.IGNORECASE | re.DOTALL, ) if mname: s = _clean(mname.group(1)) if _ok(s): return s # Try first meaningful

for mh3 in re.finditer(r"]*>(.*?)

", html, flags=re.IGNORECASE | re.DOTALL): s = _clean(mh3.group(1)) if _ok(s): return s return _extract_first_h1(html) # ----------------------------- # Scraping logic # ----------------------------- def _is_product_link(href: str) -> bool: href = (href or "").strip() if not href: return False low = href.lower() if low.startswith("javascript:") or low.startswith("mailto:") or low.startswith("tel:"): return False if "/safety-boot/" in low: return True try: parsed = urlparse(href) path = (parsed.path or "").lower().rstrip("/") query = parse_qs(parsed.query or "") except Exception: path = low query = {} if "/footwear-" in path: # Keep only actual product routes, but allow query-driven product URLs. if re.fullmatch(r".*/footwear-[^/]+", path): for key in ("style", "styleNumber", "sku", "item", "pid", "productId"): if query.get(key) or query.get(key.lower()): return True return False return True # Some site variants expose product URLs under generic paths. if "/product/" in path and re.search(r"\d{3,6}", path): return True # Red Wing wholesale product detail pages can be root-level numeric html paths: /02478.html if re.search(r"/\d{3,6}\.html$", path): return True return False def _find_product_anchors(driver): """Locate product anchors on the catalog page. The Red Wing wholesale catalog renders product tiles with specific CSS classes. Each tile (
  • ) contains two links pointing at the same product detail page: one wrapping the image and one wrapping the product name. Both have an href that ends in ".html" with the numeric style number. On the current site the anchor for the name has the classes ``c-product-tile__pdp-link js-product-name``, while the image link has no special class but is still within the tile. This helper first attempts to extract anchors using those specific selectors and falls back to scanning all anchors when necessary. Returns a list of WebElement objects (anchors) with unique hrefs. """ anchors: List = [] seen: Set[str] = set() # First try targeted selectors for the product name link. try: elements = driver.find_elements( By.CSS_SELECTOR, "li.js-product-grid-item a.c-product-tile__pdp-link.js-product-name[href]", ) for a in elements: try: href = (a.get_attribute("href") or "").strip() except Exception: href = "" if not href or href in seen: continue if _is_product_link(href): anchors.append(a) seen.add(href) except Exception: pass # Then collect image links from the same tiles. try: elements2 = driver.find_elements( By.CSS_SELECTOR, "li.js-product-grid-item a[href$='.html']" ) for a in elements2: try: href = (a.get_attribute("href") or "").strip() except Exception: href = "" if not href or href in seen: continue if _is_product_link(href): anchors.append(a) seen.add(href) except Exception: pass # Fallback: generic anchor scanning if targeted selectors found nothing. if not anchors: try: for a in driver.find_elements(By.XPATH, "//a[@href]"): try: href = (a.get_attribute("href") or "").strip() except Exception: href = "" if not href or href in seen: continue if _is_product_link(href): anchors.append(a) seen.add(href) except Exception: pass return anchors def _extract_product_links_from_page_source(driver) -> List[str]: """Fallback link discovery when clickable anchors are sparse or JS-rendered.""" try: base_url = driver.current_url or CATALOG_URL_PRIMARY except Exception: base_url = CATALOG_URL_PRIMARY try: html = driver.page_source or "" except Exception: html = "" if not html: return [] candidates: Set[str] = set() # 1) Parse tile anchors directly from source. try: soup = BeautifulSoup(html, "html.parser") # Preferred selector discovered on /footwear-rwbr product grid. for a in soup.select("li.js-product-grid-item a.c-product-tile__pdp-link.js-product-name[href]"): href = (a.get("href") or "").strip() if href: candidates.add(href) # Fallback for image-wrapped links inside the same tile. for a in soup.select("li.js-product-grid-item a[href$='.html']"): href = (a.get("href") or "").strip() if href: candidates.add(href) # Final generic fallback. for a in soup.find_all("a", href=True): href = (a.get("href") or "").strip() if href: candidates.add(href) except Exception: pass # 2) Regex fallback for URLs embedded in JSON/script blobs. blob = html_lib.unescape(html) patterns = [ r"https?://[^\s\"'<>]+", r"/(?:footwear-rwbr|safety-boot|product)/[^\s\"'<>]+", r"/\d{3,6}\.html(?:[?#][^\s\"'<>]*)?", ] for pat in patterns: for m in re.finditer(pat, blob, flags=re.IGNORECASE): u = (m.group(0) or "").strip() if u: candidates.add(u) out: List[str] = [] seen: Set[str] = set() for raw in candidates: abs_url = urljoin(base_url, raw) if not _is_product_link(abs_url): continue if abs_url in seen: continue seen.add(abs_url) out.append(abs_url) return out def ensure_logged_in(driver, target_url: str) -> None: """Attempt a best-effort login for order.redwingshoes.com if prompted.""" _polite_delay() safe_get(driver, target_url) dismiss_popups(driver) login_user_selectors = [ (By.CSS_SELECTOR, "input[type='email']"), (By.CSS_SELECTOR, "input[name='email']"), (By.CSS_SELECTOR, "input[name='username']"), (By.CSS_SELECTOR, "input[id*='email' i]"), (By.CSS_SELECTOR, "input[id*='user' i]"), (By.CSS_SELECTOR, "input[autocomplete='username']"), ] login_pass_selectors = [ (By.CSS_SELECTOR, "input[type='password']"), (By.CSS_SELECTOR, "input[name='password']"), (By.CSS_SELECTOR, "input[id*='password' i]"), (By.CSS_SELECTOR, "input[autocomplete='current-password']"), ] submit_selectors = [ (By.CSS_SELECTOR, "button[type='submit']"), (By.XPATH, "//button[contains(translate(.,'LOGINSGNIN','loginsgnin'),'login') or contains(translate(.,'LOGINSGNIN','loginsgnin'),'sign in')]"), (By.XPATH, "//input[@type='submit']"), ] # If product links are already visible, skip login attempts. try: if _find_product_anchors(driver): return except Exception: pass user_el = None pass_el = None for by, sel in login_user_selectors: try: cand = driver.find_element(by, sel) if cand.is_displayed(): user_el = cand break except Exception: pass for by, sel in login_pass_selectors: try: cand = driver.find_element(by, sel) if cand.is_displayed(): pass_el = cand break except Exception: pass if not user_el or not pass_el: return try: user_el.clear() except Exception: pass user_el.send_keys(RW_SITE_USERNAME) try: pass_el.clear() except Exception: pass pass_el.send_keys(RW_SITE_PASSWORD) submitted = False for by, sel in submit_selectors: try: btn = driver.find_element(by, sel) if btn.is_displayed() and btn.is_enabled(): btn.click() submitted = True break except Exception: pass if not submitted: pass_el.send_keys(Keys.ENTER) # Wait for either product links to appear or password field to disappear. end = time.time() + 25 while time.time() < end: dismiss_popups(driver) try: if _find_product_anchors(driver): return except Exception: pass try: for by, sel in login_pass_selectors: pe = driver.find_element(by, sel) if pe.is_displayed(): break else: return except Exception: return time.sleep(0.4) def open_footwear_section(driver) -> None: """Open the Footwear section after login when required by the site UI.""" selectors = [ (By.ID, "footwear-label"), (By.CSS_SELECTOR, "#footwear-label"), (By.XPATH, "//button[@id='footwear-label' or @aria-controls='footwear']"), (By.XPATH, "//a[@id='footwear-label']"), (By.XPATH, "//*[self::button or self::a or @role='tab'][contains(normalize-space(.), 'Footwear')]"), ] for by, sel in selectors: try: el = WebDriverWait(driver, 10).until(EC.element_to_be_clickable((by, sel))) driver.execute_script("arguments[0].scrollIntoView({block:'center', inline:'center'});", el) time.sleep(0.15) try: el.click() except Exception: driver.execute_script("arguments[0].click();", el) time.sleep(0.8) return except Exception: continue def open_red_wing_footwear_page(driver) -> None: """Use Footwear > Red Wing navigation described in docs, then ensure /footwear-rwbr.""" # Prefer clicking the Red Wing submenu entry if visible. selectors = [ (By.XPATH, "//a[contains(@href,'/footwear-rwbr')]"), (By.XPATH, "//button[contains(@data-target,'footwear-rwbr')]"), (By.XPATH, "//*[self::a or self::button][contains(normalize-space(.), 'Red Wing') and not(contains(normalize-space(.), 'Heritage'))]"), ] clicked = False for by, sel in selectors: try: el = WebDriverWait(driver, 6).until(EC.element_to_be_clickable((by, sel))) driver.execute_script("arguments[0].scrollIntoView({block:'center', inline:'center'});", el) time.sleep(0.15) try: el.click() except Exception: driver.execute_script("arguments[0].click();", el) clicked = True break except Exception: continue # Always navigate directly to the Red Wing footwear catalog. # The current site sometimes renders a quick-order dashboard first even # after clicking menu items, so we explicitly load the slug. try: safe_get(driver, CATALOG_URL_PRIMARY) except Exception: pass try: WebDriverWait(driver, 12).until( lambda d: FOOTWEAR_RWB_SLUG in ((urlparse(d.current_url).path or "").lower()) ) except Exception: try: safe_get(driver, CATALOG_URL_PRIMARY) except Exception: pass def discover_footwear_catalog_urls(driver) -> List[str]: """Discover footwear brand catalog paths from RW site navigation/elements.""" try: base_url = driver.current_url or CATALOG_URL_PRIMARY except Exception: base_url = CATALOG_URL_PRIMARY candidates: List[str] = [] # Pull links from visible navigation. dom_selectors = [ "a[href*='/footwear-']", "[id*='footwear' i] a[href]", "nav a[href*='footwear']", ] for sel in dom_selectors: try: for el in driver.find_elements(By.CSS_SELECTOR, sel): href = (el.get_attribute("href") or "").strip() if href: candidates.append(href) except Exception: pass # Some menus store the target in data attributes. try: for el in driver.find_elements(By.CSS_SELECTOR, "[data-target], [data-url], [data-href]"): for attr in ("data-target", "data-url", "data-href"): v = (el.get_attribute(attr) or "").strip() if "/footwear-" in v: candidates.append(v) except Exception: pass # Parse page source for hidden/off-canvas menu URLs. try: html = driver.page_source or "" except Exception: html = "" if html: blob = html_lib.unescape(html) for m in re.finditer(r"/footwear-[a-z0-9\-]+", blob, flags=re.IGNORECASE): candidates.append(m.group(0)) for m in re.finditer(r"https?://[^\s\"'<>]*/footwear-[a-z0-9\-]+", blob, flags=re.IGNORECASE): candidates.append(m.group(0)) urls: List[str] = [] seen: Set[str] = set() for raw in candidates: u = urljoin(base_url, raw) try: p = urlparse(u) host = (p.netloc or "").lower() path = (p.path or "").lower() except Exception: continue if SITE_HOST not in host: continue if not re.search(r"/footwear-[a-z0-9\-]+$", path): continue norm = f"{p.scheme}://{p.netloc}{path}" if norm not in seen: seen.add(norm) urls.append(norm) # Always include configured default as fallback. if CATALOG_URL_PRIMARY not in seen: urls.insert(0, CATALOG_URL_PRIMARY) return urls def wait_for_footwear_catalog_ready(driver, timeout: int = 30) -> None: """Wait for shared footwear-page elements instead of requiring product links immediately.""" def _ready(d) -> bool: """Check that the catalog UI has loaded. On the current Red Wing site, the grid of products is present in ``li.js-product-grid-item`` elements. Earlier heuristics looked for text such as "Sort By", "Results" or "Filters"; those still apply but may not appear until the user scrolls. We consider the catalog ready when either at least one product tile exists or legacy text heuristics are satisfied. """ try: # If product tiles are present, we are ready. tiles = d.find_elements(By.CSS_SELECTOR, "li.js-product-grid-item") if tiles: return True except Exception: pass try: txt = (d.find_element(By.TAG_NAME, "body").text or "").lower() except Exception: return False has_results = "result" in txt has_sort = "sort by" in txt has_filters = "filters" in txt has_more = "more results" in txt has_search = "search" in txt has_footwear_ctx = "footwear" in txt or "/footwear-" in (d.current_url or "").lower() return has_footwear_ctx and (has_sort or has_results or has_filters or has_more or has_search) WebDriverWait(driver, timeout).until(_ready) def extract_style_and_name_from_html(html: str, url: str) -> Tuple[str, str, str]: """Extract (style_number, name, style_text) from legacy and RW order-site pages.""" soup = BeautifulSoup(html, "html.parser") style_number = extract_style_from_url(url) name = "" style_text = "" guide_div = soup.find("div", class_="shoeguide") or soup.find(id="printSpacing") if guide_div: h3_tag = guide_div.find("h3") strong_tag = guide_div.find("strong") if h3_tag: style_text = h3_tag.get_text(" ", strip=True) match = re.search(r"#\s*(\d+)", style_text) if match: style_number = match.group(1) if strong_tag: name = strong_tag.get_text(" ", strip=True) if not name: for script in soup.select("script[type='application/ld+json']"): try: payload = json.loads(script.get_text(strip=True)) except Exception: continue objs = payload if isinstance(payload, list) else [payload] for obj in objs: if not isinstance(obj, dict): continue if not name: nm = (obj.get("name") or "").strip() if nm: name = nm if not style_number: for key in ("sku", "mpn", "productID"): raw = str(obj.get(key) or "").strip() m = re.search(r"\b(\d{3,6})\b", raw) if m: style_number = m.group(1) break if name and style_number: break if name and style_number: break if not name: for sel in ("h1", "h2", "title", "meta[property='og:title']"): try: el = soup.select_one(sel) except Exception: el = None if not el: continue if sel.startswith("meta"): txt = (el.get("content") or "").strip() else: txt = el.get_text(" ", strip=True) if txt and "red wing for business" not in txt.lower(): name = txt break if not style_text: style_text = f"Style #{style_number}" if style_number else (name or "") if not style_number: style_number = style_text.strip() if not name: name = style_text return style_number, name, style_text def dismiss_popups(driver) -> None: xpaths = [ "//button[contains(.,'Agree') or contains(.,'Accept') or contains(.,'Proceed')]", "//a[contains(.,'Agree') or contains(.,'Accept') or contains(.,'Proceed')]", "//button[contains(.,'Close') or contains(.,'×') or contains(@aria-label,'Close')]", ] end = time.time() + 2.0 while time.time() < end: clicked = False for xp in xpaths: try: el = driver.find_element(By.XPATH, xp) if el.is_displayed() and el.is_enabled(): el.click() clicked = True break except Exception: pass if not clicked: break time.sleep(0.2) def scroll_to_load_all(driver, max_rounds: int = 80) -> None: stable_rounds = 0 last_count = -1 for _ in range(max_rounds): # Close any transient popups. dismiss_popups(driver) clicked_more = False # Attempt to click the "More Results" button if it exists. The site renders # this as either a