How to Build a Walmart Product Tracker

import os, requests, re, json, time
from datetime import datetime

SCAVIO_KEY = os.environ['SCAVIO_API_KEY']
URL = 'https://api.scavio.dev/api/v1/search'
H = {'x-api-key': SCAVIO_KEY, 'Content-Type': 'application/json'}

def search_walmart(product: str, num: int = 5) -> list:
    resp = requests.post(URL, headers=H,
        json={'query': f'site:walmart.com {product}',
              'country_code': 'us', 'num_results': num})
    resp.raise_for_status()
    results = []
    for r in resp.json().get('organic_results', []):
        title = r.get('title', '').replace(' - Walmart.com', '')
        snippet = r.get('snippet', '')
        # Extract price
        price_match = re.search(r'\$([\d,]+\.\d{2})', snippet + ' ' + title)
        price = float(price_match.group(1).replace(',', '')) if price_match else None
        # Extract rating
        rating_match = re.search(r'(\d+\.?\d*)\s*(?:out of 5|stars?|/5)', snippet)
        rating = float(rating_match.group(1)) if rating_match else None
        results.append({
            'title': title,
            'price': price,
            'rating': rating,
            'url': r['link'],
            'snippet': snippet[:200],
            'timestamp': datetime.now().isoformat(),
        })
    return results

products = search_walmart('4K TV 55 inch')
print(f'Found {len(products)} Walmart products')
for p in products:
    price_str = f'${p["price"]:.2f}' if p['price'] else 'N/A'
    print(f'  {price_str:>10s} | {p["title"][:50]}')

DB_FILE = 'walmart_tracker.json'

def load_db() -> dict:
    if os.path.exists(DB_FILE):
        with open(DB_FILE) as f:
            return json.load(f)
    return {'products': {}, 'last_updated': None}

def save_db(db: dict):
    db['last_updated'] = datetime.now().isoformat()
    with open(DB_FILE, 'w') as f:
        json.dump(db, f, indent=2)

def track_product(product_name: str) -> dict:
    """Search and record current price."""
    db = load_db()
    results = search_walmart(product_name, num=3)
    if not results:
        return {'product': product_name, 'status': 'not_found'}
    best = min((r for r in results if r['price']), key=lambda x: x['price'], default=results[0])
    key = product_name.lower().replace(' ', '_')
    if key not in db['products']:
        db['products'][key] = {'name': product_name, 'history': [], 'target_price': None}
    entry = db['products'][key]
    entry['history'].append({
        'price': best['price'],
        'title': best['title'],
        'url': best['url'],
        'timestamp': datetime.now().isoformat(),
    })
    # Keep last 30 snapshots
    entry['history'] = entry['history'][-30:]
    entry['current_price'] = best['price']
    entry['current_url'] = best['url']
    save_db(db)
    return {
        'product': product_name,
        'current_price': best['price'],
        'history_count': len(entry['history']),
        'url': best['url'],
    }

result = track_product('Samsung 55 inch 4K TV')
print(f"Tracked: {result['product']}")
print(f"Price: ${result['current_price']:.2f}")
print(f"History: {result['history_count']} snapshots")

def check_price_drops(threshold_pct: float = 5.0) -> list:
    """Check all tracked products for price drops."""
    db = load_db()
    alerts = []
    for key, product in db['products'].items():
        history = product['history']
        if len(history) < 2:
            continue
        current = history[-1]['price']
        previous = history[-2]['price']
        if current is None or previous is None:
            continue
        # Check percentage drop
        if previous > 0:
            drop_pct = ((previous - current) / previous) * 100
            if drop_pct >= threshold_pct:
                alerts.append({
                    'product': product['name'],
                    'type': 'price_drop',
                    'previous': previous,
                    'current': current,
                    'drop_pct': drop_pct,
                    'url': history[-1]['url'],
                })
        # Check target price
        target = product.get('target_price')
        if target and current <= target:
            alerts.append({
                'product': product['name'],
                'type': 'target_hit',
                'target': target,
                'current': current,
                'url': history[-1]['url'],
            })
    return alerts

def set_target_price(product_name: str, target: float):
    db = load_db()
    key = product_name.lower().replace(' ', '_')
    if key in db['products']:
        db['products'][key]['target_price'] = target
        save_db(db)
        print(f'Target set: {product_name} at ${target:.2f}')
    else:
        print(f'Product not tracked: {product_name}')

# Set a target and check
set_target_price('Samsung 55 inch 4K TV', 399.99)
alerts = check_price_drops(threshold_pct=3.0)
for a in alerts:
    if a['type'] == 'price_drop':
        print(f'PRICE DROP: {a["product"]} ${a["previous"]:.2f} -> ${a["current"]:.2f} ({a["drop_pct"]:.1f}%)')
    else:
        print(f'TARGET HIT: {a["product"]} now ${a["current"]:.2f} (target: ${a["target"]:.2f})')

def track_all_products() -> dict:
    """Update prices for all tracked products."""
    db = load_db()
    updated = 0
    errors = 0
    for key, product in db['products'].items():
        try:
            result = track_product(product['name'])
            if result.get('current_price'):
                updated += 1
            time.sleep(0.5)
        except Exception as e:
            errors += 1
            print(f'Error tracking {product["name"]}: {e}')
    # Check for alerts
    alerts = check_price_drops()
    report = {
        'timestamp': datetime.now().isoformat(),
        'tracked': len(db['products']),
        'updated': updated,
        'errors': errors,
        'alerts': len(alerts),
        'cost': updated * 0.005,
    }
    print(f'Tracking Report ({report["timestamp"]})')
    print(f'  Products: {report["tracked"]}')
    print(f'  Updated: {report["updated"]}')
    print(f'  Alerts: {report["alerts"]}')
    print(f'  Cost: ${report["cost"]:.3f}')
    for a in alerts:
        print(f'  ALERT: {a["product"]} - {a["type"]}')
    return report

# Add products to track
for product in ['Samsung 55 inch 4K TV', 'iPad Air 2026', 'Dyson V15 vacuum']:
    track_product(product)
    time.sleep(0.5)

# Run tracking update
report = track_all_products()