Data Collection
& Extraction

import asyncio
import aiohttp
from concurrent.futures import ThreadPoolExecutor
import pandas as pd
import time

class DataCollector:
    def __init__(self, proxy_pool):
        self.proxy_pool = proxy_pool
        self.session_pool = []
        self.collected_data = []
        
    async def create_sessions(self, pool_size=100):
        """Create a pool of aiohttp sessions with different proxies"""
        for i in range(pool_size):
            proxy = self.proxy_pool[i % len(self.proxy_pool)]
            connector = aiohttp.TCPConnector(limit=10)
            session = aiohttp.ClientSession(
                connector=connector,
                timeout=aiohttp.ClientTimeout(total=30),
                headers={'User-Agent': self.get_random_user_agent()}
            )
            session._proxy = f"http://{proxy['user']}:{proxy['pass']}@{proxy['host']}:{proxy['port']}"
            self.session_pool.append(session)
    
    async def collect_batch(self, urls_batch):
        """Collect data from a batch of URLs concurrently"""
        tasks = []
        for i, url in enumerate(urls_batch):
            session = self.session_pool[i % len(self.session_pool)]
            task = self.collect_single(session, url)
            tasks.append(task)
        
        results = await asyncio.gather(*tasks, return_exceptions=True)
        return [r for r in results if not isinstance(r, Exception)]
    
    async def collect_single(self, session, url):
        """Collect data from a single URL"""
        try:
            async with session.get(url, proxy=session._proxy) as response:
                if response.status == 200:
                    html = await response.text()
                    data = self.parse_data(html, url)
                    return data
        except Exception as e:
            print(f"Error collecting {url}: {e}")
            return None
    
    def parse_data(self, html, url):
        """Parse data from HTML - implement your parsing logic"""
        # Your parsing logic here
        return {
            'url': url,
            'title': 'Extracted Title',
            'price': 'Extracted Price',
            'timestamp': time.time()
        }
    
    async def run_collection(self, urls, batch_size=1000):
        """Run large-scale data collection"""
        await self.create_sessions()
        
        for i in range(0, len(urls), batch_size):
            batch = urls[i:i + batch_size]
            batch_results = await self.collect_batch(batch)
            self.collected_data.extend(batch_results)
            
            print(f"Collected batch {i//batch_size + 1}, Total: {len(self.collected_data)}")
            
            # Rate limiting between batches
            await asyncio.sleep(1)
        
        # Save to DataFrame
        df = pd.DataFrame(self.collected_data)
        return df

# Usage
proxy_pool = [
    {'host': 'proxy1.nyronproxies.com', 'port': 8000, 'user': 'username', 'pass': 'password'},
    {'host': 'proxy2.nyronproxies.com', 'port': 8000, 'user': 'username', 'pass': 'password'},
    # Add more proxies...
]

urls = ['https://example.com/page1', 'https://example.com/page2']  # Your target URLs
collector = DataCollector(proxy_pool)

# Run collection
async def main():
    df = await collector.run_collection(urls)
    df.to_csv('collected_data.csv', index=False)
    print(f"Collected {len(df)} records")

asyncio.run(main())

const axios = require('axios');
const cheerio = require('cheerio');
const HttpsProxyAgent = require('https-proxy-agent');
const fs = require('fs').promises;

class DataPipeline {
  constructor(proxyPool, concurrency = 50) {
    this.proxyPool = proxyPool;
    this.concurrency = concurrency;
    this.collectedData = [];
    this.activeRequests = 0;
    this.requestQueue = [];
  }

  createProxyClient(proxy) {
    const proxyUrl = `http://${proxy.user}:${proxy.pass}@${proxy.host}:${proxy.port}`;
    return axios.create({
      httpsAgent: new HttpsProxyAgent(proxyUrl),
      timeout: 30000,
      headers: {
        'User-Agent': this.getRandomUserAgent()
      }
    });
  }

  async processQueue() {
    while (this.requestQueue.length > 0 && this.activeRequests < this.concurrency) {
      const request = this.requestQueue.shift();
      this.activeRequests++;
      
      this.processRequest(request)
        .finally(() => {
          this.activeRequests--;
          this.processQueue(); // Process next in queue
        });
    }
  }

  async processRequest({ url, proxy, retries = 3 }) {
    const client = this.createProxyClient(proxy);
    
    try {
      const response = await client.get(url);
      const data = this.parseData(response.data, url);
      
      if (data) {
        this.collectedData.push(data);
        console.log(`Collected: ${url} (Total: ${this.collectedData.length})`);
      }
      
    } catch (error) {
      if (retries > 0) {
        console.log(`Retrying ${url}, attempts left: ${retries}`);
        // Add back to queue with different proxy
        const newProxy = this.proxyPool[Math.floor(Math.random() * this.proxyPool.length)];
        this.requestQueue.push({ url, proxy: newProxy, retries: retries - 1 });
      } else {
        console.error(`Failed to collect ${url}: ${error.message}`);
      }
    }
  }

  parseData(html, url) {
    try {
      const $ = cheerio.load(html);
      
      return {
        url: url,
        title: $('title').text().trim(),
        description: $('meta[name="description"]').attr('content') || '',
        price: $('.price').text().trim(),
        timestamp: new Date().toISOString()
      };
    } catch (error) {
      console.error(`Parse error for ${url}: ${error.message}`);
      return null;
    }
  }

  async collect(urls) {
    console.log(`Starting collection of ${urls.length} URLs`);
    
    // Add all URLs to queue with random proxy assignment
    urls.forEach(url => {
      const proxy = this.proxyPool[Math.floor(Math.random() * this.proxyPool.length)];
      this.requestQueue.push({ url, proxy });
    });

    // Start processing
    this.processQueue();

    // Wait for completion
    return new Promise((resolve) => {
      const checkCompletion = () => {
        if (this.requestQueue.length === 0 && this.activeRequests === 0) {
          resolve(this.collectedData);
        } else {
          setTimeout(checkCompletion, 1000);
        }
      };
      checkCompletion();
    });
  }

  async exportData(filename = 'collected_data.json') {
    await fs.writeFile(filename, JSON.stringify(this.collectedData, null, 2));
    console.log(`Data exported to ${filename}`);
  }

  getRandomUserAgent() {
    const userAgents = [
      'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
      'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36',
      'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36'
    ];
    return userAgents[Math.floor(Math.random() * userAgents.length)];
  }
}

// Usage
const proxyPool = [
  { host: 'proxy1.nyronproxies.com', port: 8000, user: 'username', pass: 'password' },
  { host: 'proxy2.nyronproxies.com', port: 8000, user: 'username', pass: 'password' },
  // Add more proxies...
];

const urls = [
  'https://example1.com/data',
  'https://example2.com/data',
  // Add your target URLs...
];

const pipeline = new DataPipeline(proxyPool, 100); // 100 concurrent requests

pipeline.collect(urls)
  .then(data => {
    console.log(`Collection completed: ${data.length} records`);
    return pipeline.exportData();
  })
  .catch(console.error);

<?php
require_once 'vendor/autoload.php';

use GuzzleHttp\Client;
use GuzzleHttp\Pool;
use GuzzleHttp\Psr7\Request;
use Symfony\Component\DomCrawler\Crawler;

class DataHarvester {
    private $proxyPool;
    private $concurrency;
    private $collectedData = [];
    private $clients = [];
    
    public function __construct($proxyPool, $concurrency = 50) {
        $this->proxyPool = $proxyPool;
        $this->concurrency = $concurrency;
        $this->initializeClients();
    }
    
    private function initializeClients() {
        foreach ($this->proxyPool as $index => $proxy) {
            $proxyUrl = "http://{$proxy['user']}:{$proxy['pass']}@{$proxy['host']}:{$proxy['port']}";
            
            $this->clients[$index] = new Client([
                'proxy' => $proxyUrl,
                'timeout' => 30,
                'headers' => [
                    'User-Agent' => $this->getRandomUserAgent()
                ],
                'verify' => false
            ]);
        }
    }
    
    public function harvest($urls) {
        $requests = $this->createRequests($urls);
        
        $pool = new Pool($this->getRandomClient(), $requests, [
            'concurrency' => $this->concurrency,
            'fulfilled' => [$this, 'onFulfilled'],
            'rejected' => [$this, 'onRejected']
        ]);
        
        $promise = $pool->promise();
        $promise->wait();
        
        return $this->collectedData;
    }
    
    private function createRequests($urls) {
        $requests = [];
        
        foreach ($urls as $url) {
            $requests[] = new Request('GET', $url);
        }
        
        return $requests;
    }
    
    public function onFulfilled($response, $index) {
        $body = $response->getBody()->getContents();
        $url = $response->getHeaderLine('X-Request-URL') ?: 'unknown';
        
        $data = $this->parseData($body, $url);
        
        if ($data) {
            $this->collectedData[] = $data;
            echo "Collected: {$url} (Total: " . count($this->collectedData) . ")\n";
        }
    }
    
    public function onRejected($reason, $index) {
        echo "Request failed: " . $reason->getMessage() . "\n";
    }
    
    private function parseData($html, $url) {
        try {
            $crawler = new Crawler($html);
            
            $title = $crawler->filter('title')->count() > 0 
                ? $crawler->filter('title')->text() 
                : '';
                
            $description = $crawler->filter('meta[name="description"]')->count() > 0
                ? $crawler->filter('meta[name="description"]')->attr('content')
                : '';
                
            $price = $crawler->filter('.price, .cost, .amount')->count() > 0
                ? $crawler->filter('.price, .cost, .amount')->first()->text()
                : '';
            
            return [
                'url' => $url,
                'title' => trim($title),
                'description' => trim($description),
                'price' => trim($price),
                'timestamp' => date('Y-m-d H:i:s'),
                'data_size' => strlen($html)
            ];
            
        } catch (Exception $e) {
            echo "Parse error for {$url}: " . $e->getMessage() . "\n";
            return null;
        }
    }
    
    private function getRandomClient() {
        $randomIndex = array_rand($this->clients);
        return $this->clients[$randomIndex];
    }
    
    private function getRandomUserAgent() {
        $userAgents = [
            'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
            'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36',
            'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36'
        ];
        
        return $userAgents[array_rand($userAgents)];
    }
    
    public function exportToCSV($filename = 'harvested_data.csv') {
        if (empty($this->collectedData)) {
            echo "No data to export\n";
            return;
        }
        
        $fp = fopen($filename, 'w');
        
        // Write headers
        fputcsv($fp, array_keys($this->collectedData[0]));
        
        // Write data
        foreach ($this->collectedData as $row) {
            fputcsv($fp, $row);
        }
        
        fclose($fp);
        echo "Data exported to {$filename}\n";
    }
    
    public function getStats() {
        return [
            'total_records' => count($this->collectedData),
            'unique_urls' => count(array_unique(array_column($this->collectedData, 'url'))),
            'average_size' => array_sum(array_column($this->collectedData, 'data_size')) / count($this->collectedData),
            'collection_time' => date('Y-m-d H:i:s')
        ];
    }
}

// Usage
$proxyPool = [
    ['host' => 'proxy1.nyronproxies.com', 'port' => 8000, 'user' => 'username', 'pass' => 'password'],
    ['host' => 'proxy2.nyronproxies.com', 'port' => 8000, 'user' => 'username', 'pass' => 'password'],
    // Add more proxies...
];

$urls = [
    'https://example1.com/products',
    'https://example2.com/catalog',
    // Add your target URLs...
];

$harvester = new DataHarvester($proxyPool, 100);

echo "Starting data harvest...\n";
$data = $harvester->harvest($urls);

$harvester->exportToCSV('enterprise_data.csv');
$stats = $harvester->getStats();

echo "Harvest completed:\n";
echo "Records collected: {$stats['total_records']}\n";
echo "Unique URLs: {$stats['unique_urls']}\n";
echo "Average page size: " . round($stats['average_size']) . " bytes\n";
?>