import logging import re import time from datetime import datetime from typing import Any, Dict, Optional from urllib.parse import urljoin from browser_session import BrowserSession logger = logging.getLogger("bot") BASE_URL = "https://www.adultdvdmarketplace.com/xcart/" class ProductScraper: def __init__(self, session: BrowserSession, retry_count: int = 3, stop_event=None, product_timeout: int = 20): self.session = session self.retry_count = retry_count self.stop_event = stop_event self.product_timeout = product_timeout # Timeout per product page extraction @property def page(self): return self.session.page def _meta_value_safe(self, label: str, timeout: int = 8) -> Optional[str]: """Extract metadata value with direct Playwright calls.""" try: return self._meta_value(label) except Exception as e: logger.debug(f"Meta value extraction error for '{label}': {e}") return None def _get_title_safe(self, timeout: int = 8) -> str: """Extract title with direct Playwright calls.""" try: return self._get_title() except Exception as e: logger.debug(f"Title extraction error: {e}") return "" def _get_price_safe(self, timeout: int = 8) -> str: """Extract price with direct Playwright calls.""" try: return self._get_price() except Exception as e: logger.debug(f"Price extraction error: {e}") return "" def _get_category_safe(self, timeout: int = 8) -> str: """Extract category with direct Playwright calls.""" try: return self._get_category() except Exception as e: logger.debug(f"Category extraction error: {e}") return "" # ------------------------------------------------------------------ # Metadata extraction # ------------------------------------------------------------------ def _meta_value(self, label: str) -> Optional[str]: """ Extract the value that follows a metadata label in the product detail page. Handles table rows (td/th), definition lists (dt/dd), and inline bold/span labels. """ page = self.page # XPath patterns: label cell → sibling value cell xpaths = [ f"//td[normalize-space(.)='{label}:']/following-sibling::td[1]", f"//td[normalize-space(.)='{label}']/following-sibling::td[1]", f"//th[normalize-space(.)='{label}:']/following-sibling::td[1]", f"//th[normalize-space(.)='{label}']/following-sibling::td[1]", f"//dt[normalize-space(.)='{label}:']/following-sibling::dd[1]", f"//dt[normalize-space(.)='{label}']/following-sibling::dd[1]", f"//b[normalize-space(.)='{label}:']/following-sibling::*[1]", f"//strong[normalize-space(.)='{label}:']/following-sibling::*[1]", f"//span[normalize-space(.)='{label}:']/following-sibling::span[1]", ] for xpath in xpaths: try: els = page.locator(f"xpath={xpath}").all() for el in els: text = el.inner_text(timeout=500).strip() if text: return text except Exception: continue # Fallback: regex on raw HTML try: html = page.content() pattern = rf"{re.escape(label)}[:\s]+([^\n<]{{1,150}})" m = re.search(pattern, html, re.IGNORECASE) if m: value = re.sub(r"<[^>]+>", "", m.group(1)).strip() if value: return value except Exception: pass return None def _get_title(self) -> str: """Extract title from product page H1 tag.""" page = self.page selectors = [ "div.row h1", # Direct row > h1 "h1.product-title", "h1", # Any h1 on page ".product-title h1", ".product-name h1", "#product-title", "h1[class*='title']", "h1[class*='name']", ] for sel in selectors: try: elem = page.locator(sel).first if elem: text = elem.inner_text(timeout=2000).strip() if len(text) > 2: logger.debug(f"Found H1 title via '{sel}': {text}") return text except Exception as e: logger.debug(f"H1 selector '{sel}' failed: {e}") continue logger.debug("No H1 title found") return "" def _get_price(self) -> str: page = self.page for sel in [".product-price", ".price", "span.price", "[class*='price']", ".ProductPrice"]: try: text = page.locator(sel).first.inner_text(timeout=1000).strip() if text: return text except Exception: continue return "" def _get_category(self) -> str: page = self.page # Try explicit metadata field first cat = self._meta_value("Category") or self._meta_value("Categories") if cat: return cat # Breadcrumb fallback try: crumbs = page.locator(".breadcrumb a, .breadcrumbs a, nav.breadcrumb a").all() if len(crumbs) >= 2: return " > ".join( c.inner_text(timeout=500).strip() for c in crumbs[-2:] if c.inner_text(timeout=500).strip() ) except Exception: pass return "" # ------------------------------------------------------------------ # Main scraper # ------------------------------------------------------------------ def scrape_product(self, url: str) -> Dict[str, Any]: if not url.startswith("http"): url = urljoin(BASE_URL, url) result: Dict[str, Any] = { "title": "", "upc": "", "price": "", "sku": "", "release_date": "", "category": "", "studio_name": "", "product_url": url, "scraped_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), "error": "", } for attempt in range(1, self.retry_count + 1): try: if self.stop_event and self.stop_event.is_set(): result["error"] = "Stopped by user" return result logger.debug(f"Scraping (attempt {attempt}): {url}") if not self.session.goto(url): raise RuntimeError("Navigation failed or timed out") time.sleep(0.4) if self.stop_event and self.stop_event.is_set(): result["error"] = "Stopped by user" return result # Use safe extraction methods with timeouts result["title"] = self._get_title_safe() result["price"] = self._get_price_safe() result["upc"] = (self._meta_value_safe("UPC") or "").strip() result["sku"] = ( self._meta_value_safe("SKU") or self._meta_value_safe("Product code") or self._meta_value_safe("Item #") or "" ).strip() result["studio_name"] = ( self._meta_value_safe("Studio") or self._meta_value_safe("Manufacturer") or self._meta_value_safe("Brand") or "" ).strip() result["release_date"] = ( self._meta_value_safe("Release Date") or self._meta_value_safe("Released") or "" ).strip() result["category"] = self._get_category_safe() logger.info( f" ✓ '{result['title']}' | UPC: {result['upc'] or '—'}" ) return result except Exception as e: logger.warning(f" Attempt {attempt} failed for {url}: {e}") if attempt < self.retry_count: time.sleep(2) else: result["error"] = str(e) logger.error(f" Gave up on {url}") self.session.screenshot("product_error") return result