feat: Rebuild search system with modular, site-specific parsers

Complete refactoring of the search module for robustness and reliability:

**New Architecture:**
- Modular parser system with specialized parsers for each major site
- Factory pattern for automatic parser selection
- Generic fallback parser for sites without specific implementation

**Specialized Parsers:**
- ✅ Amazon.fr - Robust parser with multiple fallback selectors
  - Handles sponsored products, Prime eligibility, ratings, reviews
  - Based on proven amazon_scraper_v2.py implementation
- ✅ Cdiscount.com - Optimized for Cdiscount's HTML structure
- ✅ Fnac.com - Handles Fnac's picture elements and article structure
- ✅ Darty.com - Specialized for electronics retailer
- ✅ Boulanger.com - Optimized for Boulanger's product cards

**Key Features:**
- Multiple fallback selectors for each element (title, price, image)
- Proper price parsing for French format (1 234,56 €)
- Rating and review count extraction
- Stock status detection
- Image URL extraction with multiple attribute fallbacks
- Query filtering to ensure relevant results

**Benefits:**
- Much more robust than previous generic parsing
- Easier to maintain (one parser per site)
- Easily extensible (add new parser for new site)
- Better error handling and logging
- Higher success rate for product extraction

**Updated Services:**
- improved_search_service.py now uses ParserFactory
- Legacy _parse_results marked as deprecated
- Conversion layer between ProductResult and SearchResult

This implements the Amazon France model across all major sites
for consistent, reliable product search results.
This commit is contained in:
Claude committed 2025-11-30 16:09:05 +00:00
1 parent 548de46d2b
commit d2c492ff5c
10 files changed
+1257 -5

No files matched your search

+28 -5
View File
@@ -1,14 +1,15 @@
"""
Improved Search Service - Using Persistent Browser Connection
Based on ScraperService pattern for better session management and reliability
UPDATED: Now uses modular, site-specific parsers for robust product extraction
"""
import asyncio
import logging
from typing import AsyncGenerator
from urllib.parse import quote_plus, urljoin
from urllib.parse import quote_plus
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
@@ -16,6 +17,7 @@ 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
from app.services.parsers import ParserFactory, ProductResult
logger = logging.getLogger(__name__)
@@ -245,8 +247,12 @@ class ImprovedSearchService:
logger.warning(f"⚠️ Page too small - possibly blocked")
return []
# Parse results
results = cls._parse_results(html_content, site_key, search_url, query)
# Parse results using specialized parser
parser = ParserFactory.get_parser(site_key)
parsed_products = parser.parse_search_results(html_content, query, search_url)
# Convert ProductResult to SearchResult
results = [cls._convert_to_search_result(p) for p in parsed_products]
# Scrape details for each result (in parallel)
if results:
@@ -271,9 +277,26 @@ class ImprovedSearchService:
logger.info(f"✅ Successfully found {len(results)} products from {config['name']}")
return results
@staticmethod
def _convert_to_search_result(product: ProductResult) -> SearchResult:
"""Convert ProductResult to SearchResult"""
return SearchResult(
url=product.url,
title=product.title,
snippet=product.snippet,
source=product.source,
price=product.price,
currency=product.currency,
in_stock=product.in_stock,
image_url=product.image_url,
)
@staticmethod
def _parse_results(html: str, site_key: str, base_url: str, query: str) -> list[SearchResult]:
"""Parse HTML content to extract search results"""
"""
DEPRECATED: Legacy parsing method - now using specialized parsers
This method is kept for backward compatibility but should not be used
"""
config = SITE_CONFIGS[site_key]
soup = BeautifulSoup(html, "html.parser")
results = []
+25
View File
@@ -0,0 +1,25 @@
"""
Site-Specific Parsers for Robust Product Search
Each parser implements specialized logic for extracting search results and product details
"""
from .base_parser import BaseParser, ProductResult
from .parser_factory import ParserFactory
from .amazon_parser import AmazonParser
from .cdiscount_parser import CdiscountParser
from .fnac_parser import FnacParser
from .darty_parser import DartyParser
from .boulanger_parser import BoulangerParser
from .generic_parser import GenericParser
__all__ = [
"BaseParser",
"ProductResult",
"ParserFactory",
"AmazonParser",
"CdiscountParser",
"FnacParser",
"DartyParser",
"BoulangerParser",
"GenericParser",
]
+199
View File
@@ -0,0 +1,199 @@
"""
Amazon France Parser - Robust and specialized parser for Amazon.fr
Based on amazon_scraper_v2.py with proven selectors and logic
"""
from bs4 import BeautifulSoup
from .base_parser import BaseParser, ProductResult
class AmazonParser(BaseParser):
"""
Specialized parser for Amazon France
Features:
- Multiple fallback selectors for each element
- Sponsored product detection
- Prime eligibility detection
- Rating and reviews extraction
- Original price detection (discounts)
"""
def __init__(self):
super().__init__(
site_name="Amazon",
base_url="https://www.amazon.fr"
)
def parse_search_results(self, html: str, query: str, search_url: str) -> list[ProductResult]:
"""
Parse Amazon search results page
Amazon uses data-component-type="s-search-result" for product cards
"""
soup = BeautifulSoup(html, 'html.parser')
products = []
# Primary selector: data-component-type
product_cards = soup.find_all('div', {'data-component-type': 's-search-result'})
if not product_cards:
self.logger.warning("No products with primary selector, trying alternative")
# Alternative: data-asin
product_cards = soup.find_all('div', {'data-asin': True, 'data-index': True})
if not product_cards:
# Check for blocks
if 'captcha' in html.lower():
self.logger.error("🚫 CAPTCHA detected")
elif 'robot' in html.lower() or 'bot' in html.lower():
self.logger.error("🤖 Bot detection triggered")
else:
self.logger.warning("📦 No results found")
return []
self.logger.info(f"Found {len(product_cards)} product cards on Amazon")
for idx, card in enumerate(product_cards):
try:
product = self._parse_product_card(card)
if product:
products.append(product)
except Exception as e:
self.logger.error(f"Error parsing card {idx}: {e}")
continue
return products
def _parse_product_card(self, card) -> ProductResult | None:
"""Parse a single Amazon product card"""
# Extract ASIN
asin = card.get('data-asin', '')
if not asin:
return None
# Check if sponsored
sponsored = bool(card.select_one('[data-component-type="sp-sponsored-result"]'))
# Extract title - multiple selectors
title = None
title_selectors = [
'h2 a span',
'h2 span',
'h2.s-line-clamp-2 span',
'.s-title-instructions-style span',
]
for selector in title_selectors:
title_elem = card.select_one(selector)
if title_elem:
title = title_elem.get_text(strip=True)
if title:
break
if not title:
return None
# Extract URL
link_elem = card.select_one('h2 a')
if not link_elem:
link_elem = card.select_one('a.s-link-style')
if not link_elem:
return None
href = link_elem.get('href', '')
product_url = self.make_absolute_url(href)
# Extract price (current price)
price = None
price_selectors = [
'.a-price .a-offscreen',
'.a-price-whole',
'span.a-price span.a-offscreen',
]
for selector in price_selectors:
price_elem = card.select_one(selector)
if price_elem:
price_text = price_elem.get_text(strip=True)
price = self.parse_price_text(price_text)
if price:
break
# Extract original price (if discounted)
original_price = None
original_price_elem = card.select_one('.a-price.a-text-price .a-offscreen')
if original_price_elem:
original_price = self.parse_price_text(original_price_elem.get_text(strip=True))
# Extract rating
rating = None
rating_selectors = [
'[aria-label*="étoile"]',
'[aria-label*="star"]',
'i.a-icon-star-small span',
]
for selector in rating_selectors:
rating_elem = card.select_one(selector)
if rating_elem:
aria_label = rating_elem.get('aria-label', '')
if aria_label:
rating = self.parse_rating_text(aria_label)
if rating:
break
# Extract reviews count
reviews_count = None
reviews_selectors = [
'[aria-label*="étoile"] + span',
'[aria-label*="star"] + span',
'span.s-underline-text',
]
for selector in reviews_selectors:
reviews_elem = card.select_one(selector)
if reviews_elem:
reviews_count = self.parse_reviews_count_text(reviews_elem.get_text(strip=True))
if reviews_count:
break
# Extract image
image_url = None
img_selectors = [
'img.s-image',
'img[data-image-latency="s-product-image"]',
'img',
]
for selector in img_selectors:
img_elem = card.select_one(selector)
if img_elem:
image_url = self._get_image_src(img_elem)
if image_url:
break
# Check Prime eligibility
prime = bool(
card.select_one('[aria-label*="Prime"]') or
card.select_one('i.a-icon-prime')
)
# Check availability
in_stock = True
unavailable_elem = card.select_one('[aria-label*="Indisponible"]') or \
card.select_one('[aria-label*="Unavailable"]')
if unavailable_elem:
in_stock = False
return ProductResult(
title=title,
url=product_url,
source=self.site_name,
price=price,
original_price=original_price,
currency="EUR",
in_stock=in_stock,
image_url=image_url,
rating=rating,
reviews_count=reviews_count,
sponsored=sponsored,
snippet=f"Prime: {'✓' if prime else '✗'} | {'Sponsored' if sponsored else 'Organic'}"
)
+222
View File
@@ -0,0 +1,222 @@
"""
Base Parser - Abstract interface for site-specific parsers
Defines the common structure all parsers must implement
"""
import logging
import re
from abc import ABC, abstractmethod
from dataclasses import dataclass
from urllib.parse import urljoin
from bs4 import BeautifulSoup
logger = logging.getLogger(__name__)
@dataclass
class ProductResult:
"""Unified product result data structure"""
title: str
url: str
source: str
price: float | None = None
original_price: float | None = None
currency: str = "EUR"
in_stock: bool | None = None
image_url: str | None = None
rating: float | None = None
reviews_count: int | None = None
sponsored: bool = False
snippet: str = ""
class BaseParser(ABC):
"""
Abstract base class for site-specific parsers
Each parser must implement:
- parse_search_results: Extract products from search page HTML
- parse_product_details: Extract details from product page HTML (optional)
"""
def __init__(self, site_name: str, base_url: str):
self.site_name = site_name
self.base_url = base_url
self.logger = logging.getLogger(f"{__name__}.{self.__class__.__name__}")
@abstractmethod
def parse_search_results(self, html: str, query: str, search_url: str) -> list[ProductResult]:
"""
Parse search results page HTML to extract products
Args:
html: Raw HTML content of search page
query: Original search query
search_url: URL of the search page
Returns:
List of ProductResult objects
"""
pass
def parse_product_details(self, html: str, product_url: str) -> dict:
"""
Parse product detail page to extract price, stock, etc.
Optional: Override this if you need to scrape individual product pages
Args:
html: Raw HTML of product page
product_url: URL of the product
Returns:
Dict with price, in_stock, etc.
"""
return {}
# ==================== HELPER METHODS ====================
def make_absolute_url(self, url: str, base_url: str | None = None) -> str:
"""Convert relative URL to absolute URL"""
if not url:
return ""
if url.startswith("http"):
return url
return urljoin(base_url or self.base_url, url)
def parse_price_text(self, price_text: str) -> float | None:
"""
Parse French price format to float
Handles:
- "12,99 €"
- "12,99€"
- "1 234,99 €"
- "12.99 EUR"
"""
if not price_text:
return None
# Remove currency symbols
cleaned = price_text.strip().replace('€', '').replace('EUR', '').strip()
# Remove thousands separators
cleaned = cleaned.replace(' ', '').replace('\xa0', '')
# Replace French decimal separator
cleaned = cleaned.replace(',', '.')
# Extract first number
match = re.search(r'(\d+\.?\d*)', cleaned)
if match:
try:
price = float(match.group(1))
# Validate range
if 0.01 <= price <= 100000:
return price
except ValueError:
pass
return None
def parse_rating_text(self, rating_text: str) -> float | None:
"""Parse rating like '4,5 sur 5 étoiles'"""
if not rating_text:
return None
match = re.search(r'(\d+[,.]\d+)', rating_text)
if match:
try:
return float(match.group(1).replace(',', '.'))
except ValueError:
pass
return None
def parse_reviews_count_text(self, reviews_text: str) -> int | None:
"""Parse review count like '1 234'"""
if not reviews_text:
return None
cleaned = re.sub(r'[^\d\s]', '', reviews_text)
cleaned = cleaned.replace(' ', '').replace('\xa0', '')
try:
return int(cleaned)
except ValueError:
return None
def filter_by_query(self, title: str, query: str, strict: bool = False) -> bool:
"""
Check if title matches query words
Args:
title: Product title
query: Search query
strict: If True, ALL query words must be in title. If False, at least ONE word.
"""
if not query:
return True
title_lower = title.lower()
query_words = [w.lower() for w in query.split() if len(w) > 2]
if not query_words:
return True
if strict:
# All words must be present
return all(word in title_lower for word in query_words)
else:
# At least one word must be present
return any(word in title_lower for word in query_words)
def extract_image_url(self, element, selectors: list[str] = None) -> str | None:
"""
Extract image URL from element with fallback selectors
Args:
element: BeautifulSoup element
selectors: Optional CSS selectors to try
"""
if selectors:
for selector in selectors:
img = element.select_one(selector)
if img:
url = self._get_image_src(img)
if url:
return url
# Fallback: find any img
img = element.find("img")
if img:
return self._get_image_src(img)
# Try picture element
picture = element.find("picture")
if picture:
source = picture.find("source")
if source and source.get("srcset"):
return source.get("srcset").split(",")[0].split()[0]
img = picture.find("img")
if img:
return self._get_image_src(img)
return None
def _get_image_src(self, img_element) -> str | None:
"""Extract image source from img element"""
attrs = ['src', 'data-src', 'data-lazy-src', 'data-original', 'data-lazy']
for attr in attrs:
url = img_element.get(attr)
if url and not url.startswith('data:'):
return url
# Try srcset
srcset = img_element.get('srcset')
if srcset:
return srcset.split(",")[0].split()[0]
return None
+130
View File
@@ -0,0 +1,130 @@
"""
Boulanger Parser - Specialized parser for Boulanger.com
"""
from bs4 import BeautifulSoup
from .base_parser import BaseParser, ProductResult
class BoulangerParser(BaseParser):
"""
Parser for Boulanger.com
Key selectors:
- Product links: a[href*='/ref/'][href*='_']
- Images: img.product-image
"""
def __init__(self):
super().__init__(
site_name="Boulanger",
base_url="https://www.boulanger.com"
)
def parse_search_results(self, html: str, query: str, search_url: str) -> list[ProductResult]:
"""Parse Boulanger search results"""
soup = BeautifulSoup(html, 'html.parser')
products = []
# Try multiple selectors
selectors = [
"a[href*='/ref/'][href*='_']",
".product-card a",
"[class*='product'] a[href*='/ref/']",
]
links = []
for selector in selectors:
links = soup.select(selector)
if links:
self.logger.info(f"Found {len(links)} products with selector: {selector}")
break
if not links:
self.logger.warning("No products found on Boulanger")
return []
seen_urls = set()
for link in links:
try:
href = link.get("href")
if not href:
continue
full_url = self.make_absolute_url(href)
if full_url in seen_urls:
continue
seen_urls.add(full_url)
# Extract title
title = link.get_text(strip=True)
if not title:
title = link.get("title", "")
# Try in parent
if not title or len(title) < 3:
parent = link.find_parent(["article", "li", "div"])
if parent:
title_elem = parent.select_one("h2, h3, .product-title, [class*='title']")
if title_elem:
title = title_elem.get_text(strip=True)
if not title or len(title) < 3:
continue
# Filter by query
if not self.filter_by_query(title, query):
continue
# Extract image
image_url = None
parent = link.find_parent(["article", "li", "div"])
if parent:
img_selectors = [
"img.product-image",
"img[class*='product']",
"img",
]
for img_sel in img_selectors:
img = parent.select_one(img_sel)
if img:
image_url = self._get_image_src(img)
if image_url:
image_url = self.make_absolute_url(image_url)
break
# Extract price
price = None
if parent:
price_selectors = [
".price",
"[class*='price'][class*='current']",
"[data-testid='price']",
"span[class*='prix']",
]
for price_sel in price_selectors:
price_elem = parent.select_one(price_sel)
if price_elem:
price = self.parse_price_text(price_elem.get_text(strip=True))
if price:
break
products.append(ProductResult(
title=title,
url=full_url,
source=self.site_name,
price=price,
currency="EUR",
image_url=image_url,
snippet=f"Product from {self.site_name}"
))
except Exception as e:
self.logger.error(f"Error parsing Boulanger product: {e}")
continue
self.logger.info(f"Extracted {len(products)} products from Boulanger")
return products
+133
View File
@@ -0,0 +1,133 @@
"""
Cdiscount Parser - Specialized parser for Cdiscount.com
"""
from bs4 import BeautifulSoup
from .base_parser import BaseParser, ProductResult
class CdiscountParser(BaseParser):
"""
Parser for Cdiscount.com
Key selectors:
- Product links: a.prdtBILnk, a[href*='/f-'][href*='.html']
- Images: img.lazy, img[data-src]
- Prices: Multiple selectors for current/old prices
"""
def __init__(self):
super().__init__(
site_name="Cdiscount",
base_url="https://www.cdiscount.com"
)
def parse_search_results(self, html: str, query: str, search_url: str) -> list[ProductResult]:
"""Parse Cdiscount search results"""
soup = BeautifulSoup(html, 'html.parser')
products = []
# Try multiple selectors
selectors = [
"a.prdtBILnk",
"a[href*='/f-'][href*='.html']",
".prdtBIL a.prdtBIL",
"article a[href*='/f-']",
]
links = []
for selector in selectors:
links = soup.select(selector)
if links:
self.logger.info(f"Found {len(links)} products with selector: {selector}")
break
if not links:
self.logger.warning("No products found on Cdiscount")
return []
seen_urls = set()
for link in links:
try:
href = link.get("href")
if not href:
continue
full_url = self.make_absolute_url(href)
if full_url in seen_urls:
continue
seen_urls.add(full_url)
# Extract title
title = link.get_text(strip=True)
if not title:
title = link.get("title", "")
# Try to find title in nearby elements
if not title or len(title) < 3:
parent = link.find_parent(["article", "li", "div"])
if parent:
title_elem = parent.select_one(".prdtBTitle, h2, h3, .product-title")
if title_elem:
title = title_elem.get_text(strip=True)
if not title or len(title) < 3:
continue
# Filter by query
if not self.filter_by_query(title, query):
continue
# Extract image
image_url = None
parent = link.find_parent(["article", "li", "div"])
if parent:
img_selectors = [
"img.lazy",
"img[data-src]",
"img[src*='image']",
"img",
]
for img_sel in img_selectors:
img = parent.select_one(img_sel)
if img:
image_url = self._get_image_src(img)
if image_url:
image_url = self.make_absolute_url(image_url)
break
# Extract price
price = None
if parent:
price_selectors = [
".price",
".prdtPrice",
"[class*='price'][class*='current']",
"span[class*='prix']",
]
for price_sel in price_selectors:
price_elem = parent.select_one(price_sel)
if price_elem:
price = self.parse_price_text(price_elem.get_text(strip=True))
if price:
break
products.append(ProductResult(
title=title,
url=full_url,
source=self.site_name,
price=price,
currency="EUR",
image_url=image_url,
snippet=f"Product from {self.site_name}"
))
except Exception as e:
self.logger.error(f"Error parsing Cdiscount product: {e}")
continue
self.logger.info(f"Extracted {len(products)} products from Cdiscount")
return products
+131
View File
@@ -0,0 +1,131 @@
"""
Darty Parser - Specialized parser for Darty.com
"""
from bs4 import BeautifulSoup
from .base_parser import BaseParser, ProductResult
class DartyParser(BaseParser):
"""
Parser for Darty.com
Key selectors:
- Product links: a[href*='/nav/achat/'][href*='.html']
- Images: img.product_img
"""
def __init__(self):
super().__init__(
site_name="Darty",
base_url="https://www.darty.com"
)
def parse_search_results(self, html: str, query: str, search_url: str) -> list[ProductResult]:
"""Parse Darty search results"""
soup = BeautifulSoup(html, 'html.parser')
products = []
# Try multiple selectors
selectors = [
"a[href*='/nav/achat/'][href*='.html']",
".product-card a",
"[class*='product'] a[href*='/nav/achat/']",
]
links = []
for selector in selectors:
links = soup.select(selector)
if links:
self.logger.info(f"Found {len(links)} products with selector: {selector}")
break
if not links:
self.logger.warning("No products found on Darty")
return []
seen_urls = set()
for link in links:
try:
href = link.get("href")
if not href:
continue
full_url = self.make_absolute_url(href)
if full_url in seen_urls:
continue
seen_urls.add(full_url)
# Extract title
title = link.get_text(strip=True)
if not title:
title = link.get("title", "")
# Try in parent
if not title or len(title) < 3:
parent = link.find_parent(["article", "li", "div"])
if parent:
title_elem = parent.select_one("h2, h3, .product-title, [class*='title']")
if title_elem:
title = title_elem.get_text(strip=True)
if not title or len(title) < 3:
continue
# Filter by query
if not self.filter_by_query(title, query):
continue
# Extract image
image_url = None
parent = link.find_parent(["article", "li", "div"])
if parent:
img_selectors = [
"img.product_img",
"img.product-image",
"img[class*='product']",
"img",
]
for img_sel in img_selectors:
img = parent.select_one(img_sel)
if img:
image_url = self._get_image_src(img)
if image_url:
image_url = self.make_absolute_url(image_url)
break
# Extract price
price = None
if parent:
price_selectors = [
".price",
"[class*='price'][class*='current']",
"[data-testid='price']",
"span[class*='prix']",
]
for price_sel in price_selectors:
price_elem = parent.select_one(price_sel)
if price_elem:
price = self.parse_price_text(price_elem.get_text(strip=True))
if price:
break
products.append(ProductResult(
title=title,
url=full_url,
source=self.site_name,
price=price,
currency="EUR",
image_url=image_url,
snippet=f"Product from {self.site_name}"
))
except Exception as e:
self.logger.error(f"Error parsing Darty product: {e}")
continue
self.logger.info(f"Extracted {len(products)} products from Darty")
return products
+138
View File
@@ -0,0 +1,138 @@
"""
Fnac Parser - Specialized parser for Fnac.com
"""
from bs4 import BeautifulSoup
from .base_parser import BaseParser, ProductResult
class FnacParser(BaseParser):
"""
Parser for Fnac.com
Key selectors:
- Product links: a[href*='/a'], a.Article-title, article a
- Images: img, picture source
"""
def __init__(self):
super().__init__(
site_name="Fnac",
base_url="https://www.fnac.com"
)
def parse_search_results(self, html: str, query: str, search_url: str) -> list[ProductResult]:
"""Parse Fnac search results"""
soup = BeautifulSoup(html, 'html.parser')
products = []
# Try multiple selectors
selectors = [
"a[href*='/a']",
"a.Article-title",
"article a",
".Article-item a",
]
links = []
for selector in selectors:
links = soup.select(selector)
if links:
self.logger.info(f"Found {len(links)} products with selector: {selector}")
break
if not links:
self.logger.warning("No products found on Fnac")
return []
seen_urls = set()
for link in links:
try:
href = link.get("href")
if not href:
continue
# Filter Fnac URLs
if "/a" not in href:
continue
full_url = self.make_absolute_url(href)
if full_url in seen_urls:
continue
seen_urls.add(full_url)
# Extract title
title = link.get_text(strip=True)
if not title:
title = link.get("title", "")
# Try aria-label
if not title:
title = link.get("aria-label", "")
if not title or len(title) < 3:
continue
# Filter by query
if not self.filter_by_query(title, query):
continue
# Extract image
image_url = None
parent = link.find_parent(["article", "li", "div"])
if parent:
# Try picture > source first
picture = parent.find("picture")
if picture:
source = picture.find("source")
if source and source.get("srcset"):
image_url = source.get("srcset").split(",")[0].split()[0]
if not image_url:
img = picture.find("img")
if img:
image_url = self._get_image_src(img)
# Fallback to img
if not image_url:
img = parent.find("img")
if img:
image_url = self._get_image_src(img)
if image_url:
image_url = self.make_absolute_url(image_url)
# Extract price
price = None
if parent:
price_selectors = [
".Article-price",
"[class*='price']",
"span[class*='prix']",
"[data-testid='price']",
]
for price_sel in price_selectors:
price_elem = parent.select_one(price_sel)
if price_elem:
price = self.parse_price_text(price_elem.get_text(strip=True))
if price:
break
products.append(ProductResult(
title=title,
url=full_url,
source=self.site_name,
price=price,
currency="EUR",
image_url=image_url,
snippet=f"Product from {self.site_name}"
))
except Exception as e:
self.logger.error(f"Error parsing Fnac product: {e}")
continue
self.logger.info(f"Extracted {len(products)} products from Fnac")
return products
+155
View File
@@ -0,0 +1,155 @@
"""
Generic Parser - Fallback parser for sites without specific implementation
Uses configuration from search_config.py
"""
from bs4 import BeautifulSoup
from urllib.parse import urljoin
from .base_parser import BaseParser, ProductResult
from app.core.search_config import SITE_CONFIGS
class GenericParser(BaseParser):
"""
Generic parser using configured selectors from search_config.py
This parser is used as a fallback for sites that don't have
a specialized parser implementation.
"""
def __init__(self, site_key: str):
"""
Initialize with site configuration
Args:
site_key: Key in SITE_CONFIGS dict
"""
config = SITE_CONFIGS.get(site_key)
if not config:
raise ValueError(f"Unknown site key: {site_key}")
super().__init__(
site_name=config.get("name", site_key),
base_url=config.get("search_url", "").split("/")[0:3] # Extract base URL
)
self.config = config
self.site_key = site_key
# Extract base_url properly
if "search_url" in config:
parts = config["search_url"].split("/")
if len(parts) >= 3:
self.base_url = f"{parts[0]}//{parts[2]}"
def parse_search_results(self, html: str, query: str, search_url: str) -> list[ProductResult]:
"""
Parse search results using configured selectors
Uses:
- product_selector: CSS selector for product links
- product_image_selector: CSS selector for images (optional)
"""
soup = BeautifulSoup(html, 'html.parser')
products = []
# Get product links using configured selector
product_selector = self.config.get("product_selector")
if not product_selector:
self.logger.error(f"No product_selector configured for {self.site_key}")
return []
links = soup.select(product_selector)
if not links:
self.logger.warning(f"No products found for {self.site_key} with selector: {product_selector}")
return []
self.logger.info(f"Found {len(links)} product links for {self.site_name}")
seen_urls = set()
for link in links:
try:
href = link.get("href")
if not href:
continue
full_url = urljoin(search_url, href)
if full_url in seen_urls:
continue
seen_urls.add(full_url)
# Extract title
title = link.get_text(strip=True)
# Fallback: try title attribute or alt in img
if not title:
title = link.get("title", "")
if not title:
img = link.find("img")
if img:
title = img.get("alt", "")
if not title or len(title) < 3:
continue
# Filter by query (relaxed: at least one word match)
if not self.filter_by_query(title, query, strict=False):
continue
# Extract image using configured selector
image_url = None
image_selector = self.config.get("product_image_selector")
if image_selector:
# Try in link first
img_elem = link.select_one(image_selector)
# Try in parent container
if not img_elem:
parent = link.find_parent(["article", "li", "div"])
if parent:
img_elem = parent.select_one(image_selector)
if img_elem:
image_url = self._get_image_src(img_elem)
# Fallback: any img in link or parent
if not image_url:
img = link.find("img")
if img:
image_url = self._get_image_src(img)
if not image_url:
parent = link.find_parent(["article", "li", "div"])
if parent:
img = parent.find("img")
if img:
image_url = self._get_image_src(img)
# Make image URL absolute
if image_url:
# Skip invalid images
if image_url.startswith("data:") or "1x1" in image_url or "placeholder" in image_url.lower():
image_url = None
else:
image_url = urljoin(search_url, image_url)
products.append(ProductResult(
title=title,
url=full_url,
source=self.site_name,
image_url=image_url,
currency="EUR",
snippet=f"Product from {self.site_name}"
))
except Exception as e:
self.logger.error(f"Error parsing product from {self.site_key}: {e}")
continue
self.logger.info(f"Extracted {len(products)} products from {self.site_name}")
return products
+96
View File
@@ -0,0 +1,96 @@
"""
Parser Factory - Selects the appropriate parser for each site
"""
import logging
from .base_parser import BaseParser
from .amazon_parser import AmazonParser
from .cdiscount_parser import CdiscountParser
from .fnac_parser import FnacParser
from .darty_parser import DartyParser
from .boulanger_parser import BoulangerParser
from .generic_parser import GenericParser
logger = logging.getLogger(__name__)
class ParserFactory:
"""
Factory for creating site-specific parsers
Automatically selects the specialized parser based on the site key,
or falls back to the generic parser for sites without specific implementation.
"""
# Mapping of site keys to their specialized parsers
_PARSERS = {
"amazon.fr": AmazonParser,
"cdiscount.com": CdiscountParser,
"fnac.com": FnacParser,
"darty.com": DartyParser,
"boulanger.com": BoulangerParser,
}
# Cache for parser instances (singleton pattern)
_instances = {}
@classmethod
def get_parser(cls, site_key: str) -> BaseParser:
"""
Get the appropriate parser for a site
Args:
site_key: Site identifier (e.g., "amazon.fr", "cdiscount.com")
Returns:
Parser instance (specialized or generic)
"""
# Check cache first
if site_key in cls._instances:
return cls._instances[site_key]
# Check if we have a specialized parser
parser_class = cls._PARSERS.get(site_key)
if parser_class:
logger.info(f"✓ Using specialized parser for {site_key}: {parser_class.__name__}")
parser = parser_class()
else:
logger.info(f"⚠ No specialized parser for {site_key}, using GenericParser")
parser = GenericParser(site_key)
# Cache the instance
cls._instances[site_key] = parser
return parser
@classmethod
def register_parser(cls, site_key: str, parser_class: type[BaseParser]):
"""
Register a new parser for a site
Args:
site_key: Site identifier
parser_class: Parser class (must inherit from BaseParser)
"""
if not issubclass(parser_class, BaseParser):
raise ValueError(f"Parser class must inherit from BaseParser")
cls._PARSERS[site_key] = parser_class
# Clear cache for this site
if site_key in cls._instances:
del cls._instances[site_key]
logger.info(f"Registered parser {parser_class.__name__} for {site_key}")
@classmethod
def list_specialized_parsers(cls) -> list[str]:
"""Get list of sites with specialized parsers"""
return list(cls._PARSERS.keys())
@classmethod
def clear_cache(cls):
"""Clear parser instance cache"""
cls._instances.clear()
logger.info("Parser cache cleared")