feat: Implement direct product search service using Browserless for e-commerce sites and add a search flow verification script.

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]}...)"
@@ -423,19 +435,55 @@ 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 sur {len(sites)} sites pour '{query}'")
 
-        if len(all_results) >= max_results:
-            break
+    try:
+        async with async_playwright() as p:
+            # Se connecter à Browserless UNE SEULE FOIS pour tous les sites
+            logger.info(f"Connexion à Browserless: {BROWSERLESS_URL}")
+            browser = await p.chromium.connect_over_cdp(BROWSERLESS_URL)
 
-    logger.info(f"Recherche Browserless: {len(all_results)} résultats totaux")
+            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,  # Passer l'instance du navigateur
+                    )
+                    tasks.append(task)
+
+                # Exécuter en parallèle avec asyncio.gather
+                # return_exceptions=True permet de continuer même si un site plante
+                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")
+
+            finally:
+                # Fermer le navigateur à la fin
+                if browser:
+                    await browser.close()
+                    logger.info("Navigateur Browserless fermé")
+
+    except Exception as e:
+        logger.error(f"Erreur globale recherche Browserless: {e}")
+
+    logger.info(f"Recherche Browserless terminée: {len(all_results)} résultats totaux")
     return all_results[:max_results]
 
 
@@ -444,6 +492,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 +513,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 +612,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 +625,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/search_service.py

@@ -89,18 +89,29 @@ 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:
+    try:
+        search_results = await direct_search_service.search(
+            query=query,
+            sites=sites_data,
+            max_results=max_results,
+        )
+    except Exception as e:
+        logger.error(f"Erreur critique lors de la recherche directe: {e}")
         yield SearchProgress(
             status="error",
             total=0,
             completed=0,
-            message="Aucun résultat trouvé",
+            message=f"Erreur de recherche: {str(e)}",
+            results=[],
+        )
+        return
+
+    if not search_results:
+        yield SearchProgress(
+            status="completed", # Changed from error to completed to avoid scary red alerts for just 0 results
+            total=0,
+            completed=0,
+            message="Aucun résultat trouvé sur les sites configurés",
             results=[],
         )
         return

verify_search.py

@@ -0,0 +1,111 @@
+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 direct_search_service.search
+    mock_results = [
+        MockSearchResult("http://site1.com/p1", "Product 1", "site1.com"),
+        MockSearchResult("http://site2.com/p2", "Product 2", "site2.com"),
+    ]
+    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...")
+    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)}")
+            
+    print("--- Verification Complete ---")
+
+if __name__ == "__main__":
+    asyncio.run(test_search_flow())