#!/usr/bin/env python3 """ Script de monitoring pour le scraper agenda du libre Affiche les statistiques et l'état du scraper """ import json import os import sys from datetime import datetime, timedelta from typing import Dict, Any class AgendaDuLibreMonitor: def __init__(self, data_file: str = "agendadulibre_events.json"): self.data_file = data_file self.events_data = self.load_events_data() def load_events_data(self) -> Dict[str, Any]: """Charge les données d'événements""" if os.path.exists(self.data_file): try: with open(self.data_file, 'r', encoding='utf-8') as f: return json.load(f) except Exception as e: print(f"❌ Erreur lors du chargement de {self.data_file}: {e}") return {"events": {}, "last_update": None} return {"events": {}, "last_update": None} def get_statistics(self) -> Dict[str, Any]: """Calcule les statistiques des événements""" events = self.events_data.get("events", {}) stats = { "total_events": len(events), "saved": 0, "already_exists": 0, "error": 0, "unknown": 0, "recent_errors": 0, "last_update": self.events_data.get("last_update"), "events_by_status": {}, "recent_events": [] } # Analyser les statuts for event_id, event_data in events.items(): status = event_data.get("status", "unknown") stats["events_by_status"][status] = stats["events_by_status"].get(status, 0) + 1 if status == "saved": stats["saved"] += 1 elif status == "already_exists": stats["already_exists"] += 1 elif status == "error": stats["error"] += 1 else: stats["unknown"] += 1 # Vérifier les erreurs récentes (dernières 24h) last_attempt = event_data.get("last_attempt") if last_attempt and status == "error": try: attempt_time = datetime.fromisoformat(last_attempt.replace('Z', '+00:00')) if datetime.now() - attempt_time.replace(tzinfo=None) < timedelta(hours=24): stats["recent_errors"] += 1 except: pass # Collecter les événements récents (derniers 10) if len(stats["recent_events"]) < 10: event_info = { "id": event_id, "label": event_data.get("event", {}).get("properties", {}).get("label", "Sans titre"), "status": status, "last_attempt": last_attempt, "message": event_data.get("message", "") } stats["recent_events"].append(event_info) return stats def display_statistics(self): """Affiche les statistiques de manière formatée""" stats = self.get_statistics() print("📊 Statistiques du scraper agenda du libre") print("=" * 50) # Informations générales print(f"📁 Fichier de données: {self.data_file}") print(f"📅 Dernière mise à jour: {stats['last_update'] or 'Jamais'}") print(f"📈 Total d'événements traités: {stats['total_events']}") print() # Répartition par statut print("📋 Répartition par statut:") for status, count in stats["events_by_status"].items(): emoji = { "saved": "✅", "already_exists": "⚠️", "error": "❌", "unknown": "❓" }.get(status, "❓") print(f" {emoji} {status}: {count}") print() # Erreurs récentes if stats["recent_errors"] > 0: print(f"🚨 Erreurs récentes (24h): {stats['recent_errors']}") print() # Événements récents if stats["recent_events"]: print("🕒 Événements récents:") for event in stats["recent_events"][:5]: emoji = { "saved": "✅", "already_exists": "⚠️", "error": "❌", "unknown": "❓" }.get(event["status"], "❓") print(f" {emoji} {event['label'][:50]}{'...' if len(event['label']) > 50 else ''}") if event["status"] == "error": print(f" 💬 {event['message']}") print() # Recommandations self.display_recommendations(stats) def display_recommendations(self, stats: Dict[str, Any]): """Affiche des recommandations basées sur les statistiques""" print("💡 Recommandations:") if stats["total_events"] == 0: print(" - Aucun événement traité. Exécutez le scraper pour commencer.") elif stats["error"] > stats["saved"]: print(" - Beaucoup d'erreurs détectées. Vérifiez la connectivité API.") elif stats["recent_errors"] > 5: print(" - Erreurs récentes nombreuses. Vérifiez les logs.") elif stats["saved"] > 0: print(" - Scraper fonctionne correctement.") if stats["already_exists"] > stats["saved"]: print(" - Beaucoup d'événements déjà existants. Le système de déduplication fonctionne.") print() def check_file_status(self): """Vérifie l'état du fichier de données""" if not os.path.exists(self.data_file): print(f"❌ Fichier de données non trouvé: {self.data_file}") return False try: stat = os.stat(self.data_file) size = stat.st_size mtime = datetime.fromtimestamp(stat.st_mtime) print(f"📁 État du fichier de données:") print(f" - Taille: {size:,} octets") print(f" - Dernière modification: {mtime}") print(f" - Lisible: {'✅' if os.access(self.data_file, os.R_OK) else '❌'}") print(f" - Écriture: {'✅' if os.access(self.data_file, os.W_OK) else '❌'}") print() return True except Exception as e: print(f"❌ Erreur lors de la vérification du fichier: {e}") return False def show_help(self): """Affiche l'aide""" print("🔍 Monitor agenda du libre - Aide") print("=" * 40) print("Usage: python3 monitor_agendadulibre.py [options]") print() print("Options:") print(" --stats, -s Afficher les statistiques (défaut)") print(" --file, -f Vérifier l'état du fichier") print(" --help, -h Afficher cette aide") print() print("Exemples:") print(" python3 monitor_agendadulibre.py") print(" python3 monitor_agendadulibre.py --file") print(" python3 monitor_agendadulibre.py --stats") def main(): """Fonction principale""" import argparse parser = argparse.ArgumentParser(description="Monitor pour le scraper agenda du libre") parser.add_argument("--stats", "-s", action="store_true", default=True, help="Afficher les statistiques") parser.add_argument("--file", "-f", action="store_true", help="Vérifier l'état du fichier de données") parser.add_argument("--data-file", default="agendadulibre_events.json", help="Fichier de données à analyser") args = parser.parse_args() monitor = AgendaDuLibreMonitor(args.data_file) if args.file: monitor.check_file_status() if args.stats: monitor.display_statistics() if __name__ == "__main__": main()