From 62b145ca8f464c008442c468e51085fe3ddbe7e8 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 22 Nov 2025 15:27:21 +0000 Subject: [PATCH] feat: Auto-discover search URLs when adding sites - Add search_url_discovery.py service that visits sites and finds search forms - Automatically discover search_url when a new site is created without one - Add POST /{site_id}/discover endpoint to force rediscovery - Discovery runs in background to not block the API response - Tries multiple methods: form analysis, common URL patterns --- app/routers/search_sites.py | 100 +++++++++- app/services/search_url_discovery.py | 263 +++++++++++++++++++++++++++ 2 files changed, 360 insertions(+), 3 deletions(-) create mode 100644 app/services/search_url_discovery.py diff --git a/app/routers/search_sites.py b/app/routers/search_sites.py index e8accfe..d440463 100644 --- a/app/routers/search_sites.py +++ b/app/routers/search_sites.py @@ -2,12 +2,17 @@ API Router pour la gestion des sites de recherche """ -from fastapi import APIRouter, Depends, HTTPException +import logging + +from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException from sqlalchemy.orm import Session from app.database import 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,11 +33,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""" + from app.database import SessionLocal + + 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: + import os + 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)) @@ -60,3 +115,42 @@ async def delete_site(site_id: int, db: Session = Depends(get_db)): 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", + } diff --git a/app/services/search_url_discovery.py b/app/services/search_url_discovery.py new file mode 100644 index 0000000..22bb40a --- /dev/null +++ b/app/services/search_url_discovery.py @@ -0,0 +1,263 @@ +""" +Search URL Discovery Service - Découvre automatiquement l'URL de recherche d'un site +""" + +import asyncio +import logging +import os +import re +from urllib.parse import urljoin, urlparse + +from bs4 import BeautifulSoup +from playwright.async_api import async_playwright + +logger = logging.getLogger(__name__) + +BROWSERLESS_URL = os.getenv("BROWSERLESS_URL", "ws://browserless:3000") + +# 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: + """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_field = await form.query_selector('input[type="search"], input[type="text"], input[name*="q"], input[name*="query"], input[name*="search"]') + 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 == 200: + # 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) >= 2: # Au moins 2 produits trouvés + 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