Self-Healing Scrapers

# Système de Self-Healing Scraper avec LLM
from dataclasses import dataclass, field
from typing import Dict, List, Optional, Callable
from datetime import datetime
import json

@dataclass
class SelectorConfig:
    """Configuration d'un sélecteur avec historique"""
    field_name: str
    current_selector: str
    selector_history: List[str] = field(default_factory=list)
    last_success: Optional[datetime] = None
    failure_count: int = 0
    
class SelfHealingScraper:
    """Scraper avec capacités d'auto-réparation"""
    
    def __init__(self, base_url: str, selectors: Dict[str, str]):
        self.base_url = base_url
        self.selectors = {
            name: SelectorConfig(field_name=name, current_selector=sel)
            for name, sel in selectors.items()
        }
        self.failure_threshold = 3
        # self.llm_client = OpenAI()
    
    def extract(self, html: str) -> Dict[str, any]:
        """Extraction avec détection de pannes"""
        from selectolax.parser import HTMLParser
        tree = HTMLParser(html)
        
        results = {}
        failed_fields = []
        
        for field_name, config in self.selectors.items():
            value = self._try_extract(tree, config)
            
            if value:
                config.last_success = datetime.utcnow()
                config.failure_count = 0
                results[field_name] = value
            else:
                config.failure_count += 1
                failed_fields.append(field_name)
                
                if config.failure_count >= self.failure_threshold:
                    # Déclencher l'auto-réparation
                    new_selector = self._heal_selector(html, field_name, config)
                    if new_selector:
                        config.selector_history.append(config.current_selector)
                        config.current_selector = new_selector
                        # Réessayer avec le nouveau sélecteur
                        value = self._try_extract(tree, config)
                        if value:
                            results[field_name] = value
        
        return results
    
    def _try_extract(self, tree, config: SelectorConfig) -> Optional[str]:
        """Tente d'extraire une valeur avec le sélecteur actuel"""
        try:
            node = tree.css_first(config.current_selector)
            if node:
                text = node.text(strip=True)
                return text if text else None
        except Exception:
            pass
        return None
    
    def _heal_selector(self, html: str, field_name: str, config: SelectorConfig) -> Optional[str]:
        """Utilise un LLM pour trouver le nouveau sélecteur"""
        
        # Simplifier le HTML
        from selectolax.parser import HTMLParser
        tree = HTMLParser(html)
        for tag in ['script', 'style']:
            for node in tree.css(tag):
                node.decompose()
        
        simplified_html = tree.html[:15000]  # Limite tokens
        
        prompt = f'''Le sélecteur CSS "{config.current_selector}" ne fonctionne plus pour extraire "{field_name}".

Historique des sélecteurs qui fonctionnaient avant :
{json.dumps(config.selector_history[-3:], indent=2)}

HTML actuel de la page :
{simplified_html}

Analyse le HTML et trouve le nouveau sélecteur CSS qui permet d'extraire "{field_name}".
Retourne UNIQUEMENT le sélecteur CSS, rien d'autre.'''

        # En production:
        # response = self.llm_client.chat.completions.create(
        #     model="gpt-4o-mini",
        #     messages=[{"role": "user", "content": prompt}],
        #     temperature=0
        # )
        # new_selector = response.choices[0].message.content.strip()
        
        # Placeholder - simulation de réponse LLM
        new_selector = f"[data-{field_name}], .{field_name}-new"
        
        # Valider que le nouveau sélecteur fonctionne
        if tree.css_first(new_selector):
            print(f"🔧 Healed selector for '{field_name}': {new_selector}")
            return new_selector
        
        return None
    
    def get_health_report(self) -> Dict:
        """Rapport de santé des sélecteurs"""
        return {
            field: {
                'selector': config.current_selector,
                'failures': config.failure_count,
                'last_success': config.last_success.isoformat() if config.last_success else None,
                'history_depth': len(config.selector_history)
            }
            for field, config in self.selectors.items()
        }

# Usage
scraper = SelfHealingScraper(
    base_url="https://shop.example.com",
    selectors={
        'product_name': 'h1.product-title',
        'price': '.price-box .regular-price',
        'availability': '.stock-status'
    }
)

# Le scraper s'auto-répare si les sélecteurs cassent
# results = scraper.extract(html_content)
# print(scraper.get_health_report())