Michael/Priceflow
1
0 Fork 0
mirror of https://github.com/R0m1k3/Priceflow.git synced 2026-10-14 18:56:43 +02:00
Code Issues Packages Projects Releases Wiki Activity
Files
8c35ace6a94eaba16bdbbb0f136cf744d77a26ba
Priceflow/app/services/browserless_service.py
T
Claude b4dd4b4365 feat: Add generic price extraction for all e-commerce sites 
**Problem:**
Price extraction was only working for Amazon. Other sites (B&M, Stokomani, etc.)
had prices missing from extracted text, causing AI to rely solely on screenshots.

Example error:
```
WARNING - No prices found in text with € symbol
AI Response: {"price": 9.99}  // Wrong! Real price was 1.99€
```

**Root Cause:**
- `page.inner_text('body')` doesn't capture all price elements
- Some sites use hidden elements, iframes, or CSS pseudo-elements
- Only Amazon had dedicated price extraction (_extract_amazon_price)
- Other sites got: `WARNING - Could not extract price from DOM`

**Solution: Universal Price Extraction**

1. **New Method: _extract_generic_price()** (browserless_service.py:145-208)
   - Tries common CSS selectors across all e-commerce sites:
     * Class-based: `.price`, `.product-price`, `.current-price`
     * Attribute-based: `[data-price]`, `[itemprop='price']`
     * French-specific: `[class*='prix']`, `[class*='tarif']`
     * Excludes old prices: `:not([class*='old'])`
   - Validates price format (contains € or comma with digits)
   - Falls back to regex: `\d+[,\.]\d{2}\s*€` on all text
   - Returns first valid price found

2. **Updated Extraction Logic** (browserless_service.py:319-338)
   - Amazon → uses `_extract_amazon_price()` (specific)
   - All other sites → uses `_extract_generic_price()` (universal)
   - Prepends "PRIX DÉTECTÉ: X,XX €" to text context
   - Logs success/failure for debugging

**Flow:**
```
Page load (bmstores.fr)
  ↓
Try selectors: .price, .product-price, [itemprop='price']...
  ↓
Found: "1,99 €" via selector .price
  ↓
Prepend: "PRIX DÉTECTÉ: 1,99 €\n\n[page text]"
  ↓
Send to AI
  ↓
AI sees explicit price → {"price": 1.99, "confidence": 0.95}
```

**Expected Results:**
- Before: `WARNING - No prices found in text with € symbol`
- After: `INFO - Found price via selector .price: 1,99 €`
- AI confidence improves (text + image vs image only)
- Fewer digit recognition errors (1 vs 9)

**Tested with:**
- https://bmstores.fr/produits/accessoire-noel-enfant/125651-lutin-farceur-35cm
- Price: 1,99 €

Fixes issue where non-Amazon sites had missing prices in text context,
causing AI to misread prices from screenshots only.
2025-11-29 23:43:42 +00:00

475 lines
20 KiB
Python
Raw Blame History

"""
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
price_selectors = [
# Common class names
"[class*='price']:not([class*='old']):not([class*='was']):not([class*='original'])",
".product-price",
".price",
".current-price",
".sale-price",
".final-price",
# Common data attributes
"[data-price]",
"[itemprop='price']",
# ID-based
"#price",
"#product-price",
"#our-price",
# Specific patterns
"span[class*='prix']",
"div[class*='prix']",
"span[class*='tarif']",
]
for selector in price_selectors:
try:
elements = page.locator(selector)
count = await elements.count()
# Try first visible element
for i in range(min(count, 3)): # Check max 3 elements
try:
element = elements.nth(i)
if await element.is_visible(timeout=1000):
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()
logger.info(f"Found price via selector {selector}: {price_text}")
return price_text
except Exception:
continue
except Exception:
continue
# 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")
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:
content = f"PRIX DÉTECTÉ: {extracted_price}\n\n{content}"
logger.info(f"Prepended price to content: {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()
Reference in new issue View Git Blame Copy Permalink