""" FFA Calendar Scraper Un système de scraping pour récupérer les données du site de la Fédération Française d'Athlétisme (FFA) et les organiser dans des fichiers CSV. """ import requests from bs4 import BeautifulSoup import pandas as pd import time import json import re from urllib.parse import urljoin, urlparse, urlencode, parse_qs from selenium import webdriver from selenium.webdriver.chrome.service import Service from selenium.webdriver.common.by import By from selenium.webdriver.support.ui import WebDriverWait from selenium.webdriver.support import expected_conditions as EC from webdriver_manager.chrome import ChromeDriverManager import os from tqdm import tqdm import logging from dotenv import load_dotenv from concurrent.futures import ThreadPoolExecutor, as_completed from threading import Lock import threading # Charger les variables d'environnement load_dotenv() # Configuration du logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler('ffa_scraper.log'), logging.StreamHandler() ] ) class FFAScraper: """Classe principale pour scraper les données de la FFA""" def __init__(self, output_dir=None): """Initialiser le scraper avec répertoire de sortie""" # Charger les URLs depuis config.env ou utiliser les valeurs par défaut self.BASE_URL = os.getenv('BASE_URL', 'https://athle.fr') self.CLUBS_URL = os.getenv('CLUBS_URL', 'https://monclub.athle.fr') self.CALENDAR_URL = os.getenv('CALENDAR_URL', "https://www.athle.fr/bases/liste.aspx?frmpostback=true&frmbase=calendrier&frmmode=1&frmespace=0&frmsaisonffa=2026&frmdate1=2025-09-01&frmdate2=2026-08-31&frmtype1=&frmniveau=&frmligue=&frmdepartement=&frmniveaulab=&frmepreuve=&frmtype2=&frmtype3=&frmtype4=&frmposition=4") self.RESULTS_URL = os.getenv('RESULTS_URL', 'https://athle.fr/les-resultats') # Configuration du scraping self.output_dir = output_dir or os.getenv('OUTPUT_DIR', 'data') self.request_delay = float(os.getenv('REQUEST_DELAY', '1.5')) self.max_retries = int(os.getenv('MAX_RETRIES', '3')) self.timeout = int(os.getenv('TIMEOUT', '30')) self.max_workers = int(os.getenv('MAX_WORKERS', '4')) # Initialiser la session HTTP self.session = requests.Session() self.session.headers.update({ 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36' }) # Créer les répertoires nécessaires os.makedirs(self.output_dir, exist_ok=True) os.makedirs(f"{self.output_dir}/clubs", exist_ok=True) os.makedirs(f"{self.output_dir}/courses", exist_ok=True) os.makedirs(f"{self.output_dir}/resultats", exist_ok=True) # Thread-local storage pour Selenium drivers (pour le multithreading) self._thread_local = threading.local() self._data_lock = Lock() # Pour protéger l'accès aux données partagées self.driver = None # Pour compatibilité avec le code existant self._current_calendar_url = self.CALENDAR_URL # URL actuelle à utiliser (modifiable) def _init_selenium(self): """Initialiser Selenium pour les pages dynamiques""" if self.driver is None: try: # Options pour le mode headless si configuré options = webdriver.ChromeOptions() if os.getenv('HEADLESS', 'True').lower() == 'true': options.add_argument('--headless') options.add_argument('--disable-gpu') window_size = os.getenv('WINDOW_SIZE', '1920,1080') if window_size: width, height = window_size.split(',') options.add_argument(f'--window-size={width},{height}') service = Service(ChromeDriverManager().install()) self.driver = webdriver.Chrome(service=service, options=options) logging.info("Driver Chrome initialisé avec succès") except Exception as e: logging.error(f"Impossible d'initialiser le driver Chrome: {e}") raise def _get_thread_selenium(self): """Récupérer ou créer un driver Selenium pour le thread actuel""" if not hasattr(self._thread_local, 'driver') or self._thread_local.driver is None: try: options = webdriver.ChromeOptions() if os.getenv('HEADLESS', 'True').lower() == 'true': options.add_argument('--headless') options.add_argument('--disable-gpu') options.add_argument('--no-sandbox') options.add_argument('--disable-dev-shm-usage') service = Service(ChromeDriverManager().install()) self._thread_local.driver = webdriver.Chrome(service=service, options=options) except Exception as e: logging.error(f"Impossible d'initialiser le driver Chrome pour le thread: {e}") raise return self._thread_local.driver def _close_thread_selenium(self): """Fermer le driver Selenium du thread actuel""" if hasattr(self._thread_local, 'driver') and self._thread_local.driver: self._thread_local.driver.quit() self._thread_local.driver = None def _close_selenium(self): """Fermer Selenium""" if self.driver: self.driver.quit() self.driver = None def _close_all_selenium(self): """Fermer tous les drivers Selenium (pour le multithreading)""" if hasattr(self, '_thread_local') and hasattr(self._thread_local, 'driver') and self._thread_local.driver: self._thread_local.driver.quit() self._thread_local.driver = None def get_page(self, url, use_selenium=False): """Récupérer le contenu d'une page""" if use_selenium: self._init_selenium() self.driver.get(url) # Attendre que la page charge time.sleep(self.request_delay) return self.driver.page_source else: for attempt in range(self.max_retries): try: response = self.session.get(url, timeout=self.timeout) response.raise_for_status() return response.text except requests.RequestException as e: if attempt < self.max_retries - 1: logging.warning(f"Tentative {attempt + 1}/{self.max_retries} échouée pour {url}: {e}") time.sleep(self.request_delay) else: logging.error(f"Erreur lors de l'accès à {url}: {e}") return None def save_to_csv(self, data, filename, index=False): """Sauvegarder les données en CSV""" if not data: logging.warning(f"Pas de données à sauvegarder dans {filename}") return None filepath = os.path.join(self.output_dir, filename) df = pd.DataFrame(data) encoding = os.getenv('CSV_ENCODING', 'utf-8-sig') df.to_csv(filepath, index=index, encoding=encoding) logging.info(f"Données sauvegardées dans {filepath} ({len(data)} enregistrements)") return filepath def get_clubs_list(self): """Récupérer la liste des clubs (NON FONCTIONNEL - nécessite authentification)""" logging.warning("La récupération des clubs n'est pas disponible - nécessite une authentification sur monclub.athle.fr") return [] def _get_page_url(self, page_num): """Générer l'URL pour une page spécifique""" calendar_url = self._current_calendar_url or self.CALENDAR_URL if page_num == 1: return calendar_url parsed_url = urlparse(calendar_url) query_params = parse_qs(parsed_url.query) query_params['page'] = [str(page_num)] new_query = urlencode(query_params, doseq=True) return f"{parsed_url.scheme}://{parsed_url.netloc}{parsed_url.path}?{new_query}" def _detect_pagination_info(self): """Détecter le nombre total de pages et de courses depuis la première page""" logging.info("Détection des informations de pagination...") calendar_url = self._current_calendar_url or self.CALENDAR_URL html = self.get_page(calendar_url, use_selenium=True) if not html: return None, None, None soup = BeautifulSoup(html, 'html.parser') # Chercher les informations de pagination total_courses = None total_pages = None # Chercher un texte contenant le nombre total de résultats for element in soup.find_all(text=re.compile(r'\d+\s*résultats|total|compétitions', re.IGNORECASE)): match = re.search(r'(\d+(?:\s*\d+)*)\s*(?:résultats|compétitions)', str(element), re.IGNORECASE) if match: total_courses = int(match.group(1).replace(' ', '')) logging.info(f"Total des courses détecté: {total_courses}") break # Chercher les liens de pagination pour obtenir le nombre de pages pagination_select = soup.find('select', class_=re.compile(r'page|pagination', re.IGNORECASE)) if pagination_select: options = pagination_select.find_all('option') if options: total_pages = int(options[-1].get_text(strip=True)) logging.info(f"Total des pages détecté: {total_pages}") # Sinon, chercher les liens de pagination if not total_pages: pagination_links = soup.find_all('a', href=re.compile(r'page', re.IGNORECASE)) page_numbers = [] for link in pagination_links: text = link.get_text(strip=True) if text.isdigit(): page_numbers.append(int(text)) if page_numbers: total_pages = max(page_numbers) # Compter les courses sur la première page pour estimer main_table = soup.find('table', class_='reveal-table') if main_table: rows = main_table.find_all('tr') first_page_count = sum(1 for row in rows if len(row.find_all('td')) >= 5 and not row.find('table', class_='detail-inner-table')) if not total_courses and total_pages: total_courses = first_page_count * total_pages if not total_pages and total_courses and first_page_count: total_pages = (total_courses + first_page_count - 1) // first_page_count return total_pages, total_courses, first_page_count if 'first_page_count' in locals() else 250 def _scrape_page(self, page_num): """Scraper une page spécifique (méthode thread-safe)""" try: page_url = self._get_page_url(page_num) driver = self._get_thread_selenium() driver.get(page_url) time.sleep(self.request_delay) html = driver.page_source if not html: logging.warning(f"Impossible d'accéder à la page {page_num}") return [] soup = BeautifulSoup(html, 'html.parser') main_table = soup.find('table', class_='reveal-table') if not main_table: logging.warning(f"Table principale non trouvée (page {page_num})") return [] rows = main_table.find_all('tr') page_courses = [] for row in rows: cells = row.find_all('td') if len(cells) < 5 or row.find('table', class_='detail-inner-table'): continue try: date = cells[0].get_text(strip=True) libelle_cell = cells[1] link_element = libelle_cell.find('a', href=True) if link_element: nom = link_element.get_text(strip=True) course_url = link_element['href'] if not course_url.startswith('http'): course_url = urljoin(self.BASE_URL, course_url) else: nom = libelle_cell.get_text(strip=True) course_url = '' lieu = cells[2].get_text(strip=True) type_competition = cells[3].get_text(strip=True) niveau = cells[4].get_text(strip=True) detail_url = '' result_url = '' for cell in cells: links = cell.find_all('a', href=True) for link in links: href = link.get('href', '') if '/competitions/' in href: detail_url = urljoin(self.BASE_URL, href) if not href.startswith('http') else href elif 'frmbase=resultats' in href: result_url = urljoin(self.BASE_URL, href) if not href.startswith('http') else href course = { 'nom': nom, 'date': date, 'lieu': lieu, 'discipline': '', 'type': type_competition, 'niveau': niveau, 'label': '', 'lien': course_url, 'fiche_detail': detail_url, 'resultats_url': result_url, 'page': page_num } page_courses.append(course) except Exception as e: logging.warning(f"Erreur lors du parsing d'une course (page {page_num}): {e}") continue logging.info(f"Page {page_num}: {len(page_courses)} courses trouvées") return page_courses except Exception as e: logging.error(f"Erreur lors du scraping de la page {page_num}: {e}") return [] def get_courses_list(self, start_date=None, end_date=None, max_pages=1, use_multithreading=True, calendar_url=None): """Récupérer la liste des courses/calendrier avec support du multithreading""" # Utiliser l'URL personnalisée si fournie if calendar_url: self._current_calendar_url = calendar_url # Détecter d'abord le nombre total de pages et de courses total_pages, total_courses, courses_per_page = self._detect_pagination_info() if total_pages: logging.info(f"Pagination détectée: {total_pages} pages, ~{total_courses} courses totales") if max_pages is None or max_pages > total_pages: max_pages = total_pages logging.info(f"Récupération du calendrier des compétitions (max {max_pages} pages, {'multithreading' if use_multithreading else 'séquentiel'})...") all_courses = [] if use_multithreading and max_pages > 1: # Utiliser le multithreading logging.info(f"Démarrage du multithreading avec {self.max_workers} workers...") with ThreadPoolExecutor(max_workers=self.max_workers, thread_name_prefix='scraper') as executor: # Soumettre toutes les tâches future_to_page = {executor.submit(self._scrape_page, page_num): page_num for page_num in range(1, max_pages + 1)} # Récupérer les résultats avec une barre de progression with tqdm(total=max_pages, desc="Pages scrapées", unit="page") as pbar: for future in as_completed(future_to_page): page_num = future_to_page[future] try: page_courses = future.result() with self._data_lock: all_courses.extend(page_courses) pbar.update(1) pbar.set_postfix({'total': len(all_courses)}) except Exception as e: logging.error(f"Erreur sur la page {page_num}: {e}") pbar.update(1) else: # Mode séquentiel for page_num in range(1, max_pages + 1): logging.info(f"Page {page_num}/{max_pages}") page_courses = self._scrape_page(page_num) all_courses.extend(page_courses) if page_num < max_pages: time.sleep(self.request_delay) logging.info(f"Total: {len(all_courses)} courses trouvées") return all_courses def get_course_details(self, course_url): """Récupérer les détails d'une course spécifique""" full_url = urljoin(self.BASE_URL, course_url) html = self.get_page(full_url) if not html: return None soup = BeautifulSoup(html, 'html.parser') try: details = { 'nom': soup.find('h1').get_text(strip=True) if soup.find('h1') else '', 'date': soup.find('span', class_='date').get_text(strip=True) if soup.find('span', class_='date') else '', 'lieu': soup.find('span', class_='location').get_text(strip=True) if soup.find('span', class_='location') else '', 'organisateur': soup.find('span', class_='organizer').get_text(strip=True) if soup.find('span', class_='organizer') else '', 'description': soup.find('div', class_='description').get_text(strip=True) if soup.find('div', class_='description') else '', 'epreuves': [] } # Récupérer les épreuves epreuve_elements = soup.find_all('div', class_='epreuve-item') for epreuve in epreuve_elements: details['epreuves'].append({ 'nom': epreuve.find('span', class_='epreuve-name').get_text(strip=True) if epreuve.find('span', class_='epreuve-name') else '', 'categorie': epreuve.find('span', class_='category').get_text(strip=True) if epreuve.find('span', class_='category') else '', 'horaires': epreuve.find('span', class_='time').get_text(strip=True) if epreuve.find('span', class_='time') else '' }) return details except Exception as e: logging.warning(f"Erreur lors du parsing des détails de la course {course_url}: {e}") return None def get_course_results(self, course_url): """Récupérer les résultats d'une course""" full_url = urljoin(self.BASE_URL, course_url) html = self.get_page(full_url, use_selenium=True) if not html: return [] soup = BeautifulSoup(html, 'html.parser') results = [] # Trouver la table principale avec les résultats main_table = soup.find('table', class_='reveal-table') if not main_table: logging.warning("Table principale des résultats non trouvée") return [] # Parser les lignes de résultats rows = main_table.find_all('tr') header_found = False for row in rows: cells = row.find_all(['td', 'th']) # Ignorer les lignes qui sont des tables imbriquées (détails de club) if row.find('table', class_='detail-inner-table'): continue # Chercher la ligne d'en-tête if not header_found and cells: first_cell_text = cells[0].get_text(strip=True) if first_cell_text == 'Place' and len(cells) >= 9: header_found = True continue # Après l'en-tête, parser les lignes de résultats if header_found and cells and len(cells) >= 9: try: # Extraire les données place = cells[0].get_text(strip=True) resultat = cells[1].get_text(strip=True) # Le nom peut être dans la cellule 2 ou combiné nom_cell = cells[2].get_text(strip=True) if len(cells) > 2 else '' # Séparer nom et prénom si possible # Le format FFA est généralement "NOM Prénom" if nom_cell: parts = nom_cell.split(' ') if len(parts) >= 2: nom = parts[0] # Le nom de famille est généralement en premier prenom = ' '.join(parts[1:]) else: nom = nom_cell prenom = '' else: nom = '' prenom = '' club = cells[3].get_text(strip=True) if len(cells) > 3 else '' dept = cells[4].get_text(strip=True) if len(cells) > 4 else '' ligue = cells[5].get_text(strip=True) if len(cells) > 5 else '' infos = cells[6].get_text(strip=True) if len(cells) > 6 else '' niveau = cells[7].get_text(strip=True) if len(cells) > 7 else '' points = cells[8].get_text(strip=True) if len(cells) > 8 else '' # Vérifier que c'est bien une ligne de résultat (place doit être un nombre) try: int(place) except (ValueError, TypeError): continue result = { 'place': place, 'resultat': resultat, 'nom': nom, 'prenom': prenom, 'club': club, 'dept': dept, 'ligue': ligue, 'categorie': infos, # Les infos contiennent souvent la catégorie 'niveau': niveau, 'points': points, 'temps': resultat, # Le résultat est généralement le temps 'course_url': course_url } results.append(result) except Exception as e: logging.warning(f"Erreur lors du parsing d'un résultat: {e}") continue logging.info(f"Trouvé {len(results)} résultats") return results def scrap_all_data(self, limit_courses=None, limit_results=None, fetch_details=False, max_pages=10, use_multithreading=True): """Scraper toutes les données principales""" logging.info("Début du scraping complet des données FFA") # Récupérer les clubs clubs = self.get_clubs_list() if clubs: self.save_to_csv(clubs, 'clubs/clubs_list.csv') # Récupérer les courses courses = self.get_courses_list(max_pages=max_pages, use_multithreading=use_multithreading) if courses: self.save_to_csv(courses, 'courses/courses_list.csv') # Limiter le nombre de courses si spécifié if limit_courses: courses = courses[:limit_courses] # Récupérer les détails et résultats de chaque course (si demandé) all_results = [] course_details = [] if fetch_details: for idx, course in enumerate(tqdm(courses, desc="Scraping des courses")): # Utiliser le lien vers la fiche détaillée si disponible detail_url = course.get('fiche_detail', '') or course.get('lien', '') if detail_url: # Détails de la course details = self.get_course_details(detail_url) if details: details['url'] = detail_url course_details.append(details) # Utiliser le lien vers les résultats si disponible result_url = course.get('resultats_url', '') if result_url: # Résultats de la course results = self.get_course_results(result_url) if results: # Ajouter l'URL de la course aux résultats for result in results: result['course_url'] = result_url all_results.extend(results) # Pause pour éviter de surcharger le serveur time.sleep(1) # Sauvegarder progressivement toutes les 10 courses if (idx + 1) % 10 == 0: if all_results: self.save_to_csv(all_results, 'resultats/results.csv') logging.info(f"Sauvegarde intermédiaire: {len(all_results)} résultats") # Sauvegarder les détails et résultats if course_details: self.save_to_csv(course_details, 'courses/courses_details.csv') if all_results: # Limiter les résultats si spécifié if limit_results: all_results = all_results[:limit_results] self.save_to_csv(all_results, 'resultats/results.csv') self._close_all_selenium() logging.info("Scraping complet terminé") return { 'clubs_count': len(clubs), 'courses_count': len(courses), 'results_count': len(all_results) if 'all_results' in locals() else 0 } if __name__ == "__main__": scraper = FFAScraper() # D'abord vérifier le nombre de courses disponibles total_pages, total_courses, _ = scraper._detect_pagination_info() print("\n" + "="*60) print("📊 Informations de pagination") print("="*60) if total_pages: print(f"Nombre total de pages: {total_pages}") print(f"Estimation du nombre total de courses: ~{total_courses}") print(f"\n⏱️ Estimation du temps de scraping:") print(f" - Multithreading (4 workers): ~{total_pages / 4 * 2:.0f} secondes") print(f" - Séquentiel: ~{total_pages * 2:.0f} secondes") print("="*60) # Lancer le scraping avec multithreading activé par défaut stats = scraper.scrap_all_data(fetch_details=False, max_pages=total_pages if total_pages else 10, use_multithreading=True) print(f"\nScraping terminé:") print(f"- Clubs: {stats['clubs_count']}") print(f"- Courses: {stats['courses_count']}") print(f"- Résultats: {stats['results_count']}")