import logging import re import time from typing import Callable, Any, Dict, List, Optional from urllib.parse import urljoin from browser_session import BrowserSession logger = logging.getLogger("bot") def _parse_price(text: str) -> Optional[float]: """Extract price from 'Lowest Price: $X.XX' format.""" if not text: return None try: # Match "Lowest Price: $7.49" pattern match = re.search(r"Lowest\s+Price\s*:\s*\$?\s*([\d,.]+)", text, re.IGNORECASE) if match: return float(match.group(1).replace(",", "")) # Fallback: any dollar amount match = re.search(r"\$\s*([\d,.]+)", text) if match: return float(match.group(1).replace(",", "")) # Last resort: any number match = re.search(r"([\d,.]+)", text) if match: return float(match.group(1).replace(",", "")) except Exception: pass return None class ListingScraper: def __init__( self, session: BrowserSession, min_price: float, max_items: Optional[int] = None, stop_event=None, status_callback: Optional[Callable[[Dict[str, Any]], None]] = None, max_pages: int = 10000, page_timeout: int = 30, total_timeout: int = 3600, ): self.session = session self.min_price = min_price self.max_items = max_items self.stop_event = stop_event self.status_callback = status_callback self.checked = 0 self.collected = 0 self.pages_scanned = 0 self.skipped_below_threshold = 0 self.skipped_unknown_price = 0 self.max_pages = max_pages self.page_timeout = page_timeout self.total_timeout = total_timeout self.scan_start_time = None self.last_product_found_time = None @property def page(self): return self.session.page def _get_products_on_page(self) -> List[Dict[str, Any]]: """ Extract products from the listing page with timeout protection. - Find each div.row containing a product - Extract title from div.caption h4 a - Extract price from div.price strong (Lowest Price: $X.XX) - Extract link from product href - ONLY return products where price >= min_price """ page = self.page products: List[Dict[str, Any]] = [] logger.debug("Extracting products from div.row structure") try: rows = page.locator("div.row").all() logger.debug(f"Found {len(rows)} potential product rows") for idx, row in enumerate(rows): try: try: title_elem = row.locator("div.caption h4 a").first title = (title_elem.inner_text(timeout=500) or "").strip() link = title_elem.get_attribute("href") or "" except Exception: logger.debug(f"Row {idx}: Could not extract title") continue if not title or not link: logger.debug(f"Row {idx}: Missing title or link") continue price = None try: price_elem = row.locator("div.price strong").first price_text = (price_elem.inner_text(timeout=500) or "").strip() price = _parse_price(price_text) logger.debug(f"Row {idx}: Extracted price text: {price_text} -> ${price}") except Exception as e: logger.debug(f"Row {idx}: Could not extract price: {e}") if price is None: self.skipped_unknown_price += 1 logger.debug(f"Row {idx}: Skipping '{title}' - no price found") continue if price < self.min_price: self.skipped_below_threshold += 1 logger.debug(f"Row {idx}: Skip '{title}' (${price:.2f} < ${self.min_price:.2f})") continue logger.debug(f"Row {idx}: ✓ QUALIFY '{title}' @ ${price:.2f}") products.append({"url": link, "title": title, "price": price}) except Exception as e: logger.debug(f"Row {idx}: Error processing row: {e}") continue logger.info(f"Extracted {len(products)} qualifying product(s) from page") return products except Exception as e: logger.error(f"Failed to extract products: {e}") try: self.session.screenshot("product_extraction_error") except Exception: pass return [] def _next_page_url(self, current_url: str) -> Optional[str]: """Find next page link in pagination.""" page = self.page # Prefer the 'Next' link inside pagination containers to avoid jumping to numbered anchors container_selectors = [ "nav ul.pager", "ul.pager", "ul.pagination", "div.pagination", "nav.pagination", "div.pagenav", "div.pages", "div.pager", "aside.pagination", ] def _valid_href(href: str) -> bool: if not href: return False href_l = href.lower() # Avoid direct product pages if "dvd_view_" in href_l or "/dvd_view_" in href_l: return False # Prefer listing/search pages or links that include pagination params if any(token in href_l for token in ("dvd_search.php", "search_studioid", "page=", "order_by=", "search=")): return True return False for container in container_selectors: try: cont = page.locator(container).first if cont and cont.is_visible(timeout=800): anchors = cont.locator("a").all() for a in anchors: try: href = a.get_attribute("href") or "" rel = (a.get_attribute("rel") or "").lower() title = (a.get_attribute("title") or "").lower() aria = (a.get_attribute("aria-label") or "").lower() text = (a.inner_text() or "").strip().lower() is_next = False if rel == "next": is_next = True if "next" in title or "next" in aria: is_next = True # text may contain 'next' or start with an arrow symbol if text.startswith("next") or text in (">", "»", ">>") or "next" in text: is_next = True if is_next and _valid_href(href): return urljoin(current_url, href) except Exception: continue except Exception: continue # Specific fallback: look for Next link inside nav ul.pager try: el = page.locator("nav ul.pager a:has-text('Next')").first if el and el.is_visible(timeout=800): href = el.get_attribute("href") or "" if _valid_href(href): return urljoin(current_url, href) except Exception: pass # Broad fallback: look for any anchor with rel=next or text 'Next' anywhere on page try: el = page.locator("a[rel='next']").first if el and el.is_visible(timeout=800): href = el.get_attribute("href") or "" if _valid_href(href): return urljoin(current_url, href) except Exception: pass try: el = page.locator("a:has-text('Next')").first if el and el.is_visible(timeout=800): href = el.get_attribute("href") or "" if _valid_href(href): return urljoin(current_url, href) except Exception: pass return None def _resolve_next_page_url(self, current_url: str) -> Optional[str]: """Resolve the next page URL using the page counter first, then link-based fallback.""" page = self.page try: results_el = page.locator("div.col-sm-3.results").first txt = (results_el.inner_text(timeout=800) or "").strip() m = re.search(r"Page\s*(\d+)\s*of\s*(\d+)", txt, re.IGNORECASE) if m: cur = int(m.group(1)) total = int(m.group(2)) logger.debug(f"Pagination indicator: page {cur} of {total}") if cur < total: from urllib.parse import urlparse, parse_qs, urlencode, urlunparse parsed = urlparse(current_url) qs = parse_qs(parsed.query) qs["page"] = [str(cur + 1)] new_query = urlencode(qs, doseq=True) next_url = urlunparse( (parsed.scheme, parsed.netloc, parsed.path, parsed.params, new_query, parsed.fragment) ) logger.debug(f"Sequential next page URL -> {next_url}") return next_url logger.debug("Reached last page according to indicator") return None except Exception: pass next_url = self._next_page_url(current_url) if next_url: logger.debug(f"Link-based next page URL -> {next_url}") return next_url def _resolve_next_page_url_with_timeout(self, current_url: str, timeout_ms: int = 5000) -> Optional[str]: """Resolve next page URL without crossing thread boundaries.""" try: return self._resolve_next_page_url(current_url) except Exception as e: logger.warning(f"Error resolving next page URL: {e}") return None def _load_page_with_retry(self, url: str, page_num: int, max_retries: int = 3) -> bool: """ Load a page with exponential backoff retry logic and timeout protection. Returns True if successful, False if all retries exhausted or timeout. """ retry_delays = [2, 5, 10] # Exponential backoff: 2s, 5s, 10s for attempt in range(1, max_retries + 1): try: logger.info(f"Page load attempt {attempt}/{max_retries} for page {page_num}: {url}") if self.session.goto(url): logger.info(f"✓ Successfully loaded page {page_num} on attempt {attempt}") try: self.session.cleanup_page() import gc gc.collect() except Exception as e: logger.debug(f"Cleanup warning (non-critical): {e}") return True logger.warning(f"✗ Page load failed for page {page_num}, attempt {attempt}") except Exception as e: logger.warning(f"✗ Exception during page load for page {page_num}, attempt {attempt}: {e}") if attempt < max_retries: delay = retry_delays[attempt - 1] logger.info(f"Waiting {delay}s before retry...") time.sleep(delay) logger.error(f"Failed to load page {page_num} after {max_retries} retries: {url}") return False def iter_qualifying_products(self, studio_url: str): """Stream qualifying products page by page with comprehensive timeout protection.""" current_url = studio_url page_num = 1 visited_urls = set() consecutive_failures = 0 max_consecutive_failures = 3 # Stop if 3 pages in a row fail pages_without_products = 0 max_empty_pages = 5 # Stop if 5 consecutive pages have no products self.scan_start_time = time.time() self.last_product_found_time = self.scan_start_time logger.info(f"Starting listing scan with min_price=${self.min_price:.2f}") logger.info(f"Limits: max_pages={self.max_pages}, page_timeout={self.page_timeout}s, total_timeout={self.total_timeout}s") while current_url and page_num <= self.max_pages: # Check for user stop signal if self.stop_event and self.stop_event.is_set(): logger.info("Listing scan stopped by user") break # Check if we've exceeded total scan time elapsed_time = time.time() - self.scan_start_time if elapsed_time > self.total_timeout: logger.warning(f"Total scan timeout exceeded ({elapsed_time:.0f}s > {self.total_timeout}s). Stopping scan.") break # Check for pagination loop if current_url in visited_urls: logger.warning("Pagination loop detected. Stopping scan.") break # Check if too many pages with no results if pages_without_products >= max_empty_pages: logger.warning(f"Too many empty pages ({pages_without_products}/{max_empty_pages}). Stopping scan.") break # Check for no products found in extended time if self.collected == 0 and elapsed_time > 300: # 5 minutes logger.warning("No products found after 5 minutes. Stopping scan.") break visited_urls.add(current_url) logger.info(f"Scanning listing page {page_num}: {current_url}") # Attempt to load page with retry logic if not self._load_page_with_retry(current_url, page_num, max_retries=3): consecutive_failures += 1 logger.warning(f"Page load failed. Consecutive failures: {consecutive_failures}/{max_consecutive_failures}") if consecutive_failures >= max_consecutive_failures: logger.error(f"Too many consecutive page failures ({consecutive_failures}). Stopping scan.") break # Try to proceed to next page instead of breaking try: next_url = self._resolve_next_page_url_with_timeout(current_url) if next_url: logger.info(f"Skipping failed page {page_num}, attempting next page") current_url = next_url page_num += 1 time.sleep(2) # Extra delay after failure continue else: break except Exception as e: logger.error(f"Could not resolve next page after failure: {e}") break # Reset failure counter on successful page load consecutive_failures = 0 time.sleep(1) self.pages_scanned += 1 if self.status_callback: try: self.status_callback({ "state": "Scanning listing", "current_page": page_num, "found_items": self.collected, "checked_items": self.checked, }) except Exception: pass products = self._get_products_on_page() logger.info(f" Found {len(products)} qualifying product(s) on page {page_num}") if not products: pages_without_products += 1 if pages_without_products <= 2: # Only screenshot the first couple empty pages try: self.session.screenshot(f"empty_page_{page_num}") except Exception: pass logger.warning(f"No qualifying products found on page {page_num} (empty pages: {pages_without_products}/{max_empty_pages})") else: pages_without_products = 0 self.last_product_found_time = time.time() next_url = self._resolve_next_page_url_with_timeout(current_url) if next_url: logger.debug(f"Next listing page resolved before yielding products -> {next_url}") else: logger.debug("No next page found—scan may end after this page") for p in products: if self.stop_event and self.stop_event.is_set(): logger.info("Listing scan stopped by user") return self.checked += 1 self.collected += 1 p["url"] = urljoin(current_url, p.get("url") or "") p["page_num"] = page_num p["listing_url"] = current_url if self.status_callback: try: self.status_callback({ "state": "Scanning listing", "current_page": page_num, "found_items": self.collected, "checked_items": self.checked, }) except Exception: pass yield p if self.max_items and self.collected >= self.max_items: logger.info(f"Reached max_items={self.max_items}; stopping listing scan") return if not next_url: logger.info("No next page link found. Pagination complete.") break current_url = next_url page_num += 1 # Extra safety: check elapsed time again before next iteration elapsed_time = time.time() - self.scan_start_time if elapsed_time > self.total_timeout: logger.warning(f"Total scan timeout reached ({elapsed_time:.0f}s). Stopping after {page_num-1} pages.") break def scan_listing(self, studio_url: str) -> List[Dict[str, Any]]: """ Scan all pages of the studio listing and collect qualifying products. Only returns products where price >= min_price. """ qualifying = list(self.iter_qualifying_products(studio_url)) if self.status_callback: try: self.status_callback({ "state": "Listing scan complete", "current_page": self.pages_scanned, "found_items": len(qualifying), "checked_items": self.checked, }) except Exception: pass logger.info( f"Listing scan complete: pages={self.pages_scanned}, " f"checked={self.checked}, below_min={self.skipped_below_threshold}, " f"no_price={self.skipped_unknown_price}, qualifying={len(qualifying)}" ) return qualifying