Spaces:
Sleeping
Sleeping
| """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() |