feat: Improve Search & Comparateur with persistent browser ScraperService pattern

- Created improved_search_service.py using persistent browser pattern
- Persistent browser connection with auto-reconnect capability
- Proper popup/cookie handling across all sites
- Price extraction with validation (reject unrealistic prices)
- Stock status detection
- Concurrent scraping with semaphore limits (2 sites, 2 products)
- Reuses browser context for better session management
- Screenshot capture for product images
- Updated search router to use improved service
- Added initialization/shutdown in main.py lifespan

Benefits:
✓ More reliable scraping with persistent connections
✓ Better anti-detection (consistent sessions)
✓ Improved price accuracy with validation
✓ Faster performance (reuses browser contexts)
✓ Auto-recovery from connection failures
This commit is contained in:
Claude committed 2025-11-30 10:51:22 +00:00
1 parent 85b19e7680
commit 68db9ad15e
3 files changed
+629 -2

No files matched your search

+6
View File
@@ -18,6 +18,7 @@ from app.services.scheduler_service import scheduled_refresh, scheduler
from app.services import auth_service, search_service, seed_enseignes
from app.services.scheduler import start_scheduler as start_catalog_scheduler, stop_scheduler as stop_catalog_scheduler
from app.services.amazon_scraper_service import amazon_scraper_service
from app.services.improved_search_service import improved_search_service
# Configure logging
logging.basicConfig(
@@ -145,12 +146,17 @@ async def lifespan(app: FastAPI):
logger.info("Initializing Amazon scraper service...")
await amazon_scraper_service.initialize()
# Initialize Improved Search Service (persistent browser for Search/Comparatif)
logger.info("Initializing Improved Search Service...")
await improved_search_service.initialize()
logger.info("Application started")
yield
logger.info("Shutting down schedulers and services...")
scheduler.shutdown(wait=True)
stop_catalog_scheduler()
await amazon_scraper_service.shutdown()
await improved_search_service.shutdown()
logger.info("Application shutdown complete")
+2 -2
View File
@@ -11,7 +11,7 @@ from fastapi.responses import StreamingResponse
from sqlalchemy.orm import Session
from app.database import get_db
from app.services import search_service
from app.services import improved_search_service
logger = logging.getLogger(__name__)
@@ -44,7 +44,7 @@ async def search_products(
async def generate():
try:
async for progress in search_service.search_products(
async for progress in improved_search_service.search_products(
query=q,
db=db,
site_ids=site_ids,
+621
View File
@@ -0,0 +1,621 @@
"""
Improved Search Service - Using Persistent Browser Connection
Based on ScraperService pattern for better session management and reliability
"""
import asyncio
import logging
from typing import AsyncGenerator
from urllib.parse import quote_plus, urljoin
from bs4 import BeautifulSoup
from playwright.async_api import Browser, BrowserContext, Page, async_playwright
from playwright.async_api import TimeoutError as PlaywrightTimeoutError
from sqlalchemy.orm import Session
from app.core.search_config import SITE_CONFIGS, BROWSERLESS_URL
from app.models import SearchSite
from app.schemas import SearchProgress, SearchResultItem
logger = logging.getLogger(__name__)
# Common popup/cookie selectors
COMMON_POPUP_SELECTORS = [
"#sp-cc-accept", # Cookie banner
"#onetrust-accept-btn-handler", # OneTrust
".cookie-consent-accept",
"[data-action='accept-cookies']",
"button[id*='accept']",
"button[class*='accept']",
]
class SearchResult:
"""Search result data class"""
def __init__(
self,
url: str,
title: str,
snippet: str,
source: str,
price: float | None = None,
currency: str = "EUR",
in_stock: bool | None = None,
image_url: str | None = None,
):
self.url = url
self.title = title
self.snippet = snippet
self.source = source
self.price = price
self.currency = currency
self.in_stock = in_stock
self.image_url = image_url
def to_dict(self):
return {
"url": self.url,
"title": self.title,
"snippet": self.snippet,
"source": self.source,
"price": self.price,
"currency": self.currency,
"in_stock": self.in_stock,
"image_url": self.image_url,
}
class ImprovedSearchService:
"""Persistent browser service for e-commerce search scraping"""
_playwright = None
_browser: Browser | None = None
_lock = asyncio.Lock()
@classmethod
async def initialize(cls):
"""Initialize shared browser (Thread-Safe)"""
async with cls._lock:
await cls._initialize()
@classmethod
async def _initialize(cls):
"""Internal initialization"""
if cls._browser is None:
logger.info("Initializing ImprovedSearchService shared browser...")
cls._playwright = await async_playwright().start()
cls._browser = await cls._connect_browser(cls._playwright)
logger.info("ImprovedSearchService initialized.")
@classmethod
async def shutdown(cls):
"""Shutdown shared browser"""
async with cls._lock:
if cls._browser:
logger.info("Shutting down ImprovedSearchService...")
await cls._browser.close()
cls._browser = None
if cls._playwright:
await cls._playwright.stop()
cls._playwright = None
logger.info("ImprovedSearchService shutdown complete.")
@classmethod
async def _ensure_browser_connected(cls) -> bool:
"""Ensure browser is connected, reconnect if needed"""
async with cls._lock:
try:
if cls._browser is None:
logger.warning("Browser not initialized, initializing...")
await cls._initialize()
return cls._browser is not None
# Test connection
try:
test_context = await cls._browser.new_context()
await test_context.close()
return True
except Exception as e:
logger.error(f"Browser connection test failed: {e}")
logger.info("Attempting to reconnect...")
cls._browser = None
if cls._playwright:
try:
await cls._playwright.stop()
except Exception:
pass
cls._playwright = None
await cls._initialize()
return cls._browser is not None
except Exception as e:
logger.error(f"Failed to ensure browser connection: {e}")
return False
@staticmethod
async def _connect_browser(p) -> Browser:
"""Connect to Browserless"""
logger.info(f"Connecting to Browserless at {BROWSERLESS_URL}")
return await p.chromium.connect_over_cdp(BROWSERLESS_URL)
@staticmethod
async def _create_context(browser: Browser) -> BrowserContext:
"""Create browser context with stealth settings"""
context = await browser.new_context(
viewport={"width": 1920, "height": 1080},
user_agent=(
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
"AppleWebKit/537.36 (KHTML, like Gecko) "
"Chrome/131.0.0.0 Safari/537.36"
),
locale="fr-FR",
timezone_id="Europe/Paris",
)
# Stealth mode
await context.add_init_script("""
Object.defineProperty(navigator, 'webdriver', { get: () => undefined });
window.chrome = { runtime: {} };
""")
await context.route("**/*", lambda route: route.continue_())
return context
@staticmethod
async def _handle_popups(page: Page):
"""Close common popups/cookies"""
logger.debug("Handling popups...")
for selector in COMMON_POPUP_SELECTORS:
try:
if await page.locator(selector).count() > 0:
logger.debug(f"Found popup: {selector}")
await page.locator(selector).first.click(timeout=2000)
await page.wait_for_timeout(500)
except Exception:
pass
try:
await page.keyboard.press("Escape")
except Exception:
pass
@classmethod
async def search_site(cls, site_key: str, query: str) -> list[SearchResult]:
"""
Search a single site using persistent browser
Args:
site_key: Site configuration key
query: Search query
Returns:
List of SearchResult objects
"""
config = SITE_CONFIGS.get(site_key)
if not config:
logger.error(f"Unknown site: {site_key}")
return []
search_url = config["search_url"].format(query=quote_plus(query))
logger.info(f"🔍 Searching {config['name']} at {search_url}")
# Ensure browser is connected
if not await cls._ensure_browser_connected():
logger.error("Failed to establish browser connection")
return []
results = []
try:
context = await cls._create_context(cls._browser)
page = await context.new_page()
try:
# Navigate to search page
logger.debug(f"Navigating to {search_url}")
await page.goto(search_url, wait_until="domcontentloaded", timeout=30000)
logger.debug("Page loaded (domcontentloaded)")
# Wait for network idle
try:
await page.wait_for_load_state("networkidle", timeout=10000)
logger.debug("Network idle reached")
except PlaywrightTimeoutError:
logger.debug("Network idle timed out (non-critical)")
# Handle popups
await cls._handle_popups(page)
# Wait for content to load
wait_selector = config.get("wait_selector")
if wait_selector:
try:
await page.wait_for_selector(wait_selector, timeout=5000)
logger.debug(f"Wait selector found: {wait_selector}")
except PlaywrightTimeoutError:
logger.warning(f"Wait selector not found: {wait_selector}")
# Small delay for JS rendering
await page.wait_for_timeout(2000)
# Get HTML content
html_content = await page.content()
logger.info(f"✅ Page content extracted ({len(html_content)} bytes)")
if len(html_content) < 5000:
logger.warning(f"⚠️ Page too small - possibly blocked")
return []
# Parse results
results = cls._parse_results(html_content, site_key, search_url, query)
# Scrape details for each result (in parallel)
if results:
logger.info(f"📦 Found {len(results)} initial results, enriching with details...")
semaphore = asyncio.Semaphore(2) # Limit concurrency
async def scrape_with_limit(res):
async with semaphore:
return await cls._scrape_item_details(res, context)
tasks = [scrape_with_limit(r) for r in results]
enriched_results = await asyncio.gather(*tasks)
results = [r for r in enriched_results if r] # Filter None
finally:
await context.close()
except Exception as e:
logger.error(f"❌ Error during search: {e}", exc_info=True)
return []
logger.info(f"✅ Successfully found {len(results)} products from {config['name']}")
return results
@staticmethod
def _parse_results(html: str, site_key: str, base_url: str, query: str) -> list[SearchResult]:
"""Parse HTML content to extract search results"""
config = SITE_CONFIGS[site_key]
soup = BeautifulSoup(html, "html.parser")
results = []
# Prepare query words for filtering
query_words = [w.lower() for w in query.split() if len(w) > 2]
# Select product links
links = soup.select(config["product_selector"])
# Deduplicate links
seen_urls = set()
for link in links:
href = link.get("href")
if not href:
continue
full_url = urljoin(base_url, href)
# Basic cleanup
if full_url in seen_urls:
continue
seen_urls.add(full_url)
# Extract title
title = link.get_text(strip=True)
# If no text, check title attribute or nested image alt
if not title:
if link.get("title"):
title = link.get("title")
else:
img = link.find("img")
if img and img.get("alt"):
title = img.get("alt")
if not title or len(title) < 3:
continue
# STRICT FILTERING: Check if all query words are in the title
title_lower = title.lower()
if query_words:
all_words_found = True
for word in query_words:
if word not in title_lower:
all_words_found = False
break
if not all_words_found:
continue
# Extract Image URL
image_url = None
if "product_image_selector" in config:
container = link.find_parent("article") or link.find_parent("div", class_=lambda x: x and "product" in x)
if container:
img_el = container.select_one(config["product_image_selector"])
if img_el:
image_url = img_el.get("src") or img_el.get("data-src")
# Fallback image extraction
if not image_url:
img = link.find("img")
if img:
image_url = img.get("src") or img.get("data-src")
if image_url and not image_url.startswith("http"):
image_url = urljoin(base_url, image_url)
# Create result
results.append(SearchResult(
url=full_url,
title=title,
snippet=f"Product from {config['name']}",
source=config["name"],
image_url=image_url
))
logger.debug(f"Parsed {len(results)} results from HTML")
return results
@classmethod
async def _scrape_item_details(cls, result: SearchResult, context: BrowserContext) -> SearchResult | None:
"""Scrape price and details for a single item using same context"""
try:
page = await context.new_page()
try:
logger.debug(f"Scraping details for: {result.title[:50]}...")
# Navigate to product page
await page.goto(result.url, wait_until="domcontentloaded", timeout=20000)
# Wait for network idle
try:
await page.wait_for_load_state("networkidle", timeout=5000)
except PlaywrightTimeoutError:
pass
# Handle popups
await cls._handle_popups(page)
# Wait for content
await page.wait_for_timeout(1500)
# Extract price using multiple selectors
price = await cls._extract_price(page)
result.price = price
# Extract stock status
in_stock = await cls._extract_stock_status(page)
result.in_stock = in_stock
# Take screenshot for image
screenshot_path = f"/home/user/Priceflow/screenshots/{result.title[:30].replace('/', '_')}_{id(result)}.png"
try:
await page.screenshot(path=screenshot_path, full_page=False)
import os
filename = os.path.basename(screenshot_path)
result.image_url = f"/screenshots/{filename}"
except Exception as e:
logger.debug(f"Screenshot failed: {e}")
logger.debug(f" ✓ {result.title[:40]}... - {price}€ - Stock: {in_stock}")
return result
finally:
await page.close()
except Exception as e:
logger.error(f"Error scraping item details {result.url}: {e}")
return result # Return original result without price
@staticmethod
async def _extract_price(page: Page) -> float | None:
"""Extract price from product page using multiple selectors"""
price_selectors = [
'.price',
'[data-testid="price"]',
'.prix-actuel',
'.price-current',
'[itemprop="price"]',
'.product-price',
'.a-price .a-offscreen',
'.a-price-whole',
'span[class*="price"]',
]
for selector in price_selectors:
try:
elements = await page.query_selector_all(selector)
for elem in elements:
price_text = await elem.inner_text()
if price_text:
# Parse price
import re
cleaned = price_text.strip().replace('€', '').replace('EUR', '').strip()
cleaned = cleaned.replace(' ', '').replace('\xa0', '')
cleaned = cleaned.replace(',', '.')
match = re.search(r'(\d+\.?\d*)', cleaned)
if match:
try:
price_val = float(match.group(1))
# Validate price
if 0.01 < price_val < 100000:
logger.debug(f"Found price: {price_val}€ from {selector}")
return price_val
except ValueError:
continue
except Exception:
continue
logger.debug("No price found")
return None
@staticmethod
async def _extract_stock_status(page: Page) -> bool | None:
"""Extract stock status from product page"""
# Check for out of stock indicators
out_of_stock_texts = [
"rupture de stock",
"indisponible",
"out of stock",
"unavailable",
"épuisé",
"non disponible"
]
try:
page_text = await page.inner_text("body")
page_text_lower = page_text.lower()
for text in out_of_stock_texts:
if text in page_text_lower:
logger.debug(f"Out of stock detected: '{text}'")
return False
# If we find "add to cart" or similar, assume in stock
add_to_cart_texts = ["ajouter au panier", "add to cart", "acheter", "buy now"]
for text in add_to_cart_texts:
if text in page_text_lower:
logger.debug(f"In stock detected: '{text}'")
return True
except Exception as e:
logger.debug(f"Stock check error: {e}")
return None # Unknown
@classmethod
async def search_site_generator(cls, site_key: str, query: str) -> AsyncGenerator[SearchResult, None]:
"""Search a single site and yield results as they are scraped"""
results = await cls.search_site(site_key, query)
for result in results:
yield result
@classmethod
async def search_all(cls, query: str) -> list[SearchResult]:
"""Search all configured sites"""
tasks = []
for site_key in SITE_CONFIGS.keys():
tasks.append(cls.search_site(site_key, query))
results_list = await asyncio.gather(*tasks)
all_results = []
for r in results_list:
all_results.extend(r)
return all_results
# ==========================================
# COMPATIBILITY LAYER FOR API ROUTERS
# ==========================================
async def search_products(
query: str,
db: Session,
site_ids: list[int] | None = None,
max_results: int | None = None,
) -> AsyncGenerator[SearchProgress, None]:
"""
Compatibility wrapper for search_products using improved service.
Yields SearchProgress events incrementally.
"""
# 1. Get sites to search
sites = db.query(SearchSite).order_by(SearchSite.priority).all()
if site_ids:
sites = [s for s in sites if s.id in site_ids]
active_sites = [s for s in sites if s.is_active]
# Initial event
yield SearchProgress(
status="searching",
total=len(active_sites),
completed=0,
message=f"Démarrage de la recherche sur {len(active_sites)} sites...",
results=[],
)
# 2. Map DB sites to Config keys
site_keys = []
for site in active_sites:
for key in SITE_CONFIGS.keys():
if key in site.domain or site.domain in key:
site_keys.append(key)
break
# 3. Execute searches and stream results
generators = [ImprovedSearchService.search_site_generator(key, query) for key in site_keys]
queue = asyncio.Queue()
active_producers = len(generators)
# Limit concurrent sites
site_semaphore = asyncio.Semaphore(2)
async def producer(gen):
async with site_semaphore:
try:
async for item in gen:
await queue.put(item)
except Exception as e:
logger.error(f"Error in search producer: {e}")
finally:
await queue.put(None) # Sentinel
# Start producers
for gen in generators:
asyncio.create_task(producer(gen))
# Consumer loop
results_so_far = []
completed_sites = 0
while active_producers > 0:
item = await queue.get()
if item is None:
active_producers -= 1
completed_sites += 1
yield SearchProgress(
status="searching",
total=len(active_sites),
completed=completed_sites,
message=f"Recherche en cours... ({completed_sites}/{len(active_sites)} sites terminés)",
results=results_so_far,
)
else:
# Convert to SearchResultItem
api_item = SearchResultItem(
url=item.url,
title=item.title,
price=item.price,
currency=item.currency,
in_stock=item.in_stock,
site_name=item.source,
site_domain=item.source,
image_url=item.image_url,
)
results_so_far.append(api_item)
# Yield update with new result
yield SearchProgress(
status="searching",
total=len(active_sites),
completed=completed_sites,
message=f"Trouvé: {item.title[:30]}...",
results=results_so_far,
)
# Final event
yield SearchProgress(
status="completed",
total=len(active_sites),
completed=len(active_sites),
message=f"Terminé. {len(results_so_far)} résultats trouvés.",
results=results_so_far,
)
# Global instance
improved_search_service = ImprovedSearchService()