"""Main scraper with balanced classification (lenient + strict combined).""" # src/core/balanced_scraper.py import time from typing import Dict, Optional from .config import ScraperConfig from .catalog_optimizer import CatalogOptimizer from ..utils.http_client import HTTPClient from ..extractors.link_extractor import LinkExtractor from ..extractors.product_extractor import ProductExtractor from ..extractors.configurator_detector import ConfiguratorDetector from ..extractors.external_configurator_scraper import ExternalConfiguratorScraper from ..classifiers.balanced_classifier import BalancedClassifier, StrictnessLevel from ..crawlers.web_crawler import WebCrawler from ..storage.json_storage import JSONStorage from ..storage.csv_storage import CSVStorage from ..storage.google_sheets import GoogleSheetsStorage from ..storage.quotation_template import QuotationTemplate class BalancedScraper: """ Main scraper with balanced classification. Combines lenient and strict approaches for optimal results. Configurable strictness: LENIENT, BALANCED, or STRICT. """ def __init__(self, config: ScraperConfig, strictness: str = "balanced", optimize_results: bool = True): """ Initialize the scraper. Args: config: Scraper configuration strictness: Classification strictness ("lenient", "balanced", or "strict") optimize_results: Enable post-processing optimization to remove duplicates and invalid entries """ self.config = config self.optimize_results = optimize_results # Map strictness string to enum strictness_map = { "lenient": StrictnessLevel.LENIENT, "balanced": StrictnessLevel.BALANCED, "strict": StrictnessLevel.STRICT } strictness_level = strictness_map.get(strictness.lower(), StrictnessLevel.BALANCED) # Initialize components self.http_client = HTTPClient(timeout=config.request_timeout) self.link_extractor = LinkExtractor() self.product_extractor = ProductExtractor() self.configurator_detector = ConfiguratorDetector() self.external_scraper = ExternalConfiguratorScraper( http_client=self.http_client, product_extractor=self.product_extractor ) # Initialize BALANCED classifier self.classifier = BalancedClassifier(strictness=strictness_level) print(f"✓ Using BALANCED classifier (strictness: {strictness_level.value})") print(f" Threshold: {self.classifier.PRODUCT_THRESHOLD}") print(f" Content signals required: {self.classifier.CONTENT_SIGNAL_REQUIREMENT}") # Initialize crawler self.crawler = WebCrawler( base_url=config.base_url, http_client=self.http_client, link_extractor=self.link_extractor, classifier=self.classifier, crawl_delay=config.crawl_delay ) # Initialize storage self.json_storage = JSONStorage() self.csv_storage = CSVStorage() self.quotation_template = QuotationTemplate() self.google_sheets = None # Initialized on demand # Initialize optimizer if optimize_results: try: self.optimizer = CatalogOptimizer( gemini_api_key=config.gemini_api_key, user_intent=config.user_intent ) except Exception as e: print(f"⚠️ Optimizer initialization failed: {e}") self.optimize_results = False self.optimizer = None else: self.optimizer = None def scrape_product(self, url: str) -> Optional[Dict]: """ Scrape a single product page with intelligent configurator handling. Args: url: The product page URL Returns: Product data dictionary or None """ print(f"\n{'─'*80}") print(f"Scraping product: {url}") print(f"{'─'*80}") # Scrape the page markdown = self.http_client.scrape_with_jina(url) if not markdown: return None # NOTE: Classification already done by crawler - trust the crawler's judgment # No need to re-classify here as each crawler (AI, Web, etc.) uses its own method # Detect configurator dynamically configurator_info = self.configurator_detector.has_configurator(url, markdown) # Log configurator detection print(f" 🔧 Configurator:") print(f" Detected: {configurator_info['has_configurator']}") if configurator_info['has_configurator']: print(f" Type: {configurator_info['configurator_type']}") print(f" Confidence: {configurator_info['confidence']:.0%}") if configurator_info['configurator_url']: print(f" URL: {configurator_info['configurator_url']}") # Determine if we should scrape customizations should_scrape, reason = self.configurator_detector.should_scrape_configurator(configurator_info) # Extract basic product information product_name = self.product_extractor.extract_product_name(url, markdown) base_price = self.product_extractor.extract_base_price(markdown) specifications = self.product_extractor.extract_specifications(markdown) customizations = {} scrape_source = "none" external_platform = None if should_scrape: print(f" → Extracting customizations ({reason})...") # Strategy 1: External configurator if configurator_info['configurator_type'] == 'external' and configurator_info['configurator_url']: time.sleep(10) external_result = self.external_scraper.scrape_external_configurator( url=configurator_info['configurator_url'], product_name=product_name, delay=self.config.crawl_delay ) if external_result['success']: customizations = external_result['customizations'] scrape_source = external_result['source'] external_platform = external_result.get('platform', 'unknown') print(f" ✓ Extracted from external configurator ({external_platform})") else: print(f" ✗ External configurator failed: {external_result.get('error', 'Unknown error')}") print(f" → Falling back to product page") customizations = self.product_extractor.extract_customizations(markdown) scrape_source = "product_page_fallback" # Strategy 2: Embedded configurator URL elif configurator_info['configurator_url'] and configurator_info['configurator_type'] == 'embedded': print(f" → Following embedded configurator URL...") time.sleep(self.config.crawl_delay) config_markdown = self.http_client.scrape_with_jina(configurator_info['configurator_url']) if config_markdown: customizations = self.product_extractor.extract_customizations(config_markdown) scrape_source = "embedded_configurator_page" print(f" ✓ Extracted from embedded configurator page") else: print(f" ✗ Failed to load configurator page, using current page") customizations = self.product_extractor.extract_customizations(markdown) scrape_source = "product_page_fallback" # Strategy 3: Extract from current page else: customizations = self.product_extractor.extract_customizations(markdown) scrape_source = "product_page" print(f" ✓ Extracted from product page") else: print(f" → Skipping customization extraction: {reason}") # Build product data product_data = { "product_name": product_name, "url": url, "base_price": base_price, "specifications": specifications, # Configurator metadata "has_configurator": configurator_info['has_configurator'], "configurator_type": configurator_info['configurator_type'], "configurator_url": configurator_info['configurator_url'], "configurator_confidence": configurator_info['confidence'], "configurator_signals": configurator_info['signals'], # External platform info "external_platform": external_platform, # Customization data "customization_source": scrape_source, "customization_categories": list(customizations.keys()), "customizations": customizations, "total_customization_options": sum(len(opts) for opts in customizations.values()) } # Print summary print(f"\n 📦 Product Summary:") print(f" Name: {product_name}") print(f" Price: {base_price or 'Not found'}") print(f" Specifications: {len(specifications)}") print(f" Categories: {len(customizations)}") print(f" Total Options: {product_data['total_customization_options']}") return product_data def scrape_all_products(self, product_urls: set = None) -> Dict[str, Dict]: """ Main method: scrape all products. Args: product_urls: Optional pre-discovered product URLs (from any crawler) If None, uses WebCrawler to discover URLs Returns: Complete product catalog """ print("\n" + "="*80) print("PRODUCT CATALOG SCRAPER - BALANCED MODE") print("="*80) print(f"Strictness: {self.classifier.strictness.value.upper()}") print(f"Threshold: {self.classifier.PRODUCT_THRESHOLD}") print("="*80) # Step 1: Get product URLs (either provided or crawl) if product_urls is None: product_urls = self.crawler.crawl( max_pages=self.config.max_pages, max_depth=self.config.max_depth ) else: print(f"\nUsing {len(product_urls)} pre-discovered product URLs") if not product_urls: print("\n⚠️ No product pages found!") return {} # Step 2: Scrape each product print("\n" + "="*80) print("SCRAPING PRODUCT DETAILS") print("="*80) catalog = {} for url in product_urls: product_data = self.scrape_product(url) if product_data: product_id = product_data['product_name'].lower().replace(' ', '_').replace('-', '_') catalog[product_id] = product_data # Delay between product scrapes time.sleep(self.config.crawl_delay) return catalog async def save_catalog(self, catalog: Dict, export_formats: list = ['json']): """ Save catalog to file(s) with optional optimization. Args: catalog: The catalog to save export_formats: List of formats ['json', 'csv', 'csv_prices', 'google_sheets', 'quotation'] """ # Optimize catalog if enabled if self.optimize_results and self.optimizer: print(f"\n{'='*80}") print("OPTIMIZING RESULTS") print(f"{'='*80}") catalog = await self.optimizer.optimize_catalog(catalog) import os for fmt in export_formats: if fmt == 'json': self.json_storage.save(catalog, self.config.full_output_path) elif fmt == 'csv': csv_path = self.config.full_output_path.replace('.json', '.csv') self.csv_storage.save_simple(catalog, csv_path) elif fmt == 'csv_prices': csv_path = self.config.full_output_path.replace('.json', '_with_prices.csv') self.csv_storage.save_with_prices(catalog, csv_path) elif fmt == 'quotation': quot_path = self.config.full_output_path.replace('.json', '_quotation_template.json') self.quotation_template.create(catalog, quot_path) elif fmt == 'google_sheets': if self.google_sheets is None: credentials_file = os.getenv('GOOGLE_CREDENTIALS_FILE', 'credentials.json') self.google_sheets = GoogleSheetsStorage(credentials_file) spreadsheet_id = os.getenv('GOOGLE_SPREADSHEET_ID') if self.google_sheets.service: self.google_sheets.save_catalog( catalog, spreadsheet_id=spreadsheet_id, title="Product Catalog" ) def print_summary(self, catalog: Dict): """ Print comprehensive catalog summary. Args: catalog: The catalog to summarize """ print("\n" + "="*80) print("CATALOG SUMMARY") print("="*80 + "\n") # Handle both dict format (product_name: data) and list format (products: [...]) if isinstance(catalog, dict) and 'products' in catalog: # New format from LAM products = catalog['products'] else: # Old format (dict of product_name: data) products = list(catalog.values()) if isinstance(catalog, dict) else [] total_products = len(products) if total_products == 0: print("No products found.") return total_with_configurator = sum(1 for p in products if p.get('has_configurator')) total_embedded = sum(1 for p in products if p.get('configurator_type') == 'embedded') total_external = sum(1 for p in products if p.get('configurator_type') == 'external') total_categories = sum(len(p.get('customization_categories', [])) for p in products) total_options = sum(p.get('total_customization_options', 0) for p in products) # Classification stats (may not exist in LAM format) avg_confidence = sum(p.get('classification_confidence', 0) for p in products) / total_products if total_products > 0 else 0 avg_score = sum(p.get('classification_score', 0) for p in products) / total_products if total_products > 0 else 0 print(f"📊 Overview:") print(f" Total Products: {total_products}") if avg_confidence > 0: print(f" Avg Classification Confidence: {avg_confidence:.0%}") if avg_score > 0: print(f" Avg Classification Score: {avg_score:.1f}") print(f" Products with Configurator: {total_with_configurator}") print(f" → Embedded: {total_embedded}") print(f" → External: {total_external}") print(f" Total Customization Categories: {total_categories}") print(f" Total Customization Options: {total_options}\n") print(f"{'─'*80}\n") for p in products: print(f"📦 {p.get('product_name', 'Unknown')}") print(f" URL: {p.get('url', 'N/A')}") print(f" Price: {p.get('base_price') or 'N/A'}") if p.get('classification_confidence'): print(f" Classification: {p.get('page_type', 'N/A')} ({p['classification_confidence']:.0%} confidence)") print(f" Configurator: {p.get('configurator_type', 'none')} (confidence: {p.get('configurator_confidence', 0):.0%})") if p.get('configurator_url'): print(f" Configurator URL: {p['configurator_url']}") if p.get('external_platform'): print(f" External Platform: {p['external_platform']}") if p.get('extraction_method'): print(f" Extraction Method: {p['extraction_method']}") if p.get('customization_source'): print(f" Source: {p['customization_source']}") print(f" Categories: {len(p.get('customization_categories', []))}") print(f" Options: {p.get('total_customization_options', 0)}") print()