diff --git a/app/services/improved_search_service.py b/app/services/improved_search_service.py index b7fea94..788f838 100644 --- a/app/services/improved_search_service.py +++ b/app/services/improved_search_service.py @@ -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 = [] diff --git a/app/services/parsers/__init__.py b/app/services/parsers/__init__.py new file mode 100644 index 0000000..14a75e9 --- /dev/null +++ b/app/services/parsers/__init__.py @@ -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", +] diff --git a/app/services/parsers/amazon_parser.py b/app/services/parsers/amazon_parser.py new file mode 100644 index 0000000..5474a7a --- /dev/null +++ b/app/services/parsers/amazon_parser.py @@ -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'}" + ) diff --git a/app/services/parsers/base_parser.py b/app/services/parsers/base_parser.py new file mode 100644 index 0000000..a341f93 --- /dev/null +++ b/app/services/parsers/base_parser.py @@ -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 diff --git a/app/services/parsers/boulanger_parser.py b/app/services/parsers/boulanger_parser.py new file mode 100644 index 0000000..5e39cc7 --- /dev/null +++ b/app/services/parsers/boulanger_parser.py @@ -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 diff --git a/app/services/parsers/cdiscount_parser.py b/app/services/parsers/cdiscount_parser.py new file mode 100644 index 0000000..2fad32d --- /dev/null +++ b/app/services/parsers/cdiscount_parser.py @@ -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 diff --git a/app/services/parsers/darty_parser.py b/app/services/parsers/darty_parser.py new file mode 100644 index 0000000..8ce3174 --- /dev/null +++ b/app/services/parsers/darty_parser.py @@ -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 diff --git a/app/services/parsers/fnac_parser.py b/app/services/parsers/fnac_parser.py new file mode 100644 index 0000000..faa42a1 --- /dev/null +++ b/app/services/parsers/fnac_parser.py @@ -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 diff --git a/app/services/parsers/generic_parser.py b/app/services/parsers/generic_parser.py new file mode 100644 index 0000000..866e205 --- /dev/null +++ b/app/services/parsers/generic_parser.py @@ -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 diff --git a/app/services/parsers/parser_factory.py b/app/services/parsers/parser_factory.py new file mode 100644 index 0000000..9e621f6 --- /dev/null +++ b/app/services/parsers/parser_factory.py @@ -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")