tilbudgivern/test_bygma_scraper_multi.py

#!/usr/bin/env python3
"""
Test Bygma Scraper with Multiple Products from Different Categories
Selects 1 product from each major category and attempts to scrape installation manuals
"""

import asyncio
import sys
import subprocess
import json

# Get products from different categories
def get_test_products():
    # Get one product from each of the top 5 product groups by count
    query = """
    SELECT p.varegrp, p.vareNr, p.tekst 
    FROM bygma_products p
    WHERE p.is_active = 1 
      AND p.tekst IS NOT NULL 
      AND p.tekst != ''
    GROUP BY p.varegrp
    LIMIT 5
    """
    
    cmd = f'sudo mysql tilbudgivern -N -e "{query}"'
    result = subprocess.run(cmd, shell=True, capture_output=True, text=True)
    
    products = []
    for line in result.stdout.strip().split('\n'):
        if line and not line.startswith('varegrp'):
            parts = line.split('\t')
            if len(parts) >= 3:
                products.append({
                    'varegrp': parts[0],
                    'vareNr': parts[1],
                    'tekst': parts[2]
                })
    
    return products

async def search_bygma_for_product(varenr):
    """
    Search Bygma.dk for a product by vareNr and return the product URL
    This is a simplified version - in production you'd want to use Playwright to search
    """
    from playwright.async_api import async_playwright
    
    async with async_playwright() as p:
        browser = await p.chromium.launch(headless=True)
        context = await browser.new_context(viewport={'width': 1920, 'height': 1080})
        page = await context.new_page()
        
        try:
            # Go to Bygma search
            search_url = f"https://www.bygma.dk/proff/search?query={varenr}"
            await page.goto(search_url, wait_until='networkidle', timeout=30000)
            await page.wait_for_timeout(3000)
            
            # Find first product link
            product_links = await page.query_selector_all('a[href*="/proff/"][href*="?selectedM3Number="]')
            
            if product_links:
                href = await product_links[0].get_attribute('href')
                if href:
                    if not href.startswith('http'):
                        href = 'https://www.bygma.dk' + href
                    await browser.close()
                    return href
            
            await browser.close()
            return None
            
        except Exception as e:
            await browser.close()
            return None

async def main():
    print("🔍 Bygma Scraper Multi-Category Test")
    print("="*80)
    
    # Get test products
    print("\n1️⃣ Getting test products from database...")
    products = get_test_products()
    
    if not products:
        print("❌ No products found!")
        return
    
    print(f"✅ Found {len(products)} products:\n")
    for p in products:
        print(f"   📦 {p['vareNr']} - {p['tekst'][:60]}... (Group: {p['varegrp']})")
    
    # For each product, search and scrape
    results = []
    
    for i, product in enumerate(products, 1):
        print(f"\n{'='*80}")
        print(f"[{i}/{len(products)}] Processing: {product['tekst'][:50]}...")
        print(f"VareNr: {product['vareNr']}")
        
        # Search for product URL
        print("   🔍 Searching Bygma for product URL...")
        product_url = await search_bygma_for_product(product['vareNr'])
        
        if not product_url:
            print(f"   ❌ Could not find product URL on Bygma")
            results.append({
                'product': product,
                'success': False,
                'error': 'Product URL not found'
            })
            continue
        
        print(f"   ✅ Found URL: {product_url[:80]}...")
        
        # Try to scrape installation manual
        print("   📥 Attempting to scrape installation manual...")
        
        try:
            scrape_cmd = ['python3', 'scrape_bygma_api.py', product_url]
            scrape_result = subprocess.run(
                scrape_cmd,
                cwd='/mnt/HC_Volume_103713257/tilbudgivern',
                capture_output=True,
                text=True,
                timeout=120  # 2 minutes max per product
            )
            
            if scrape_result.returncode == 0:
                try:
                    result_json = json.loads(scrape_result.stdout)
                    if result_json.get('success'):
                        manual = result_json.get('manual', {})
                        print(f"   ✅ SUCCESS! Found manual:")
                        print(f"      Product: {manual.get('product_name', 'N/A')}")
                        print(f"      Steps: {len(manual.get('installation_steps', []))}")
                        print(f"      Time: {manual.get('time_estimate', 'N/A')}")
                        results.append({
                            'product': product,
                            'success': True,
                            'url': product_url,
                            'manual': manual
                        })
                    else:
                        error = result_json.get('error', 'Unknown error')
                        print(f"   ⚠️  Scraper returned error: {error}")
                        results.append({
                            'product': product,
                            'success': False,
                            'url': product_url,
                            'error': error
                        })
                except json.JSONDecodeError:
                    print(f"   ❌ Could not parse scraper output")
                    results.append({
                        'product': product,
                        'success': False,
                        'url': product_url,
                        'error': 'JSON parse error'
                    })
            else:
                print(f"   ❌ Scraper failed with exit code {scrape_result.returncode}")
                print(f"      Error: {scrape_result.stderr[:200]}")
                results.append({
                    'product': product,
                    'success': False,
                    'url': product_url,
                    'error': scrape_result.stderr[:200]
                })
                
        except subprocess.TimeoutExpired:
            print(f"   ⏱️  Timeout after 2 minutes")
            results.append({
                'product': product,
                'success': False,
                'url': product_url,
                'error': 'Timeout'
            })
        except Exception as e:
            print(f"   ❌ Exception: {str(e)}")
            results.append({
                'product': product,
                'success': False,
                'url': product_url,
                'error': str(e)
            })
    
    # Summary
    print(f"\n{'='*80}")
    print("📊 SUMMARY")
    print("="*80)
    
    successful = sum(1 for r in results if r['success'])
    print(f"Total products tested: {len(results)}")
    print(f"Successful scrapes: {successful}")
    print(f"Failed scrapes: {len(results) - successful}")
    
    if successful > 0:
        print("\n✅ Successfully scraped manuals:")
        for r in results:
            if r['success']:
                print(f"   - {r['product']['vareNr']}: {r['manual'].get('product_name', 'N/A')}")
    
    if len(results) - successful > 0:
        print("\n❌ Failed scrapes:")
        for r in results:
            if not r['success']:
                print(f"   - {r['product']['vareNr']}: {r.get('error', 'Unknown error')[:80]}")
    
    print("\n" + "="*80)

if __name__ == "__main__":
    asyncio.run(main())