From 23afe4bc6caf0ea1dce84835835b6951de01ebe1 Mon Sep 17 00:00:00 2001 From: Michael SCHAL Date: Sun, 30 Nov 2025 12:39:26 +0100 Subject: [PATCH 1/3] feat: Add new services for search orchestration, browser automation, and scheduling. --- app/services/browserless_service.py | 492 +++++++++++++++----------- app/services/cataloguemate_scraper.py | 6 +- app/services/scheduler_service.py | 4 +- app/services/search_service.py | 8 +- 4 files changed, 295 insertions(+), 215 deletions(-) diff --git a/app/services/browserless_service.py b/app/services/browserless_service.py index e9b3723..d206d9b 100644 --- a/app/services/browserless_service.py +++ b/app/services/browserless_service.py @@ -1,77 +1,139 @@ """ -Browserless Service -Manages connections to Browserless.io with advanced stealth capabilities. -Handles: -- Connection lifecycle -- Stealth injection -- Proxy rotation -- Human behavior simulation +Browserless Service - Enhanced Version +Manages connections to Browserless.io with advanced robustness features: +- Auto-reconnection on browser disconnection +- Advanced popup handling +- Flexible configuration +- Thread-safe operations """ import asyncio import logging -import random -import time -from typing import Optional, Any +import os +from dataclasses import dataclass +from datetime import datetime + +from playwright.async_api import Browser, BrowserContext, Page, async_playwright +from playwright.async_api import TimeoutError as PlaywrightTimeoutError -from playwright.async_api import async_playwright, Browser, BrowserContext, Page from app.core.search_config import ( - BROWSERLESS_URL, - get_amazon_proxies, + BROWSERLESS_URL, get_random_user_agent, - COOKIE_ACCEPT_SELECTORS ) logger = logging.getLogger(__name__) -# Stealth JS to inject -STEALTH_JS = """ -// Advanced Stealth Mode -Object.defineProperty(navigator, 'webdriver', { get: () => false }); -delete Object.getPrototypeOf(navigator).webdriver; -window.chrome = { runtime: {} }; -Object.defineProperty(navigator, 'plugins', { get: () => [1, 2, 3, 4, 5] }); -Object.defineProperty(navigator, 'languages', { get: () => ['fr-FR', 'fr', 'en-US', 'en'] }); -""" +# Comprehensive popup selectors +POPUP_SELECTORS = [ + "button[aria-label='Close']", + "button[aria-label='close']", + "button[aria-label='Fermer']", + ".close-button", + ".modal-close", + "svg[data-name='Close']", + "[class*='popup'] button", + "[class*='modal'] button", + "button:has-text('No, thanks')", + "button:has-text('No thanks')", + "a:has-text('No, thanks')", + "div[role='dialog'] button[aria-label='Close']", + # Cookie banners (kept for compatibility) + "#sp-cc-accept", + "#onetrust-accept-btn-handler", + "button:has-text('Tout accepter')", + "button:has-text('Accepter')", +] + + +@dataclass +class ScrapeConfig: + """Configuration for scraping parameters.""" + + smart_scroll: bool = False + scroll_pixels: int = 350 + text_length: int = 0 + timeout: int = 90000 + class BrowserlessService: - def __init__(self): - self._playwright = None - self._browser: Optional[Browser] = None - self._proxies = get_amazon_proxies() + """Enhanced browserless service with auto-reconnection and robust error handling.""" - async def start(self): - """Initialize the browser connection""" - if not self._playwright: - self._playwright = await async_playwright().start() - - if not self._browser or not self._browser.is_connected(): + _playwright = None + _browser: Browser | None = None + _lock = asyncio.Lock() + + @classmethod + async def initialize(cls): + """Initialize the shared browser instance (Public, Thread-Safe).""" + async with cls._lock: + await cls._initialize() + + @classmethod + async def _initialize(cls): + """Internal initialization logic (Assumes lock is held).""" + if cls._browser is None: + logger.info("Initializing BrowserlessService shared browser...") + cls._playwright = await async_playwright().start() + cls._browser = await cls._connect_browser(cls._playwright) + logger.info("BrowserlessService initialized.") + + @classmethod + async def shutdown(cls): + """Shutdown the shared browser instance.""" + async with cls._lock: + if cls._browser: + logger.info("Shutting down BrowserlessService shared browser...") + await cls._browser.close() + cls._browser = None + if cls._playwright: + await cls._playwright.stop() + cls._playwright = None + logger.info("BrowserlessService shutdown complete.") + + @classmethod + async def _ensure_browser_connected(cls) -> bool: + """Ensure browser is connected, reconnect if needed.""" + async with cls._lock: try: - logger.info(f"Connecting to Browserless at {BROWSERLESS_URL}") - # Use the stealth endpoint if possible, otherwise standard - # Note: /chromium/stealth might need specific token or config, - # falling back to standard connect_over_cdp which is safer for generic browserless - self._browser = await self._playwright.chromium.connect_over_cdp( - BROWSERLESS_URL, - timeout=60000 - ) + if cls._browser is None: + logger.warning("Browser not initialized, initializing...") + await cls._initialize() + return cls._browser is not None + + # Test if browser is still alive by trying to create a context + try: + test_context = await cls._browser.new_context() + await test_context.close() + return True + except Exception as e: + logger.error(f"Browser connection test failed: {e}") + logger.info("Attempting to reconnect browser...") + # Clear the old browser + cls._browser = None + if cls._playwright: + try: + await cls._playwright.stop() + except Exception: + pass + cls._playwright = None + # Reconnect + await cls._initialize() + return cls._browser is not None except Exception as e: - logger.error(f"Failed to connect to Browserless: {e}") - raise + logger.error(f"Failed to ensure browser connection: {e}") + return False - async def stop(self): - """Close resources""" - if self._browser: - await self._browser.close() - if self._playwright: - await self._playwright.stop() + @staticmethod + async def _connect_browser(p) -> Browser: + """Connect to Browserless.""" + logger.info(f"Connecting to Browserless at {BROWSERLESS_URL}") + return await p.chromium.connect_over_cdp(BROWSERLESS_URL, timeout=60000) - async def get_context(self, use_proxy: bool = False) -> BrowserContext: - """Create a new context with stealth settings""" - await self.start() - + @staticmethod + async def _create_context(browser: Browser, use_proxy: bool = False) -> BrowserContext: + """Create a new browser context with stealth settings.""" user_agent = get_random_user_agent() - + options = { "viewport": {"width": 1920, "height": 1080}, "user_agent": user_agent, @@ -83,174 +145,192 @@ class BrowserlessService: "Accept-Language": "fr-FR,fr;q=0.9,en-US;q=0.8,en;q=0.7", "Sec-Ch-Ua-Platform": '"Windows"', "Upgrade-Insecure-Requests": "1", - } + }, } - if use_proxy and self._proxies: - proxy = random.choice(self._proxies) - options["proxy"] = proxy - logger.info(f"Using proxy: {proxy['server'].split('//')[1]}") + # Proxy support (can be enhanced with proxy pool in future) + if use_proxy: + logger.info("Proxy support requested (not implemented in this version)") - context = await self._browser.new_context(**options) - - # Block aggressive tracking but keep images for visual verification if needed - # (Optimized for speed vs detection) + context = await browser.new_context(**options) + + # Block aggressive tracking but keep images await context.route("**/*", lambda route: route.continue_()) - + return context - async def simulate_human_behavior(self, page: Page): - """Simulate random mouse movements and scrolling""" + @staticmethod + async def _navigate_and_wait(page: Page, url: str, timeout: int): + """Navigate to URL and wait for page load.""" + logger.info(f"Navigating to {url} (Timeout: {timeout}ms)") try: - # Mouse movements - for _ in range(random.randint(2, 5)): - await page.mouse.move( - random.randint(100, 1800), - random.randint(100, 900) - ) - await asyncio.sleep(random.uniform(0.1, 0.3)) - - # Scrolling - await page.evaluate("window.scrollBy(0, 300)") - await asyncio.sleep(random.uniform(0.5, 1.0)) - await page.evaluate("window.scrollBy(0, -100)") - except Exception: - pass + await page.goto(url, wait_until="domcontentloaded", timeout=timeout) + logger.info(f"Page loaded (domcontentloaded): {url}") - async def handle_popups(self, page: Page): - """Attempt to close cookie banners and popups""" - try: - # Check for multiple popups (cookies + promo) - # We iterate through all selectors and click any that are visible - # We do this in a loop to handle sequential popups - for _ in range(3): # Try up to 3 times for sequential popups - clicked_something = False - for selector in COOKIE_ACCEPT_SELECTORS: - try: - # Use a short timeout for check - element = page.locator(selector).first - if await element.is_visible(timeout=100): - await element.click() - logger.info(f"Closed popup/banner with {selector}") - await asyncio.sleep(0.5) # Wait for animation - clicked_something = True - except Exception: - continue - - if not clicked_something: - break - except Exception: - pass - - async def get_page_content(self, url: str, use_proxy: bool = False, wait_selector: str = None) -> tuple[str, str]: - """ - Fetch page content with full stealth lifecycle. - Returns (html_content, screenshot_path) - """ - context = await self.get_context(use_proxy=use_proxy) - page = await context.new_page() - - try: - logger.info(f"Navigating to {url}") - # Random delay before start - await asyncio.sleep(random.uniform(0.5, 1.5)) - - response = await page.goto(url, wait_until="domcontentloaded", timeout=60000) - - if response.status == 503: - logger.warning(f"503 Detected on {url}") - # Retry logic could be handled here or by caller, - # for now we return what we have, caller decides - - # Handle pre-search interaction - from app.core.search_config import SITE_CONFIGS - for config in SITE_CONFIGS.values(): - if config["search_url"].split("/")[2] in url: - # Handle complex interaction list - if config.get("pre_search_interaction"): - try: - logger.info(f"Executing pre-search interaction sequence for {config['name']}") - for step in config["pre_search_interaction"]: - step_type = step.get("type") - selector = step.get("selector") - - if step_type == "input": - logger.info(f"Inputting text into {selector}") - await page.fill(selector, step["value"]) - - elif step_type == "click": - logger.info(f"Clicking {selector}") - # Use force=True to bypass potential overlays - await page.click(selector, timeout=5000) - - elif step_type == "wait": - seconds = step.get("seconds", 1) - logger.info(f"Waiting {seconds}s") - await asyncio.sleep(seconds) - - logger.info("Pre-search sequence completed") - await asyncio.sleep(2) # Final stabilization wait - except Exception as e: - logger.warning(f"Pre-search sequence failed: {e}") - - # Handle simple selector (legacy support) - elif config.get("pre_search_selector"): - try: - selector = config["pre_search_selector"] - logger.info(f"Attempting pre-search interaction: {selector}") - if await page.locator(selector).is_visible(timeout=5000): - await page.click(selector) - logger.info("Clicked pre-search selector") - await asyncio.sleep(2) # Wait for transition - except Exception as e: - logger.warning(f"Pre-search interaction failed: {e}") - break - - await self.handle_popups(page) - await self.simulate_human_behavior(page) - - if wait_selector: - try: - await page.wait_for_selector(wait_selector, timeout=10000) - except Exception: - logger.warning(f"Wait selector {wait_selector} timed out") - - # Wait for network idle to ensure dynamic content loads try: await page.wait_for_load_state("networkidle", timeout=5000) + logger.info("Network idle reached") + except PlaywrightTimeoutError: + logger.info("Network idle timed out (non-critical), proceeding...") + + except Exception as e: + logger.warning(f"Navigation warning for {url}: {e}") + + # Wait a bit for dynamic content + await page.wait_for_timeout(2000) + + @staticmethod + async def _handle_popups(page: Page): + """Attempt to close popups and cookie banners.""" + logger.info("Attempting to close popups...") + for popup_selector in POPUP_SELECTORS: + try: + if await page.locator(popup_selector).count() > 0: + logger.info(f"Found popup close button: {popup_selector}") + await page.locator(popup_selector).first.click(timeout=2000) + await page.wait_for_timeout(1000) except Exception: pass - content = await page.content() - - # Take screenshot if needed for AI analysis - screenshot_path = "" - try: - import os - from datetime import datetime - - # Ensure directory exists - screenshots_dir = "/app/screenshots" - if not os.path.exists(screenshots_dir): - os.makedirs(screenshots_dir, exist_ok=True) - - timestamp = int(time.time() * 1000) - safe_name = "".join(c if c.isalnum() else "_" for c in url.split("//")[-1])[:50] - filename = f"{safe_name}_{timestamp}.jpg" - screenshot_path = f"{screenshots_dir}/{filename}" - - await page.screenshot(path=screenshot_path, full_page=False, quality=80, type="jpeg") - logger.info(f"Screenshot saved to {screenshot_path}") - except Exception as e: - logger.warning(f"Failed to take screenshot: {e}") + try: + await page.keyboard.press("Escape") + except Exception: + pass - return content, screenshot_path + @staticmethod + async def _wait_for_selector(page: Page, selector: str): + """Wait for selector and scroll into view.""" + try: + logger.info(f"Waiting for selector: {selector}") + await page.wait_for_selector(selector, timeout=5000) + element = page.locator(selector).first + await element.scroll_into_view_if_needed() + logger.info(f"Scrolled to selector: {selector}") + except Exception as e: + logger.warning(f"Selector {selector} not found or timed out: {e}") + + @staticmethod + async def _auto_detect_price(page: Page): + """Attempt to auto-detect and scroll to price element.""" + logger.info("No selector provided. Attempting to find price element...") + try: + price_locator = page.locator("text=/$[0-9,]+(\\.[0-9]{2})?/") + if await price_locator.count() > 0: + await price_locator.first.scroll_into_view_if_needed() + logger.info("Scrolled to potential price element") + except Exception as e: + logger.warning(f"Auto-price detection failed: {e}") + + @staticmethod + async def _smart_scroll(page: Page, scroll_pixels: int): + """Perform smart scrolling.""" + logger.info(f"Performing smart scroll ({scroll_pixels}px)...") + try: + await page.evaluate(f"window.scrollBy(0, {scroll_pixels})") + await page.wait_for_timeout(1000) + except Exception as e: + logger.warning(f"Smart scroll failed: {e}") + + @staticmethod + async def _extract_text(page: Page, text_length: int) -> str: + """Extract text from page body.""" + if text_length <= 0: + return "" + + try: + logger.info(f"Extracting text (limit: {text_length} chars)...") + raw_text = await page.inner_text("body") + page_text = raw_text[:text_length] + logger.info(f"Extracted {len(page_text)} characters") + return page_text + except Exception as e: + logger.error(f"Text extraction failed: {e}") + return "" + + @staticmethod + async def _take_screenshot(page: Page, url: str, item_id: int | None = None) -> str: + """Take screenshot and save to disk.""" + screenshot_dir = "/app/screenshots" + os.makedirs(screenshot_dir, exist_ok=True) + + if item_id: + filename = f"{screenshot_dir}/item_{item_id}.png" + else: + url_part = url.split("//")[-1].replace("/", "_") + timestamp = datetime.now().timestamp() + filename = f"{screenshot_dir}/{url_part}_{timestamp}.png" + + await page.screenshot(path=filename, full_page=False) + logger.info(f"Screenshot saved to {filename}") + return filename + + @classmethod + async def get_page_content( + cls, + url: str, + use_proxy: bool = False, + wait_selector: str | None = None, + config: ScrapeConfig | None = None, + ) -> tuple[str, str]: + """ + Fetch page content with full stealth lifecycle. + + Args: + url: URL to scrape + use_proxy: Whether to use proxy (reserved for future use) + wait_selector: Optional CSS selector to wait for + config: Optional scraping configuration + + Returns: + tuple[screenshot_path, page_text]: Screenshot path and extracted text + """ + if config is None: + config = ScrapeConfig() + + # Input validation & defaults + scroll_pixels = max(350, config.scroll_pixels if config.scroll_pixels > 0 else 350) + timeout = max(30000, config.timeout if config.timeout > 0 else 90000) + + # Ensure browser is connected and healthy + if not await cls._ensure_browser_connected(): + logger.error("Failed to establish browser connection") + return "", "" + + try: + context = await cls._create_context(cls._browser, use_proxy=use_proxy) + page = await context.new_page() + + try: + await cls._navigate_and_wait(page, url, timeout) + await cls._handle_popups(page) + + if wait_selector: + await cls._wait_for_selector(page, wait_selector) + else: + await cls._auto_detect_price(page) + + if config.smart_scroll: + await cls._smart_scroll(page, scroll_pixels) + + page_text = await cls._extract_text(page, config.text_length) + + # Extract item_id from URL if it matches pattern + item_id = None + if "item_" in url: + # This is a simplified extraction, adjust based on your URL patterns + pass + + screenshot_path = await cls._take_screenshot(page, url, item_id) + + return screenshot_path, page_text + + finally: + await context.close() except Exception as e: - logger.error(f"Error fetching {url}: {e}") + logger.error(f"Error scraping {url}: {e}") return "", "" - finally: - await context.close() -# Global instance + +# Global instance for backward compatibility browserless_service = BrowserlessService() diff --git a/app/services/cataloguemate_scraper.py b/app/services/cataloguemate_scraper.py index 00fc51f..7c012d3 100644 --- a/app/services/cataloguemate_scraper.py +++ b/app/services/cataloguemate_scraper.py @@ -41,7 +41,7 @@ async def scrape_catalog_list(enseigne: Enseigne) -> list[dict[str, Any]]: logger.info(f"Scraping catalog list from: {url}") # Use browserless service to get content - html_content, _ = await browserless_service.get_page_content(url) + _, html_content = await browserless_service.get_page_content(url) if not html_content: logger.error(f"Failed to fetch list {url}") @@ -111,7 +111,7 @@ async def scrape_catalog_pages(catalog_url: str) -> list[dict[str, Any]]: # IMPORTANT: Pagination links only appear on page 2+, not on page 1 # So we fetch page 2 first to detect the max pages page_2_url = f"{catalog_url}?page=2" - page_2_html, _ = await browserless_service.get_page_content(page_2_url) + _, page_2_html = await browserless_service.get_page_content(page_2_url) if page_2_html: soup = BeautifulSoup(page_2_html, 'html.parser') @@ -162,7 +162,7 @@ async def scrape_catalog_pages(catalog_url: str) -> list[dict[str, Any]]: if page_num == 2 and page_2_html: html_content = page_2_html else: - html_content, _ = await browserless_service.get_page_content(current_url) + _, html_content = await browserless_service.get_page_content(current_url) if not html_content: logger.warning(f"Failed to fetch page {page_num}") diff --git a/app/services/scheduler_service.py b/app/services/scheduler_service.py index 04eef86..b38ec97 100644 --- a/app/services/scheduler_service.py +++ b/app/services/scheduler_service.py @@ -105,9 +105,9 @@ async def process_item_check(item_id: int): logger.info(f"Checking item: {item_data['name']} ({item_data['url']})") # Use new browserless service # Note: smart_scroll is handled internally by browserless_service - page_text, screenshot_path = await browserless_service.get_page_content( + screenshot_path, page_text = await browserless_service.get_page_content( item_data["url"], - use_proxy="amazon" in item_data["url"], # Simple heuristic for now + use_proxy="amazon" in item_data["url"], # Simple heuristic for now wait_selector=item_data["selector"] ) diff --git a/app/services/search_service.py b/app/services/search_service.py index 779cce4..3079e86 100644 --- a/app/services/search_service.py +++ b/app/services/search_service.py @@ -68,7 +68,7 @@ class NewSearchService: # Use proxy if required by config use_proxy = config.get("requires_proxy", False) - html_content, _ = await browserless_service.get_page_content( + _, html_content = await browserless_service.get_page_content( search_url, use_proxy=use_proxy, wait_selector=config.get("wait_selector") @@ -96,7 +96,7 @@ class NewSearchService: use_proxy = config.get("requires_proxy", False) if config else False # Use browserless to get content and screenshot - html, screenshot_path = await browserless_service.get_page_content( + screenshot_path, html = await browserless_service.get_page_content( result.url, use_proxy=use_proxy, wait_selector=None @@ -140,7 +140,7 @@ class NewSearchService: # Use proxy if required by config use_proxy = config.get("requires_proxy", False) - html_content, _ = await browserless_service.get_page_content( + _, html_content = await browserless_service.get_page_content( search_url, use_proxy=use_proxy, wait_selector=config.get("wait_selector") @@ -285,7 +285,7 @@ class NewSearchService: # Use proxy if required by config use_proxy = config.get("requires_proxy", False) - html_content, _ = await browserless_service.get_page_content( + _, html_content = await browserless_service.get_page_content( search_url, use_proxy=use_proxy, wait_selector=config.get("wait_selector") From 93d73dbed87804d62251d39df24e1605f4c56848 Mon Sep 17 00:00:00 2001 From: Michael SCHAL Date: Sun, 30 Nov 2025 12:43:22 +0100 Subject: [PATCH 2/3] feat: Implement BrowserlessService for robust browser automation with auto-reconnection, popup handling, and price extraction capabilities. --- app/services/browserless_service.py | 293 ++++++++++++++++++---------- 1 file changed, 194 insertions(+), 99 deletions(-) diff --git a/app/services/browserless_service.py b/app/services/browserless_service.py index d206d9b..9ed981b 100644 --- a/app/services/browserless_service.py +++ b/app/services/browserless_service.py @@ -5,11 +5,14 @@ Manages connections to Browserless.io with advanced robustness features: - Advanced popup handling - Flexible configuration - Thread-safe operations +- Amazon & Generic price extraction """ import asyncio import logging import os +import re +import time from dataclasses import dataclass from datetime import datetime @@ -37,7 +40,7 @@ POPUP_SELECTORS = [ "button:has-text('No thanks')", "a:has-text('No, thanks')", "div[role='dialog'] button[aria-label='Close']", - # Cookie banners (kept for compatibility) + # Cookie banners "#sp-cc-accept", "#onetrust-accept-btn-handler", "button:has-text('Tout accepter')", @@ -100,7 +103,7 @@ class BrowserlessService: await cls._initialize() return cls._browser is not None - # Test if browser is still alive by trying to create a context + # Test if browser is still alive try: test_context = await cls._browser.new_context() await test_context.close() @@ -108,7 +111,6 @@ class BrowserlessService: except Exception as e: logger.error(f"Browser connection test failed: {e}") logger.info("Attempting to reconnect browser...") - # Clear the old browser cls._browser = None if cls._playwright: try: @@ -116,7 +118,6 @@ class BrowserlessService: except Exception: pass cls._playwright = None - # Reconnect await cls._initialize() return cls._browser is not None except Exception as e: @@ -148,13 +149,10 @@ class BrowserlessService: }, } - # Proxy support (can be enhanced with proxy pool in future) if use_proxy: logger.info("Proxy support requested (not implemented in this version)") context = await browser.new_context(**options) - - # Block aggressive tracking but keep images await context.route("**/*", lambda route: route.continue_()) return context @@ -176,7 +174,6 @@ class BrowserlessService: except Exception as e: logger.warning(f"Navigation warning for {url}: {e}") - # Wait a bit for dynamic content await page.wait_for_timeout(2000) @staticmethod @@ -198,71 +195,126 @@ class BrowserlessService: pass @staticmethod - async def _wait_for_selector(page: Page, selector: str): - """Wait for selector and scroll into view.""" + async def _extract_amazon_price(page: Page) -> str: + """Extract price from Amazon product page.""" + price_selectors = [ + ".a-price .a-offscreen", + "#corePrice_desktop .a-price .a-offscreen", + "#corePriceDisplay_desktop_feature_div .a-price .a-offscreen", + ".a-price[data-a-color='price'] .a-offscreen", + "#priceblock_ourprice", + "#priceblock_dealprice", + "span.a-price-whole", + ] + + for selector in price_selectors: + try: + element = page.locator(selector).first + price_text = await element.inner_text(timeout=2000) + if price_text and price_text.strip(): + logger.info(f"Extracted Amazon price via {selector}: {price_text}") + return price_text.strip() + except Exception: + continue + + # Try combination: whole + fraction try: - logger.info(f"Waiting for selector: {selector}") - await page.wait_for_selector(selector, timeout=5000) - element = page.locator(selector).first - await element.scroll_into_view_if_needed() - logger.info(f"Scrolled to selector: {selector}") - except Exception as e: - logger.warning(f"Selector {selector} not found or timed out: {e}") + whole = await page.locator("span.a-price-whole").first.inner_text() + fraction = await page.locator("span.a-price-fraction").first.inner_text() + if whole and fraction: + price_text = f"{whole}{fraction}" + logger.info(f"Extracted Amazon price from whole+fraction: {price_text}") + return price_text + except Exception: + pass + + logger.warning("Could not extract Amazon price") + return "" @staticmethod - async def _auto_detect_price(page: Page): - """Attempt to auto-detect and scroll to price element.""" - logger.info("No selector provided. Attempting to find price element...") + async def _extract_generic_price(page: Page) -> str: + """Extract price from generic e-commerce pages.""" + price_selectors = [ + "[itemprop='price']", + "[data-testid='price']", + "[data-test='price']", + ".current-price", + ".sale-price", + ".final-price", + ".product-price", + ".special-price", + "[data-price]", + ".price-current", + ".price-now", + ".price-sales", + "[class*='price']:not([class*='old']):not([class*='was']):not([class*='original']):not([class*='before']):not([class*='regular']):not([class*='strike']):not([class*='barre'])", + "#price", + "#product-price", + "span[class*='prix']:not([class*='ancien']):not([class*='barre'])", + ".prix-actuel", + ".price:not(.old-price):not(.was-price)", + ] + + found_prices = [] + + for selector in price_selectors: + try: + elements = page.locator(selector) + count = await elements.count() + + for i in range(min(count, 3)): + try: + element = elements.nth(i) + if await element.is_visible(timeout=1000): + # Check if strikethrough + try: + text_decoration = await element.evaluate("el => window.getComputedStyle(el).textDecoration") + if "line-through" in text_decoration: + continue + except Exception: + pass + + price_text = await element.inner_text() + if price_text and ('€' in price_text or (',' in price_text and any(c.isdigit() for c in price_text))): + price_text = price_text.strip() + + # Validate price range + numeric_match = re.search(r'(\d+[.,]?\d*)', price_text.replace(' ', '').replace('\xa0', '')) + if numeric_match: + try: + price_val = float(numeric_match.group(1).replace(',', '.')) + if 0.01 <= price_val <= 100000: + found_prices.append((selector, price_text)) + logger.info(f"Found valid price via {selector}: {price_text}") + + # Return high-priority immediately + if selector in ["[itemprop='price']", "[data-testid='price']", ".current-price", ".sale-price", ".final-price", ".product-price"]: + return price_text + except (ValueError, AttributeError): + continue + except Exception: + continue + except Exception: + continue + + if found_prices: + best_price = found_prices[-1][1] + if len(found_prices) > 1: + logger.info(f"Multiple prices found, selecting last: {best_price}") + return best_price + + # Regex fallback try: - price_locator = page.locator("text=/$[0-9,]+(\\.[0-9]{2})?/") - if await price_locator.count() > 0: - await price_locator.first.scroll_into_view_if_needed() - logger.info("Scrolled to potential price element") - except Exception as e: - logger.warning(f"Auto-price detection failed: {e}") + all_text = await page.inner_text('body') + price_matches = re.findall(r'\d+[,\.]\d{2}\s*€', all_text) + if price_matches: + logger.info(f"Found price via regex: {price_matches[0]}") + return price_matches[0] + except Exception: + pass - @staticmethod - async def _smart_scroll(page: Page, scroll_pixels: int): - """Perform smart scrolling.""" - logger.info(f"Performing smart scroll ({scroll_pixels}px)...") - try: - await page.evaluate(f"window.scrollBy(0, {scroll_pixels})") - await page.wait_for_timeout(1000) - except Exception as e: - logger.warning(f"Smart scroll failed: {e}") - - @staticmethod - async def _extract_text(page: Page, text_length: int) -> str: - """Extract text from page body.""" - if text_length <= 0: - return "" - - try: - logger.info(f"Extracting text (limit: {text_length} chars)...") - raw_text = await page.inner_text("body") - page_text = raw_text[:text_length] - logger.info(f"Extracted {len(page_text)} characters") - return page_text - except Exception as e: - logger.error(f"Text extraction failed: {e}") - return "" - - @staticmethod - async def _take_screenshot(page: Page, url: str, item_id: int | None = None) -> str: - """Take screenshot and save to disk.""" - screenshot_dir = "/app/screenshots" - os.makedirs(screenshot_dir, exist_ok=True) - - if item_id: - filename = f"{screenshot_dir}/item_{item_id}.png" - else: - url_part = url.split("//")[-1].replace("/", "_") - timestamp = datetime.now().timestamp() - filename = f"{screenshot_dir}/{url_part}_{timestamp}.png" - - await page.screenshot(path=filename, full_page=False) - logger.info(f"Screenshot saved to {filename}") - return filename + logger.warning("Could not extract generic price") + return "" @classmethod async def get_page_content( @@ -270,28 +322,21 @@ class BrowserlessService: url: str, use_proxy: bool = False, wait_selector: str | None = None, - config: ScrapeConfig | None = None, + extract_text: bool = False ) -> tuple[str, str]: """ Fetch page content with full stealth lifecycle. Args: - url: URL to scrape - use_proxy: Whether to use proxy (reserved for future use) - wait_selector: Optional CSS selector to wait for - config: Optional scraping configuration + url: URL to fetch + use_proxy: Whether to use proxy + wait_selector: CSS selector to wait for + extract_text: If True, returns visible text; if False, returns HTML Returns: - tuple[screenshot_path, page_text]: Screenshot path and extracted text + tuple[content, screenshot_path]: Content (HTML or text) and screenshot path """ - if config is None: - config = ScrapeConfig() - - # Input validation & defaults - scroll_pixels = max(350, config.scroll_pixels if config.scroll_pixels > 0 else 350) - timeout = max(30000, config.timeout if config.timeout > 0 else 90000) - - # Ensure browser is connected and healthy + # Ensure browser connected if not await cls._ensure_browser_connected(): logger.error("Failed to establish browser connection") return "", "" @@ -301,28 +346,78 @@ class BrowserlessService: page = await context.new_page() try: - await cls._navigate_and_wait(page, url, timeout) + await cls._navigate_and_wait(page, url, 90000) await cls._handle_popups(page) - if wait_selector: - await cls._wait_for_selector(page, wait_selector) + # Amazon-specific wait + if "amazon" in url.lower() and "/dp/" in url: + amazon_selectors = [".a-price .a-offscreen", "#corePriceDisplay_desktop_feature_div"] + for selector in amazon_selectors: + try: + await page.wait_for_selector(selector, timeout=5000, state="visible") + logger.info(f"Amazon price element found: {selector}") + break + except Exception: + continue + + # Extract content + if extract_text: + content = await page.inner_text('body') + logger.info(f"Extracted {len(content)} chars of visible text") + + # Normalize French prices + content = re.sub(r'(\d+)€(\d{2})\b', r'\1.\2 €', content) + content = re.sub(r'(\d+),(\d{2})\s*€', r'\1.\2 €', content) + content = re.sub(r'(\d+)\s(\d{3})', r'\1\2', content) + + # Extract price + extracted_price = "" + if "amazon" in url.lower() and "/dp/" in url: + extracted_price = await cls._extract_amazon_price(page) + else: + extracted_price = await cls._extract_generic_price(page) + + # Prepend normalized price + if extracted_price: + normalized_price = re.sub(r'(\d+)€(\d{2})', r'\1.\2 €', extracted_price) + normalized_price = re.sub(r'(\d+),(\d{2})', r'\1.\2', normalized_price) + normalized_price = normalized_price.replace(' ', '') + content = f"PRIX DÉTECTÉ: {normalized_price}\n\n{content}" + logger.info(f"Prepended price: {normalized_price}") else: - await cls._auto_detect_price(page) + content = await page.content() - if config.smart_scroll: - await cls._smart_scroll(page, scroll_pixels) + # Take screenshot + screenshot_path = "" + try: + os.makedirs("/app/screenshots", exist_ok=True) + timestamp = int(time.time() * 1000) + safe_name = "".join(c if c.isalnum() else "_" for c in url.split("//")[-1])[:50] + screenshot_path = f"/app/screenshots/{safe_name}_{timestamp}.jpg" - page_text = await cls._extract_text(page, config.text_length) - - # Extract item_id from URL if it matches pattern - item_id = None - if "item_" in url: - # This is a simplified extraction, adjust based on your URL patterns - pass - - screenshot_path = await cls._take_screenshot(page, url, item_id) + is_amazon = "amazon" in url.lower() and "/dp/" in url - return screenshot_path, page_text + if is_amazon: + # Focused Amazon screenshot + for selector in ["#dp-container", "#ppd", "#centerCol"]: + try: + element = page.locator(selector).first + if await element.is_visible(): + await element.screenshot(path=screenshot_path, quality=85, type="jpeg") + logger.info(f"Amazon focused screenshot: {selector}") + break + except Exception: + continue + else: + await page.screenshot(path=screenshot_path, full_page=False, quality=80, type="jpeg") + else: + await page.screenshot(path=screenshot_path, full_page=False, quality=80, type="jpeg") + + logger.info(f"Screenshot saved to {screenshot_path}") + except Exception as e: + logger.warning(f"Screenshot failed: {e}") + + return content, screenshot_path finally: await context.close() @@ -332,5 +427,5 @@ class BrowserlessService: return "", "" -# Global instance for backward compatibility +# Global instance browserless_service = BrowserlessService() From 2710388faf47373bc9c4a317f18d21f6d61765e9 Mon Sep 17 00:00:00 2001 From: Michael SCHAL Date: Sun, 30 Nov 2025 12:48:30 +0100 Subject: [PATCH 3/3] feat: Implement scheduled item checks with AI price extraction, database updates, and price change notifications. --- app/services/scheduler_service.py | 10 +++++----- app/services/search_service.py | 8 ++++---- 2 files changed, 9 insertions(+), 9 deletions(-) diff --git a/app/services/scheduler_service.py b/app/services/scheduler_service.py index b38ec97..67b620a 100644 --- a/app/services/scheduler_service.py +++ b/app/services/scheduler_service.py @@ -103,12 +103,12 @@ async def process_item_check(item_id: int): try: logger.info(f"Checking item: {item_data['name']} ({item_data['url']})") - # Use new browserless service - # Note: smart_scroll is handled internally by browserless_service - screenshot_path, page_text = await browserless_service.get_page_content( + # Use browserless service with extract_text=True for monitoring + page_text, screenshot_path = await browserless_service.get_page_content( item_data["url"], - use_proxy="amazon" in item_data["url"], # Simple heuristic for now - wait_selector=item_data["selector"] + use_proxy="amazon" in item_data["url"], + wait_selector=item_data["selector"], + extract_text=True # Get visible text for AI analysis ) # Determine availability based on content presence diff --git a/app/services/search_service.py b/app/services/search_service.py index 3079e86..779cce4 100644 --- a/app/services/search_service.py +++ b/app/services/search_service.py @@ -68,7 +68,7 @@ class NewSearchService: # Use proxy if required by config use_proxy = config.get("requires_proxy", False) - _, html_content = await browserless_service.get_page_content( + html_content, _ = await browserless_service.get_page_content( search_url, use_proxy=use_proxy, wait_selector=config.get("wait_selector") @@ -96,7 +96,7 @@ class NewSearchService: use_proxy = config.get("requires_proxy", False) if config else False # Use browserless to get content and screenshot - screenshot_path, html = await browserless_service.get_page_content( + html, screenshot_path = await browserless_service.get_page_content( result.url, use_proxy=use_proxy, wait_selector=None @@ -140,7 +140,7 @@ class NewSearchService: # Use proxy if required by config use_proxy = config.get("requires_proxy", False) - _, html_content = await browserless_service.get_page_content( + html_content, _ = await browserless_service.get_page_content( search_url, use_proxy=use_proxy, wait_selector=config.get("wait_selector") @@ -285,7 +285,7 @@ class NewSearchService: # Use proxy if required by config use_proxy = config.get("requires_proxy", False) - _, html_content = await browserless_service.get_page_content( + html_content, _ = await browserless_service.get_page_content( search_url, use_proxy=use_proxy, wait_selector=config.get("wait_selector")