Concurrency Control

# Système de concurrence adaptatif avec rate limiting intelligent
import asyncio
import time
from dataclasses import dataclass, field
from typing import Dict, Optional
import httpx

@dataclass
class DomainState:
    """État de rate limiting pour un domaine"""
    concurrent_requests: int = 0
    max_concurrent: int = 5
    last_request_time: float = 0
    min_delay: float = 0.5
    current_delay: float = 0.5
    consecutive_errors: int = 0
    avg_response_time: float = 0.5
    request_count: int = 0

class AdaptiveConcurrencyController:
    """Contrôleur de concurrence adaptatif par domaine"""
    
    def __init__(self):
        self.domains: Dict[str, DomainState] = {}
        self.locks: Dict[str, asyncio.Lock] = {}
    
    def _get_domain(self, url: str) -> str:
        from urllib.parse import urlparse
        return urlparse(url).netloc
    
    def _get_state(self, domain: str) -> DomainState:
        if domain not in self.domains:
            self.domains[domain] = DomainState()
            self.locks[domain] = asyncio.Lock()
        return self.domains[domain]
    
    async def acquire(self, url: str):
        """Attend et acquiert un slot de concurrence"""
        domain = self._get_domain(url)
        state = self._get_state(domain)
        
        async with self.locks[domain]:
            # Attendre si trop de requêtes en cours
            while state.concurrent_requests >= state.max_concurrent:
                await asyncio.sleep(0.1)
            
            # Respecter le délai minimum
            elapsed = time.time() - state.last_request_time
            if elapsed < state.current_delay:
                await asyncio.sleep(state.current_delay - elapsed)
            
            state.concurrent_requests += 1
            state.last_request_time = time.time()
    
    def release(self, url: str, response_time: float, success: bool):
        """Libère un slot et ajuste les paramètres"""
        domain = self._get_domain(url)
        state = self._get_state(domain)
        
        state.concurrent_requests -= 1
        state.request_count += 1
        
        # Mise à jour du temps de réponse moyen (moyenne mobile)
        alpha = 0.2
        state.avg_response_time = alpha * response_time + (1 - alpha) * state.avg_response_time
        
        if success:
            state.consecutive_errors = 0
            # Accélérer si le serveur répond vite
            if state.avg_response_time < 0.3 and state.current_delay > state.min_delay:
                state.current_delay = max(state.min_delay, state.current_delay * 0.9)
            # Augmenter la concurrence si tout va bien
            if state.request_count % 100 == 0 and state.avg_response_time < 0.5:
                state.max_concurrent = min(20, state.max_concurrent + 1)
        else:
            state.consecutive_errors += 1
            # Backoff exponentiel
            state.current_delay = min(30, state.current_delay * 2)
            # Réduire la concurrence en cas d'erreurs
            if state.consecutive_errors >= 3:
                state.max_concurrent = max(1, state.max_concurrent // 2)
    
    def get_stats(self) -> Dict:
        return {
            domain: {
                'concurrent': state.concurrent_requests,
                'max_concurrent': state.max_concurrent,
                'current_delay': round(state.current_delay, 2),
                'avg_response': round(state.avg_response_time, 3),
                'total_requests': state.request_count
            }
            for domain, state in self.domains.items()
        }

# Client HTTP avec concurrence contrôlée
class PoliteClient:
    def __init__(self):
        self.controller = AdaptiveConcurrencyController()
        self.client = httpx.AsyncClient(timeout=30)
    
    async def get(self, url: str) -> httpx.Response:
        await self.controller.acquire(url)
        
        start = time.time()
        success = False
        try:
            response = await self.client.get(url)
            success = response.status_code < 400
            return response
        finally:
            response_time = time.time() - start
            self.controller.release(url, response_time, success)
    
    async def close(self):
        await self.client.aclose()

# Usage
async def polite_scrape(urls: list):
    client = PoliteClient()
    
    async def fetch(url):
        try:
            response = await client.get(url)
            return {'url': url, 'status': response.status_code}
        except Exception as e:
            return {'url': url, 'error': str(e)}
    
    results = await asyncio.gather(*[fetch(url) for url in urls])
    
    print("Stats:", client.controller.get_stats())
    await client.close()
    return results