mirror of
https://github.com/R0m1k3/Priceflow.git
synced 2026-10-12 01:39:25 +02:00
Claude
66d9046de2
**Problem:** L'Incroyable product showing "3€99" was detected as 399.00 EUR instead of 3.99 EUR. URL: https://www.lincroyable.fr/p95361-porte-magique-lutin/ **Root Cause:** Previous normalization only handled "3,99 €" (with comma separator). Some French sites use alternative format: "3€99" (euro symbol as separator). Existing regex: ```python re.sub(r'(\d+),(\d{2})\s*€', r'\1.\2 €', content) # Only matches "3,99 €" ``` Missed format: ``` "3€99" → No match → AI sees "399" → Returns 399.00 ``` **Solution: Add Pattern for Euro-as-Separator Format** 1. **Text Normalization** (browserless_service.py:408-411) ```python # NEW Pattern 1: digits€XX → digits.XX € content = re.sub(r'(\d+)€(\d{2})\b', r'\1.\2 €', content) # "3€99" → "3.99 €" # "19€50" → "19.50 €" # Pattern 2: digits,XX € → digits.XX € (existing) content = re.sub(r'(\d+),(\d{2})\s*€', r'\1.\2 €', content) # "3,99 €" → "3.99 €" ``` 2. **Extracted Price Normalization** (browserless_service.py:438-441) - Same patterns applied to directly extracted prices - Ensures consistency regardless of extraction method **Regex Details:** - `(\d+)€(\d{2})\b` matches: * `\d+` = one or more digits (euros) * `€` = euro symbol as separator * `(\d{2})` = exactly 2 digits (cents) * `\b` = word boundary (prevents matching "10€999") - Replacement: `\1.\2 €` = "digits.cents €" **Supported Formats After Fix:** ``` "3,99 €" → "3.99 €" ✓ (existing) "3€99" → "3.99 €" ✓ (NEW) "19,50 €" → "19.50 €" ✓ (existing) "19€50" → "19.50 €" ✓ (NEW) "1 234,56" → "1234.56" ✓ (existing) ``` **Flow:** ``` Page text: "Porte Magique 3€99" ↓ Normalization: "Porte Magique 3.99 €" ↓ Extracted: "3€99" → Normalized: "3.99 €" ↓ Sent to AI: "PRIX DÉTECTÉ: 3.99 €\n\nPorte Magique 3.99 €" ↓ AI sees: "3.99" (NOT "399") ↓ Returns: 3.99 ✓ ``` **Expected Results:** - Before: "3€99" → AI returns 399.00 ❌ - After: "3€99" → Normalized to "3.99 €" → AI returns 3.99 ✅ Fixes L'Incroyable and any other sites using euro-as-separator format. Complements existing comma-separator normalization.
522 lines
24 KiB
Python
522 lines
24 KiB
Python
"""
|
|
Browserless Service
|
|
Manages connections to Browserless.io with advanced stealth capabilities.
|
|
Handles:
|
|
- Connection lifecycle
|
|
- Stealth injection
|
|
- Proxy rotation
|
|
- Human behavior simulation
|
|
"""
|
|
|
|
import asyncio
|
|
import logging
|
|
import random
|
|
import time
|
|
from typing import Optional, Any
|
|
|
|
from playwright.async_api import async_playwright, Browser, BrowserContext, Page
|
|
from app.core.search_config import (
|
|
BROWSERLESS_URL,
|
|
get_amazon_proxies,
|
|
get_random_user_agent,
|
|
COOKIE_ACCEPT_SELECTORS
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Stealth JS to inject
|
|
STEALTH_JS = """
|
|
// Advanced Stealth Mode
|
|
Object.defineProperty(navigator, 'webdriver', { get: () => false });
|
|
delete Object.getPrototypeOf(navigator).webdriver;
|
|
window.chrome = { runtime: {} };
|
|
Object.defineProperty(navigator, 'plugins', { get: () => [1, 2, 3, 4, 5] });
|
|
Object.defineProperty(navigator, 'languages', { get: () => ['fr-FR', 'fr', 'en-US', 'en'] });
|
|
"""
|
|
|
|
class BrowserlessService:
|
|
def __init__(self):
|
|
self._playwright = None
|
|
self._browser: Optional[Browser] = None
|
|
self._proxies = get_amazon_proxies()
|
|
|
|
async def start(self):
|
|
"""Initialize the browser connection"""
|
|
if not self._playwright:
|
|
self._playwright = await async_playwright().start()
|
|
|
|
if not self._browser or not self._browser.is_connected():
|
|
try:
|
|
logger.info(f"Connecting to Browserless at {BROWSERLESS_URL}")
|
|
# Use the stealth endpoint if possible, otherwise standard
|
|
# Note: /chromium/stealth might need specific token or config,
|
|
# falling back to standard connect_over_cdp which is safer for generic browserless
|
|
self._browser = await self._playwright.chromium.connect_over_cdp(
|
|
BROWSERLESS_URL,
|
|
timeout=60000
|
|
)
|
|
except Exception as e:
|
|
logger.error(f"Failed to connect to Browserless: {e}")
|
|
raise
|
|
|
|
async def stop(self):
|
|
"""Close resources"""
|
|
if self._browser:
|
|
await self._browser.close()
|
|
if self._playwright:
|
|
await self._playwright.stop()
|
|
|
|
async def get_context(self, use_proxy: bool = False) -> BrowserContext:
|
|
"""Create a new context with stealth settings"""
|
|
await self.start()
|
|
|
|
user_agent = get_random_user_agent()
|
|
|
|
options = {
|
|
"viewport": {"width": 1920, "height": 1080},
|
|
"user_agent": user_agent,
|
|
"locale": "fr-FR",
|
|
"timezone_id": "Europe/Paris",
|
|
"java_script_enabled": True,
|
|
"bypass_csp": True,
|
|
"extra_http_headers": {
|
|
"Accept-Language": "fr-FR,fr;q=0.9,en-US;q=0.8,en;q=0.7",
|
|
"Sec-Ch-Ua-Platform": '"Windows"',
|
|
"Upgrade-Insecure-Requests": "1",
|
|
}
|
|
}
|
|
|
|
if use_proxy and self._proxies:
|
|
proxy = random.choice(self._proxies)
|
|
options["proxy"] = proxy
|
|
logger.info(f"Using proxy: {proxy['server'].split('//')[1]}")
|
|
|
|
context = await self._browser.new_context(**options)
|
|
|
|
# Block aggressive tracking but keep images for visual verification if needed
|
|
# (Optimized for speed vs detection)
|
|
await context.route("**/*", lambda route: route.continue_())
|
|
|
|
return context
|
|
|
|
async def simulate_human_behavior(self, page: Page):
|
|
"""Simulate random mouse movements and scrolling"""
|
|
try:
|
|
# Mouse movements
|
|
for _ in range(random.randint(2, 5)):
|
|
await page.mouse.move(
|
|
random.randint(100, 1800),
|
|
random.randint(100, 900)
|
|
)
|
|
await asyncio.sleep(random.uniform(0.1, 0.3))
|
|
|
|
# Scrolling
|
|
await page.evaluate("window.scrollBy(0, 300)")
|
|
await asyncio.sleep(random.uniform(0.5, 1.0))
|
|
await page.evaluate("window.scrollBy(0, -100)")
|
|
except Exception:
|
|
pass
|
|
|
|
async def handle_popups(self, page: Page):
|
|
"""Attempt to close cookie banners and popups"""
|
|
try:
|
|
# Check for multiple popups (cookies + promo)
|
|
# We iterate through all selectors and click any that are visible
|
|
# We do this in a loop to handle sequential popups
|
|
for _ in range(3): # Try up to 3 times for sequential popups
|
|
clicked_something = False
|
|
for selector in COOKIE_ACCEPT_SELECTORS:
|
|
try:
|
|
# Use a short timeout for check
|
|
element = page.locator(selector).first
|
|
if await element.is_visible(timeout=100):
|
|
await element.click()
|
|
logger.info(f"Closed popup/banner with {selector}")
|
|
await asyncio.sleep(0.5) # Wait for animation
|
|
clicked_something = True
|
|
except Exception:
|
|
continue
|
|
|
|
if not clicked_something:
|
|
break
|
|
except Exception:
|
|
pass
|
|
|
|
async def _extract_generic_price(self, page: Page) -> str:
|
|
"""
|
|
Extract price from generic e-commerce pages using common selectors and patterns.
|
|
Returns price text (e.g., "1,99 €") or empty string if not found.
|
|
"""
|
|
# Common price selectors used across e-commerce sites (ordered by priority)
|
|
price_selectors = [
|
|
# Highest priority: semantic and explicit current prices
|
|
"[itemprop='price']",
|
|
".current-price",
|
|
".sale-price",
|
|
".final-price",
|
|
".product-price",
|
|
"[data-price]",
|
|
# Medium priority: generic price classes (exclude old/was/original)
|
|
"[class*='price']:not([class*='old']):not([class*='was']):not([class*='original']):not([class*='before']):not([class*='regular'])",
|
|
# ID-based
|
|
"#price",
|
|
"#product-price",
|
|
"#our-price",
|
|
# French-specific
|
|
"span[class*='prix']:not([class*='ancien']):not([class*='barre'])",
|
|
"div[class*='prix']:not([class*='ancien'])",
|
|
"span[class*='tarif']",
|
|
# Lower priority: generic .price (might catch old prices)
|
|
".price",
|
|
]
|
|
|
|
found_prices = []
|
|
|
|
for selector in price_selectors:
|
|
try:
|
|
elements = page.locator(selector)
|
|
count = await elements.count()
|
|
|
|
# Check up to 3 elements per selector
|
|
for i in range(min(count, 3)):
|
|
try:
|
|
element = elements.nth(i)
|
|
if await element.is_visible(timeout=1000):
|
|
# Check if element is strikethrough (old price)
|
|
try:
|
|
text_decoration = await element.evaluate("el => window.getComputedStyle(el).textDecoration")
|
|
if "line-through" in text_decoration:
|
|
logger.debug(f"Skipping strikethrough price at {selector}")
|
|
continue # Skip crossed-out prices
|
|
except Exception:
|
|
pass # If we can't check, continue anyway
|
|
|
|
price_text = await element.inner_text()
|
|
# Check if it looks like a price (contains € or digits with comma/dot)
|
|
if price_text and ('€' in price_text or (',' in price_text and any(c.isdigit() for c in price_text))):
|
|
# Clean up price text
|
|
price_text = price_text.strip()
|
|
found_prices.append((selector, price_text))
|
|
logger.info(f"Found price via selector {selector}: {price_text}")
|
|
|
|
# Return immediately if we found with high-priority selector
|
|
if selector in ["[itemprop='price']", ".current-price", ".sale-price", ".final-price", ".product-price"]:
|
|
return price_text
|
|
except Exception:
|
|
continue
|
|
except Exception:
|
|
continue
|
|
|
|
# If we found prices with lower-priority selectors
|
|
if found_prices:
|
|
# When multiple prices found, the LAST one in DOM order is usually the current price
|
|
# (websites typically show: old price first, then current price)
|
|
best_price = found_prices[-1][1] # Take last (most recently added)
|
|
if len(found_prices) > 1:
|
|
prices_list = [p[1] for p in found_prices]
|
|
logger.info(f"Multiple prices found: {prices_list}, selecting last (most likely current): {best_price}")
|
|
return best_price
|
|
|
|
# Fallback: Use regex to find price patterns in all text
|
|
try:
|
|
all_text = await page.inner_text('body')
|
|
import re
|
|
# Match French price formats: "1,99 €", "12,99€", "1.234,56 €"
|
|
price_matches = re.findall(r'\d+[,\.]\d{2}\s*€', all_text)
|
|
if price_matches:
|
|
# Return first price found
|
|
logger.info(f"Found price via regex: {price_matches[0]}")
|
|
return price_matches[0]
|
|
except Exception:
|
|
pass
|
|
|
|
logger.warning("Could not extract generic price with any method")
|
|
return ""
|
|
|
|
async def _extract_amazon_price(self, page: Page) -> str:
|
|
"""
|
|
Extract price directly from Amazon product page using CSS selectors.
|
|
Returns price text (e.g., "89,99 €") or empty string if not found.
|
|
"""
|
|
# Amazon price selectors in priority order
|
|
# Note: .a-offscreen elements are hidden but contain the full price for screen readers
|
|
price_selectors = [
|
|
".a-price .a-offscreen", # Most reliable - hidden text with full price
|
|
"#corePrice_desktop .a-price .a-offscreen",
|
|
"#corePriceDisplay_desktop_feature_div .a-price .a-offscreen",
|
|
".a-price[data-a-color='price'] .a-offscreen",
|
|
"#priceblock_ourprice", # Older Amazon layout
|
|
"#priceblock_dealprice", # Deal prices
|
|
"span.a-price-whole", # Visible whole number part
|
|
]
|
|
|
|
for selector in price_selectors:
|
|
try:
|
|
# Don't check visibility for .a-offscreen elements (they're hidden by design)
|
|
element = page.locator(selector).first
|
|
|
|
# For offscreen elements, just check if they exist and have text
|
|
if "offscreen" in selector or "priceblock" in selector:
|
|
price_text = await element.inner_text(timeout=2000)
|
|
else:
|
|
# For visible elements, check visibility first
|
|
if await element.is_visible(timeout=2000):
|
|
price_text = await element.inner_text()
|
|
else:
|
|
continue
|
|
|
|
if price_text and price_text.strip():
|
|
logger.info(f"Extracted Amazon price via {selector}: {price_text}")
|
|
return price_text.strip()
|
|
except Exception:
|
|
continue
|
|
|
|
# Try combination: whole + fraction
|
|
try:
|
|
whole = await page.locator("span.a-price-whole").first.inner_text()
|
|
fraction = await page.locator("span.a-price-fraction").first.inner_text()
|
|
if whole and fraction:
|
|
price_text = f"{whole}{fraction}"
|
|
logger.info(f"Extracted Amazon price from whole+fraction: {price_text}")
|
|
return price_text
|
|
except Exception:
|
|
pass
|
|
|
|
logger.warning("Could not extract Amazon price with any selector")
|
|
return ""
|
|
|
|
async def get_page_content(
|
|
self,
|
|
url: str,
|
|
use_proxy: bool = False,
|
|
wait_selector: str = None,
|
|
extract_text: bool = False
|
|
) -> tuple[str, str]:
|
|
"""
|
|
Fetch page content with full stealth lifecycle.
|
|
|
|
Args:
|
|
url: URL to fetch
|
|
use_proxy: Whether to use proxy rotation
|
|
wait_selector: CSS selector to wait for before capturing
|
|
extract_text: If True, returns visible text; if False, returns HTML source
|
|
|
|
Returns:
|
|
(content, screenshot_path) where content is either HTML or visible text
|
|
"""
|
|
context = await self.get_context(use_proxy=use_proxy)
|
|
page = await context.new_page()
|
|
|
|
try:
|
|
logger.info(f"Navigating to {url}")
|
|
# Random delay before start
|
|
await asyncio.sleep(random.uniform(0.5, 1.5))
|
|
|
|
response = await page.goto(url, wait_until="domcontentloaded", timeout=60000)
|
|
|
|
if response.status == 503:
|
|
logger.warning(f"503 Detected on {url}")
|
|
# Retry logic could be handled here or by caller,
|
|
# for now we return what we have, caller decides
|
|
|
|
# Handle pre-search interaction
|
|
from app.core.search_config import SITE_CONFIGS
|
|
for config in SITE_CONFIGS.values():
|
|
if config["search_url"].split("/")[2] in url:
|
|
# Handle complex interaction list
|
|
if config.get("pre_search_interaction"):
|
|
try:
|
|
logger.info(f"Executing pre-search interaction sequence for {config['name']}")
|
|
for step in config["pre_search_interaction"]:
|
|
step_type = step.get("type")
|
|
selector = step.get("selector")
|
|
|
|
if step_type == "input":
|
|
logger.info(f"Inputting text into {selector}")
|
|
await page.fill(selector, step["value"])
|
|
|
|
elif step_type == "click":
|
|
logger.info(f"Clicking {selector}")
|
|
# Use force=True to bypass potential overlays
|
|
await page.click(selector, timeout=5000)
|
|
|
|
elif step_type == "wait":
|
|
seconds = step.get("seconds", 1)
|
|
logger.info(f"Waiting {seconds}s")
|
|
await asyncio.sleep(seconds)
|
|
|
|
logger.info("Pre-search sequence completed")
|
|
await asyncio.sleep(2) # Final stabilization wait
|
|
except Exception as e:
|
|
logger.warning(f"Pre-search sequence failed: {e}")
|
|
|
|
# Handle simple selector (legacy support)
|
|
elif config.get("pre_search_selector"):
|
|
try:
|
|
selector = config["pre_search_selector"]
|
|
logger.info(f"Attempting pre-search interaction: {selector}")
|
|
if await page.locator(selector).is_visible(timeout=5000):
|
|
await page.click(selector)
|
|
logger.info("Clicked pre-search selector")
|
|
await asyncio.sleep(2) # Wait for transition
|
|
except Exception as e:
|
|
logger.warning(f"Pre-search interaction failed: {e}")
|
|
break
|
|
|
|
await self.handle_popups(page)
|
|
await self.simulate_human_behavior(page)
|
|
|
|
if wait_selector:
|
|
try:
|
|
await page.wait_for_selector(wait_selector, timeout=10000)
|
|
except Exception:
|
|
logger.warning(f"Wait selector {wait_selector} timed out")
|
|
|
|
# For Amazon product pages, wait for price to load
|
|
if "amazon" in url.lower() and "/dp/" in url:
|
|
amazon_price_selectors = [
|
|
".a-price .a-offscreen", # Main price element
|
|
"#corePriceDisplay_desktop_feature_div", # Price section
|
|
"#corePrice_desktop", # Alternative price container
|
|
".a-price-whole", # Price number
|
|
]
|
|
for selector in amazon_price_selectors:
|
|
try:
|
|
await page.wait_for_selector(selector, timeout=5000, state="visible")
|
|
logger.info(f"Amazon price element found: {selector}")
|
|
break
|
|
except Exception:
|
|
continue
|
|
else:
|
|
logger.warning("No Amazon price selector found, proceeding anyway")
|
|
|
|
# Wait for network idle to ensure dynamic content loads
|
|
try:
|
|
await page.wait_for_load_state("networkidle", timeout=5000)
|
|
except Exception:
|
|
pass
|
|
|
|
# Extract content based on extract_text parameter
|
|
if extract_text:
|
|
# Extract visible text for AI analysis (monitoring use case)
|
|
try:
|
|
content = await page.inner_text('body')
|
|
logger.info(f"Extracted {len(content)} chars of visible text from page")
|
|
|
|
# Normalize French prices to English format for AI
|
|
import re
|
|
# Pattern 1: digits€XX → digits.XX € (e.g., "3€99" → "3.99 €")
|
|
content = re.sub(r'(\d+)€(\d{2})\b', r'\1.\2 €', content)
|
|
# Pattern 2: digits,XX € → digits.XX € (e.g., "3,99 €" → "3.99 €")
|
|
content = re.sub(r'(\d+),(\d{2})\s*€', r'\1.\2 €', content)
|
|
# Remove spaces in thousands: 1 234.56 → 1234.56
|
|
content = re.sub(r'(\d+)\s(\d{3})', r'\1\2', content)
|
|
logger.debug("Normalized French price formats to English")
|
|
|
|
except Exception as e:
|
|
# Fallback to HTML content if inner_text fails
|
|
logger.warning(f"Failed to extract inner_text, falling back to HTML: {e}")
|
|
content = await page.content()
|
|
|
|
# Extract price directly from DOM for better accuracy
|
|
extracted_price = None
|
|
|
|
# For Amazon, use specific selectors
|
|
if "amazon" in url.lower() and "/dp/" in url:
|
|
extracted_price = await self._extract_amazon_price(page)
|
|
if extracted_price:
|
|
logger.info(f"Extracted Amazon price: {extracted_price}")
|
|
else:
|
|
# For other sites, try common price selectors
|
|
extracted_price = await self._extract_generic_price(page)
|
|
if extracted_price:
|
|
logger.info(f"Extracted generic price: {extracted_price}")
|
|
|
|
# Prepend extracted price to content if found
|
|
if extracted_price:
|
|
# Convert French format to English for AI
|
|
import re
|
|
normalized_price = extracted_price
|
|
# Pattern 1: digits€digits → digits.digits € (e.g., "3€99" → "3.99 €")
|
|
normalized_price = re.sub(r'(\d+)€(\d{2})', r'\1.\2 €', normalized_price)
|
|
# Pattern 2: digits,digits → digits.digits (e.g., "3,99" → "3.99")
|
|
normalized_price = re.sub(r'(\d+),(\d{2})', r'\1.\2', normalized_price)
|
|
# Remove spaces in thousands separators if any (1 234,56 → 1234.56)
|
|
normalized_price = normalized_price.replace(' ', '')
|
|
|
|
content = f"PRIX DÉTECTÉ: {normalized_price}\n\n{content}"
|
|
logger.info(f"Prepended normalized price to content: {normalized_price} (original: {extracted_price})")
|
|
else:
|
|
logger.warning("Could not extract price from DOM, relying on text/image only")
|
|
else:
|
|
# Extract raw HTML for parsing (search use case)
|
|
content = await page.content()
|
|
logger.debug(f"Extracted {len(content)} chars of HTML from page")
|
|
|
|
|
|
# Take screenshot if needed for AI analysis
|
|
screenshot_path = ""
|
|
try:
|
|
import os
|
|
from datetime import datetime
|
|
|
|
# Ensure directory exists
|
|
screenshots_dir = "/app/screenshots"
|
|
if not os.path.exists(screenshots_dir):
|
|
os.makedirs(screenshots_dir, exist_ok=True)
|
|
|
|
timestamp = int(time.time() * 1000)
|
|
safe_name = "".join(c if c.isalnum() else "_" for c in url.split("//")[-1])[:50]
|
|
filename = f"{safe_name}_{timestamp}.jpg"
|
|
screenshot_path = f"{screenshots_dir}/{filename}"
|
|
|
|
# For Amazon, take focused screenshot of product info area
|
|
# Full-page screenshots of Amazon are too large and price becomes tiny when resized
|
|
is_amazon = "amazon" in url.lower() and "/dp/" in url
|
|
|
|
if is_amazon:
|
|
# Try to screenshot just the main product area containing price
|
|
try:
|
|
# Look for main product container
|
|
product_selectors = [
|
|
"#dp-container", # Main product container
|
|
"#ppd", # Product page data
|
|
"#centerCol", # Center column with price
|
|
]
|
|
screenshot_taken = False
|
|
for selector in product_selectors:
|
|
try:
|
|
element = page.locator(selector).first
|
|
if await element.is_visible():
|
|
await element.screenshot(path=screenshot_path, quality=85, type="jpeg")
|
|
logger.info(f"Amazon focused screenshot saved using {selector}")
|
|
screenshot_taken = True
|
|
break
|
|
except Exception:
|
|
continue
|
|
|
|
if not screenshot_taken:
|
|
# Fallback to viewport screenshot (better than full-page for Amazon)
|
|
await page.screenshot(path=screenshot_path, full_page=False, quality=80, type="jpeg")
|
|
logger.info("Amazon viewport screenshot saved (fallback)")
|
|
except Exception as e:
|
|
logger.warning(f"Amazon focused screenshot failed: {e}, using viewport")
|
|
await page.screenshot(path=screenshot_path, full_page=False, quality=80, type="jpeg")
|
|
else:
|
|
# For non-Amazon sites, use full_page to capture entire page
|
|
await page.screenshot(path=screenshot_path, full_page=True, quality=80, type="jpeg")
|
|
logger.info(f"Full-page screenshot saved to {screenshot_path}")
|
|
except Exception as e:
|
|
logger.warning(f"Failed to take screenshot: {e}")
|
|
|
|
return content, screenshot_path
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error fetching {url}: {e}")
|
|
return "", ""
|
|
finally:
|
|
await context.close()
|
|
|
|
# Global instance
|
|
browserless_service = BrowserlessService()
|