From 9a6ebf6f0637687fa5778378007d0c36d664872d Mon Sep 17 00:00:00 2001 From: Michael SCHAL Date: Sun, 23 Nov 2025 15:45:09 +0100 Subject: [PATCH 1/2] 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 | 152 ++++++++++++++++++-------- app/services/search_service.py | 27 +++-- verify_search.py | 111 +++++++++++++++++++ 3 files changed, 234 insertions(+), 56 deletions(-) create mode 100644 verify_search.py diff --git a/app/services/direct_search_service.py b/app/services/direct_search_service.py index 4821dac..62d512d 100644 --- a/app/services/direct_search_service.py +++ b/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: diff --git a/app/services/search_service.py b/app/services/search_service.py index ed3ce94..2e4a14d 100644 --- a/app/services/search_service.py +++ b/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 diff --git a/verify_search.py b/verify_search.py new file mode 100644 index 0000000..f40aa86 --- /dev/null +++ b/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()) From e551927327b44b6839e459e6d3cd05d7b4159e5e Mon Sep 17 00:00:00 2001 From: Michael SCHAL Date: Sun, 23 Nov 2025 16:06:16 +0100 Subject: [PATCH 2/2] feat: Add direct e-commerce search service using Playwright/Browserless with site-specific configurations. --- app/services/direct_search_service.py | 101 +++++++----- app/services/scraper_service.py | 55 +++++-- app/services/search_service.py | 229 +++++++++++++++----------- verify_search.py | 24 ++- 4 files changed, 246 insertions(+), 163 deletions(-) diff --git a/app/services/direct_search_service.py b/app/services/direct_search_service.py index 62d512d..1e0e0ea 100644 --- a/app/services/direct_search_service.py +++ b/app/services/direct_search_service.py @@ -415,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. @@ -424,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 @@ -435,56 +437,69 @@ async def search( all_results = [] results_per_site = max(5, max_results // len(sites)) - logger.info(f"Démarrage recherche parallèle sur {len(sites)} sites pour '{query}'") + logger.info(f"Démarrage recherche parallèle v2 (Shared Browser) sur {len(sites)} sites pour '{query}'") + # 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: - # 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"Connexion autonome à Browserless: {BROWSERLESS_URL}") + local_browser = await p.chromium.connect_over_cdp(BROWSERLESS_URL) 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") - + return await _execute_search_with_browser(query, sites, results_per_site, timeout, local_browser) finally: - # Fermer le navigateur à la fin - if browser: - await browser.close() - logger.info("Navigateur Browserless fermé") - + await local_browser.close() + logger.info("Navigateur autonome fermé") except Exception as e: - logger.error(f"Erreur globale recherche Browserless: {e}") + logger.error(f"Erreur globale recherche autonome: {e}") + return [] - logger.info(f"Recherche Browserless terminée: {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 diff --git a/app/services/scraper_service.py b/app/services/scraper_service.py index 0d4bb0a..68602db 100644 --- a/app/services/scraper_service.py +++ b/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() diff --git a/app/services/search_service.py b/app/services/search_service.py index 2e4a14d..312968f 100644 --- a/app/services/search_service.py +++ b/app/services/search_service.py @@ -89,125 +89,154 @@ async def search_products( results=[], ) + browser = None + playwright = None + try: - search_results = await direct_search_service.search( - query=query, - sites=sites_data, - max_results=max_results, + # 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 ) - 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 - 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 + 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 - total = len(search_results) - logger.info(f"Recherche directe: {total} URLs trouvées pour '{query}'") + if not search_results: + yield SearchProgress( + status="completed", + total=0, + completed=0, + message="Aucun résultat trouvé sur les sites configurés", + results=[], + ) + return - # Phase 2: Scraping des URLs - results: list[SearchResultItem] = [] - completed = 0 + total = len(search_results) + logger.info(f"Recherche directe: {total} URLs trouvées pour '{query}'") - # Créer un sémaphore pour limiter la concurrence - semaphore = asyncio.Semaphore(parallel_limit) + # Phase 2: Scraping des URLs + results: list[SearchResultItem] = [] + completed = 0 - 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) + # Créer un sémaphore pour limiter la concurrence + semaphore = asyncio.Semaphore(parallel_limit) - 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 + 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) - site_name = site.name if site else domain - requires_js = site.requires_js if site else False + 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 - # 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) + site_name = site.name if site else domain + requires_js = site.requires_js if site else False - if light_result.success and light_result.price is not None: + # 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( @@ -215,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, @@ -226,6 +256,7 @@ async def _scrape_with_browserless( scroll_pixels=350, text_length=3000, timeout=30000, + browser=browser, ) if not screenshot_path: diff --git a/verify_search.py b/verify_search.py index f40aa86..9b79c1c 100644 --- a/verify_search.py +++ b/verify_search.py @@ -71,11 +71,22 @@ async def test_search_flow(): # 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 @@ -100,10 +111,15 @@ async def test_search_flow(): # 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)}") + 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 ---")