Ethical Scraping / Responsible Crawling

# Framework de scraping éthique complet
from dataclasses import dataclass
from typing import Optional, List
from urllib.parse import urlparse, urljoin
from urllib.robotparser import RobotFileParser
import httpx
import asyncio

@dataclass
class EthicalScrapingConfig:
    """Configuration de scraping éthique"""
    bot_name: str = "MyCompanyBot"
    bot_version: str = "1.0"
    contact_url: str = "https://mycompany.com/bot"
    contact_email: str = "bot@mycompany.com"
    respect_robots_txt: bool = True
    max_requests_per_second: float = 1.0
    min_delay_between_requests: float = 1.0
    max_retries: int = 3
    collect_personal_data: bool = False

class EthicalScraper:
    """Scraper respectant les bonnes pratiques éthiques"""
    
    def __init__(self, config: EthicalScrapingConfig):
        self.config = config
        self.robots_cache: dict[str, RobotFileParser] = {}
        self.user_agent = f"{config.bot_name}/{config.bot_version} (+{config.contact_url})"
    
    async def _get_robots_parser(self, base_url: str) -> RobotFileParser:
        """Récupère et parse le robots.txt"""
        if base_url in self.robots_cache:
            return self.robots_cache[base_url]
        
        robots_url = urljoin(base_url, '/robots.txt')
        parser = RobotFileParser()
        
        try:
            async with httpx.AsyncClient() as client:
                response = await client.get(robots_url, timeout=10)
                if response.status_code == 200:
                    parser.parse(response.text.splitlines())
        except Exception:
            pass  # Pas de robots.txt = tout autorisé
        
        self.robots_cache[base_url] = parser
        return parser
    
    async def can_fetch(self, url: str) -> bool:
        """Vérifie si on peut crawler cette URL selon robots.txt"""
        if not self.config.respect_robots_txt:
            return True
        
        parsed = urlparse(url)
        base_url = f"{parsed.scheme}://{parsed.netloc}"
        
        parser = await self._get_robots_parser(base_url)
        return parser.can_fetch(self.user_agent, url)
    
    def get_crawl_delay(self, url: str) -> float:
        """Retourne le Crawl-Delay du robots.txt ou la config par défaut"""
        parsed = urlparse(url)
        base_url = f"{parsed.scheme}://{parsed.netloc}"
        
        if base_url in self.robots_cache:
            delay = self.robots_cache[base_url].crawl_delay(self.user_agent)
            if delay:
                return max(delay, self.config.min_delay_between_requests)
        
        return self.config.min_delay_between_requests
    
    def _sanitize_data(self, data: dict) -> dict:
        """Supprime les données personnelles si configuré"""
        if self.config.collect_personal_data:
            return data
        
        # Patterns de données personnelles à supprimer
        personal_patterns = ['email', 'phone', 'address', 'name', 'ssn', 'credit_card']
        
        sanitized = {}
        for key, value in data.items():
            key_lower = key.lower()
            if any(pattern in key_lower for pattern in personal_patterns):
                continue  # Skip personal data
            sanitized[key] = value
        
        return sanitized
    
    async def fetch(self, url: str) -> Optional[dict]:
        """Effectue une requête éthique"""
        
        # 1. Vérifier robots.txt
        if not await self.can_fetch(url):
            print(f"⛔ Blocked by robots.txt: {url}")
            return None
        
        # 2. Respecter le délai
        delay = self.get_crawl_delay(url)
        await asyncio.sleep(delay)
        
        # 3. Requête avec User-Agent identifiable
        headers = {
            'User-Agent': self.user_agent,
            'From': self.config.contact_email,
            'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8',
        }
        
        try:
            async with httpx.AsyncClient() as client:
                response = await client.get(url, headers=headers, timeout=30)
                
                if response.status_code == 200:
                    return {
                        'url': url,
                        'status': response.status_code,
                        'content': response.text,
                        'scraped_ethically': True
                    }
                elif response.status_code == 429:
                    print(f"⚠️ Rate limited on {url}, backing off...")
                    await asyncio.sleep(60)
                    return None
                    
        except Exception as e:
            print(f"❌ Error fetching {url}: {e}")
            return None
    
    def generate_compliance_report(self, urls_scraped: List[str]) -> dict:
        """Génère un rapport de conformité"""
        return {
            'bot_identity': self.user_agent,
            'contact': self.config.contact_email,
            'robots_txt_respected': self.config.respect_robots_txt,
            'personal_data_collected': self.config.collect_personal_data,
            'urls_scraped_count': len(urls_scraped),
            'compliance_statement': (
                "This scraping operation was conducted following ethical guidelines: "
                "robots.txt was respected, rate limiting was applied, and our bot is identifiable."
            )
        }

# Usage
config = EthicalScrapingConfig(
    bot_name="CrawlersFR-Research",
    contact_url="https://crawlers.fr/bot-info",
    contact_email="tech@crawlers.fr",
    min_delay_between_requests=2.0
)

scraper = EthicalScraper(config)