diff --git a/app/routers/search_sites.py b/app/routers/search_sites.py index d440463..ff08a04 100644 --- a/app/routers/search_sites.py +++ b/app/routers/search_sites.py @@ -3,11 +3,12 @@ API Router pour la gestion des sites de recherche """ 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 @@ -35,8 +36,6 @@ async def get_site(site_id: int, db: Session = Depends(get_db)): 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) @@ -78,7 +77,6 @@ async def create_site( # 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, @@ -90,7 +88,7 @@ async def create_site( 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) @@ -114,7 +112,6 @@ 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) diff --git a/app/services/search_url_discovery.py b/app/services/search_url_discovery.py index 22bb40a..f7d5265 100644 --- a/app/services/search_url_discovery.py +++ b/app/services/search_url_discovery.py @@ -5,16 +5,18 @@ Search URL Discovery Service - Découvre automatiquement l'URL de recherche d'un import asyncio import logging import os -import re -from urllib.parse import urljoin, urlparse +from urllib.parse import urljoin -from bs4 import BeautifulSoup 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}", @@ -101,7 +103,7 @@ async def discover_search_url(domain: str) -> dict: return result -async def _find_search_form(page, base_url: str, domain: str) -> str | None: +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 @@ -125,7 +127,11 @@ async def _find_search_form(page, base_url: str, domain: str) -> str | None: 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"]') + 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" @@ -192,7 +198,7 @@ async def _try_common_patterns(page, base_url: str, domain: str) -> str | None: 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: + 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(): @@ -229,7 +235,7 @@ async def _discover_product_selector(page, search_url: str, domain: str) -> str for selector in product_selectors: try: elements = await page.query_selector_all(selector) - if len(elements) >= 2: # Au moins 2 produits trouvés + if len(elements) >= MIN_PRODUCTS_TO_MATCH: return selector except Exception: continue