"""Catalog Optimizer: Post-processing layer for scraper results. Uses Gemini to optimize and clean extracted catalogs by: - Removing duplicate products/services/projects - Filtering out non-product content (questions, FAQs, generic text) - Validating that items are actual products/services/projects - Standardizing data structure Workflow: 1. Load catalog JSON from scraper 2. Send to Gemini for analysis and filtering 3. Remove duplicates and invalid entries 4. Return optimized catalog 5. Export to JSON and Google Sheets Usage: optimizer = CatalogOptimizer(gemini_api_key="...") optimized = await optimizer.optimize_catalog(catalog) optimizer.save_results(optimized, formats=['json', 'sheets']) """ import json import os from typing import Dict, List, Optional, Set import google.generativeai as genai from ..storage.json_storage import JSONStorage from ..storage.csv_storage import CSVStorage from ..storage.google_sheets import GoogleSheetsStorage class CatalogOptimizer: """Post-processing optimizer for scraper results.""" def __init__( self, gemini_api_key: Optional[str] = None, user_intent: Optional[str] = None ): """ Initialize catalog optimizer. Args: gemini_api_key: Gemini API key user_intent: Original user intent for context """ self.gemini_api_key = gemini_api_key or os.getenv('GEMINI_API_KEY') or os.getenv('GEMINI_API_KEY') self.user_intent = user_intent or "Extract product information" if not self.gemini_api_key: raise ValueError("GEMINI_API_KEY not found in environment") # Initialize Gemini genai.configure(api_key=self.gemini_api_key) self.model = genai.GenerativeModel("gemini-2.5-flash") # Initialize storage self.json_storage = JSONStorage() self.csv_storage = CSVStorage() self.google_sheets = None # Statistics self.stats = { 'input_items': 0, 'duplicates_removed': 0, 'invalid_items_removed': 0, 'output_items': 0, 'optimization_time': 0 } print(f"\033[32m[✓]\033[0m Catalog Optimizer initialized") print(f" Mode: AI-Powered Quality Control (Gemini)") print(f" Intent: {self.user_intent}") async def optimize_catalog(self, catalog: Dict) -> Dict: """ Optimize catalog by removing duplicates and invalid entries. Args: catalog: Raw catalog from scraper Returns: Optimized catalog with cleaned data """ print(f"\n{'='*80}") print("CATALOG OPTIMIZATION") print(f"{'='*80}") import time start_time = time.time() # Extract products list if isinstance(catalog, dict) and 'products' in catalog: products = catalog['products'] metadata = {k: v for k, v in catalog.items() if k != 'products'} else: # Old format: dict of product_name: data products = list(catalog.values()) if isinstance(catalog, dict) else [] metadata = {} self.stats['input_items'] = len(products) print(f"Input items: {len(products)}") if not products: print("⚠️ No items to optimize") return catalog # Step 1: Analyze with Gemini in batches print("\nPhase 1: Analyzing items with Gemini...") optimized_products = await self._analyze_and_filter(products) # Step 2: Remove duplicates print("\nPhase 2: Removing duplicates...") deduplicated = self._remove_duplicates(optimized_products) self.stats['output_items'] = len(deduplicated) self.stats['duplicates_removed'] = len(optimized_products) - len(deduplicated) self.stats['invalid_items_removed'] = len(products) - len(optimized_products) self.stats['optimization_time'] = time.time() - start_time # Build optimized catalog optimized_catalog = { **metadata, 'products': deduplicated, 'total_products': len(deduplicated), 'optimization_stats': self.stats, 'optimized': True } self._print_optimization_summary() return optimized_catalog async def _analyze_and_filter( self, products: List[Dict], batch_size: int = 10 ) -> List[Dict]: """ Analyze products with Gemini and filter invalid entries. Args: products: List of product dictionaries batch_size: Number of products per batch Returns: Filtered list of valid products """ valid_products = [] # Process in batches for i in range(0, len(products), batch_size): batch = products[i:i+batch_size] print(f" Analyzing batch {i//batch_size + 1}/{(len(products)-1)//batch_size + 1}...") try: filtered_batch = await self._filter_batch(batch) valid_products.extend(filtered_batch) except Exception as e: print(f" ⚠️ Batch failed: {e}") # On error, keep original batch (fail-safe) valid_products.extend(batch) return valid_products async def _filter_batch(self, products: List[Dict]) -> List[Dict]: """Filter a batch of products using Gemini.""" # Prepare product summaries for Gemini product_summaries = [] for idx, p in enumerate(products): summary = { 'index': idx, 'product_name': p.get('product_name', 'Unknown'), 'url': p.get('url', ''), 'base_price': p.get('base_price', 'N/A'), 'description': p.get('description', '')[:200], # First 200 chars 'page_type': p.get('page_type', 'PRODUCT'), 'specifications': list(p.get('specifications', {}).keys())[:5], # First 5 specs 'features': p.get('features', [])[:5] # First 5 features } product_summaries.append(summary) prompt = f""" You are validating extracted product/service/project data. ORIGINAL USER INTENT: {self.user_intent} EXTRACTED ITEMS: {json.dumps(product_summaries, indent=2)} TASK: For each item, determine if it is VALID or INVALID. VALID ITEMS: - Actual products for sale - Services offered - Project examples - Portfolio items - Offerings that match user intent - Items with substantive information INVALID ITEMS (REMOVE THESE): - FAQ questions or answers - Generic text snippets - Navigation elements - "Contact us" pages - Generic questions like "What is X?" - Generic informational content - Duplicate or near-duplicate entries - Items that are clearly NOT products/services/projects - Irrelevant content that doesn't match user intent DECISION RULES: 1. If item looks like a question or FAQ → INVALID 2. If item has no product name or generic name → INVALID 3. If item matches user intent with real product/service data → VALID 4. If item is duplicate of another item → mark which one to keep 5. When in doubt, check if it provides value to the user based on intent Return JSON array with validation results: [ {{ "index": 0, "status": "VALID" | "INVALID", "reason": "brief explanation", "keep": true | false }} ] CRITICAL: Return valid JSON only, no explanations outside the JSON. """ try: # Try with response_mime_type (newer API) try: response = self.model.generate_content( prompt, generation_config={ "temperature": 0.1, "response_mime_type": "application/json" } ) except (TypeError, Exception) as e: # Fallback for older API versions if 'response_mime_type' in str(e): response = self.model.generate_content( prompt, generation_config={ "temperature": 0.1 } ) else: raise # Clean response text (remove markdown code blocks if present) response_text = response.text.strip() if response_text.startswith('```'): lines = response_text.split('\n') response_text = '\n'.join(lines[1:-1]) if len(lines) > 2 else response_text validations = json.loads(response_text) # Filter products based on validations valid_products = [] for validation in validations: idx = validation['index'] if validation.get('keep', True) and validation.get('status') == 'VALID': valid_products.append(products[idx]) else: print(f" ✗ Filtered out: {products[idx].get('product_name', 'Unknown')} - {validation.get('reason', 'Invalid')}") return valid_products except json.JSONDecodeError as e: print(f" ⚠️ JSON parse error: {e}") return products # Return original on error except Exception as e: print(f" ⚠️ Validation error: {e}") return products # Return original on error def _remove_duplicates(self, products: List[Dict]) -> List[Dict]: """ Remove duplicate products based on name and URL similarity. Args: products: List of products Returns: Deduplicated list """ seen_signatures = set() deduplicated = [] for product in products: # Create signature from product name and URL name = product.get('product_name', '').lower().strip() url = product.get('url', '').lower().strip() # Normalize signature signature = f"{name}|{url}" if signature not in seen_signatures: seen_signatures.add(signature) deduplicated.append(product) else: print(f" ✗ Duplicate removed: {product.get('product_name', 'Unknown')}") return deduplicated def save_results( self, catalog: Dict, export_formats: List[str] = None, output_filename: str = "optimized_catalog" ): """ Save optimized catalog in multiple formats. Args: catalog: Optimized catalog export_formats: List of formats ('json', 'csv', 'sheets') output_filename: Base filename (without extension) """ if export_formats is None: export_formats = ['json', 'sheets'] print(f"\n{'='*80}") print("SAVING OPTIMIZED RESULTS") print(f"{'='*80}") # JSON if 'json' in export_formats: json_path = f"results/{output_filename}.json" self.json_storage.save(catalog, json_path) print(f"✓ JSON saved: {json_path}") # CSV if 'csv' in export_formats: csv_path = f"results/{output_filename}.csv" self.csv_storage.save(catalog, csv_path) print(f"✓ CSV saved: {csv_path}") # Google Sheets if 'sheets' in export_formats: try: if self.google_sheets is None: self.google_sheets = GoogleSheetsStorage() sheet_url = self.google_sheets.save(catalog, f"{output_filename}") print(f"✓ Google Sheets saved: {sheet_url}") except Exception as e: print(f"⚠️ Google Sheets export failed: {e}") def _print_optimization_summary(self): """Print optimization statistics.""" print(f"\n{'='*80}") print("OPTIMIZATION COMPLETE") print(f"{'='*80}") print(f"📊 Statistics:") print(f" Input items: {self.stats['input_items']}") print(f" Invalid items removed: {self.stats['invalid_items_removed']}") print(f" Duplicates removed: {self.stats['duplicates_removed']}") print(f" Output items: {self.stats['output_items']}") print(f" Optimization time: {self.stats['optimization_time']:.2f}s") if self.stats['input_items'] > 0: retention_rate = (self.stats['output_items'] / self.stats['input_items']) * 100 print(f" Retention rate: {retention_rate:.1f}%") print(f"{'='*80}\n") def load_and_optimize( self, json_path: str, export_formats: List[str] = None ) -> Dict: """ Load catalog from JSON file, optimize, and save. Args: json_path: Path to input JSON catalog export_formats: Export formats for optimized results Returns: Optimized catalog """ print(f"\n{'='*80}") print(f"LOADING CATALOG FROM FILE") print(f"{'='*80}") print(f"Source: {json_path}") # Load catalog with open(json_path, 'r', encoding='utf-8') as f: catalog = json.load(f) # Optimize import asyncio optimized = asyncio.run(self.optimize_catalog(catalog)) # Save base_name = os.path.splitext(os.path.basename(json_path))[0] output_name = f"{base_name}_optimized" self.save_results(optimized, export_formats, output_name) return optimized # Standalone CLI for optimizing existing catalogs async def main(): """CLI entry point for catalog optimization.""" import argparse parser = argparse.ArgumentParser(description="Optimize scraper results") parser.add_argument('input', help='Input JSON catalog file') parser.add_argument('--formats', nargs='+', default=['json', 'sheets'], help='Export formats (json, csv, sheets)') parser.add_argument('--intent', help='Original user intent for context') args = parser.parse_args() # Initialize optimizer optimizer = CatalogOptimizer(user_intent=args.intent) # Load and optimize optimized = optimizer.load_and_optimize( args.input, export_formats=args.formats ) print("\n✓ Optimization complete!") return optimized if __name__ == "__main__": import asyncio asyncio.run(main())