Michael/Priceflow
1
0 Fork 0
mirror of https://github.com/R0m1k3/Priceflow.git synced 2026-10-11 17:29:14 +02:00
Code Issues Packages Projects Releases Wiki Activity

test: Update search verification query and target site.

Browse Source
This commit is contained in:
Michael committed 2025-12-22 15:42:33 +01:00
1 parent 0ff229de13
commit 39d4107e6d
2 files changed
+96 -71
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

app/services/improved_search_service.py
+94 -69
View File
@@ -36,6 +36,7 @@ COMMON_POPUP_SELECTORS = [
class SearchResult:
"""Search result data class"""
def __init__(
self,
url: str,
@@ -177,8 +178,6 @@ class ImprovedSearchService:
except Exception:
pass
@staticmethod
def _parse_results(html: str, site_key: str, base_url: str, query: str) -> list[SearchResult]:
"""
@@ -193,7 +192,9 @@ class ImprovedSearchService:
query_words = [w.lower() for w in query.split() if len(w) > 2]
# Select product links
logger.debug(f"Parsing content for {site_key} (length: {len(html)}) with selector: {config['product_selector']}")
logger.debug(
f"Parsing content for {site_key} (length: {len(html)}) with selector: {config['product_selector']}"
)
links = soup.select(config["product_selector"])
logger.debug(f"Found {len(links)} raw items for {site_key}")
@@ -207,7 +208,7 @@ class ImprovedSearchService:
logger.debug(f" Processing Centrakor item: {item.name}, classes: {item.get('class')}")
# If selector targets the container (div.product-card), we need to find the link inside
if item.name != 'a':
if item.name != "a":
# Try to find the link using product_link_selector if available
if "product_link_selector" in config:
found_link = item.select_one(config["product_link_selector"])
@@ -215,7 +216,7 @@ class ImprovedSearchService:
link = found_link
else:
# Fallback: find first 'a' tag
found_link = item.find('a')
found_link = item.find("a")
if found_link:
link = found_link
else:
@@ -223,7 +224,7 @@ class ImprovedSearchService:
continue
else:
# No link selector specified, try to find first 'a'
found_link = item.find('a')
found_link = item.find("a")
if found_link:
link = found_link
else:
@@ -273,10 +274,10 @@ class ImprovedSearchService:
for img_el in img_els:
# Try multiple attributes in order of priority
candidate_url = (
img_el.get("src") or
img_el.get("data-src") or
img_el.get("data-lazy-src") or
img_el.get("data-original")
img_el.get("src")
or img_el.get("data-src")
or img_el.get("data-lazy-src")
or img_el.get("data-original")
)
# Handle srcset (use first URL)
@@ -293,7 +294,9 @@ class ImprovedSearchService:
logger.debug(f" ⏭️ Skipping placeholder: {candidate_url[:50]}")
continue
# Filter only tiny pictos
if 'picto' in candidate_url.lower() and ('width=60' in candidate_url or 'height=80' in candidate_url):
if "picto" in candidate_url.lower() and (
"width=60" in candidate_url or "height=80" in candidate_url
):
logger.debug(f" ⏭️ Skipping tiny picto: {candidate_url[:50]}")
continue
image_url = candidate_url
@@ -302,13 +305,14 @@ class ImprovedSearchService:
# Normal filtering for other sites
# Filter out obvious pictos and small icons
if any(keyword in candidate_url.lower() for keyword in ['picto', 'icon', 'logo', 'badge']):
if any(keyword in candidate_url.lower() for keyword in ["picto", "icon", "logo", "badge"]):
logger.debug(f" ⏭️ Skipping picto/icon: {candidate_url[:50]}")
continue
# Filter out VERY small images (less than 100px)
import re
width_match = re.search(r'width=(\d+)', candidate_url)
height_match = re.search(r'height=(\d+)', candidate_url)
width_match = re.search(r"width=(\d+)", candidate_url)
height_match = re.search(r"height=(\d+)", candidate_url)
if width_match and height_match:
width = int(width_match.group(1))
height = int(height_match.group(1))
@@ -326,10 +330,7 @@ class ImprovedSearchService:
img = link.find("img")
if img:
image_url = (
img.get("src") or
img.get("data-src") or
img.get("data-lazy-src") or
img.get("data-original")
img.get("src") or img.get("data-src") or img.get("data-lazy-src") or img.get("data-original")
)
# Handle srcset
@@ -360,10 +361,12 @@ class ImprovedSearchService:
# Clean up and validate image URL
if image_url:
# Remove data URIs, 1x1 pixels, placeholders
if (image_url.startswith("data:") or
"1x1" in image_url or
"placeholder" in image_url.lower() or
image_url.strip() == ""):
if (
image_url.startswith("data:")
or "1x1" in image_url
or "placeholder" in image_url.lower()
or image_url.strip() == ""
):
logger.debug(f" ⏭️ Skipping invalid image: {image_url[:50]}")
image_url = None
elif not image_url.startswith("http"):
@@ -378,8 +381,9 @@ class ImprovedSearchService:
if site_key == "gifi.fr":
# Gifi: Extract price from product tile HTML
import re
container_html = str(container)
price_match = re.search(r'(\d+)[,.](\d+)\s*€', container_html)
price_match = re.search(r"(\d+)[,.](\d+)\s*€", container_html)
if price_match:
euros = int(price_match.group(1))
cents = int(price_match.group(2))
@@ -387,14 +391,16 @@ class ImprovedSearchService:
logger.debug(f" 💰 Extracted price from search: {product_price}€")
# Create result
results.append(SearchResult(
results.append(
SearchResult(
url=full_url,
title=title,
snippet=f"Product from {config['name']}",
source=config["name"],
price=product_price, # Set price if extracted from search
image_url=image_url
))
image_url=image_url,
)
)
logger.debug(f"Parsed {len(results)} results from HTML")
return results
@@ -406,8 +412,11 @@ class ImprovedSearchService:
# SPECIAL CASE: L'Incroyable - Price is in the title
if "lincroyable.fr" in result.url:
import re
# Extract price from title (e.g., "34€99" or "59€99")
price_match = re.search(r'(\d+)€(\d+)', result.title)
# Extract price from title (e.g., "34€99" or "59€99")
# Fix: Use non-greedy regex and ensure only 2 decimals
price_match = re.search(r"(\d+)€(\d{2})(?!\d)", result.title)
if price_match:
# Convert to float (e.g., "34€99" -> 34.99)
price_euros = int(price_match.group(1))
@@ -415,7 +424,7 @@ class ImprovedSearchService:
result.price = float(f"{price_euros}.{price_cents}")
# Clean title by removing price
result.title = re.sub(r'\d+€\d+', '', result.title).strip()
result.title = re.sub(r"\d+€\d+", "", result.title).strip()
logger.debug(f"L'Incroyable - Extracted price {result.price}€ from title")
return result
@@ -483,13 +492,13 @@ class ImprovedSearchService:
data = data[0] if data else {}
# Check for Product type
if data.get('@type') == 'Product':
offers = data.get('offers')
if data.get("@type") == "Product":
offers = data.get("offers")
if isinstance(offers, list):
offers = offers[0]
if offers and 'price' in offers:
price = float(offers['price'])
if offers and "price" in offers:
price = float(offers["price"])
logger.debug(f" ✅ Found price via JSON-LD: {price}€")
return price
except:
@@ -516,8 +525,8 @@ class ImprovedSearchService:
# If found, we trust them and return immediately.
strict_selectors = [
'[itemprop="price"]', # Schema.org standard
'.current-price-value',
'.product-price .price',
".current-price-value",
".product-price .price",
'meta[property="product:price:amount"]',
'meta[name="twitter:data1"]', # Sometimes used for price
]
@@ -525,13 +534,13 @@ class ImprovedSearchService:
for selector in strict_selectors:
try:
# Check meta tags first
if selector.startswith('meta'):
if selector.startswith("meta"):
element = await page.query_selector(selector)
if element:
content = await element.get_attribute('content')
content = await element.get_attribute("content")
if content:
try:
price_val = float(content.strip().replace(',', '.'))
price_val = float(content.strip().replace(",", "."))
logger.debug(f" ✅ Found price via Meta Tag {selector}: {price_val}€")
return price_val
except:
@@ -547,9 +556,9 @@ class ImprovedSearchService:
price_text = await elem.inner_text()
if price_text:
cleaned = price_text.strip().replace('€', '').replace('EUR', '').strip()
cleaned = cleaned.replace(' ', '').replace('\xa0', '').replace(',', '.')
match = re.search(r'(\d+\.?\d*)', cleaned)
cleaned = price_text.strip().replace("€", "").replace("EUR", "").strip()
cleaned = cleaned.replace(" ", "").replace("\xa0", "").replace(",", ".")
match = re.search(r"(\d+\.?\d*)", cleaned)
if match:
try:
price_val = float(match.group(1))
@@ -566,10 +575,10 @@ class ImprovedSearchService:
# PRIORITY 1: Selectors for sale/promotional prices
sale_price_selectors = [
'.price-current',
'.prix-actuel',
'.sale-price',
'.promo-price',
".price-current",
".prix-actuel",
".sale-price",
".promo-price",
'[class*="promo"]',
'[class*="sale"]',
'[class*="discount"]',
@@ -577,30 +586,41 @@ class ImprovedSearchService:
# PRIORITY 2: Standard price selectors
price_selectors = [
'.price',
".price",
'[data-testid="price"]',
'.product-price',
'.a-price .a-offscreen',
'.a-price-whole',
'span[class*="price"]',
".product-price",
".a-price .a-offscreen",
".a-price-whole",
# REMOVED: 'span[class*="price"]' - Too broad, picks up 'price-tax-excluded' (HT)
]
# Blacklist words that indicate installments or taxes
FORBIDDEN_WORDS = ["mois", "fois", "x", "ht", "hors tax", "eco-part", "sans"]
# Try sale prices first
for selector in sale_price_selectors:
try:
elements = await page.query_selector_all(selector)
for elem in elements:
# Skip if element is strikethrough (old price)
parent_html = await elem.evaluate('el => el.parentElement.outerHTML')
if 'text-decoration: line-through' in parent_html or 'text-decoration-line: line-through' in parent_html:
parent_html = await elem.evaluate("el => el.parentElement.outerHTML")
if (
"text-decoration: line-through" in parent_html
or "text-decoration-line: line-through" in parent_html
):
continue
price_text = await elem.inner_text()
if price_text:
cleaned = price_text.strip().replace('€', '').replace('EUR', '').strip()
cleaned = cleaned.replace(' ', '').replace('\xa0', '').replace(',', '.')
text_lower = price_text.lower()
# Filter out forbidden words
if any(w in text_lower for w in FORBIDDEN_WORDS):
continue
match = re.search(r'(\d+\.?\d*)', cleaned)
cleaned = price_text.strip().replace("€", "").replace("EUR", "").strip()
cleaned = cleaned.replace(" ", "").replace("\xa0", "").replace(",", ".")
match = re.search(r"(\d+\.?\d*)", cleaned)
if match:
try:
price_val = float(match.group(1))
@@ -612,7 +632,6 @@ class ImprovedSearchService:
except Exception:
continue
# Fallback to standard price selectors
all_prices = []
@@ -622,21 +641,24 @@ class ImprovedSearchService:
for elem in elements:
# Skip if element is strikethrough (old price)
try:
parent_html = await elem.evaluate('el => el.parentElement.outerHTML')
if 'text-decoration: line-through' in parent_html or 'text-decoration-line: line-through' in parent_html:
parent_html = await elem.evaluate("el => el.parentElement.outerHTML")
if (
"text-decoration: line-through" in parent_html
or "text-decoration-line: line-through" in parent_html
):
continue
elem_style = await elem.evaluate('el => window.getComputedStyle(el).textDecoration')
if 'line-through' in elem_style:
elem_style = await elem.evaluate("el => window.getComputedStyle(el).textDecoration")
if "line-through" in elem_style:
continue
except:
pass
price_text = await elem.inner_text()
if price_text:
cleaned = price_text.strip().replace('€', '').replace('EUR', '').strip()
cleaned = cleaned.replace(' ', '').replace('\xa0', '').replace(',', '.')
cleaned = price_text.strip().replace("€", "").replace("EUR", "").strip()
cleaned = cleaned.replace(" ", "").replace("\xa0", "").replace(",", ".")
match = re.search(r'(\d+\.?\d*)', cleaned)
match = re.search(r"(\d+\.?\d*)", cleaned)
if match:
try:
price_val = float(match.group(1))
@@ -668,7 +690,7 @@ class ImprovedSearchService:
"out of stock",
"unavailable",
"épuisé",
"non disponible"
"non disponible",
]
try:
@@ -692,9 +714,6 @@ class ImprovedSearchService:
return None # Unknown
@classmethod
async def search_site_generator(cls, site_key: str, query: str) -> AsyncGenerator[SearchResult, None]:
"""Search a single site and yield results as they are scraped"""
@@ -742,10 +761,11 @@ class ImprovedSearchService:
logger.warning(f"No results found for {site_key}")
# Dump HTML for debugging
import os
dump_dir = "/app/debug_dumps"
os.makedirs(dump_dir, exist_ok=True)
dump_path = f"{dump_dir}/{site_key.replace('.', '_')}_no_results.html"
with open(dump_path, 'w', encoding='utf-8') as f:
with open(dump_path, "w", encoding="utf-8") as f:
f.write(content)
logger.warning(f"HTML dumped to {dump_path} for inspection")
return
@@ -788,6 +808,7 @@ class ImprovedSearchService:
# COMPATIBILITY LAYER FOR API ROUTERS
# ==========================================
async def search_products(
query: str,
db: Session,
@@ -820,7 +841,9 @@ async def search_products(
matched_key = None
# Normalize domain for comparison (remove www., lowercase, etc.)
site_domain_normalized = site.domain.lower().replace("www.", "").replace("http://", "").replace("https://", "").strip("/")
site_domain_normalized = (
site.domain.lower().replace("www.", "").replace("http://", "").replace("https://", "").strip("/")
)
# Try multiple matching strategies:
for key in SITE_CONFIGS.keys():
@@ -859,7 +882,9 @@ async def search_products(
if matched_key:
site_keys.append(matched_key)
else:
logger.warning(f"❌ No config found for {site.name} (domain: {site.domain}, normalized: {site_domain_normalized})")
logger.warning(
f"❌ No config found for {site.name} (domain: {site.domain}, normalized: {site_domain_normalized})"
)
# 3. Execute searches and stream results
generators = [ImprovedSearchService.search_site_generator(key, query) for key in site_keys]
verify_search_price.py
+2 -2
View File
@@ -30,8 +30,8 @@ ImprovedSearchService._connect_browser = _connect_browser_local
async def test_search():
await ImprovedSearchService.initialize()
query = "nintendo switch"
target_site = "stokomani.fr"
query = "Mange-debout" # Common item on their site
target_site = "lincroyable.fr"
print(f"Searching for: {query} on {target_site}")