Merge pull request #43 from R0m1k3/antigravity

feat: Implement direct product search service using Browserless for e…

app/services/direct_search_service.py

@@ -28,11 +28,23 @@ USER_AGENT = (
 class SearchResult:
     """Résultat de recherche"""
 
-    def __init__(self, url: str, title: str, snippet: str, source: str):
+    def __init__(
+        self,
+        url: str,
+        title: str,
+        snippet: str,
+        source: str,
+        price: float | None = None,
+        currency: str = "EUR",
+        in_stock: bool | None = None,
+    ):
         self.url = url
         self.title = title
         self.snippet = snippet
         self.source = source
+        self.price = price
+        self.currency = currency
+        self.in_stock = in_stock
 
     def __repr__(self):
         return f"SearchResult(url={self.url}, title={self.title[:30]}...)"
@@ -403,6 +415,7 @@ async def search(
     sites: list[dict],
     max_results: int = 20,
     timeout: float = 30.0,
+    browser=None,  # Instance de navigateur partagée optionnelle
 ) -> list[SearchResult]:
     """
     Recherche sur les sites e-commerce via Browserless.
@@ -412,6 +425,7 @@ async def search(
         sites: Liste de dicts avec {domain, search_url, product_link_selector, name}
         max_results: Nombre maximum de résultats
         timeout: Timeout par site
+        browser: Instance Playwright browser partagée (optionnel)
 
     Returns:
         Liste de SearchResult
@@ -423,20 +437,69 @@ async def search(
     all_results = []
     results_per_site = max(5, max_results // len(sites))
 
-    # Rechercher sur chaque site en séquence (pour éviter de surcharger Browserless)
-    for site in sites:
-        try:
-            results = await _search_site_browserless(query, site, results_per_site, timeout)
-            all_results.extend(results)
-            logger.info(f"Site {site.get('domain')}: {len(results)} résultats")
-        except Exception as e:
-            logger.error(f"Erreur recherche sur {site.get('domain')}: {e}")
+    logger.info(f"Démarrage recherche parallèle v2 (Shared Browser) sur {len(sites)} sites pour '{query}'")
 
-        if len(all_results) >= max_results:
-            break
+    # Si un navigateur est fourni, on l'utilise directement
+    if browser:
+        return await _execute_search_with_browser(query, sites, results_per_site, timeout, browser)
+    
+    # Sinon on crée notre propre instance (comportement autonome)
+    try:
+        async with async_playwright() as p:
+            logger.info(f"Connexion autonome à Browserless: {BROWSERLESS_URL}")
+            local_browser = await p.chromium.connect_over_cdp(BROWSERLESS_URL)
+            try:
+                return await _execute_search_with_browser(query, sites, results_per_site, timeout, local_browser)
+            finally:
+                await local_browser.close()
+                logger.info("Navigateur autonome fermé")
+    except Exception as e:
+        logger.error(f"Erreur globale recherche autonome: {e}")
+        return []
 
-    logger.info(f"Recherche Browserless: {len(all_results)} résultats totaux")
-    return all_results[:max_results]
+async def _execute_search_with_browser(
+    query: str,
+    sites: list[dict],
+    results_per_site: int,
+    timeout: float,
+    browser,
+) -> list[SearchResult]:
+    """Exécute la recherche avec un navigateur donné"""
+    all_results = []
+    try:
+        # Créer des tâches pour chaque site
+        tasks = []
+        for site in sites:
+            task = _search_site_browserless(
+                query=query,
+                site=site,
+                max_results=results_per_site,
+                timeout=timeout,
+                browser=browser,
+            )
+            tasks.append(task)
+
+        # Exécuter en parallèle avec asyncio.gather
+        results_list = await asyncio.gather(*tasks, return_exceptions=True)
+
+        # Agréger les résultats
+        for i, result in enumerate(results_list):
+            site_domain = sites[i].get("domain", "inconnu")
+            
+            if isinstance(result, Exception):
+                logger.error(f"Erreur fatale recherche {site_domain}: {result}")
+                continue
+                
+            if result:
+                all_results.extend(result)
+                logger.info(f"Site {site_domain}: {len(result)} résultats")
+            else:
+                logger.info(f"Site {site_domain}: 0 résultat")
+                
+    except Exception as e:
+        logger.error(f"Erreur exécution recherche: {e}")
+        
+    return all_results
 
 
 async def _search_site_browserless(  # noqa: PLR0912, PLR0915
@@ -444,6 +507,7 @@ async def _search_site_browserless(  # noqa: PLR0912, PLR0915
     site: dict,
     max_results: int,
     timeout: float,
+    browser,  # Instance de navigateur partagée
 ) -> list[SearchResult]:
     """Recherche sur un site via Browserless (Playwright)"""
     raw_domain = site.get("domain", "").lower()
@@ -464,47 +528,42 @@ async def _search_site_browserless(  # noqa: PLR0912, PLR0915
 
     logger.info(f"Recherche Browserless sur {domain}: {final_url}")
 
+    context = None
+    page = None
+
     try:
-        async with async_playwright() as p:
-            # Se connecter à Browserless
-            browser = await p.chromium.connect_over_cdp(BROWSERLESS_URL)
+        # Créer un nouveau contexte isolé pour ce site
+        context = await browser.new_context(
+            viewport={"width": 1920, "height": 1080},
+            user_agent=USER_AGENT,
+            locale="fr-FR",
+        )
+        
+        # Bloquer les ressources inutiles
+        await context.route("**/*.{png,jpg,jpeg,gif,webp,svg,ico}", lambda route: route.abort())
+        await context.route("**/analytics*", lambda route: route.abort())
+        await context.route("**/tracking*", lambda route: route.abort())
+        
+        page = await context.new_page()
 
+        # Naviguer vers la page de recherche
+        await page.goto(final_url, wait_until="domcontentloaded", timeout=int(timeout * 1000))
+
+        # Accepter les cookies si nécessaire
+        await _accept_cookies(page, domain)
+
+        # Attendre que les résultats se chargent
+        if wait_selector:
             try:
-                context = await browser.new_context(
-                    viewport={"width": 1920, "height": 1080},
-                    user_agent=USER_AGENT,
-                    locale="fr-FR",
-                )
-                page = await context.new_page()
+                await page.wait_for_selector(wait_selector, timeout=10000)
+            except Exception:
+                logger.debug(f"Selector {wait_selector} non trouvé sur {domain}, on continue...")
 
-                # Bloquer les ressources inutiles pour accélérer
-                await page.route("**/*.{png,jpg,jpeg,gif,webp,svg,ico}", lambda route: route.abort())
-                await page.route("**/analytics*", lambda route: route.abort())
-                await page.route("**/tracking*", lambda route: route.abort())
+        # Attendre un peu pour le JS (réduit pour parallélisation)
+        await asyncio.sleep(1.5)
 
-                # Naviguer vers la page de recherche
-                await page.goto(final_url, wait_until="domcontentloaded", timeout=int(timeout * 1000))
-
-                # Accepter les cookies si nécessaire
-                await _accept_cookies(page, domain)
-
-                # Attendre que les résultats se chargent
-                if wait_selector:
-                    try:
-                        await page.wait_for_selector(wait_selector, timeout=10000)
-                    except Exception:
-                        logger.debug(f"Selector {wait_selector} non trouvé, on continue...")
-
-                # Attendre un peu pour le JS
-                await asyncio.sleep(2)
-
-                # Extraire le HTML
-                html_content = await page.content()
-
-                await context.close()
-
-            finally:
-                await browser.close()
+        # Extraire le HTML
+        html_content = await page.content()
 
         # Parser le HTML avec BeautifulSoup
         soup = BeautifulSoup(html_content, "lxml")
@@ -568,6 +627,8 @@ async def _search_site_browserless(  # noqa: PLR0912, PLR0915
                 title=title,
                 snippet="",
                 source=domain,
+                # On ne peut pas extraire le prix facilement ici sans scraper chaque page
+                # Le scraping détaillé se fera dans la phase 2
             ))
 
             if len(results) >= max_results:
@@ -579,6 +640,16 @@ async def _search_site_browserless(  # noqa: PLR0912, PLR0915
     except Exception as e:
         logger.error(f"Erreur Browserless sur {domain}: {e}")
         return []
+        
+    finally:
+        # Fermer le contexte et la page, mais PAS le navigateur
+        try:
+            if page:
+                await page.close()
+            if context:
+                await context.close()
+        except Exception as e:
+            logger.debug(f"Erreur fermeture contexte {domain}: {e}")
 
 
 def _get_default_search_url(domain: str) -> str | None:

app/services/scraper_service.py

@@ -88,6 +88,7 @@ class ScraperService:
         scroll_pixels: int = 350,
         text_length: int = 0,
         timeout: int = 90000,
+        browser=None,  # Instance de navigateur partagée optionnelle
     ) -> tuple[str | None, str, bool]:
         """
         Scrapes the given URL using Browserless and Playwright.
@@ -101,6 +102,7 @@ class ScraperService:
             scroll_pixels: Number of pixels to scroll (must be positive)
             text_length: Number of characters to extract (0 = disabled)
             timeout: Page load timeout in milliseconds
+            browser: Optional shared Playwright browser instance
         """
         # Simplifier l'URL avant le scraping
         original_url = url
@@ -130,6 +132,7 @@ class ScraperService:
                 scroll_pixels=scroll_pixels,
                 text_length=text_length,
                 timeout=timeout,
+                browser=browser,
             )
 
             if result[0] is not None:  # Screenshot réussi
@@ -150,25 +153,36 @@ class ScraperService:
         scroll_pixels: int = 350,
         text_length: int = 0,
         timeout: int = 90000,
+        browser=None,
     ) -> tuple[str | None, str, bool]:
         """Exécute le scraping réel (appelé par scrape_item avec retries)."""
-        async with async_playwright() as p:
-            browser = None
+        
+        # Si un navigateur est fourni, on l'utilise directement sans async_playwright context manager
+        # sinon on crée tout de zéro
+        playwright_manager = None
+        local_browser = None
+        
+        try:
+            if browser:
+                # Utiliser le navigateur partagé
+                current_browser = browser
+            else:
+                # Créer un nouveau navigateur
+                playwright_manager = async_playwright()
+                p = await playwright_manager.start()
+                logger.info(f"Connecting to Browserless at {BROWSERLESS_URL}")
+                current_browser = await p.chromium.connect_over_cdp(
+                    BROWSERLESS_URL,
+                    timeout=60000,
+                )
+                local_browser = current_browser
+
             context = None
             page = None
             is_available = True  # Par défaut, le produit est disponible
             
             try:
-                logger.info(f"Connecting to Browserless at {BROWSERLESS_URL}")
-
-                # Utiliser l'endpoint WebSocket de Browserless
-                # Format: ws://browserless:3000?token=xxx ou ws://browserless:3000
-                browser = await p.chromium.connect_over_cdp(
-                    BROWSERLESS_URL,
-                    timeout=60000,  # 60s pour la connexion
-                )
-
-                context = await browser.new_context(
+                context = await current_browser.new_context(
                     viewport={"width": 1920, "height": 1080},
                     user_agent=(
                         "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
@@ -427,8 +441,15 @@ class ScraperService:
                 except Exception as e:
                     logger.debug(f"Error closing context: {e}")
 
-                try:
-                    if browser and browser.is_connected():
-                        await browser.close()
-                except Exception as e:
-                    logger.debug(f"Error closing browser: {e}")
+                # On ne ferme le navigateur que s'il est local (non partagé)
+                if local_browser:
+                    try:
+                        if local_browser.is_connected():
+                            await local_browser.close()
+                    except Exception as e:
+                        logger.debug(f"Error closing browser: {e}")
+                        
+        finally:
+            # Si on a créé un manager playwright local, on l'arrête
+            if playwright_manager:
+                await playwright_manager.stop()

app/services/search_service.py

@@ -89,114 +89,154 @@ async def search_products(
         results=[],
     )
 
-    search_results = await direct_search_service.search(
-        query=query,
-        sites=sites_data,
-        max_results=max_results,
-    )
-
-    if not search_results:
-        yield SearchProgress(
-            status="error",
-            total=0,
-            completed=0,
-            message="Aucun résultat trouvé",
-            results=[],
+    browser = None
+    playwright = None
+    
+    try:
+        # Initialiser Playwright et Browserless pour TOUTE la session de recherche
+        logger.info(f"Initialisation session Browserless globale pour '{query}'")
+        playwright = await async_playwright().start()
+        browser = await playwright.chromium.connect_over_cdp(
+            os.getenv("BROWSERLESS_URL", "ws://browserless:3000"),
+            timeout=60000
         )
-        return
 
-    total = len(search_results)
-    logger.info(f"Recherche directe: {total} URLs trouvées pour '{query}'")
+        try:
+            search_results = await direct_search_service.search(
+                query=query,
+                sites=sites_data,
+                max_results=max_results,
+                browser=browser,  # Utiliser le navigateur partagé
+            )
+        except Exception as e:
+            logger.error(f"Erreur critique lors de la recherche directe: {e}")
+            yield SearchProgress(
+                status="error",
+                total=0,
+                completed=0,
+                message=f"Erreur de recherche: {str(e)}",
+                results=[],
+            )
+            return
 
-    # Phase 2: Scraping des URLs
-    results: list[SearchResultItem] = []
-    completed = 0
+        if not search_results:
+            yield SearchProgress(
+                status="completed",
+                total=0,
+                completed=0,
+                message="Aucun résultat trouvé sur les sites configurés",
+                results=[],
+            )
+            return
 
-    # Créer un sémaphore pour limiter la concurrence
-    semaphore = asyncio.Semaphore(parallel_limit)
+        total = len(search_results)
+        logger.info(f"Recherche directe: {total} URLs trouvées pour '{query}'")
 
-    async def process_url(search_result: direct_search_service.SearchResult) -> SearchResultItem | None:
-        async with semaphore:
-            url = search_result.url
-            domain = search_result.source
-            site = site_map.get(domain)
+        # Phase 2: Scraping des URLs
+        results: list[SearchResultItem] = []
+        completed = 0
 
-            if not site:
-                # Trouver le site par correspondance partielle
-                for d, s in site_map.items():
-                    if d in domain or domain in d:
-                        site = s
-                        break
+        # Créer un sémaphore pour limiter la concurrence
+        semaphore = asyncio.Semaphore(parallel_limit)
 
-            site_name = site.name if site else domain
-            requires_js = site.requires_js if site else False
+        async def process_url(search_result: direct_search_service.SearchResult) -> SearchResultItem | None:
+            async with semaphore:
+                url = search_result.url
+                domain = search_result.source
+                site = site_map.get(domain)
 
-            # Essayer le scraping léger d'abord (sauf si le site requiert JS)
-            if not requires_js:
-                light_result = await light_scraper_service.scrape_url(url)
+                if not site:
+                    # Trouver le site par correspondance partielle
+                    for d, s in site_map.items():
+                        if d in domain or domain in d:
+                            site = s
+                            break
 
-                if light_result.success and light_result.price is not None:
+                site_name = site.name if site else domain
+                requires_js = site.requires_js if site else False
+
+                # Essayer le scraping léger d'abord (sauf si le site requiert JS)
+                if not requires_js:
+                    light_result = await light_scraper_service.scrape_url(url)
+
+                    if light_result.success and light_result.price is not None:
+                        return SearchResultItem(
+                            url=url,
+                            title=light_result.title or search_result.title,
+                            price=light_result.price,
+                            currency=light_result.currency,
+                            in_stock=light_result.in_stock,
+                            image_url=light_result.image_url,
+                            site_name=site_name,
+                            site_domain=domain,
+                            confidence=0.7,  # Confiance moyenne pour le scraping léger
+                        )
+
+                # Fallback: Browserless + IA
+                try:
+                    # Passer le navigateur partagé
+                    result = await _scrape_with_browserless(url, site_name, domain, search_result.title, browser)
+                    return result
+                except Exception as e:
+                    logger.error(f"Erreur scraping {url}: {e}")
                     return SearchResultItem(
                         url=url,
-                        title=light_result.title or search_result.title,
-                        price=light_result.price,
-                        currency=light_result.currency,
-                        in_stock=light_result.in_stock,
-                        image_url=light_result.image_url,
+                        title=search_result.title,
+                        price=None,
                         site_name=site_name,
                         site_domain=domain,
-                        confidence=0.7,  # Confiance moyenne pour le scraping léger
+                        error=str(e),
                     )
 
-            # Fallback: Browserless + IA
+        # Traiter les URLs en parallèle avec mises à jour progressives
+        tasks = [asyncio.create_task(process_url(r)) for r in search_results]
+
+        for coro in asyncio.as_completed(tasks):
             try:
-                result = await _scrape_with_browserless(url, site_name, domain, search_result.title)
-                return result
+                result = await coro
+                completed += 1
+
+                if result:
+                    # Ajouter tous les résultats, même sans prix (le frontend affichera "Prix non disponible")
+                    results.append(result)
+
+                    yield SearchProgress(
+                        status="scraping",
+                        total=total,
+                        completed=completed,
+                        current_site=result.site_name if result else None,
+                        results=results.copy(),
+                        message=f"Extraction {completed}/{total}...",
+                    )
             except Exception as e:
-                logger.error(f"Erreur scraping {url}: {e}")
-                return SearchResultItem(
-                    url=url,
-                    title=search_result.title,
-                    price=None,
-                    site_name=site_name,
-                    site_domain=domain,
-                    error=str(e),
-                )
+                completed += 1
+                logger.error(f"Erreur lors du traitement: {e}")
 
-    # Traiter les URLs en parallèle avec mises à jour progressives
-    tasks = [asyncio.create_task(process_url(r)) for r in search_results]
+        # Trier les résultats par prix
+        results.sort(key=lambda x: x.price if x.price else float("inf"))
 
-    for coro in asyncio.as_completed(tasks):
-        try:
-            result = await coro
-            completed += 1
-
-            if result:
-                # Ajouter tous les résultats, même sans prix (le frontend affichera "Prix non disponible")
-                results.append(result)
-
-                yield SearchProgress(
-                    status="scraping",
-                    total=total,
-                    completed=completed,
-                    current_site=result.site_name if result else None,
-                    results=results.copy(),
-                    message=f"Extraction {completed}/{total}...",
-                )
-        except Exception as e:
-            completed += 1
-            logger.error(f"Erreur lors du traitement: {e}")
-
-    # Trier les résultats par prix
-    results.sort(key=lambda x: x.price if x.price else float("inf"))
-
-    yield SearchProgress(
-        status="completed",
-        total=total,
-        completed=completed,
-        results=results,
-        message=f"{len(results)} produits trouvés avec prix",
-    )
+        yield SearchProgress(
+            status="completed",
+            total=total,
+            completed=completed,
+            results=results,
+            message=f"{len(results)} produits trouvés avec prix",
+        )
+        
+    finally:
+        # Nettoyage global des ressources
+        if browser:
+            try:
+                await browser.close()
+                logger.info("Navigateur global fermé")
+            except Exception as e:
+                logger.error(f"Erreur fermeture navigateur: {e}")
+                
+        if playwright:
+            try:
+                await playwright.stop()
+            except Exception as e:
+                logger.error(f"Erreur arrêt Playwright: {e}")
 
 
 async def _scrape_with_browserless(
@@ -204,10 +244,11 @@ async def _scrape_with_browserless(
     site_name: str,
     domain: str,
     fallback_title: str,
+    browser=None,  # Navigateur partagé
 ) -> SearchResultItem:
     """Scrape une URL avec Browserless + extraction IA"""
     try:
-        # Scraper avec Browserless
+        # Scraper avec Browserless (utiliser le navigateur partagé)
         screenshot_path, page_text, is_available = await ScraperService.scrape_item(
             url=url,
             item_id=None,
@@ -215,6 +256,7 @@ async def _scrape_with_browserless(
             scroll_pixels=350,
             text_length=3000,
             timeout=30000,
+            browser=browser,
         )
 
         if not screenshot_path:

verify_search.py

@@ -0,0 +1,127 @@
+import asyncio
+import logging
+import sys
+from unittest.mock import MagicMock, AsyncMock
+
+# Configure logging
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+# Mock dependencies
+sys.modules["sqlalchemy"] = MagicMock()
+sys.modules["sqlalchemy.orm"] = MagicMock()
+sys.modules["app.database"] = MagicMock()
+sys.modules["app.models"] = MagicMock()
+sys.modules["app.services.ai_service"] = MagicMock()
+sys.modules["app.services.scraper_service"] = MagicMock()
+sys.modules["app.services.light_scraper_service"] = MagicMock()
+sys.modules["app.services.settings_service"] = MagicMock()
+
+# Define dummy classes for schemas
+class MockSearchProgress:
+    def __init__(self, status, total, completed, message, results, current_site=None):
+        self.status = status
+        self.total = total
+        self.completed = completed
+        self.message = message
+        self.results = results
+        self.current_site = current_site
+        
+    def model_dump_json(self):
+        return "json"
+
+class MockSearchResultItem:
+    def __init__(self, **kwargs):
+        for k, v in kwargs.items():
+            setattr(self, k, v)
+
+# Setup schema mocks
+schemas_mock = MagicMock()
+schemas_mock.SearchProgress = MockSearchProgress
+schemas_mock.SearchResultItem = MockSearchResultItem
+sys.modules["app.schemas"] = schemas_mock
+
+# Import services after mocking
+# We need to mock direct_search_service before importing search_service
+# because search_service imports it.
+direct_search_mock = MagicMock()
+sys.modules["app.services.direct_search_service"] = direct_search_mock
+
+# Now we can import search_service
+# We might need to mock other things that search_service imports
+from app.services import search_service
+
+# Define a mock SearchResult class matching the one in direct_search_service
+class MockSearchResult:
+    def __init__(self, url, title, source, price=None, currency="EUR", in_stock=None):
+        self.url = url
+        self.title = title
+        self.source = source
+        self.snippet = "snippet"
+        self.price = price
+        self.currency = currency
+        self.in_stock = in_stock
+
+async def test_search_flow():
+    print("--- Starting Search Flow Verification ---")
+    
+    # Setup mocks
+    db = MagicMock()
+    
+    # Mock SettingsService
+    search_service.SettingsService.get_setting_value.side_effect = lambda db, key, default: default
+    
+    # Mock async_playwright in search_service
+    mock_browser = AsyncMock()
+    mock_playwright_obj = AsyncMock()
+    mock_playwright_obj.chromium.connect_over_cdp.return_value = mock_browser
+    
+    mock_playwright_manager = MagicMock()
+    mock_playwright_manager.start = AsyncMock(return_value=mock_playwright_obj)
+    
+    search_service.async_playwright = MagicMock(return_value=mock_playwright_manager)
+    
+    # Mock direct_search_service.search
+    mock_results = [
+        MockSearchResult("http://site1.com/p1", "Product 1", "site1.com"),
+        MockSearchResult("http://site2.com/p2", "Product 2", "site2.com"),
+    ]
+    # Accept any arguments including browser
+    direct_search_mock.search = AsyncMock(return_value=mock_results)
+    direct_search_mock.SearchResult = MockSearchResult
+    
+    # Mock light_scraper_service
+    search_service.light_scraper_service.scrape_url = AsyncMock(return_value=MagicMock(success=False))
+    
+    # Mock _scrape_with_browserless (to avoid actual scraping)
+    # We need to patch it in the module
+    original_scrape = search_service._scrape_with_browserless
+    search_service._scrape_with_browserless = AsyncMock(return_value=MagicMock(
+        url="http://site1.com/p1",
+        title="Product 1",
+        price=10.0,
+        site_name="Site 1",
+        site_domain="site1.com"
+    ))
+    
+    # Mock _get_sites to return some dummy sites
+    mock_site1 = MagicMock(domain="site1.com", name="Site 1", requires_js=True)
+    mock_site2 = MagicMock(domain="site2.com", name="Site 2", requires_js=True)
+    search_service._get_sites = MagicMock(return_value=[mock_site1, mock_site2])
+    
+    # Run the search
+    print("Running search_products...")
+    try:
+        async for progress in search_service.search_products("test query", db):
+            print(f"Event: {progress.status} - {progress.message}")
+            if progress.results:
+                print(f"  Results: {len(progress.results)}")
+    except Exception as e:
+        print(f"Caught exception during search: {e}")
+        import traceback
+        traceback.print_exc()
+            
+    print("--- Verification Complete ---")
+
+if __name__ == "__main__":
+    asyncio.run(test_search_flow())