mydatascraper/competitor_compare
0
1import asyncio
2import re
3from typing import List, Optional
4from datetime import datetime, timezone
5from thefuzz import fuzz
6from app.scrapers import WalmartScraper, AldiScraper, TargetScraper, WegmansScraper
7from app.schemas import (
8 ComparisonResponse,
9 RetailerResult,
10 ProductResult,
11 MultiCompareResponse,
12)
13
14
15# ── Size normalization ─────────────────────────────────────────────────────────
16
17# Convert everything to a common unit (fl oz for liquids, oz for weight)
18SIZE_CONVERSIONS = {
19 "gallon": 128, # 1 gallon = 128 fl oz
20 "gal": 128,
21 "half gallon": 64,
22 "0.5 gallon": 64,
23 "quart": 32,
24 "qt": 32,
25 "pint": 16,
26 "pt": 16,
27 "liter": 33.814,
28 "litre": 33.814,
29 "l": 33.814,
30 "ml": 0.033814,
31 "lb": 16, # 1 lb = 16 oz
32 "pound": 16,
33 "kg": 35.274,
34 "g": 0.035274,
35}
36
37
38def normalize_size(size_str: str) -> Optional[float]:
39 """
40 Convert size string to a normalized numeric value in base units.
41 Returns fl oz for liquids, oz for weights, or count.
42
43 Examples:
44 "1 gallon" → 128.0 (fl oz)
45 "0.5 gallon" → 64.0 (fl oz)
46 "59 fl. oz." → 59.0
47 "64 fl oz" → 64.0
48 "52 fl oz" → 52.0
49 "1 gal" → 128.0
50 "16 oz" → 16.0
51 "12 count" → 12.0
52 """
53 if not size_str:
54 return None
55
56 text = size_str.lower().strip()
57
58 # Remove parentheses, periods
59 text = text.replace(".", " ").replace("(", "").replace(")", "")
60
61 # Special case: "0.5 gallon" or "half gallon"
62 if "half gallon" in text or "1/2 gallon" in text or "1/2 gal" in text:
63 return 64.0
64
65 # Try to extract number + unit
66 # Pattern: number (with optional decimal) followed by unit
67 m = re.search(
68 r"(\d+(?:\.\d+)?)\s*"
69 r"(fl\.?\s*oz|fluid\s*ounce|oz|ounce|gallon|gal|quart|qt|"
70 r"pint|pt|liter|litre|l|ml|lb|pound|kg|g|count|ct|each|ea|pk|pack)",
71 text,
72 )
73
74 if not m:
75 # Try just a number
76 m2 = re.search(r"(\d+(?:\.\d+)?)", text)
77 if m2:
78 val = float(m2.group(1))
79 # Check for gallon in rest of text
80 if "gallon" in text or "gal" in text:
81 return val * 128
82 return val
83 return None
84
85 value = float(m.group(1))
86 unit = m.group(2).strip()
87
88 # Normalize unit
89 unit = re.sub(r"\s+", " ", unit)
90
91 if unit in ("fl oz", "fl oz", "fluid ounce", "fluid ounces"):
92 return value
93 elif unit in ("oz", "ounce", "ounces"):
94 return value
95 elif unit in ("gallon", "gal", "gallons"):
96 return value * 128
97 elif unit in ("quart", "qt"):
98 return value * 32
99 elif unit in ("pint", "pt"):
100 return value * 16
101 elif unit in ("liter", "litre", "l"):
102 return value * 33.814
103 elif unit == "ml":
104 return value * 0.033814
105 elif unit in ("lb", "pound", "pounds"):
106 return value * 16
107 elif unit == "kg":
108 return value * 35.274
109 elif unit == "g":
110 return value * 0.035274
111 elif unit in ("count", "ct", "each", "ea", "pk", "pack"):
112 return value
113 else:
114 return value
115
116
117def sizes_are_comparable(size1: Optional[float], size2: Optional[float]) -> bool:
118 """
119 Check if two normalized sizes are close enough to be comparable.
120 Allows up to 30% difference (e.g., 59 fl oz vs 64 fl oz).
121 """
122 if size1 is None or size2 is None:
123 return True # If we can't determine size, still compare
124
125 if size1 == 0 or size2 == 0:
126 return True
127
128 ratio = max(size1, size2) / min(size1, size2)
129 return ratio <= 1.35 # Within 35% of each other
130
131
132def compute_cross_retailer_score(
133 product: ProductResult,
134 query: str,
135 anchor_product: Optional[ProductResult] = None,
136) -> float:
137 """
138 Score a product for cross-retailer comparison.
139
140 Considers:
141 1. Name match to query (fuzzy)
142 2. Size similarity to anchor product (if provided)
143 3. Brand similarity to anchor (if provided)
144 4. Penalizes very different sizes
145 """
146 name = product.name.lower()
147
148 # Base score: fuzzy match to query
149 ratio = fuzz.ratio(query.lower(), name)
150 partial = fuzz.partial_ratio(query.lower(), name)
151 token_sort = fuzz.token_sort_ratio(query.lower(), name)
152 token_set = fuzz.token_set_ratio(query.lower(), name)
153 base_score = ratio * 0.15 + partial * 0.25 + token_sort * 0.3 + token_set * 0.3
154
155 # If no anchor, return base score
156 if not anchor_product:
157 return base_score
158
159 # Size comparison bonus/penalty
160 size_bonus = 0
161 prod_size = normalize_size(product.size or "")
162 anchor_size = normalize_size(anchor_product.size or "")
163
164 if prod_size and anchor_size and anchor_size > 0:
165 ratio_val = prod_size / anchor_size
166 if 0.8 <= ratio_val <= 1.2:
167 size_bonus = 15 # Very similar size
168 elif 0.65 <= ratio_val <= 1.35:
169 size_bonus = 5 # Close enough
170 else:
171 size_bonus = -20 # Very different size — penalize
172
173 # Name similarity to anchor product
174 anchor_name_score = fuzz.token_set_ratio(
175 anchor_product.name.lower(), name
176 )
177 name_bonus = (anchor_name_score - 50) * 0.2 # -10 to +10
178
179 return base_score + size_bonus + name_bonus
180
181
182class ComparisonService:
183 def __init__(self):
184 self.scrapers = [
185 WalmartScraper(),
186 AldiScraper(),
187 TargetScraper(),
188 WegmansScraper(),
189 ]
190
191 async def compare(
192 self,
193 query: str,
194 retailers: Optional[List[str]] = None,
195 ) -> ComparisonResponse:
196 # Filter scrapers
197 active_scrapers = self.scrapers
198 if retailers:
199 retailers_lower = [r.lower() for r in retailers]
200 active_scrapers = [
201 s for s in self.scrapers
202 if s.RETAILER_NAME in retailers_lower
203 ]
204
205 print(f"\n{'='*60}")
206 print(f"🔍 Searching: '{query}' across {len(active_scrapers)} retailers")
207 print(f"{'='*60}")
208
209 # Run all scrapers concurrently
210 tasks = [scraper.search(query) for scraper in active_scrapers]
211 results: List[RetailerResult] = await asyncio.gather(
212 *tasks, return_exceptions=True
213 )
214
215 # Handle exceptions
216 cleaned_results = []
217 for i, result in enumerate(results):
218 if isinstance(result, Exception):
219 cleaned_results.append(
220 RetailerResult(
221 retailer=active_scrapers[i].RETAILER_NAME,
222 status="failed",
223 scrape_method="none",
224 error=str(result),
225 )
226 )
227 else:
228 cleaned_results.append(result)
229
230 # ── Smart cross-retailer matching ──────────────────────────────────
231 cleaned_results = self._align_best_matches(query, cleaned_results)
232
233 return self._build_comparison(query, cleaned_results)
234
235 async def multi_compare(
236 self,
237 products: List[str],
238 retailers: Optional[List[str]] = None,
239 ) -> MultiCompareResponse:
240 tasks = [self.compare(product, retailers) for product in products]
241 comparisons = await asyncio.gather(*tasks)
242 return MultiCompareResponse(
243 comparisons=list(comparisons),
244 timestamp=datetime.now(timezone.utc).isoformat(),
245 )
246
247 def _align_best_matches(
248 self,
249 query: str,
250 results: List[RetailerResult],
251 ) -> List[RetailerResult]:
252 """
253 Re-pick best_match for each retailer so they are comparable products.
254
255 Strategy:
256 1. Find the "anchor" — the most common/standard product across retailers
257 2. For each retailer, re-score products based on similarity to anchor
258 3. Pick the product that best matches the anchor from each retailer
259 """
260 # Collect all successful results with products
261 successful = [
262 r for r in results
263 if r.status == "success" and r.all_results
264 ]
265
266 if len(successful) < 2:
267 return results
268
269 # ── Step 1: Find the anchor product ────────────────────────────────
270 # Pick the best match from the retailer with the most results
271 # This is likely the most "standard" version of the product
272 anchor = self._find_anchor_product(query, successful)
273
274 if not anchor:
275 return results
276
277 print(f"\n[COMPARE] Anchor product: '{anchor.name}' "
278 f"size={anchor.size} price=${anchor.price}")
279
280 # ── Step 2: Re-pick best match for each retailer ──────────────────
281 for r in results:
282 if r.status != "success" or not r.all_results:
283 continue
284
285 best_score = -999
286 best_product = None
287
288 for product in r.all_results:
289 if not product.price:
290 continue
291
292 score = compute_cross_retailer_score(product, query, anchor)
293
294 if score > best_score:
295 best_score = score
296 best_product = product
297
298 if best_product:
299 old_name = r.best_match.name if r.best_match else "None"
300 r.best_match = best_product
301 r.best_match.match_score = best_score
302
303 if old_name != best_product.name:
304 print(
305 f"[COMPARE] {r.retailer}: "
306 f"'{old_name[:30]}' → '{best_product.name[:30]}' "
307 f"(${best_product.price}, size={best_product.size})"
308 )
309
310 return results
311
312 def _find_anchor_product(
313 self,
314 query: str,
315 successful: List[RetailerResult],
316 ) -> Optional[ProductResult]:
317 """
318 Find the best "anchor" product to align all retailers against.
319
320 Prefers:
321 - Standard sizes (gallon, half gallon for milk, etc.)
322 - Generic/store brand (more likely to exist everywhere)
323 - High fuzzy match to query
324 """
325 candidates = []
326
327 for r in successful:
328 for product in r.all_results:
329 if not product.price:
330 continue
331
332 score = 0
333
334 # Fuzzy match to query
335 name_score = fuzz.token_set_ratio(
336 query.lower(), product.name.lower()
337 )
338 score += name_score
339
340 # Prefer standard sizes
341 size_val = normalize_size(product.size or "")
342 if size_val:
343 # Common milk sizes: 128 oz (1 gal), 64 oz (0.5 gal), 59 oz
344 standard_sizes = [128, 64, 59, 52, 32, 16]
345 for std in standard_sizes:
346 if 0.9 <= size_val / std <= 1.1:
347 score += 10
348 break
349
350 # Prefer simpler product names (less likely to be specialty)
351 word_count = len(product.name.split())
352 if word_count <= 5:
353 score += 5
354 if word_count > 8:
355 score -= 5
356
357 # Penalize premium/specialty keywords
358 name_lower = product.name.lower()
359 premium_keywords = [
360 "organic", "grass", "a2", "ultra", "premium",
361 "specialty", "goat", "oat", "almond", "soy",
362 "chocolate", "strawberry", "vanilla",
363 "lactose", "kefir", "buttermilk",
364 ]
365 for kw in premium_keywords:
366 if kw in name_lower:
367 score -= 8
368
369 # Bonus for store brands (more likely comparable)
370 store_brands = [
371 "great value", "good & gather", "friendly farms",
372 "wegmans", "market pantry", "simply balanced",
373 ]
374 for sb in store_brands:
375 if sb in name_lower:
376 score += 5
377 break
378
379 candidates.append((product, score))
380
381 if not candidates:
382 return None
383
384 # Sort by score, pick best
385 candidates.sort(key=lambda x: -x[1])
386
387 anchor = candidates[0][0]
388 return anchor
389
390 def _build_comparison(
391 self, query: str, results: List[RetailerResult]
392 ) -> ComparisonResponse:
393 # Find cheapest/most expensive among successful results with prices
394 priced_results = []
395 for r in results:
396 if r.status == "success" and r.best_match and r.best_match.price:
397 priced_results.append(r)
398
399 cheapest = None
400 most_expensive = None
401 price_spread = None
402 summary = None
403
404 if priced_results:
405 sorted_by_price = sorted(
406 priced_results, key=lambda r: r.best_match.price
407 )
408 cheapest_r = sorted_by_price[0]
409 expensive_r = sorted_by_price[-1]
410
411 cheapest = {
412 "retailer": cheapest_r.retailer,
413 "product_name": cheapest_r.best_match.name,
414 "price": cheapest_r.best_match.price,
415 "size": cheapest_r.best_match.size,
416 "product_url": cheapest_r.best_match.product_url,
417 }
418
419 most_expensive = {
420 "retailer": expensive_r.retailer,
421 "product_name": expensive_r.best_match.name,
422 "price": expensive_r.best_match.price,
423 "size": expensive_r.best_match.size,
424 "product_url": expensive_r.best_match.product_url,
425 }
426
427 if len(priced_results) >= 2:
428 spread = round(
429 expensive_r.best_match.price - cheapest_r.best_match.price,
430 2,
431 )
432 price_spread = spread
433
434 savings_pct = round(
435 (spread / expensive_r.best_match.price) * 100, 1
436 )
437
438 # Build summary with size info
439 cheap_size = cheapest_r.best_match.size or ""
440 exp_size = expensive_r.best_match.size or ""
441
442 summary = (
443 f"Best price for '{query}': "
444 f"${cheapest_r.best_match.price:.2f} "
445 f"at {cheapest_r.retailer.title()} "
446 f"({cheapest_r.best_match.name}"
447 f"{', ' + cheap_size if cheap_size else ''}). "
448 f"${spread:.2f} cheaper than "
449 f"{expensive_r.retailer.title()} "
450 f"({expensive_r.best_match.name}"
451 f"{', ' + exp_size if exp_size else ''}"
452 f" at ${expensive_r.best_match.price:.2f}), "
453 f"save {savings_pct}%. "
454 f"Compared across {len(priced_results)} retailers."
455 )
456 elif len(priced_results) == 1:
457 summary = (
458 f"Found '{query}' at {cheapest_r.retailer.title()} "
459 f"for ${cheapest_r.best_match.price:.2f}. "
460 f"Only 1 retailer returned results."
461 )
462
463 retailers_with_results = sum(
464 1 for r in results if r.status == "success"
465 )
466
467 return ComparisonResponse(
468 query=query,
469 timestamp=datetime.now(timezone.utc).isoformat(),
470 total_retailers_searched=len(results),
471 retailers_with_results=retailers_with_results,
472 results=results,
473 cheapest=cheapest,
474 most_expensive=most_expensive,
475 price_spread=price_spread,
476 summary=summary,
477 )