'Zero to Production Scraping Pipeline: How We Built a 2.5M Company Dataset

from scrapegraph_py import Client
from scrapegraph_py.logger import sgai_logger
import asyncio
from concurrent.futures import ThreadPoolExecutor
 
sgai_logger.set_logging(level="INFO")
 
# Initialize client with enterprise configuration
sgai_client = Client(api_key="your-enterprise-key")
 
# Source discovery agent
def discover_company_data_sources(industry, region):
    discovery_prompt = f"""
    Find the top 10 most comprehensive business directory websites for {industry} 
    companies in {region}. For each source, evaluate:
    - Data completeness (1-10 scale)
    - Update frequency
    - Extraction difficulty
    - Coverage scope
    
    Return structured JSON with source URLs, quality scores, and estimated company count.
    """
    
    response = sgai_client.smartscraper(
        website_url=f"https://google.com/search?q={industry}+companies+directory+{region}",
        user_prompt=discovery_prompt
    )
    
    return response.result
 
# Automatically discover sources across all target regions
async def build_source_inventory():
    regions = ["North America", "Europe", "Asia Pacific", "Latin America", "Middle East", "Africa"]
    industries = ["Technology", "Financial Services", "Healthcare", "Manufacturing", "Retail"]
    
    source_inventory = []
    
    for region in regions:
        for industry in industries:
            sources = discover_company_data_sources(industry, region)
            source_inventory.extend(sources)
    
    # AI-powered source ranking and deduplication
    return optimize_source_selection(source_inventory)

import time
import random
from dataclasses import dataclass
from typing import List, Dict, Any
 
@dataclass
class ExtractionTask:
    company_url: str
    priority: int
    retry_count: int = 0
    estimated_duration: float = 2.0
 
class DistributedScrapeManager:
    def __init__(self, max_workers: int = 50):
        self.sgai_client = Client(api_key="your-enterprise-key")
        self.max_workers = max_workers
        self.task_queue = []
        self.completed_extractions = []
        self.failed_extractions = []
        
    async def extract_company_profile(self, task: ExtractionTask) -> Dict[str, Any]:
        """Extract comprehensive company profile with intelligent retry logic"""
        
        extraction_prompt = """
        Extract the following company information in structured JSON format:
        
        Basic Info:
        - Company name (legal and trading names)
        - Founded date
        - Headquarters location (full address)
        - Employee count (current estimate)
        - Industry classification (primary and secondary)
        
        Financial Data:
        - Annual revenue (latest year)
        - Revenue growth rate (3-year average)
        - Funding status (public/private)
        - Last funding round (if applicable)
        
        Leadership:
        - CEO name and tenure
        - Key executives (CTO, CFO, etc.)
        - Board composition (if available)
        
        Market Position:
        - Primary competitors (top 3)
        - Market share estimate
        - Geographic presence
        - Key products/services
        
        Contact & Digital:
        - Website URL
        - Primary phone number
        - LinkedIn company page
        - Key office locations
        
        For each data point, include confidence score (0-1) and last update timestamp.
        If data is not available, mark as null with explanation.
        """
        
        try:
            start_time = time.time()
            
            response = await self.sgai_client.smartscraper(
                website_url=task.company_url,
                user_prompt=extraction_prompt
            )
            
            # Add extraction metadata
            extraction_data = response.result
            extraction_data['extraction_metadata'] = {
                'source_url': task.company_url,
                'extraction_time': time.time() - start_time,
                'extraction_timestamp': time.time(),
                'retry_count': task.retry_count
            }
            
            return extraction_data
            
        except Exception as e:
            if task.retry_count < 3:
                # Intelligent retry with exponential backoff
                task.retry_count += 1
                await asyncio.sleep(2 ** task.retry_count + random.uniform(0, 1))
                return await self.extract_company_profile(task)
            else:
                self.failed_extractions.append({
                    'url': task.company_url,
                    'error': str(e),
                    'retry_count': task.retry_count
                })
                return None
    
    async def process_batch(self, batch_size: int = 1000):
        """Process extraction tasks in optimized batches"""
        
        batch_tasks = self.task_queue[:batch_size]
        self.task_queue = self.task_queue[batch_size:]
        
        # Use ThreadPoolExecutor for optimal concurrency
        with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
            loop = asyncio.get_event_loop()
            
            extraction_futures = [
                loop.run_in_executor(executor, self.extract_company_profile, task)
                for task in batch_tasks
            ]
            
            results = await asyncio.gather(*extraction_futures, return_exceptions=True)
            
            # Filter successful extractions
            successful_results = [r for r in results if r is not None and not isinstance(r, Exception)]
            self.completed_extractions.extend(successful_results)
            
            return len(successful_results)

from typing import Dict, List, Tuple
import re
from datetime import datetime, timedelta
 
class RealTimeDataValidator:
    def __init__(self):
        self.validation_rules = self._load_validation_rules()
        self.quality_thresholds = {
            'completeness': 0.85,  # 85% of fields must be populated
            'accuracy': 0.95,      # 95% accuracy on validated fields
            'consistency': 0.90    # 90% consistency across similar companies
        }
    
    def validate_company_profile(self, profile: Dict[str, Any]) -> Tuple[bool, List[str], float]:
        """Comprehensive real-time validation of extracted company data"""
        
        validation_errors = []
        quality_score = 0.0
        
        # 1. Completeness validation
        required_fields = ['company_name', 'industry', 'headquarters', 'website']
        completeness_score = self._validate_completeness(profile, required_fields)
        
        # 2. Format validation
        format_score = self._validate_formats(profile, validation_errors)
        
        # 3. Business logic validation
        logic_score = self._validate_business_logic(profile, validation_errors)
        
        # 4. Cross-reference validation (for critical data points)
        cross_ref_score = self._validate_cross_references(profile, validation_errors)
        
        # Calculate weighted quality score
        quality_score = (
            completeness_score * 0.3 +
            format_score * 0.25 +
            logic_score * 0.25 +
            cross_ref_score * 0.2
        )
        
        is_valid = (
            len(validation_errors) == 0 and 
            quality_score >= self.quality_thresholds['accuracy']
        )
        
        return is_valid, validation_errors, quality_score
    
    def _validate_completeness(self, profile: Dict[str, Any], required_fields: List[str]) -> float:
        """Validate data completeness against requirements"""
        populated_fields = sum(1 for field in required_fields if profile.get(field))
        return populated_fields / len(required_fields)
    
    def _validate_formats(self, profile: Dict[str, Any], errors: List[str]) -> float:
        """Validate data format compliance"""
        format_checks = 0
        passed_checks = 0
        
        # Website URL validation
        if profile.get('website'):
            format_checks += 1
            if re.match(r'^https?://[^\s]+\.[^\s]+', profile['website']):
                passed_checks += 1
            else:
                errors.append(f"Invalid website URL format: {profile['website']}")
        
        # Employee count validation
        if profile.get('employee_count'):
            format_checks += 1
            try:
                count = int(profile['employee_count'].replace(',', '').replace('+', ''))
                if 1 <= count <= 10000000:  # Reasonable range
                    passed_checks += 1
                else:
                    errors.append(f"Employee count out of reasonable range: {count}")
            except (ValueError, AttributeError):
                errors.append(f"Invalid employee count format: {profile['employee_count']}")
        
        # Revenue validation
        if profile.get('annual_revenue'):
            format_checks += 1
            revenue_str = str(profile['annual_revenue']).lower()
            if any(indicator in revenue_str for indicator in ['$', 'million', 'billion', 'k', 'm', 'b']):
                passed_checks += 1
            else:
                errors.append(f"Invalid revenue format: {profile['annual_revenue']}")
        
        return passed_checks / format_checks if format_checks > 0 else 1.0
    
    def _validate_business_logic(self, profile: Dict[str, Any], errors: List[str]) -> float:
        """Validate business logic consistency"""
        logic_checks = 0
        passed_checks = 0
        
        # Founded date vs company age consistency
        if profile.get('founded_date') and profile.get('employee_count'):
            logic_checks += 1
            try:
                founded_year = int(profile['founded_date'][:4])
                current_year = datetime.now().year
                company_age = current_year - founded_year
                
                employee_count = int(str(profile['employee_count']).replace(',', '').replace('+', ''))
                
                # Basic logic: very young companies shouldn't have massive employee counts
                if company_age < 5 and employee_count > 10000:
                    errors.append(f"Inconsistent: {company_age} year old company with {employee_count} employees")
                else:
                    passed_checks += 1
            except (ValueError, TypeError):
                errors.append("Could not validate founded date vs employee count consistency")
        
        return passed_checks / logic_checks if logic_checks > 0 else 1.0
    
    def _validate_cross_references(self, profile: Dict[str, Any], errors: List[str]) -> float:
        """Cross-reference critical data points with external sources"""
        # This would integrate with external APIs for validation
        # For demo purposes, we'll simulate the validation
        return 0.95  # Simulated cross-reference score
 
# Real-time quality monitoring
class QualityMonitor:
    def __init__(self):
        self.validator = RealTimeDataValidator()
        self.quality_metrics = {
            'total_processed': 0,
            'valid_records': 0,
            'quality_scores': [],
            'error_categories': {}
        }
    
    def process_extraction_result(self, extraction_result: Dict[str, Any]) -> bool:
        """Process and validate extraction result in real-time"""
        self.quality_metrics['total_processed'] += 1
        
        is_valid, errors, quality_score = self.validator.validate_company_profile(extraction_result)
        
        self.quality_metrics['quality_scores'].append(quality_score)
        
        if is_valid:
            self.quality_metrics['valid_records'] += 1
            return True
        else:
            # Categorize errors for trend analysis
            for error in errors:
                category = self._categorize_error(error)
                self.quality_metrics['error_categories'][category] = (
                    self.quality_metrics['error_categories'].get(category, 0) + 1
                )
            return False
    
    def get_real_time_metrics(self) -> Dict[str, Any]:
        """Get current quality metrics for monitoring dashboard"""
        if self.quality_metrics['total_processed'] == 0:
            return {'status': 'No data processed yet'}
        
        avg_quality = sum(self.quality_metrics['quality_scores']) / len(self.quality_metrics['quality_scores'])
        success_rate = self.quality_metrics['valid_records'] / self.quality_metrics['total_processed']
        
        return {
            'total_processed': self.quality_metrics['total_processed'],
            'success_rate': success_rate,
            'average_quality_score': avg_quality,
            'quality_trend': self.quality_metrics['quality_scores'][-100:],  # Last 100 for trending
            'top_error_categories': sorted(
                self.quality_metrics['error_categories'].items(), 
                key=lambda x: x[1], 
                reverse=True
            )[:5]
        }

class GeographicLoadBalancer:
    def __init__(self):
        self.regions = {
            'us-east-1': {'capacity': 200, 'current_load': 0, 'latency_avg': 50},
            'eu-west-1': {'capacity': 150, 'current_load': 0, 'latency_avg': 75},
            'ap-southeast-1': {'capacity': 100, 'current_load': 0, 'latency_avg': 120},
        }
    
    def select_optimal_region(self, target_url: str) -> str:
        """Select optimal region based on target geography and current load"""
        
        # Determine target region from URL
        target_region = self._determine_target_region(target_url)
        
        # Calculate efficiency scores for each region
        region_scores = {}
        for region, metrics in self.regions.items():
            load_factor = metrics['current_load'] / metrics['capacity']
            latency_penalty = metrics['latency_avg'] / 100  # Normalize latency
            
            # Prefer regions close to target, with low load and low latency
            proximity_bonus = 1.0 if region.startswith(target_region) else 0.5
            
            efficiency_score = proximity_bonus * (1 - load_factor) * (1 - latency_penalty)
            region_scores[region] = efficiency_score
        
        # Select region with highest efficiency score
        optimal_region = max(region_scores.items(), key=lambda x: x[1])[0]
        
        # Update load tracking
        self.regions[optimal_region]['current_load'] += 1
        
        return optimal_region
    
    def _determine_target_region(self, url: str) -> str:
        """Determine target region based on URL characteristics"""
        if any(tld in url for tld in ['.com', '.us', '.ca']):
            return 'us'
        elif any(tld in url for tld in ['.eu', '.uk', '.de', '.fr']):
            return 'eu'
        elif any(tld in url for tld in ['.jp', '.cn', '.sg', '.au']):
            return 'ap'
        else:
            return 'us'  # Default to US