mirror of
https://github.com/R0m1k3/Priceflow.git
synced 2026-10-24 07:38:13 +02:00
feat: Add multi-site product search system with SearXNG integration
This commit adds a comprehensive product search feature that allows users to search for products across multiple configured e-commerce sites. Backend changes: - Add SearXNG service to docker-compose for meta-search capability - Add SearchSite model and Alembic migration with default sites - Create searxng_service.py for search queries across sites - Create light_scraper_service.py for fast HTTP-based extraction - Create search_service.py orchestrator combining SearXNG + HTTP + Browserless - Add API endpoints for search (/api/search) with SSE streaming - Add CRUD endpoints for search sites (/api/search-sites) Frontend changes: - Add new /search page with real-time results display - Create SearchResultCard component for product display - Add Progress, Checkbox, and Badge UI components - Update Settings page with search sites management section - Add navigation link to search page - Create new modern SVG logo with price-flow gradient design - Update favicon to SVG format The search system uses a hybrid approach: 1. SearXNG finds product URLs across configured sites 2. Light HTTP scraper attempts fast extraction 3. Falls back to Browserless for JavaScript-heavy sites 4. Results stream progressively via SSE
This commit is contained in:
26 files changed
+1831
-10
No files matched your search
@@ -0,0 +1,52 @@
|
||||
"""Add search_sites table for product search
|
||||
|
||||
Revision ID: b2c3d4e5f6g7
|
||||
Revises: a1b2c3d4e5f6
|
||||
Create Date: 2025-11-22 10:00:00.000000
|
||||
|
||||
"""
|
||||
|
||||
from collections.abc import Sequence
|
||||
|
||||
import sqlalchemy as sa
|
||||
|
||||
from alembic import op
|
||||
|
||||
# revision identifiers, used by Alembic.
|
||||
revision: str = "b2c3d4e5f6g7"
|
||||
down_revision: str | None = "a1b2c3d4e5f6"
|
||||
branch_labels: str | Sequence[str] | None = None
|
||||
depends_on: str | Sequence[str] | None = None
|
||||
|
||||
|
||||
def upgrade() -> None:
|
||||
"""Create search_sites table for configurable product search sites."""
|
||||
op.create_table(
|
||||
"search_sites",
|
||||
sa.Column("id", sa.Integer(), primary_key=True, index=True),
|
||||
sa.Column("name", sa.String(), nullable=False),
|
||||
sa.Column("domain", sa.String(), nullable=False, unique=True, index=True),
|
||||
sa.Column("logo_url", sa.String(), nullable=True),
|
||||
sa.Column("category", sa.String(), nullable=True),
|
||||
sa.Column("is_active", sa.Boolean(), default=True),
|
||||
sa.Column("priority", sa.Integer(), default=0),
|
||||
sa.Column("requires_js", sa.Boolean(), default=False),
|
||||
sa.Column("created_at", sa.DateTime(), nullable=True),
|
||||
sa.Column("updated_at", sa.DateTime(), nullable=True),
|
||||
)
|
||||
|
||||
# Insert default popular sites
|
||||
op.execute("""
|
||||
INSERT INTO search_sites (name, domain, category, is_active, priority, requires_js, created_at, updated_at)
|
||||
VALUES
|
||||
('Amazon France', 'amazon.fr', 'Généraliste', true, 1, true, NOW(), NOW()),
|
||||
('Fnac', 'fnac.com', 'Tech / Culture', true, 2, true, NOW(), NOW()),
|
||||
('Cdiscount', 'cdiscount.com', 'Généraliste', true, 3, true, NOW(), NOW()),
|
||||
('Darty', 'darty.com', 'Électroménager', true, 4, true, NOW(), NOW()),
|
||||
('Boulanger', 'boulanger.com', 'Électronique', true, 5, true, NOW(), NOW())
|
||||
""")
|
||||
|
||||
|
||||
def downgrade() -> None:
|
||||
"""Drop search_sites table."""
|
||||
op.drop_table("search_sites")
|
||||
+5
-1
@@ -11,7 +11,7 @@ from slowapi import _rate_limit_exceeded_handler
|
||||
from slowapi.errors import RateLimitExceeded
|
||||
|
||||
from app.limiter import limiter
|
||||
from app.routers import items, jobs, notifications, openrouter, settings
|
||||
from app.routers import items, jobs, notifications, openrouter, search, search_sites, settings
|
||||
from app.services.scheduler_service import scheduled_refresh, scheduler
|
||||
|
||||
# Configure logging
|
||||
@@ -66,6 +66,10 @@ app.mount("/screenshots", StaticFiles(directory="screenshots"), name="screenshot
|
||||
for router in [notifications.router, items.router, settings.router, jobs.router, openrouter.router]:
|
||||
app.include_router(router, prefix="/api")
|
||||
|
||||
# Search routers (already have /api prefix)
|
||||
app.include_router(search.router)
|
||||
app.include_router(search_sites.router)
|
||||
|
||||
|
||||
@app.get("/api/")
|
||||
def read_root():
|
||||
|
||||
@@ -77,3 +77,19 @@ class Settings(Base):
|
||||
|
||||
key = Column(String, primary_key=True, index=True)
|
||||
value = Column(Text)
|
||||
|
||||
|
||||
class SearchSite(Base):
|
||||
"""Sites configurés pour la recherche de produits"""
|
||||
__tablename__ = "search_sites"
|
||||
|
||||
id: int = Column(Integer, primary_key=True, index=True) # type: ignore
|
||||
name: str = Column(String, nullable=False) # type: ignore # "Amazon France"
|
||||
domain: str = Column(String, unique=True, nullable=False, index=True) # type: ignore # "amazon.fr"
|
||||
logo_url: str | None = Column(String, nullable=True) # type: ignore
|
||||
category: str | None = Column(String, nullable=True) # type: ignore # "Tech", "Déco", "Généraliste"
|
||||
is_active: bool = Column(Boolean, default=True) # type: ignore
|
||||
priority: int = Column(Integer, default=0) # type: ignore # Ordre d'affichage
|
||||
requires_js: bool = Column(Boolean, default=False) # type: ignore # Force Browserless si True
|
||||
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
|
||||
@@ -0,0 +1,89 @@
|
||||
"""
|
||||
API Router pour la recherche de produits
|
||||
Utilise Server-Sent Events (SSE) pour le streaming des résultats
|
||||
"""
|
||||
|
||||
import json
|
||||
import logging
|
||||
|
||||
from fastapi import APIRouter, Depends, Query
|
||||
from fastapi.responses import StreamingResponse
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from app.database import get_db
|
||||
from app.services import search_service
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
router = APIRouter(prefix="/api/search", tags=["search"])
|
||||
|
||||
|
||||
@router.get("")
|
||||
async def search_products(
|
||||
q: str = Query(..., min_length=1, description="Terme de recherche"),
|
||||
sites: str | None = Query(None, description="IDs des sites, séparés par virgule"),
|
||||
max_results: int = Query(20, ge=1, le=50, description="Nombre max de résultats"),
|
||||
db: Session = Depends(get_db),
|
||||
):
|
||||
"""
|
||||
Recherche des produits sur les sites configurés.
|
||||
|
||||
Retourne un flux SSE avec les résultats au fur et à mesure.
|
||||
|
||||
Format des événements SSE:
|
||||
- event: progress
|
||||
- data: {"status": "...", "total": N, "completed": N, "results": [...]}
|
||||
"""
|
||||
# Parser les IDs de sites
|
||||
site_ids = None
|
||||
if sites:
|
||||
try:
|
||||
site_ids = [int(s.strip()) for s in sites.split(",") if s.strip()]
|
||||
except ValueError:
|
||||
site_ids = None
|
||||
|
||||
async def generate():
|
||||
try:
|
||||
async for progress in search_service.search_products(
|
||||
query=q,
|
||||
db=db,
|
||||
site_ids=site_ids,
|
||||
max_results=max_results,
|
||||
):
|
||||
# Format SSE
|
||||
data = progress.model_dump_json()
|
||||
yield f"event: progress\ndata: {data}\n\n"
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Erreur recherche: {e}")
|
||||
error_data = json.dumps({
|
||||
"status": "error",
|
||||
"total": 0,
|
||||
"completed": 0,
|
||||
"message": str(e),
|
||||
"results": [],
|
||||
})
|
||||
yield f"event: progress\ndata: {error_data}\n\n"
|
||||
|
||||
return StreamingResponse(
|
||||
generate(),
|
||||
media_type="text/event-stream",
|
||||
headers={
|
||||
"Cache-Control": "no-cache",
|
||||
"Connection": "keep-alive",
|
||||
"X-Accel-Buffering": "no",
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
@router.get("/health")
|
||||
async def search_health():
|
||||
"""Vérifie que le service de recherche est opérationnel"""
|
||||
from app.services import searxng_service
|
||||
|
||||
searxng_ok = await searxng_service.health_check()
|
||||
|
||||
return {
|
||||
"status": "ok" if searxng_ok else "degraded",
|
||||
"searxng": "ok" if searxng_ok else "unavailable",
|
||||
}
|
||||
@@ -0,0 +1,62 @@
|
||||
"""
|
||||
API Router pour la gestion des sites de recherche
|
||||
"""
|
||||
|
||||
from fastapi import APIRouter, 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
|
||||
|
||||
router = APIRouter(prefix="/api/search-sites", tags=["search-sites"])
|
||||
|
||||
|
||||
@router.get("", response_model=list[SearchSiteResponse])
|
||||
async def get_all_sites(db: Session = Depends(get_db)):
|
||||
"""Récupère tous les sites de recherche"""
|
||||
sites = search_service.get_all_sites(db)
|
||||
return sites
|
||||
|
||||
|
||||
@router.get("/{site_id}", response_model=SearchSiteResponse)
|
||||
async def get_site(site_id: int, db: Session = Depends(get_db)):
|
||||
"""Récupère un site par son ID"""
|
||||
site = search_service.get_site_by_id(db, site_id)
|
||||
if not site:
|
||||
raise HTTPException(status_code=404, detail="Site not found")
|
||||
return site
|
||||
|
||||
|
||||
@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"""
|
||||
try:
|
||||
new_site = search_service.create_site(db, site.model_dump())
|
||||
return new_site
|
||||
except Exception as e:
|
||||
raise HTTPException(status_code=400, detail=str(e))
|
||||
|
||||
|
||||
@router.put("/{site_id}", response_model=SearchSiteResponse)
|
||||
async def update_site(
|
||||
site_id: int,
|
||||
site: SearchSiteUpdate,
|
||||
db: Session = Depends(get_db),
|
||||
):
|
||||
"""Met à jour un site de recherche"""
|
||||
updated = search_service.update_site(
|
||||
db, site_id, site.model_dump(exclude_unset=True)
|
||||
)
|
||||
if not updated:
|
||||
raise HTTPException(status_code=404, detail="Site not found")
|
||||
return updated
|
||||
|
||||
|
||||
@router.delete("/{site_id}", status_code=204)
|
||||
async def delete_site(site_id: int, db: Session = Depends(get_db)):
|
||||
"""Supprime un site de recherche"""
|
||||
deleted = search_service.delete_site(db, site_id)
|
||||
if not deleted:
|
||||
raise HTTPException(status_code=404, detail="Site not found")
|
||||
return None
|
||||
@@ -50,3 +50,71 @@ class ItemResponse(ItemCreate):
|
||||
class SettingsUpdate(BaseModel):
|
||||
key: str
|
||||
value: str
|
||||
|
||||
|
||||
# === Search Sites Schemas ===
|
||||
|
||||
class SearchSiteCreate(BaseModel):
|
||||
name: str
|
||||
domain: str
|
||||
logo_url: str | None = None
|
||||
category: str | None = None
|
||||
is_active: bool = True
|
||||
priority: int = 0
|
||||
requires_js: bool = False
|
||||
|
||||
|
||||
class SearchSiteUpdate(BaseModel):
|
||||
name: str | None = None
|
||||
domain: str | None = None
|
||||
logo_url: str | None = None
|
||||
category: str | None = None
|
||||
is_active: bool | None = None
|
||||
priority: int | None = None
|
||||
requires_js: bool | None = None
|
||||
|
||||
|
||||
class SearchSiteResponse(BaseModel):
|
||||
id: int
|
||||
name: str
|
||||
domain: str
|
||||
logo_url: str | None = None
|
||||
category: str | None = None
|
||||
is_active: bool
|
||||
priority: int
|
||||
requires_js: bool
|
||||
created_at: datetime | None = None
|
||||
updated_at: datetime | None = None
|
||||
model_config = ConfigDict(from_attributes=True)
|
||||
|
||||
|
||||
# === Search Results Schemas ===
|
||||
|
||||
class SearchQuery(BaseModel):
|
||||
query: str
|
||||
site_ids: list[int] | None = None # If None, use all active sites
|
||||
max_results: int = 20
|
||||
|
||||
|
||||
class SearchResultItem(BaseModel):
|
||||
"""Un produit trouvé lors de la recherche"""
|
||||
url: str
|
||||
title: str
|
||||
price: float | None = None
|
||||
currency: str = "EUR"
|
||||
in_stock: bool | None = None
|
||||
image_url: str | None = None
|
||||
site_name: str
|
||||
site_domain: str
|
||||
confidence: float | None = None
|
||||
error: str | None = None
|
||||
|
||||
|
||||
class SearchProgress(BaseModel):
|
||||
"""Progression de la recherche pour SSE"""
|
||||
status: str # "searching", "scraping", "completed", "error"
|
||||
total: int
|
||||
completed: int
|
||||
current_site: str | None = None
|
||||
results: list[SearchResultItem] = []
|
||||
message: str | None = None
|
||||
@@ -0,0 +1,276 @@
|
||||
"""
|
||||
Light Scraper Service - Scraping HTTP rapide sans navigateur
|
||||
Utilisé comme première tentative avant le fallback Browserless
|
||||
"""
|
||||
|
||||
import logging
|
||||
import re
|
||||
from dataclasses import dataclass
|
||||
|
||||
import httpx
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Headers pour simuler un navigateur
|
||||
DEFAULT_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-US;q=0.8,en;q=0.7",
|
||||
"Accept-Encoding": "gzip, deflate, br",
|
||||
"Connection": "keep-alive",
|
||||
"Upgrade-Insecure-Requests": "1",
|
||||
"Cache-Control": "max-age=0",
|
||||
}
|
||||
|
||||
# Patterns regex pour extraire les prix
|
||||
PRICE_PATTERNS = [
|
||||
# Format européen : 1 234,56 € ou 1234,56€
|
||||
r'(\d{1,3}(?:\s?\d{3})*[,\.]\d{2})\s*(?:€|EUR|euros?)',
|
||||
# Format : €1,234.56 ou € 1234.56
|
||||
r'(?:€|EUR)\s*(\d{1,3}(?:[,\s]?\d{3})*[,\.]\d{2})',
|
||||
# Prix simples : 12,99 € ou 12.99€
|
||||
r'(\d+[,\.]\d{2})\s*(?:€|EUR)',
|
||||
# data-price ou price attributes
|
||||
r'data-price["\']?\s*[:=]\s*["\']?(\d+[,\.]?\d*)',
|
||||
r'price["\']?\s*[:=]\s*["\']?(\d+[,\.]?\d*)',
|
||||
]
|
||||
|
||||
# Patterns pour le stock
|
||||
STOCK_POSITIVE_PATTERNS = [
|
||||
r'en\s*stock',
|
||||
r'disponible',
|
||||
r'in\s*stock',
|
||||
r'available',
|
||||
r'ajouter\s*au\s*panier',
|
||||
r'add\s*to\s*cart',
|
||||
r'acheter',
|
||||
r'livraison',
|
||||
]
|
||||
|
||||
STOCK_NEGATIVE_PATTERNS = [
|
||||
r'rupture',
|
||||
r'indisponible',
|
||||
r'out\s*of\s*stock',
|
||||
r'épuisé',
|
||||
r'sold\s*out',
|
||||
r'plus\s*disponible',
|
||||
r'non\s*disponible',
|
||||
]
|
||||
|
||||
|
||||
@dataclass
|
||||
class LightScraperResult:
|
||||
"""Résultat du scraping léger"""
|
||||
|
||||
success: bool
|
||||
url: str
|
||||
title: str | None = None
|
||||
price: float | None = None
|
||||
currency: str = "EUR"
|
||||
in_stock: bool | None = None
|
||||
image_url: str | None = None
|
||||
error: str | None = None
|
||||
requires_js: bool = False # Indique si le site nécessite JS
|
||||
|
||||
|
||||
async def scrape_url(url: str, timeout: float = 10.0) -> LightScraperResult:
|
||||
"""
|
||||
Tente un scraping léger via HTTP simple.
|
||||
|
||||
Args:
|
||||
url: URL du produit à scraper
|
||||
timeout: Timeout en secondes
|
||||
|
||||
Returns:
|
||||
LightScraperResult avec les données extraites ou requires_js=True
|
||||
"""
|
||||
try:
|
||||
async with httpx.AsyncClient(
|
||||
timeout=timeout,
|
||||
follow_redirects=True,
|
||||
headers=DEFAULT_HEADERS,
|
||||
) as client:
|
||||
response = await client.get(url)
|
||||
response.raise_for_status()
|
||||
|
||||
html = response.text
|
||||
|
||||
# Vérifier si la page nécessite JavaScript
|
||||
if _requires_javascript(html):
|
||||
logger.debug(f"Page {url} nécessite JavaScript")
|
||||
return LightScraperResult(
|
||||
success=False,
|
||||
url=url,
|
||||
requires_js=True,
|
||||
error="Page requires JavaScript",
|
||||
)
|
||||
|
||||
# Extraire les données
|
||||
title = _extract_title(html)
|
||||
price = _extract_price(html)
|
||||
in_stock = _extract_stock_status(html)
|
||||
image_url = _extract_image(html, url)
|
||||
|
||||
# Si on n'a pas pu extraire le prix, fallback sur Browserless
|
||||
if price is None:
|
||||
logger.debug(f"Impossible d'extraire le prix de {url}")
|
||||
return LightScraperResult(
|
||||
success=False,
|
||||
url=url,
|
||||
title=title,
|
||||
requires_js=True,
|
||||
error="Could not extract price",
|
||||
)
|
||||
|
||||
logger.info(f"Scraping léger réussi pour {url}: {price}€")
|
||||
return LightScraperResult(
|
||||
success=True,
|
||||
url=url,
|
||||
title=title,
|
||||
price=price,
|
||||
in_stock=in_stock,
|
||||
image_url=image_url,
|
||||
)
|
||||
|
||||
except httpx.TimeoutException:
|
||||
logger.warning(f"Timeout HTTP pour {url}")
|
||||
return LightScraperResult(
|
||||
success=False,
|
||||
url=url,
|
||||
error="HTTP timeout",
|
||||
requires_js=True,
|
||||
)
|
||||
except httpx.HTTPStatusError as e:
|
||||
logger.warning(f"Erreur HTTP {e.response.status_code} pour {url}")
|
||||
return LightScraperResult(
|
||||
success=False,
|
||||
url=url,
|
||||
error=f"HTTP {e.response.status_code}",
|
||||
requires_js=True,
|
||||
)
|
||||
except Exception as e:
|
||||
logger.error(f"Erreur scraping léger {url}: {e}")
|
||||
return LightScraperResult(
|
||||
success=False,
|
||||
url=url,
|
||||
error=str(e),
|
||||
requires_js=True,
|
||||
)
|
||||
|
||||
|
||||
def _requires_javascript(html: str) -> bool:
|
||||
"""Détecte si la page nécessite JavaScript pour afficher le contenu"""
|
||||
indicators = [
|
||||
"Please enable JavaScript",
|
||||
"JavaScript is required",
|
||||
"You need to enable JavaScript",
|
||||
"__NEXT_DATA__", # Next.js SSR mais peut nécessiter JS
|
||||
"window.__INITIAL_STATE__",
|
||||
"React.createElement",
|
||||
'<div id="root"></div>', # React SPA vide
|
||||
'<div id="app"></div>', # Vue SPA vide
|
||||
"Loading...",
|
||||
]
|
||||
|
||||
html_lower = html.lower()
|
||||
|
||||
# Vérifier si le body est presque vide
|
||||
body_match = re.search(r"<body[^>]*>(.*?)</body>", html, re.DOTALL | re.IGNORECASE)
|
||||
if body_match:
|
||||
body_content = body_match.group(1)
|
||||
# Retirer les scripts et styles
|
||||
body_clean = re.sub(r"<script[^>]*>.*?</script>", "", body_content, flags=re.DOTALL | re.IGNORECASE)
|
||||
body_clean = re.sub(r"<style[^>]*>.*?</style>", "", body_clean, flags=re.DOTALL | re.IGNORECASE)
|
||||
body_clean = re.sub(r"<[^>]+>", "", body_clean)
|
||||
body_clean = body_clean.strip()
|
||||
|
||||
if len(body_clean) < 500: # Page très peu de contenu texte
|
||||
return True
|
||||
|
||||
for indicator in indicators:
|
||||
if indicator.lower() in html_lower:
|
||||
return True
|
||||
|
||||
return False
|
||||
|
||||
|
||||
def _extract_title(html: str) -> str | None:
|
||||
"""Extrait le titre de la page"""
|
||||
# Essayer og:title d'abord
|
||||
og_match = re.search(r'<meta[^>]+property=["\']og:title["\'][^>]+content=["\']([^"\']+)["\']', html, re.IGNORECASE)
|
||||
if og_match:
|
||||
return og_match.group(1).strip()
|
||||
|
||||
# Fallback sur <title>
|
||||
title_match = re.search(r"<title[^>]*>([^<]+)</title>", html, re.IGNORECASE)
|
||||
if title_match:
|
||||
title = title_match.group(1).strip()
|
||||
# Nettoyer les suffixes de site courants
|
||||
title = re.sub(r"\s*[-|]\s*(Amazon|Fnac|Cdiscount|Darty).*$", "", title, flags=re.IGNORECASE)
|
||||
return title
|
||||
|
||||
return None
|
||||
|
||||
|
||||
def _extract_price(html: str) -> float | None:
|
||||
"""Extrait le prix de la page"""
|
||||
prices = []
|
||||
|
||||
for pattern in PRICE_PATTERNS:
|
||||
matches = re.findall(pattern, html, re.IGNORECASE)
|
||||
for match in matches:
|
||||
try:
|
||||
# Normaliser le format
|
||||
price_str = match.replace(" ", "").replace(",", ".")
|
||||
price = float(price_str)
|
||||
if 0.01 < price < 100000: # Prix raisonnable
|
||||
prices.append(price)
|
||||
except ValueError:
|
||||
continue
|
||||
|
||||
if not prices:
|
||||
return None
|
||||
|
||||
# Retourner le prix le plus fréquent ou le premier
|
||||
from collections import Counter
|
||||
|
||||
price_counts = Counter(prices)
|
||||
most_common = price_counts.most_common(1)
|
||||
if most_common:
|
||||
return most_common[0][0]
|
||||
|
||||
return prices[0]
|
||||
|
||||
|
||||
def _extract_stock_status(html: str) -> bool | None:
|
||||
"""Extrait le statut de stock"""
|
||||
html_lower = html.lower()
|
||||
|
||||
# Chercher les indicateurs négatifs d'abord
|
||||
for pattern in STOCK_NEGATIVE_PATTERNS:
|
||||
if re.search(pattern, html_lower):
|
||||
return False
|
||||
|
||||
# Chercher les indicateurs positifs
|
||||
for pattern in STOCK_POSITIVE_PATTERNS:
|
||||
if re.search(pattern, html_lower):
|
||||
return True
|
||||
|
||||
return None # Incertain
|
||||
|
||||
|
||||
def _extract_image(html: str, base_url: str) -> str | None:
|
||||
"""Extrait l'image principale du produit"""
|
||||
from urllib.parse import urljoin
|
||||
|
||||
# Essayer og:image d'abord
|
||||
og_match = re.search(r'<meta[^>]+property=["\']og:image["\'][^>]+content=["\']([^"\']+)["\']', html, re.IGNORECASE)
|
||||
if og_match:
|
||||
return urljoin(base_url, og_match.group(1))
|
||||
|
||||
# Essayer twitter:image
|
||||
twitter_match = re.search(r'<meta[^>]+name=["\']twitter:image["\'][^>]+content=["\']([^"\']+)["\']', html, re.IGNORECASE)
|
||||
if twitter_match:
|
||||
return urljoin(base_url, twitter_match.group(1))
|
||||
|
||||
return None
|
||||
@@ -0,0 +1,311 @@
|
||||
"""
|
||||
Search Service - Orchestrateur de recherche de produits
|
||||
Combine SearXNG + Light Scraper + Browserless fallback
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
from collections.abc import AsyncGenerator
|
||||
from typing import Any
|
||||
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from app.models import SearchSite
|
||||
from app.schemas import SearchProgress, SearchResultItem
|
||||
from app.services import ai_service, light_scraper_service, searxng_service
|
||||
from app.services.scraper_service import ScraperService
|
||||
from app.services.settings_service import get_setting_value
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Configuration par défaut
|
||||
DEFAULT_MAX_RESULTS = 20
|
||||
DEFAULT_PARALLEL_LIMIT = 5
|
||||
DEFAULT_TIMEOUT = 30
|
||||
|
||||
|
||||
async def search_products(
|
||||
query: str,
|
||||
db: Session,
|
||||
site_ids: list[int] | None = None,
|
||||
max_results: int | None = None,
|
||||
) -> AsyncGenerator[SearchProgress, None]:
|
||||
"""
|
||||
Recherche des produits sur les sites configurés.
|
||||
|
||||
Flux:
|
||||
1. SearXNG recherche les URLs
|
||||
2. Light scraper tente extraction HTTP rapide
|
||||
3. Browserless fallback pour les sites JS
|
||||
|
||||
Args:
|
||||
query: Terme de recherche
|
||||
db: Session de base de données
|
||||
site_ids: IDs des sites à utiliser (None = tous les actifs)
|
||||
max_results: Nombre max de résultats
|
||||
|
||||
Yields:
|
||||
SearchProgress avec les résultats au fur et à mesure
|
||||
"""
|
||||
# Récupérer la configuration
|
||||
if max_results is None:
|
||||
max_results = int(get_setting_value(db, "search_max_results", str(DEFAULT_MAX_RESULTS)))
|
||||
|
||||
parallel_limit = int(get_setting_value(db, "search_parallel_limit", str(DEFAULT_PARALLEL_LIMIT)))
|
||||
|
||||
# Récupérer les sites
|
||||
sites = _get_sites(db, site_ids)
|
||||
if not sites:
|
||||
yield SearchProgress(
|
||||
status="error",
|
||||
total=0,
|
||||
completed=0,
|
||||
message="Aucun site de recherche configuré",
|
||||
results=[],
|
||||
)
|
||||
return
|
||||
|
||||
domains = [site.domain for site in sites]
|
||||
site_map = {site.domain: site for site in sites}
|
||||
|
||||
# Phase 1: Recherche SearXNG
|
||||
yield SearchProgress(
|
||||
status="searching",
|
||||
total=0,
|
||||
completed=0,
|
||||
message=f"Recherche sur {len(domains)} sites...",
|
||||
results=[],
|
||||
)
|
||||
|
||||
searxng_results = await searxng_service.search(
|
||||
query=query,
|
||||
domains=domains,
|
||||
max_results=max_results,
|
||||
)
|
||||
|
||||
if not searxng_results:
|
||||
yield SearchProgress(
|
||||
status="error",
|
||||
total=0,
|
||||
completed=0,
|
||||
message="Aucun résultat trouvé",
|
||||
results=[],
|
||||
)
|
||||
return
|
||||
|
||||
total = len(searxng_results)
|
||||
logger.info(f"SearXNG: {total} URLs trouvées pour '{query}'")
|
||||
|
||||
# Phase 2: Scraping des URLs
|
||||
results: list[SearchResultItem] = []
|
||||
completed = 0
|
||||
|
||||
# Créer un sémaphore pour limiter la concurrence
|
||||
semaphore = asyncio.Semaphore(parallel_limit)
|
||||
|
||||
async def process_url(searxng_result: searxng_service.SearXNGResult) -> SearchResultItem | None:
|
||||
async with semaphore:
|
||||
url = searxng_result.url
|
||||
domain = searxng_result.source
|
||||
site = site_map.get(domain)
|
||||
|
||||
if not site:
|
||||
# Trouver le site par correspondance partielle
|
||||
for d, s in site_map.items():
|
||||
if d in domain or domain in d:
|
||||
site = s
|
||||
break
|
||||
|
||||
site_name = site.name if site else domain
|
||||
requires_js = site.requires_js if site else False
|
||||
|
||||
# Essayer le scraping léger d'abord (sauf si le site requiert JS)
|
||||
if not requires_js:
|
||||
light_result = await light_scraper_service.scrape_url(url)
|
||||
|
||||
if light_result.success and light_result.price is not None:
|
||||
return SearchResultItem(
|
||||
url=url,
|
||||
title=light_result.title or searxng_result.title,
|
||||
price=light_result.price,
|
||||
currency=light_result.currency,
|
||||
in_stock=light_result.in_stock,
|
||||
image_url=light_result.image_url,
|
||||
site_name=site_name,
|
||||
site_domain=domain,
|
||||
confidence=0.7, # Confiance moyenne pour le scraping léger
|
||||
)
|
||||
|
||||
# Fallback: Browserless + IA
|
||||
try:
|
||||
result = await _scrape_with_browserless(url, site_name, domain, searxng_result.title)
|
||||
return result
|
||||
except Exception as e:
|
||||
logger.error(f"Erreur scraping {url}: {e}")
|
||||
return SearchResultItem(
|
||||
url=url,
|
||||
title=searxng_result.title,
|
||||
price=None,
|
||||
site_name=site_name,
|
||||
site_domain=domain,
|
||||
error=str(e),
|
||||
)
|
||||
|
||||
# Traiter les URLs en parallèle avec mises à jour progressives
|
||||
tasks = [asyncio.create_task(process_url(r)) for r in searxng_results]
|
||||
|
||||
for coro in asyncio.as_completed(tasks):
|
||||
try:
|
||||
result = await coro
|
||||
completed += 1
|
||||
|
||||
if result:
|
||||
# N'ajouter que les résultats avec un prix
|
||||
if result.price is not None:
|
||||
results.append(result)
|
||||
|
||||
yield SearchProgress(
|
||||
status="scraping",
|
||||
total=total,
|
||||
completed=completed,
|
||||
current_site=result.site_name if result else None,
|
||||
results=results.copy(),
|
||||
message=f"Extraction {completed}/{total}...",
|
||||
)
|
||||
except Exception as e:
|
||||
completed += 1
|
||||
logger.error(f"Erreur lors du traitement: {e}")
|
||||
|
||||
# Trier les résultats par prix
|
||||
results.sort(key=lambda x: x.price if x.price else float("inf"))
|
||||
|
||||
yield SearchProgress(
|
||||
status="completed",
|
||||
total=total,
|
||||
completed=completed,
|
||||
results=results,
|
||||
message=f"{len(results)} produits trouvés avec prix",
|
||||
)
|
||||
|
||||
|
||||
async def _scrape_with_browserless(
|
||||
url: str,
|
||||
site_name: str,
|
||||
domain: str,
|
||||
fallback_title: str,
|
||||
) -> SearchResultItem:
|
||||
"""Scrape une URL avec Browserless + extraction IA"""
|
||||
try:
|
||||
# Scraper avec Browserless
|
||||
screenshot_path, page_text = await ScraperService.scrape_item(
|
||||
url=url,
|
||||
item_id=None,
|
||||
smart_scroll=True,
|
||||
scroll_pixels=350,
|
||||
text_length=3000,
|
||||
timeout=30000,
|
||||
)
|
||||
|
||||
if not screenshot_path:
|
||||
return SearchResultItem(
|
||||
url=url,
|
||||
title=fallback_title,
|
||||
price=None,
|
||||
site_name=site_name,
|
||||
site_domain=domain,
|
||||
error="Screenshot failed",
|
||||
)
|
||||
|
||||
# Extraction IA
|
||||
ai_result = await ai_service.analyze_image(
|
||||
screenshot_path=screenshot_path,
|
||||
webpage_text=page_text,
|
||||
)
|
||||
|
||||
if ai_result and ai_result.get("price"):
|
||||
return SearchResultItem(
|
||||
url=url,
|
||||
title=fallback_title,
|
||||
price=ai_result.get("price"),
|
||||
currency=ai_result.get("currency", "EUR"),
|
||||
in_stock=ai_result.get("in_stock"),
|
||||
image_url=None, # Pas d'image depuis le screenshot
|
||||
site_name=site_name,
|
||||
site_domain=domain,
|
||||
confidence=ai_result.get("price_confidence", 0.5),
|
||||
)
|
||||
|
||||
return SearchResultItem(
|
||||
url=url,
|
||||
title=fallback_title,
|
||||
price=None,
|
||||
site_name=site_name,
|
||||
site_domain=domain,
|
||||
error="AI extraction failed",
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Browserless scraping error for {url}: {e}")
|
||||
return SearchResultItem(
|
||||
url=url,
|
||||
title=fallback_title,
|
||||
price=None,
|
||||
site_name=site_name,
|
||||
site_domain=domain,
|
||||
error=str(e),
|
||||
)
|
||||
|
||||
|
||||
def _get_sites(db: Session, site_ids: list[int] | None = None) -> list[SearchSite]:
|
||||
"""Récupère les sites de recherche actifs"""
|
||||
query = db.query(SearchSite).filter(SearchSite.is_active == True)
|
||||
|
||||
if site_ids:
|
||||
query = query.filter(SearchSite.id.in_(site_ids))
|
||||
|
||||
return query.order_by(SearchSite.priority).all()
|
||||
|
||||
|
||||
def get_all_sites(db: Session) -> list[SearchSite]:
|
||||
"""Récupère tous les sites de recherche"""
|
||||
return db.query(SearchSite).order_by(SearchSite.priority).all()
|
||||
|
||||
|
||||
def get_site_by_id(db: Session, site_id: int) -> SearchSite | None:
|
||||
"""Récupère un site par son ID"""
|
||||
return db.query(SearchSite).filter(SearchSite.id == site_id).first()
|
||||
|
||||
|
||||
def create_site(db: Session, site_data: dict[str, Any]) -> SearchSite:
|
||||
"""Crée un nouveau site de recherche"""
|
||||
site = SearchSite(**site_data)
|
||||
db.add(site)
|
||||
db.commit()
|
||||
db.refresh(site)
|
||||
return site
|
||||
|
||||
|
||||
def update_site(db: Session, site_id: int, site_data: dict[str, Any]) -> SearchSite | None:
|
||||
"""Met à jour un site de recherche"""
|
||||
site = get_site_by_id(db, site_id)
|
||||
if not site:
|
||||
return None
|
||||
|
||||
for key, value in site_data.items():
|
||||
if value is not None:
|
||||
setattr(site, key, value)
|
||||
|
||||
db.commit()
|
||||
db.refresh(site)
|
||||
return site
|
||||
|
||||
|
||||
def delete_site(db: Session, site_id: int) -> bool:
|
||||
"""Supprime un site de recherche"""
|
||||
site = get_site_by_id(db, site_id)
|
||||
if not site:
|
||||
return False
|
||||
|
||||
db.delete(site)
|
||||
db.commit()
|
||||
return True
|
||||
@@ -0,0 +1,140 @@
|
||||
"""
|
||||
SearXNG Service - Client pour le moteur de recherche auto-hébergé
|
||||
"""
|
||||
|
||||
import logging
|
||||
import os
|
||||
from urllib.parse import quote_plus
|
||||
|
||||
import httpx
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
SEARXNG_URL = os.getenv("SEARXNG_URL", "http://searxng:8080")
|
||||
|
||||
|
||||
class SearXNGResult:
|
||||
"""Résultat de recherche SearXNG"""
|
||||
|
||||
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"SearXNGResult(url={self.url}, title={self.title[:30]}...)"
|
||||
|
||||
|
||||
async def search(
|
||||
query: str,
|
||||
domains: list[str],
|
||||
max_results: int = 20,
|
||||
timeout: float = 15.0,
|
||||
) -> list[SearXNGResult]:
|
||||
"""
|
||||
Effectue une recherche SearXNG ciblée sur les domaines spécifiés.
|
||||
|
||||
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 SearXNGResult avec url, title, snippet, source
|
||||
"""
|
||||
if not domains:
|
||||
logger.warning("Aucun domaine spécifié pour la recherche")
|
||||
return []
|
||||
|
||||
# Construire la requête avec site: pour chaque domaine
|
||||
site_query = " OR ".join([f"site:{domain}" for domain in domains])
|
||||
full_query = f"{query} ({site_query})"
|
||||
|
||||
logger.info(f"Recherche SearXNG: {full_query}")
|
||||
|
||||
try:
|
||||
async with httpx.AsyncClient(timeout=timeout) as client:
|
||||
response = await client.get(
|
||||
f"{SEARXNG_URL}/search",
|
||||
params={
|
||||
"q": full_query,
|
||||
"format": "json",
|
||||
"language": "fr",
|
||||
"safesearch": 0,
|
||||
"pageno": 1,
|
||||
},
|
||||
headers={
|
||||
"Accept": "application/json",
|
||||
},
|
||||
)
|
||||
response.raise_for_status()
|
||||
data = response.json()
|
||||
|
||||
results = []
|
||||
seen_urls = set()
|
||||
|
||||
for item in data.get("results", []):
|
||||
url = item.get("url", "")
|
||||
|
||||
# Éviter les doublons
|
||||
if url in seen_urls:
|
||||
continue
|
||||
seen_urls.add(url)
|
||||
|
||||
# Vérifier que l'URL appartient à un des domaines ciblés
|
||||
url_domain = _extract_domain(url)
|
||||
if not any(domain in url_domain for domain in domains):
|
||||
continue
|
||||
|
||||
results.append(
|
||||
SearXNGResult(
|
||||
url=url,
|
||||
title=item.get("title", ""),
|
||||
snippet=item.get("content", ""),
|
||||
source=url_domain,
|
||||
)
|
||||
)
|
||||
|
||||
if len(results) >= max_results:
|
||||
break
|
||||
|
||||
logger.info(f"SearXNG: {len(results)} résultats trouvés")
|
||||
return results
|
||||
|
||||
except httpx.TimeoutException:
|
||||
logger.error(f"Timeout SearXNG après {timeout}s")
|
||||
return []
|
||||
except httpx.HTTPStatusError as e:
|
||||
logger.error(f"Erreur HTTP SearXNG: {e.response.status_code}")
|
||||
return []
|
||||
except Exception as e:
|
||||
logger.error(f"Erreur SearXNG: {e}")
|
||||
return []
|
||||
|
||||
|
||||
def _extract_domain(url: str) -> str:
|
||||
"""Extrait le domaine d'une URL"""
|
||||
try:
|
||||
from urllib.parse import urlparse
|
||||
|
||||
parsed = urlparse(url)
|
||||
domain = parsed.netloc.lower()
|
||||
# Retirer www. si présent
|
||||
if domain.startswith("www."):
|
||||
domain = domain[4:]
|
||||
return domain
|
||||
except Exception:
|
||||
return ""
|
||||
|
||||
|
||||
async def health_check() -> bool:
|
||||
"""Vérifie que SearXNG est accessible"""
|
||||
try:
|
||||
async with httpx.AsyncClient(timeout=5.0) as client:
|
||||
response = await client.get(f"{SEARXNG_URL}/healthz")
|
||||
return response.status_code == 200
|
||||
except Exception as e:
|
||||
logger.warning(f"SearXNG health check failed: {e}")
|
||||
return False
|
||||
Reference in new issue
Block a user