Crawl Budget

# Analyse du Crawl Budget via logs serveur
import re
from datetime import datetime, timedelta
from collections import defaultdict
from typing import Dict, List
from dataclasses import dataclass

@dataclass
class CrawlHit:
    timestamp: datetime
    bot: str
    url: str
    status_code: int
    response_time_ms: float

class CrawlBudgetAnalyzer:
    """Analyseur de Crawl Budget à partir des logs serveur"""
    
    BOT_PATTERNS = {
        'Googlebot': r'Googlebot(?:-Image|-News|-Video)?',
        'Bingbot': r'bingbot',
        'GPTBot': r'GPTBot',
        'ClaudeBot': r'Claude-Web|ClaudeBot',
    }
    
    def __init__(self):
        self.hits: List[CrawlHit] = []
    
    def parse_log_line(self, line: str) -> CrawlHit | None:
        """Parse une ligne de log Apache/Nginx"""
        # Format: IP - - [timestamp] "METHOD URL PROTO" STATUS SIZE "REFERER" "UA"
        pattern = r'(d+.d+.d+.d+).*[(.+?)].*"(w+) (.+?) HTTP.*" (d+) (d+|-) ".*?" "(.+?)"'
        
        match = re.match(pattern, line)
        if not match:
            return None
        
        ip, timestamp_str, method, url, status, size, user_agent = match.groups()
        
        # Détecter le bot
        bot = None
        for bot_name, pattern in self.BOT_PATTERNS.items():
            if re.search(pattern, user_agent, re.IGNORECASE):
                bot = bot_name
                break
        
        if not bot:
            return None  # Ignorer les non-bots
        
        # Parser le timestamp
        timestamp = datetime.strptime(timestamp_str.split()[0], '%d/%b/%Y:%H:%M:%S')
        
        return CrawlHit(
            timestamp=timestamp,
            bot=bot,
            url=url,
            status_code=int(status),
            response_time_ms=0  # À extraire si disponible
        )
    
    def analyze_logs(self, log_file: str):
        """Analyse un fichier de logs complet"""
        with open(log_file, 'r') as f:
            for line in f:
                hit = self.parse_log_line(line)
                if hit:
                    self.hits.append(hit)
    
    def get_crawl_stats(self, days: int = 7) -> Dict:
        """Génère les statistiques de crawl"""
        cutoff = datetime.now() - timedelta(days=days)
        recent_hits = [h for h in self.hits if h.timestamp > cutoff]
        
        stats = {
            'period_days': days,
            'total_crawls': len(recent_hits),
            'by_bot': defaultdict(int),
            'by_status': defaultdict(int),
            'crawl_rate_per_day': len(recent_hits) / days,
            'top_crawled_urls': defaultdict(int),
            'error_rate': 0
        }
        
        for hit in recent_hits:
            stats['by_bot'][hit.bot] += 1
            stats['by_status'][hit.status_code] += 1
            stats['top_crawled_urls'][hit.url] += 1
        
        # Calculer le taux d'erreur
        total = len(recent_hits)
        errors = sum(1 for h in recent_hits if h.status_code >= 400)
        stats['error_rate'] = errors / total if total > 0 else 0
        
        # Top 20 URLs les plus crawlées
        stats['top_crawled_urls'] = dict(
            sorted(stats['top_crawled_urls'].items(), 
                   key=lambda x: x[1], reverse=True)[:20]
        )
        
        return stats
    
    def detect_budget_waste(self) -> List[Dict]:
        """Détecte les gaspillages de crawl budget"""
        issues = []
        stats = self.get_crawl_stats()
        
        # Issue 1: Trop de crawl sur les mêmes URLs
        for url, count in stats['top_crawled_urls'].items():
            if count > 50:  # Seuil arbitraire
                issues.append({
                    'type': 'over_crawled',
                    'url': url,
                    'crawl_count': count,
                    'recommendation': 'Vérifier le maillage interne ou utiliser noindex'
                })
        
        # Issue 2: Taux d'erreur élevé
        if stats['error_rate'] > 0.05:
            issues.append({
                'type': 'high_error_rate',
                'error_rate': f"{stats['error_rate']*100:.1f}%",
                'recommendation': 'Corriger les erreurs 4xx/5xx pour préserver le budget'
            })
        
        # Issue 3: Paramètres URL excessifs
        param_urls = [u for u in stats['top_crawled_urls'] if '?' in u]
        if len(param_urls) > 10:
            issues.append({
                'type': 'parameter_explosion',
                'count': len(param_urls),
                'recommendation': 'Configurer les paramètres URL dans Search Console'
            })
        
        return issues

# Usage
analyzer = CrawlBudgetAnalyzer()
# analyzer.analyze_logs('/var/log/nginx/access.log')
# print(analyzer.get_crawl_stats())
# print(analyzer.detect_budget_waste())