TECHNICAL_IMPLEMENTATION.md

Technical Implementation Details

Key Code Changes

1. Listing Scraper - Thread-Based Page Extraction Timeout

Before:

def _get_products_on_page(self) -> List[Dict[str, Any]]:
    page = self.page
    products: List[Dict[str, Any]] = []
    rows = page.locator("div.row").all()  # Could hang here
    for idx, row in enumerate(rows):
        # ... extraction code ...
    return products

After:

def _get_products_on_page(self) -> List[Dict[str, Any]]:
    page = self.page
    products: List[Dict[str, Any]] = []
    
    extraction_timeout = self.page_timeout - 5  # Leave 5s buffer
    result_container = {"products": [], "error": None, "done": False}
    
    def extract_products():
        try:
            rows = page.locator("div.row").all()  # Runs in separate thread
            logger.debug(f"Found {len(rows)} potential product rows")
            for idx, row in enumerate(rows):
                # ... extraction code ...
        except Exception as e:
            result_container["error"] = str(e)
        finally:
            result_container["done"] = True
    
    # Run extraction in thread with timeout
    thread = threading.Thread(target=extract_products, daemon=True)
    thread.start()
    thread.join(timeout=extraction_timeout)  # Max 25 seconds
    
    if result_container["done"]:
        if result_container["error"]:
            logger.error(f"Product extraction error: {result_container['error']}")
        products = result_container["products"]
    else:
        logger.warning(f"Product extraction timed out after {extraction_timeout}s")
        try:
            self.session.screenshot("extraction_timeout")
        except Exception:
            pass
    
    logger.info(f"Extracted {len(products)} qualifying product(s) from page")
    return products

Why: Locator operations can hang indefinitely if the page is still loading or has JavaScript errors. Threading with timeout ensures we detect and recover from hangs.

---

2. Pagination Loop - Comprehensive Timeout System

Before:

def iter_qualifying_products(self, studio_url: str):
    current_url = studio_url
    page_num = 1
    visited_urls = set()
    max_pages = 10000  # Unlimited effectively
    
    while current_url and page_num <= max_pages:
        if current_url in visited_urls:
            logger.warning("Pagination loop detected. Stopping scan.")
            break
        
        visited_urls.add(current_url)
        products = self._get_products_on_page()
        # ... rest of loop ...

After:

def iter_qualifying_products(self, studio_url: str):
    current_url = studio_url
    page_num = 1
    visited_urls = set()
    consecutive_failures = 0
    max_consecutive_failures = 3
    pages_without_products = 0
    max_empty_pages = 5
    
    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 1: User stop signal
        if self.stop_event and self.stop_event.is_set():
            logger.info("Listing scan stopped by user")
            break
        
        # CHECK 2: Total time exceeded
        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 3: Pagination loop
        if current_url in visited_urls:
            logger.warning("Pagination loop detected. Stopping scan.")
            break
        
        # CHECK 4: Too many empty pages
        if pages_without_products >= max_empty_pages:
            logger.warning(f"Too many empty pages ({pages_without_products}/{max_empty_pages}). Stopping scan.")
            break
        
        # CHECK 5: No products found in extended time
        if self.collected == 0 and elapsed_time > 300:
            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}")
        
        # ... process page ...
        
        # CHECK 6: Page load failures
        if not self._load_page_with_retry(current_url, page_num, max_retries=3):
            consecutive_failures += 1
            if consecutive_failures >= max_consecutive_failures:
                logger.error(f"Too many consecutive page failures ({consecutive_failures}). Stopping scan.")
                break
        
        products = self._get_products_on_page()
        
        if not products:
            pages_without_products += 1
        else:
            pages_without_products = 0
            self.last_product_found_time = time.time()
        
        # ... yield products ...
        
        current_url = next_url
        page_num += 1

Why: Multiple checks create a safety net. If one check fails, others catch the problem. Total timeout prevents infinite loops on any individual check failure.

---

3. Page Loading - Thread-Protected Navigation

Before:

def _load_page_with_retry(self, url: str, page_num: int, max_retries: int = 3) -> bool:
    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):  # Could hang here
                logger.info(f"✓ Successfully loaded page {page_num} on attempt {attempt}")
                try:
                    self.session.cleanup_page()
                except Exception as e:
                    logger.debug(f"Cleanup warning: {e}")
                return True
            else:
                logger.warning(f"✗ Page load failed for page {page_num}, attempt {attempt}")
        except Exception as e:
            logger.warning(f"✗ Exception during page load: {e}")
        
        if attempt < max_retries:
            delay = retry_delays[attempt - 1]
            logger.info(f"Waiting {delay}s before retry...")
            time.sleep(delay)
    
    return False

After:

def _load_page_with_retry(self, url: str, page_num: int, max_retries: int = 3) -> bool:
    retry_delays = [2, 5, 10]
    
    for attempt in range(1, max_retries + 1):
        try:
            logger.info(f"Page load attempt {attempt}/{max_retries} for page {page_num}: {url}")
            
            # Load page with timeout thread
            load_result = {"success": False, "done": False}
            
            def load_thread():
                try:
                    load_result["success"] = self.session.goto(url)
                except Exception as e:
                    logger.warning(f"Navigation exception: {e}")
                finally:
                    load_result["done"] = True
            
            thread = threading.Thread(target=load_thread, daemon=True)
            thread.start()
            thread.join(timeout=self.page_timeout)  # Max 30 seconds
            
            if load_result["done"] and load_result["success"]:
                logger.info(f"✓ Successfully loaded page {page_num} on attempt {attempt}")
                try:
                    self.session.cleanup_page()
                    import gc
                    gc.collect()  # Force garbage collection
                except Exception as e:
                    logger.debug(f"Cleanup warning: {e}")
                return True
            else:
                if not load_result["done"]:
                    logger.warning(f"✗ Page load timed out after {self.page_timeout}s for page {page_num}, attempt {attempt}")
                else:
                    logger.warning(f"✗ Page load failed for page {page_num}, attempt {attempt}")
        except Exception as e:
            logger.warning(f"✗ Exception during page load: {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

Why: Threading ensures Playwright navigation doesn't hang. If network is slow or server doesn't respond, we detect it and move on after timeout instead of waiting forever.

---

4. Product Scraper - Timeout-Protected Field Extraction

Before:

def scrape_product(self, url: str) -> Dict[str, Any]:
    # ...
    result["title"]        = self._get_title()  # Could hang
    result["price"]        = self._get_price()  # Could hang
    result["upc"]          = (self._meta_value("UPC") or "").strip()  # Could hang
    result["category"]     = self._get_category()  # Could hang
    # ...

After:

def scrape_product(self, url: str) -> Dict[str, Any]:
    # ...
    result["title"]        = self._get_title_safe()  # 8s timeout
    result["price"]        = self._get_price_safe()  # 8s timeout
    result["upc"]          = (self._meta_value_safe("UPC") or "").strip()  # 8s timeout
    result["category"]     = self._get_category_safe()  # 8s timeout
    # ...

def _get_title_safe(self, timeout: int = 8) -> str:
    """Extract title with timeout protection."""
    result = {"value": "", "done": False}
    
    def extract():
        try:
            result["value"] = self._get_title()
        except Exception as e:
            logger.debug(f"Title extraction error: {e}")
        finally:
            result["done"] = True
    
    thread = threading.Thread(target=extract, daemon=True)
    thread.start()
    thread.join(timeout=timeout)
    
    if not result["done"]:
        logger.warning("Title extraction timed out")
    return result["value"]

Why: Individual fields might hang on slow servers. Timeout per field prevents one slow field from blocking entire scan.

---

5. Browser Session - Enhanced Cleanup

Before:

def cleanup_page(self):
    try:
        if self.page:
            self.page.evaluate("() => { localStorage.clear(); sessionStorage.clear(); }")
            logger.debug("Cleared page storage")
    except Exception as e:
        logger.debug(f"Page cleanup warning: {e}")

After:

def cleanup_page(self):
    """Cleanup page resources to prevent memory accumulation during long pagination runs."""
    try:
        if self.page:
            self.page.evaluate("""
                () => { 
                    localStorage.clear(); 
                    sessionStorage.clear();
                    // Purge cache
                    if (window.caches) {
                        caches.keys().then(names => {
                            names.forEach(name => caches.delete(name));
                        });
                    }
                }
            """)
            logger.debug("Cleared page storage and cache")
    except Exception as e:
        logger.debug(f"Page cleanup warning: {e}")

Why: On memory-constrained live servers, every bit helps. Clearing caches and storage prevents memory accumulation across pages.

---

Configuration Examples

Hugging Face Space (Limited Resources)

ListingScraper(
    session=session,
    min_price=self.min_price,
    max_pages=50,          # Conservative
    page_timeout=15,       # Shorter timeouts
    total_timeout=900,     # 15 minutes max
)

Local Development (Full Resources)

ListingScraper(
    session=session,
    min_price=self.min_price,
    max_pages=100,         # Default
    page_timeout=30,       # Default
    total_timeout=1800,    # 30 minutes
)

High-Traffic Sites (Slow Servers)

ListingScraper(
    session=session,
    min_price=self.min_price,
    max_pages=100,
    page_timeout=45,       # Longer for slow servers
    total_timeout=2400,    # 40 minutes
)

---

Troubleshooting

If scan still hangs:

1. Lower page_timeout to detect hangs faster
2. Lower max_pages limit
3. Lower total_timeout to terminate early
4. Check network connectivity
5. Look at screenshot files in logs/ directory for visual debugging

If scan terminates too early:

1. Increase page_timeout
2. Increase total_timeout
3. Check logs for "Page load timed out" messages
4. Consider increasing max_pages if legitimate pages are being skipped

Memory issues on Hugging Face:

1. Use conservative settings (see above)
2. Ensure cleanup_page() is called (happens automatically)
3. Monitor with smaller max_pages values
4. Consider running multiple smaller scans instead of one large scan