fix: Clean up linting issues in search URL discovery

- Move imports to top-level in search_sites router
- Fix line too long issue with input selector
- Define constants for magic numbers (HTTP_OK, MIN_PRODUCTS_TO_MATCH)
- Use 'raise from' pattern for proper exception chaining
This commit is contained in:
Claude committed 2025-11-22 15:31:12 +00:00
1 parent 62b145ca8f
commit aad16985d4
2 files changed
+16 -13

No files matched your search

+3 -6
View File
@@ -3,11 +3,12 @@ API Router pour la gestion des sites de recherche
""" """
import logging import logging
import os
from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException
from sqlalchemy.orm import Session 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.schemas import SearchSiteCreate, SearchSiteResponse, SearchSiteUpdate
from app.services import search_service from app.services import search_service
from app.services.search_url_discovery import discover_search_url 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): 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""" """Découvre l'URL de recherche en background et met à jour le site"""
from app.database import SessionLocal
try: try:
logger.info(f"Discovering search URL for site {site_id} ({domain})") logger.info(f"Discovering search URL for site {site_id} ({domain})")
result = await discover_search_url(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 # Si pas de search_url, lancer la découverte en background
if not site.search_url: if not site.search_url:
import os
db_url = os.getenv("DATABASE_URL", "") db_url = os.getenv("DATABASE_URL", "")
background_tasks.add_task( background_tasks.add_task(
_discover_and_update_site, _discover_and_update_site,
@@ -90,7 +88,7 @@ async def create_site(
return new_site return new_site
except Exception as e: 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) @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) deleted = search_service.delete_site(db, site_id)
if not deleted: if not deleted:
raise HTTPException(status_code=404, detail="Site not found") raise HTTPException(status_code=404, detail="Site not found")
return None
@router.post("/{site_id}/discover", response_model=dict) @router.post("/{site_id}/discover", response_model=dict)
+13 -7
View File
@@ -5,16 +5,18 @@ Search URL Discovery Service - Découvre automatiquement l'URL de recherche d'un
import asyncio import asyncio
import logging import logging
import os import os
import re from urllib.parse import urljoin
from urllib.parse import urljoin, urlparse
from bs4 import BeautifulSoup
from playwright.async_api import async_playwright from playwright.async_api import async_playwright
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
BROWSERLESS_URL = os.getenv("BROWSERLESS_URL", "ws://browserless:3000") 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 # Patterns communs pour les URLs de recherche
COMMON_SEARCH_PATTERNS = [ COMMON_SEARCH_PATTERNS = [
"/search?q={query}", "/search?q={query}",
@@ -101,7 +103,7 @@ async def discover_search_url(domain: str) -> dict:
return result 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""" """Trouve le formulaire de recherche sur la page"""
try: try:
# Chercher les formulaires de recherche # 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") action = await form.get_attribute("action")
if action: if action:
# Trouver le nom du champ de recherche # 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: if input_field:
input_name = await input_field.get_attribute("name") or "q" 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") test_url = f"https://www.{domain}{pattern}".replace("{query}", "test")
try: try:
response = await page.goto(test_url, wait_until="domcontentloaded", timeout=10000) 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 # Vérifier qu'on est sur une page de résultats
content = await page.content() content = await page.content()
if "test" in content.lower() or "résultat" in content.lower() or "result" in content.lower(): 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: for selector in product_selectors:
try: try:
elements = await page.query_selector_all(selector) 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 return selector
except Exception: except Exception:
continue continue