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

Claude/design product search 01 ea9d w zq zpu k fw bw vmp26 pg

.gitignore

@@ -0,0 +1,51 @@
+# Python
+__pycache__/
+*.py[cod]
+*$py.class
+*.so
+.Python
+build/
+develop-eggs/
+dist/
+downloads/
+eggs/
+.eggs/
+lib/
+lib64/
+parts/
+sdist/
+var/
+wheels/
+*.egg-info/
+.installed.cfg
+*.egg
+
+# Virtual environments
+.env
+.venv
+env/
+venv/
+ENV/
+
+# IDE
+.idea/
+.vscode/
+*.swp
+*.swo
+
+# Testing
+.coverage
+htmlcov/
+.tox/
+.pytest_cache/
+.mypy_cache/
+
+# Logs
+*.log
+
+# Screenshots (generated)
+screenshots/
+
+# Database
+*.db
+*.sqlite3

app/routers/search_sites.py

@@ -2,12 +2,18 @@
 API Router pour la gestion des sites de recherche
 """
 
-from fastapi import APIRouter, Depends, HTTPException
+import logging
+import os
+
+from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException
 from sqlalchemy.orm import Session
 
-from app.database import get_db
+from app.database import SessionLocal, get_db
 from app.schemas import SearchSiteCreate, SearchSiteResponse, SearchSiteUpdate
 from app.services import search_service
+from app.services.search_url_discovery import discover_search_url
+
+logger = logging.getLogger(__name__)
 
 router = APIRouter(prefix="/api/search-sites", tags=["search-sites"])
 
@@ -28,14 +34,61 @@ async def get_site(site_id: int, db: Session = Depends(get_db)):
     return site
 
 
+async def _discover_and_update_site(site_id: int, domain: str, db_url: str):
+    """Découvre l'URL de recherche en background et met à jour le site"""
+    try:
+        logger.info(f"Discovering search URL for site {site_id} ({domain})")
+        result = await discover_search_url(domain)
+
+        if result.get("discovered"):
+            # Mettre à jour le site avec l'URL découverte
+            db = SessionLocal()
+            try:
+                update_data = {}
+                if result.get("search_url"):
+                    update_data["search_url"] = result["search_url"]
+                if result.get("product_link_selector"):
+                    update_data["product_link_selector"] = result["product_link_selector"]
+
+                if update_data:
+                    search_service.update_site(db, site_id, update_data)
+                    logger.info(f"Updated site {site_id} with discovered URL: {result.get('search_url')}")
+            finally:
+                db.close()
+        else:
+            logger.warning(f"Could not discover search URL for {domain}")
+
+    except Exception as e:
+        logger.error(f"Error in background discovery for {domain}: {e}")
+
+
 @router.post("", response_model=SearchSiteResponse, status_code=201)
-async def create_site(site: SearchSiteCreate, db: Session = Depends(get_db)):
-    """Crée un nouveau site de recherche"""
+async def create_site(
+    site: SearchSiteCreate,
+    background_tasks: BackgroundTasks,
+    db: Session = Depends(get_db),
+):
+    """
+    Crée un nouveau site de recherche.
+    Si search_url n'est pas fourni, tente de le découvrir automatiquement.
+    """
     try:
         new_site = search_service.create_site(db, site.model_dump())
+
+        # Si pas de search_url, lancer la découverte en background
+        if not site.search_url:
+            db_url = os.getenv("DATABASE_URL", "")
+            background_tasks.add_task(
+                _discover_and_update_site,
+                new_site.id,
+                new_site.domain,
+                db_url,
+            )
+            logger.info(f"Started background discovery for site {new_site.id}")
+
         return new_site
     except Exception as e:
-        raise HTTPException(status_code=400, detail=str(e))
+        raise HTTPException(status_code=400, detail=str(e)) from e
 
 
 @router.put("/{site_id}", response_model=SearchSiteResponse)
@@ -59,4 +112,42 @@ async def delete_site(site_id: int, db: Session = Depends(get_db)):
     deleted = search_service.delete_site(db, site_id)
     if not deleted:
         raise HTTPException(status_code=404, detail="Site not found")
-    return None
+
+
+@router.post("/{site_id}/discover", response_model=dict)
+async def discover_site_search_url(
+    site_id: int,
+    db: Session = Depends(get_db),
+):
+    """
+    Force la découverte de l'URL de recherche pour un site existant.
+    Utile si la découverte automatique n'a pas fonctionné.
+    """
+    site = search_service.get_site_by_id(db, site_id)
+    if not site:
+        raise HTTPException(status_code=404, detail="Site not found")
+
+    result = await discover_search_url(site.domain)
+
+    if result.get("discovered"):
+        # Mettre à jour le site
+        update_data = {}
+        if result.get("search_url"):
+            update_data["search_url"] = result["search_url"]
+        if result.get("product_link_selector"):
+            update_data["product_link_selector"] = result["product_link_selector"]
+
+        if update_data:
+            search_service.update_site(db, site_id, update_data)
+
+        return {
+            "success": True,
+            "message": f"URL de recherche découverte: {result.get('search_url')}",
+            "search_url": result.get("search_url"),
+            "product_link_selector": result.get("product_link_selector"),
+        }
+    else:
+        return {
+            "success": False,
+            "message": "Impossible de découvrir l'URL de recherche automatiquement",
+        }

app/services/direct_search_service.py

@@ -67,6 +67,52 @@ DEFAULT_SITE_CONFIGS = {
         "product_selector": ".product-list__item a, a[href*='/ref/']",
         "wait_selector": ".product-list",
     },
+    # Magasins discount français
+    "stokomani.fr": {
+        "search_url": "https://www.stokomani.fr/recherche?q={query}",
+        "product_selector": "a.product-card, .product-item a, a[href*='/product']",
+        "wait_selector": ".product-card, .product-item, .search-results",
+    },
+    "gifi.fr": {
+        "search_url": "https://www.gifi.fr/catalogsearch/result/?q={query}",
+        "product_selector": ".product-item a.product-item-link, a[href*='/produit/']",
+        "wait_selector": ".product-item, .products-grid",
+    },
+    "bmstores.fr": {
+        "search_url": "https://www.bmstores.fr/search?q={query}",
+        "product_selector": "a.product-card, .product a, a[href*='/products/']",
+        "wait_selector": ".product-card, .product-grid",
+    },
+    "centrakor.com": {
+        "search_url": "https://www.centrakor.com/recherche?search={query}",
+        "product_selector": "a.product-card, .product-item a",
+        "wait_selector": ".product-card, .product-list",
+    },
+    "lafoirfouille.fr": {
+        "search_url": "https://www.lafoirfouille.fr/recherche?q={query}",
+        "product_selector": "a.product-item-link, .product-card a",
+        "wait_selector": ".product-item, .products",
+    },
+    "tedi.fr": {
+        "search_url": "https://www.tedi.fr/search?q={query}",
+        "product_selector": "a.product-card, .product a",
+        "wait_selector": ".product-card, .search-results",
+    },
+    "leclerc.fr": {
+        "search_url": "https://www.e.leclerc/recherche?q={query}",
+        "product_selector": "a.product-card, .product-tile a",
+        "wait_selector": ".product-card, .products-grid",
+    },
+    "carrefour.fr": {
+        "search_url": "https://www.carrefour.fr/s?q={query}",
+        "product_selector": "a.product-card, .product-item a",
+        "wait_selector": ".product-card, .product-list",
+    },
+    "auchan.fr": {
+        "search_url": "https://www.auchan.fr/recherche?text={query}",
+        "product_selector": "a.product-card, .product-tile a",
+        "wait_selector": ".product-card, .products",
+    },
 }
 
 

app/services/search_service.py

@@ -5,6 +5,7 @@ Combine recherche directe sur sites + Light Scraper + Browserless fallback
 
 import asyncio
 import logging
+import os
 from collections.abc import AsyncGenerator
 from typing import Any
 
@@ -233,6 +234,12 @@ async def _scrape_with_browserless(
             page_text=page_text,
         )
 
+        # Construire l'URL du screenshot pour le frontend
+        screenshot_url = None
+        if screenshot_path:
+            screenshot_filename = os.path.basename(screenshot_path)
+            screenshot_url = f"/screenshots/{screenshot_filename}"
+
         if ai_result:
             extraction, metadata = ai_result
             if extraction and extraction.price is not None:
@@ -242,7 +249,7 @@ async def _scrape_with_browserless(
                     price=extraction.price,
                     currency=extraction.currency or "EUR",
                     in_stock=extraction.in_stock,
-                    image_url=None,
+                    image_url=screenshot_url,
                     site_name=site_name,
                     site_domain=domain,
                     confidence=extraction.price_confidence or 0.5,
@@ -252,6 +259,7 @@ async def _scrape_with_browserless(
             url=url,
             title=fallback_title,
             price=None,
+            image_url=screenshot_url,
             site_name=site_name,
             site_domain=domain,
             error="AI extraction failed",

app/services/search_url_discovery.py

@@ -0,0 +1,269 @@
+"""
+Search URL Discovery Service - Découvre automatiquement l'URL de recherche d'un site
+"""
+
+import asyncio
+import logging
+import os
+from urllib.parse import urljoin
+
+from playwright.async_api import async_playwright
+
+logger = logging.getLogger(__name__)
+
+BROWSERLESS_URL = os.getenv("BROWSERLESS_URL", "ws://browserless:3000")
+
+# Constants
+HTTP_OK = 200
+MIN_PRODUCTS_TO_MATCH = 2
+
+# Patterns communs pour les URLs de recherche
+COMMON_SEARCH_PATTERNS = [
+    "/search?q={query}",
+    "/recherche?q={query}",
+    "/s?k={query}",
+    "/s?q={query}",
+    "/catalogsearch/result/?q={query}",
+    "/search?query={query}",
+    "/recherche?query={query}",
+    "/search?text={query}",
+    "/recherche?text={query}",
+    "/search/{query}",
+    "/find?q={query}",
+]
+
+
+async def discover_search_url(domain: str) -> dict:
+    """
+    Découvre l'URL de recherche d'un site en visitant sa page d'accueil.
+
+    Args:
+        domain: Le domaine du site (ex: "amazon.fr", "action.com")
+
+    Returns:
+        Dict avec search_url et product_link_selector si trouvés
+    """
+    # Nettoyer le domaine
+    domain = _clean_domain(domain)
+    base_url = f"https://www.{domain}"
+
+    logger.info(f"Discovering search URL for {domain}")
+
+    result = {
+        "search_url": None,
+        "product_link_selector": None,
+        "discovered": False,
+    }
+
+    try:
+        async with async_playwright() as p:
+            browser = await p.chromium.connect_over_cdp(BROWSERLESS_URL)
+
+            try:
+                context = await browser.new_context(
+                    viewport={"width": 1920, "height": 1080},
+                    user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
+                    locale="fr-FR",
+                )
+                page = await context.new_page()
+
+                # Visiter la page d'accueil
+                await page.goto(base_url, wait_until="domcontentloaded", timeout=30000)
+                await asyncio.sleep(2)  # Attendre le JS
+
+                # Méthode 1: Chercher un formulaire de recherche
+                search_url = await _find_search_form(page, base_url, domain)
+
+                if search_url:
+                    result["search_url"] = search_url
+                    result["discovered"] = True
+                    logger.info(f"Found search URL via form: {search_url}")
+                else:
+                    # Méthode 2: Essayer les patterns communs
+                    search_url = await _try_common_patterns(page, base_url, domain)
+                    if search_url:
+                        result["search_url"] = search_url
+                        result["discovered"] = True
+                        logger.info(f"Found search URL via pattern: {search_url}")
+
+                # Essayer de trouver un sélecteur de produits si on a une URL de recherche
+                if result["search_url"]:
+                    selector = await _discover_product_selector(page, result["search_url"], domain)
+                    if selector:
+                        result["product_link_selector"] = selector
+
+                await context.close()
+
+            finally:
+                await browser.close()
+
+    except Exception as e:
+        logger.error(f"Error discovering search URL for {domain}: {e}")
+
+    return result
+
+
+async def _find_search_form(page, base_url: str, domain: str) -> str | None:  # noqa: PLR0912
+    """Trouve le formulaire de recherche sur la page"""
+    try:
+        # Chercher les formulaires de recherche
+        search_selectors = [
+            'form[action*="search"]',
+            'form[action*="recherche"]',
+            'form[action*="/s"]',
+            'form[role="search"]',
+            'form.search-form',
+            'form.search',
+            'form#search',
+            'form#search-form',
+            '.search-form form',
+            'header form',
+        ]
+
+        for selector in search_selectors:
+            try:
+                form = await page.query_selector(selector)
+                if form:
+                    action = await form.get_attribute("action")
+                    if action:
+                        # Trouver le nom du champ de recherche
+                        input_selector = (
+                            'input[type="search"], input[type="text"], '
+                            'input[name*="q"], input[name*="query"], input[name*="search"]'
+                        )
+                        input_field = await form.query_selector(input_selector)
+                        if input_field:
+                            input_name = await input_field.get_attribute("name") or "q"
+
+                            # Construire l'URL de recherche
+                            if action.startswith("/"):
+                                search_url = f"https://www.{domain}{action}"
+                            elif action.startswith("http"):
+                                search_url = action
+                            else:
+                                search_url = urljoin(base_url, action)
+
+                            # Ajouter le paramètre de recherche
+                            separator = "&" if "?" in search_url else "?"
+                            search_url = f"{search_url}{separator}{input_name}={{query}}"
+
+                            return search_url
+            except Exception:
+                continue
+
+        # Méthode alternative: chercher les inputs de recherche
+        input_selectors = [
+            'input[type="search"]',
+            'input[name="q"]',
+            'input[name="query"]',
+            'input[name="search"]',
+            'input[name="s"]',
+            'input[placeholder*="recherche" i]',
+            'input[placeholder*="search" i]',
+        ]
+
+        for selector in input_selectors:
+            try:
+                input_el = await page.query_selector(selector)
+                if input_el:
+                    # Trouver le formulaire parent
+                    form = await input_el.evaluate_handle("el => el.closest('form')")
+                    if form:
+                        action = await form.get_attribute("action")
+                        input_name = await input_el.get_attribute("name") or "q"
+
+                        if action:
+                            if action.startswith("/"):
+                                search_url = f"https://www.{domain}{action}"
+                            elif action.startswith("http"):
+                                search_url = action
+                            else:
+                                search_url = urljoin(base_url, action)
+
+                            separator = "&" if "?" in search_url else "?"
+                            search_url = f"{search_url}{separator}{input_name}={{query}}"
+                            return search_url
+            except Exception:
+                continue
+
+    except Exception as e:
+        logger.debug(f"Error finding search form: {e}")
+
+    return None
+
+
+async def _try_common_patterns(page, base_url: str, domain: str) -> str | None:
+    """Essaye les patterns de recherche communs"""
+    for pattern in COMMON_SEARCH_PATTERNS:
+        test_url = f"https://www.{domain}{pattern}".replace("{query}", "test")
+        try:
+            response = await page.goto(test_url, wait_until="domcontentloaded", timeout=10000)
+            if response and response.status == HTTP_OK:
+                # Vérifier qu'on est sur une page de résultats
+                content = await page.content()
+                if "test" in content.lower() or "résultat" in content.lower() or "result" in content.lower():
+                    return f"https://www.{domain}{pattern}"
+        except Exception:
+            continue
+
+    return None
+
+
+async def _discover_product_selector(page, search_url: str, domain: str) -> str | None:
+    """Découvre le sélecteur CSS pour les liens produits"""
+    try:
+        # Faire une recherche test
+        test_url = search_url.replace("{query}", "test")
+        await page.goto(test_url, wait_until="domcontentloaded", timeout=20000)
+        await asyncio.sleep(2)
+
+        # Sélecteurs communs pour les produits
+        product_selectors = [
+            "a.product-card",
+            "a.product-link",
+            ".product-card a",
+            ".product-item a",
+            ".product a",
+            "a[href*='/product']",
+            "a[href*='/p/']",
+            "a[href*='/dp/']",
+            "a[href*='/item']",
+            ".search-result a",
+            ".results a.product",
+        ]
+
+        for selector in product_selectors:
+            try:
+                elements = await page.query_selector_all(selector)
+                if len(elements) >= MIN_PRODUCTS_TO_MATCH:
+                    return selector
+            except Exception:
+                continue
+
+    except Exception as e:
+        logger.debug(f"Error discovering product selector: {e}")
+
+    return None
+
+
+def _clean_domain(domain: str) -> str:
+    """Nettoie un domaine"""
+    if not domain:
+        return ""
+
+    domain = domain.strip().lower()
+
+    # Enlever le protocole
+    if domain.startswith("https://"):
+        domain = domain[8:]
+    elif domain.startswith("http://"):
+        domain = domain[7:]
+
+    # Enlever www.
+    if domain.startswith("www."):
+        domain = domain[4:]
+
+    # Enlever le slash final
+    domain = domain.rstrip("/")
+
+    return domain