Merge pull request #26 from R0m1k3/claude/design-product-search-01Ea9dWZqZpuKFwBwVmp26PG

feat: Replace DuckDuckGo with direct site search

app/main.py

@@ -53,15 +53,21 @@ def run_migrations():
             conn.commit()
             logger.info("search_sites table created")
         else:
-            # 2. Add price_selector column if missing
+            # 2. Add missing columns
-            result = conn.execute(text(
+            columns_to_add = [
-                "SELECT 1 FROM information_schema.columns WHERE table_name = 'search_sites' AND column_name = 'price_selector'"
+                ("price_selector", "VARCHAR(512)"),
-            ))
+                ("search_url", "VARCHAR(1024)"),
-            if not result.fetchone():
+                ("product_link_selector", "VARCHAR(512)"),
-                logger.info("Adding price_selector column to search_sites...")
+            ]
-                conn.execute(text("ALTER TABLE search_sites ADD COLUMN price_selector VARCHAR(512)"))
+            for col_name, col_type in columns_to_add:
-                conn.commit()
+                result = conn.execute(text(
-                logger.info("price_selector column added")
+                    f"SELECT 1 FROM information_schema.columns WHERE table_name = 'search_sites' AND column_name = '{col_name}'"
+                ))
+                if not result.fetchone():
+                    logger.info(f"Adding {col_name} column to search_sites...")
+                    conn.execute(text(f"ALTER TABLE search_sites ADD COLUMN {col_name} {col_type}"))
+                    conn.commit()
+                    logger.info(f"{col_name} column added")
 
 
 @asynccontextmanager

app/models.py

@@ -92,5 +92,7 @@ class SearchSite(Base):
     priority: int = Column(Integer, default=0)  # type: ignore  # Ordre d'affichage
     requires_js: bool = Column(Boolean, default=False)  # type: ignore  # Force Browserless si True
     price_selector: str | None = Column(String, nullable=True)  # type: ignore  # Sélecteur CSS pour le prix
+    search_url: str | None = Column(String, nullable=True)  # type: ignore  # URL de recherche avec {query} placeholder
+    product_link_selector: str | None = Column(String, nullable=True)  # type: ignore  # Sélecteur CSS pour les liens produits
     created_at: datetime = Column(DateTime, default=lambda: datetime.now(UTC))  # type: ignore
     updated_at: datetime = Column(DateTime, default=lambda: datetime.now(UTC), onupdate=lambda: datetime.now(UTC))  # type: ignore

app/schemas.py

@@ -63,6 +63,8 @@ class SearchSiteCreate(BaseModel):
     priority: int = 0
     requires_js: bool = False
     price_selector: str | None = None
+    search_url: str | None = None  # URL avec {query} placeholder, ex: https://amazon.fr/s?k={query}
+    product_link_selector: str | None = None  # Sélecteur CSS pour les liens produits
 
 
 class SearchSiteUpdate(BaseModel):
@@ -74,6 +76,8 @@ class SearchSiteUpdate(BaseModel):
     priority: int | None = None
     requires_js: bool | None = None
     price_selector: str | None = None
+    search_url: str | None = None
+    product_link_selector: str | None = None
 
 
 class SearchSiteResponse(BaseModel):
@@ -86,6 +90,8 @@ class SearchSiteResponse(BaseModel):
     priority: int
     requires_js: bool
     price_selector: str | None = None
+    search_url: str | None = None
+    product_link_selector: str | None = None
     created_at: datetime | None = None
     updated_at: datetime | None = None
     model_config = ConfigDict(from_attributes=True)

app/services/direct_search_service.py

@@ -0,0 +1,334 @@
+"""
+Direct Search Service - Recherche directement sur les sites e-commerce
+Au lieu d'utiliser un moteur de recherche externe, scrape les pages de recherche de chaque site.
+"""
+
+import asyncio
+import logging
+import re
+from urllib.parse import quote_plus, urljoin, urlparse
+
+import httpx
+from bs4 import BeautifulSoup
+
+logger = logging.getLogger(__name__)
+
+
+class SearchResult:
+    """Résultat de recherche"""
+
+    def __init__(self, url: str, title: str, snippet: str, source: str):
+        self.url = url
+        self.title = title
+        self.snippet = snippet
+        self.source = source
+
+    def __repr__(self):
+        return f"SearchResult(url={self.url}, title={self.title[:30]}...)"
+
+
+# Configuration des sites avec leurs sélecteurs par défaut
+DEFAULT_SITE_CONFIGS = {
+    "amazon.fr": {
+        "search_url": "https://www.amazon.fr/s?k={query}",
+        "product_selector": "div[data-component-type='s-search-result'] a.a-link-normal[href*='/dp/']",
+        "title_selector": "span.a-text-normal",
+    },
+    "amazon.com": {
+        "search_url": "https://www.amazon.com/s?k={query}",
+        "product_selector": "div[data-component-type='s-search-result'] a.a-link-normal[href*='/dp/']",
+        "title_selector": "span.a-text-normal",
+    },
+    "action.com": {
+        "search_url": "https://www.action.com/fr-fr/search/?q={query}",
+        "product_selector": "a.product-card__link",
+        "title_selector": ".product-card__title",
+    },
+    "fnac.com": {
+        "search_url": "https://www.fnac.com/SearchResult/ResultList.aspx?Search={query}",
+        "product_selector": "a.Article-title",
+        "title_selector": "a.Article-title",
+    },
+    "cdiscount.com": {
+        "search_url": "https://www.cdiscount.com/search/10/{query}.html",
+        "product_selector": "a.prdtBImg",
+        "title_selector": ".prdtTit",
+    },
+    "darty.com": {
+        "search_url": "https://www.darty.com/nav/recherche?text={query}",
+        "product_selector": "a.product-card__link",
+        "title_selector": ".product-card__title",
+    },
+    "boulanger.com": {
+        "search_url": "https://www.boulanger.com/resultats?tr={query}",
+        "product_selector": "a.product-list__link",
+        "title_selector": ".product-list__title",
+    },
+}
+
+
+async def search(
+    query: str,
+    sites: list[dict],
+    max_results: int = 20,
+    timeout: float = 15.0,
+) -> list[SearchResult]:
+    """
+    Recherche directement sur les sites e-commerce.
+
+    Args:
+        query: Terme de recherche
+        sites: Liste de dicts avec {domain, search_url, product_link_selector, name}
+        max_results: Nombre maximum de résultats
+        timeout: Timeout par requête
+
+    Returns:
+        Liste de SearchResult
+    """
+    if not sites:
+        logger.warning("Aucun site configuré pour la recherche")
+        return []
+
+    all_results = []
+    results_per_site = max(5, max_results // len(sites))
+
+    # Rechercher sur chaque site en parallèle
+    tasks = []
+    for site in sites:
+        task = asyncio.create_task(
+            _search_site(query, site, results_per_site, timeout)
+        )
+        tasks.append(task)
+
+    results = await asyncio.gather(*tasks, return_exceptions=True)
+
+    for site, result in zip(sites, results):
+        if isinstance(result, Exception):
+            logger.error(f"Erreur recherche sur {site.get('domain')}: {result}")
+            continue
+        if result:
+            all_results.extend(result)
+
+    logger.info(f"Recherche directe: {len(all_results)} résultats totaux")
+    return all_results[:max_results]
+
+
+async def _search_site(
+    query: str,
+    site: dict,
+    max_results: int,
+    timeout: float,
+) -> list[SearchResult]:
+    """Recherche sur un site spécifique"""
+    domain = site.get("domain", "").lower()
+    search_url = site.get("search_url") or _get_default_search_url(domain)
+    product_selector = site.get("product_link_selector") or _get_default_product_selector(domain)
+    site_name = site.get("name", domain)
+
+    if not search_url:
+        logger.warning(f"Pas d'URL de recherche configurée pour {domain}")
+        return []
+
+    # Construire l'URL de recherche
+    encoded_query = quote_plus(query)
+    final_url = search_url.replace("{query}", encoded_query)
+
+    logger.info(f"Recherche directe sur {domain}: {final_url}")
+
+    try:
+        async with httpx.AsyncClient(timeout=timeout, follow_redirects=True) as client:
+            response = await client.get(
+                final_url,
+                headers={
+                    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
+                    "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8",
+                    "Accept-Language": "fr-FR,fr;q=0.9,en;q=0.8",
+                    "Accept-Encoding": "gzip, deflate, br",
+                    "DNT": "1",
+                    "Connection": "keep-alive",
+                    "Upgrade-Insecure-Requests": "1",
+                },
+            )
+            response.raise_for_status()
+
+            # Parser le HTML
+            soup = BeautifulSoup(response.text, "html.parser")
+            results = []
+
+            # Extraire les liens produits
+            if product_selector:
+                links = soup.select(product_selector)
+            else:
+                # Fallback: chercher tous les liens qui ressemblent à des produits
+                links = _find_product_links(soup, domain)
+
+            logger.info(f"{domain}: {len(links)} liens produits trouvés")
+
+            seen_urls = set()
+            for link in links[:max_results * 2]:  # Prendre plus pour filtrer
+                href = link.get("href", "")
+                if not href:
+                    continue
+
+                # Construire l'URL complète
+                if href.startswith("/"):
+                    href = f"https://{domain}{href}"
+                elif not href.startswith("http"):
+                    href = urljoin(final_url, href)
+
+                # Nettoyer l'URL
+                href = _clean_product_url(href)
+
+                # Éviter les doublons
+                if href in seen_urls:
+                    continue
+                seen_urls.add(href)
+
+                # Vérifier que c'est bien un lien vers ce domaine
+                url_domain = _extract_domain(href)
+                if domain not in url_domain:
+                    continue
+
+                # Extraire le titre
+                title = _extract_title(link)
+                if not title or len(title) < 3:
+                    continue
+
+                results.append(SearchResult(
+                    url=href,
+                    title=title,
+                    snippet="",
+                    source=domain,
+                ))
+
+                if len(results) >= max_results:
+                    break
+
+            logger.info(f"{domain}: {len(results)} résultats extraits")
+            return results
+
+    except httpx.TimeoutException:
+        logger.error(f"Timeout lors de la recherche sur {domain}")
+        return []
+    except httpx.HTTPStatusError as e:
+        logger.error(f"Erreur HTTP {e.response.status_code} sur {domain}")
+        return []
+    except Exception as e:
+        logger.error(f"Erreur lors de la recherche sur {domain}: {e}")
+        return []
+
+
+def _get_default_search_url(domain: str) -> str | None:
+    """Retourne l'URL de recherche par défaut pour un domaine"""
+    # Chercher une correspondance exacte ou partielle
+    for key, config in DEFAULT_SITE_CONFIGS.items():
+        if key in domain or domain in key:
+            return config.get("search_url")
+    return None
+
+
+def _get_default_product_selector(domain: str) -> str | None:
+    """Retourne le sélecteur de produits par défaut pour un domaine"""
+    for key, config in DEFAULT_SITE_CONFIGS.items():
+        if key in domain or domain in key:
+            return config.get("product_selector")
+    return None
+
+
+def _find_product_links(soup: BeautifulSoup, domain: str) -> list:
+    """Trouve les liens produits de manière générique"""
+    links = []
+
+    # Patterns communs pour les liens produits
+    patterns = [
+        "a[href*='/product']",
+        "a[href*='/p/']",
+        "a[href*='/dp/']",
+        "a[href*='/item']",
+        "a[href*='/article']",
+        "a.product",
+        "a.product-link",
+        "a.product-card",
+        ".product a",
+        ".product-card a",
+        ".product-tile a",
+        ".item a",
+    ]
+
+    for pattern in patterns:
+        found = soup.select(pattern)
+        if found:
+            links.extend(found)
+
+    # Dédupliquer
+    seen = set()
+    unique_links = []
+    for link in links:
+        href = link.get("href", "")
+        if href and href not in seen:
+            seen.add(href)
+            unique_links.append(link)
+
+    return unique_links
+
+
+def _extract_title(element) -> str:
+    """Extrait le titre d'un élément"""
+    # Essayer différentes méthodes
+    title = element.get("title", "")
+    if title:
+        return title.strip()
+
+    # Texte de l'élément
+    text = element.get_text(strip=True)
+    if text and len(text) > 3:
+        return text[:200]
+
+    # Attribut aria-label
+    label = element.get("aria-label", "")
+    if label:
+        return label.strip()
+
+    # Image alt text
+    img = element.find("img")
+    if img:
+        alt = img.get("alt", "")
+        if alt:
+            return alt.strip()
+
+    return ""
+
+
+def _clean_product_url(url: str) -> str:
+    """Nettoie une URL produit (retire les paramètres de tracking, etc.)"""
+    # Pour Amazon, simplifier l'URL
+    if "amazon." in url and "/dp/" in url:
+        match = re.search(r"(/dp/[A-Z0-9]{10})", url)
+        if match:
+            domain_match = re.search(r"(https?://[^/]+)", url)
+            if domain_match:
+                return domain_match.group(1) + match.group(1)
+
+    return url
+
+
+def _extract_domain(url: str) -> str:
+    """Extrait le domaine d'une URL"""
+    try:
+        parsed = urlparse(url)
+        domain = parsed.netloc.lower()
+        if domain.startswith("www."):
+            domain = domain[4:]
+        return domain
+    except Exception:
+        return ""
+
+
+async def health_check() -> bool:
+    """Vérifie la connectivité"""
+    try:
+        async with httpx.AsyncClient(timeout=5.0) as client:
+            response = await client.get("https://www.google.com")
+            return response.status_code == 200
+    except Exception:
+        return False

app/services/duckduckgo_service.py

@@ -1,178 +0,0 @@
-"""
-DuckDuckGo Search Service - Recherche de produits via DuckDuckGo
-"""
-
-import asyncio
-import logging
-from urllib.parse import urlparse
-
-from duckduckgo_search import DDGS
-
-logger = logging.getLogger(__name__)
-
-
-class SearchResult:
-    """Résultat de recherche DuckDuckGo"""
-
-    def __init__(self, url: str, title: str, snippet: str, source: str):
-        self.url = url
-        self.title = title
-        self.snippet = snippet
-        self.source = source
-
-    def __repr__(self):
-        return f"SearchResult(url={self.url}, title={self.title[:30]}...)"
-
-
-async def search(
-    query: str,
-    domains: list[str],
-    max_results: int = 20,
-    timeout: float = 15.0,
-) -> list[SearchResult]:
-    """
-    Effectue une recherche DuckDuckGo ciblée sur les domaines spécifiés.
-    Recherche chaque domaine séparément pour de meilleurs résultats.
-
-    Args:
-        query: Terme de recherche (ex: "lutin de noel")
-        domains: Liste des domaines à cibler (ex: ["amazon.fr", "fnac.com"])
-        max_results: Nombre maximum de résultats à retourner
-        timeout: Timeout en secondes
-
-    Returns:
-        Liste de SearchResult avec url, title, snippet, source
-    """
-    if not domains:
-        logger.warning("Aucun domaine spécifié pour la recherche")
-        return []
-
-    # Nettoyer les domaines (enlever protocole, www, etc.)
-    clean_domains = [_clean_domain(d) for d in domains]
-    clean_domains = [d for d in clean_domains if d]
-
-    logger.info(f"Domaines ciblés: {clean_domains}")
-
-    all_results = []
-    seen_urls = set()
-    results_per_domain = max(5, max_results // len(clean_domains))
-
-    # Rechercher sur chaque domaine séparément
-    for domain in clean_domains:
-        # Format de requête simple: "query site:domain"
-        full_query = f"{query} site:{domain}"
-        logger.info(f"Recherche DuckDuckGo: {full_query}")
-
-        try:
-            loop = asyncio.get_event_loop()
-            results = await loop.run_in_executor(
-                None,
-                lambda d=domain, q=full_query: _do_search(q, results_per_domain, timeout),
-            )
-
-            logger.info(f"DuckDuckGo [{domain}]: {len(results)} résultats bruts")
-
-            for item in results:
-                url = item.get("href", "") or item.get("link", "")
-                if not url or url in seen_urls:
-                    continue
-
-                seen_urls.add(url)
-                url_domain = _extract_domain(url)
-
-                # Vérifier que l'URL appartient au domaine ciblé
-                if domain not in url_domain and not url_domain.endswith(domain):
-                    logger.debug(f"URL ignorée (domaine {url_domain} != {domain}): {url}")
-                    continue
-
-                all_results.append(
-                    SearchResult(
-                        url=url,
-                        title=item.get("title", ""),
-                        snippet=item.get("body", "") or item.get("snippet", ""),
-                        source=url_domain,
-                    )
-                )
-
-                if len(all_results) >= max_results:
-                    break
-
-        except Exception as e:
-            logger.error(f"Erreur recherche sur {domain}: {e}")
-            continue
-
-        if len(all_results) >= max_results:
-            break
-
-    logger.info(f"DuckDuckGo: {len(all_results)} résultats totaux")
-    return all_results
-
-
-def _do_search(query: str, max_results: int, timeout: float) -> list[dict]:
-    """Effectue la recherche synchrone DuckDuckGo"""
-    try:
-        with DDGS() as ddgs:
-            results = list(ddgs.text(
-                query,
-                region="fr-fr",
-                safesearch="off",
-                max_results=max_results,
-            ))
-            logger.info(f"DDGS.text('{query[:50]}...') retourne {len(results)} résultats")
-
-            # Log les premiers résultats pour debug
-            for i, r in enumerate(results[:3]):
-                logger.info(f"  Résultat {i+1}: {r.get('href', 'N/A')[:80]}")
-
-            return results
-    except Exception as e:
-        logger.error(f"Erreur lors de la recherche DuckDuckGo: {e}")
-        return []
-
-
-def _extract_domain(url: str) -> str:
-    """Extrait le domaine d'une URL"""
-    try:
-        parsed = urlparse(url)
-        domain = parsed.netloc.lower()
-        # Retirer www. si présent
-        if domain.startswith("www."):
-            domain = domain[4:]
-        return domain
-    except Exception:
-        return ""
-
-
-def _clean_domain(domain: str) -> str:
-    """Nettoie un domaine (enlève protocole, www, slash final, etc.)"""
-    if not domain:
-        return ""
-
-    domain = domain.strip()
-
-    # Si c'est une URL complète, extraire le domaine
-    if domain.startswith(("http://", "https://")):
-        domain = _extract_domain(domain)
-    else:
-        # Nettoyer le domaine directement
-        domain = domain.lower()
-        if domain.startswith("www."):
-            domain = domain[4:]
-        # Retirer le slash final
-        domain = domain.rstrip("/")
-
-    return domain
-
-
-async def health_check() -> bool:
-    """Vérifie que DuckDuckGo est accessible"""
-    try:
-        loop = asyncio.get_event_loop()
-        results = await loop.run_in_executor(
-            None,
-            lambda: _do_search("test", 1, 5.0),
-        )
-        return len(results) > 0
-    except Exception as e:
-        logger.warning(f"DuckDuckGo health check failed: {e}")
-        return False

app/services/search_service.py

@@ -1,6 +1,6 @@
 """
 Search Service - Orchestrateur de recherche de produits
-Combine DuckDuckGo + Light Scraper + Browserless fallback
+Combine recherche directe sur sites + Light Scraper + Browserless fallback
 """
 
 import asyncio
@@ -12,7 +12,7 @@ from sqlalchemy.orm import Session
 
 from app.models import SearchSite
 from app.schemas import SearchProgress, SearchResultItem
-from app.services import ai_service, duckduckgo_service, light_scraper_service
+from app.services import ai_service, direct_search_service, light_scraper_service
 from app.services.scraper_service import ScraperService
 from app.services.settings_service import SettingsService
 
@@ -34,7 +34,7 @@ async def search_products(
     Recherche des produits sur les sites configurés.
 
     Flux:
-    1. SearXNG recherche les URLs
+    1. Recherche directe sur les pages de recherche de chaque site
     2. Light scraper tente extraction HTTP rapide
     3. Browserless fallback pour les sites JS
 
@@ -65,21 +65,31 @@ async def search_products(
         )
         return
 
-    domains = [site.domain for site in sites]
+    # Préparer les données des sites pour la recherche directe
+    sites_data = [
+        {
+            "domain": site.domain,
+            "name": site.name,
+            "search_url": site.search_url,
+            "product_link_selector": site.product_link_selector,
+            "requires_js": site.requires_js,
+        }
+        for site in sites
+    ]
     site_map = {site.domain: site for site in sites}
 
-    # Phase 1: Recherche DuckDuckGo
+    # Phase 1: Recherche directe sur les sites
     yield SearchProgress(
         status="searching",
         total=0,
         completed=0,
-        message=f"Recherche sur {len(domains)} sites...",
+        message=f"Recherche sur {len(sites)} sites...",
         results=[],
     )
 
-    search_results = await duckduckgo_service.search(
+    search_results = await direct_search_service.search(
         query=query,
-        domains=domains,
+        sites=sites_data,
         max_results=max_results,
     )
 
@@ -94,7 +104,7 @@ async def search_products(
         return
 
     total = len(search_results)
-    logger.info(f"DuckDuckGo: {total} URLs trouvées pour '{query}'")
+    logger.info(f"Recherche directe: {total} URLs trouvées pour '{query}'")
 
     # Phase 2: Scraping des URLs
     results: list[SearchResultItem] = []
@@ -103,7 +113,7 @@ async def search_products(
     # Créer un sémaphore pour limiter la concurrence
     semaphore = asyncio.Semaphore(parallel_limit)
 
-    async def process_url(search_result: duckduckgo_service.SearchResult) -> SearchResultItem | None:
+    async def process_url(search_result: direct_search_service.SearchResult) -> SearchResultItem | None:
         async with semaphore:
             url = search_result.url
             domain = search_result.source

pyproject.toml

@@ -23,7 +23,8 @@ dependencies = [
     "alembic",
     "tenacity",
     "slowapi",
-    "duckduckgo-search",
+    "beautifulsoup4",
+    "lxml",
 ]
 
 [build-system]