#!/usr/bin/env python3 """ Script pour récupérer les détails des pages de courses FFA Récupère: - Le numéro de course (frmcompetition) - Les différentes épreuves - Le lieu - Le lien vers le site de la course """ import sys import os import time import csv import logging import re import argparse from tqdm import tqdm from selenium import webdriver from selenium.webdriver.chrome.service import Service from webdriver_manager.chrome import ChromeDriverManager from bs4 import BeautifulSoup import pandas as pd sys.path.insert(0, os.path.join(os.path.dirname(__file__), '../src')) # Configuration du logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler('scrape_course_details.log'), logging.StreamHandler() ] ) logger = logging.getLogger(__name__) class CourseDetailsScraper: """Scraper pour récupérer les détails des pages de courses""" def __init__(self, output_dir="data"): self.output_dir = output_dir self.base_url = "https://www.athle.fr" self.request_delay = 2.0 self.max_retries = 3 # Fichiers de sortie self.details_file = os.path.join(output_dir, "course_details.csv") self.epreuves_file = os.path.join(output_dir, "course_epreuves.csv") # Initialiser le driver Selenium self.driver = None # En-têtes CSV self.details_headers = [ 'course_id', 'course_url', 'course_nom', 'course_date', 'course_lieu', 'course_lieu_complet', 'course_site_web', 'competition_id', 'organisateur', 'contact', 'date_heure', 'lieu_details', 'nb_epreuves', 'url_source' ] self.epreuves_headers = [ 'course_id', 'epreuve_id', 'epreuve_nom', 'epreuve_numero', 'epreuve_distance', 'epreuve_categorie', 'epreuve_sexe', 'epreuve_type', 'participants', 'url_resultats' ] # Initialiser les fichiers CSV self._init_csv_files() def _init_csv_file(self, filename, headers): """Initialiser un fichier CSV avec les en-têtes""" if not os.path.exists(filename): with open(filename, 'w', newline='', encoding='utf-8-sig') as f: writer = csv.DictWriter(f, fieldnames=headers) writer.writeheader() logger.info(f"Fichier CSV initialisé: {filename}") def _init_csv_files(self): """Initialiser tous les fichiers CSV""" self._init_csv_file(self.details_file, self.details_headers) self._init_csv_file(self.epreuves_file, self.epreuves_headers) def _init_driver(self): """Initialiser le driver Selenium""" if self.driver is None: options = webdriver.ChromeOptions() options.add_argument('--headless') options.add_argument('--no-sandbox') options.add_argument('--disable-dev-shm-usage') options.add_argument('--disable-gpu') options.add_argument('--window-size=1920,1080') options.add_argument('user-agent=Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36') service = Service(ChromeDriverManager().install()) self.driver = webdriver.Chrome(service=service, options=options) logger.info("Driver Chrome initialisé") def extract_competition_id(self, url): """Extraire l'ID de competition depuis l'URL (frmcompetition=XXX)""" match = re.search(r'frmcompetition=(\d+)', url) if match: return match.group(1) return "" def get_courses_from_csv(self, courses_file): """Lire les courses depuis le fichier CSV""" courses = [] try: df = pd.read_csv(courses_file, encoding='utf-8-sig') for idx, row in df.iterrows(): if pd.notna(row.get('resultats_url')) and row['resultats_url']: courses.append({ 'id': idx, 'nom': row.get('nom', ''), 'date': row.get('date', ''), 'lieu': row.get('lieu', ''), 'resultats_url': row['resultats_url'], 'lien': row.get('lien', '') }) logger.info(f"{len(courses)} courses trouvées dans {courses_file}") return courses except Exception as e: logger.error(f"Erreur lecture du fichier CSV: {e}") return [] def parse_course_page(self, url, course_info): """Parser la page d'une course pour extraire les détails et épreuves""" course_details = None epreuves = [] for attempt in range(self.max_retries): try: self.driver.get(url) time.sleep(self.request_delay) html = self.driver.page_source soup = BeautifulSoup(html, 'html.parser') # Extraire l'ID de compétition competition_id = self.extract_competition_id(url) # ===== EXTRAIRE LES DÉTAILS DE LA COURSE ===== course_details = { 'course_id': course_info['id'], 'course_url': url, 'course_nom': course_info['nom'], 'course_date': course_info['date'], 'course_lieu': course_info['lieu'], 'course_lieu_complet': '', 'course_site_web': '', 'competition_id': competition_id, 'organisateur': '', 'contact': '', 'date_heure': '', 'lieu_details': '', 'nb_epreuves': 0, 'url_source': url } # Chercher les informations générales de la course # Ces informations peuvent être dans des divs spécifiques ou le titre de la page title = soup.find('h1') if title: course_details['course_nom'] = title.get_text(strip=True) # Chercher les blocs d'information (lieu, date, etc.) info_divs = soup.find_all('div', class_=re.compile(r'info|detail|location|lieu', re.IGNORECASE)) for div in info_divs: text = div.get_text(strip=True) if 'site web' in text.lower() or 'http' in text: # Extraire le lien vers le site links = div.find_all('a', href=True) for link in links: href = link['href'] if href.startswith('http'): course_details['course_site_web'] = href break elif 'contact' in text.lower(): course_details['contact'] = text elif 'organisateur' in text.lower(): course_details['organisateur'] = text # Chercher le site web dans les liens all_links = soup.find_all('a', href=True) for link in all_links: href = link['href'] link_text = link.get_text(strip=True).lower() if (href.startswith('http') and 'athle.fr' not in href and any(keyword in link_text for keyword in ['site', 'web', 'inscription', 'organisateur'])): course_details['course_site_web'] = href break # ===== EXTRAIRE LES ÉPREUVES ===== main_table = soup.find('table', class_='reveal-table') if not main_table: logger.warning(f"Pas de table trouvée pour {url}") return course_details, epreuves rows = main_table.find_all('tr') current_epreuve = None epreuve_numero = 0 for row in rows: cells = row.find_all(['td', 'th']) # Ignorer les tables imbriquées if row.find('table', class_='detail-inner-table'): continue # Détecter une nouvelle épreuve (en-tête de section) if cells and len(cells) == 1: cell = cells[0] cell_text = cell.get_text(strip=True) # Vérifier si c'est un en-tête d'épreuve # Les épreuves FFA ont souvent: distance, catégorie, sexe if cell_text and self._is_epreuve_header(cell_text): epreuve_numero += 1 current_epreuve = { 'course_id': course_info['id'], 'epreuve_id': f"{course_info['id']}_E{epreuve_numero}", 'epreuve_nom': cell_text, 'epreuve_numero': epreuve_numero, 'epreuve_distance': self._extract_distance(cell_text), 'epreuve_categorie': self._extract_categorie(cell_text), 'epreuve_sexe': self._extract_sexe(cell_text), 'epreuve_type': self._extract_type(cell_text), 'participants': 0, 'url_resultats': url } # Compter les participants si c'est une ligne de résultats if current_epreuve and cells and len(cells) >= 9: try: place = cells[0].get_text(strip=True) int(place) # Vérifier que c'est un nombre current_epreuve['participants'] += 1 except (ValueError, TypeError): continue # Ajouter l'épreuve si elle a des participants if current_epreuve and current_epreuve['participants'] > 0: epreuves.append(current_epreuve) current_epreuve = None course_details['nb_epreuves'] = len(epreuves) logger.info(f"Course: {course_info['nom']} - {len(epreuves)} épreuves") return course_details, epreuves except Exception as e: logger.error(f"Erreur tentative {attempt + 1}/{self.max_retries} pour {url}: {e}") if attempt < self.max_retries - 1: time.sleep(self.request_delay * (attempt + 1)) else: logger.warning(f"Impossible de récupérer les détails pour {url}") return course_details, epreuves return course_details, epreuves def _is_epreuve_header(self, text): """Vérifier si le texte est un en-tête d'épreuve""" # Les épreuves contiennent généralement: # - Une distance (ex: 100m, 5km, marathon) # - Un sexe (F/H, Hommes/Femmes, Masculin/Féminin) # - Une catégorie (Seniors, Vétérans, etc.) keywords_distance = ['m ', 'm$', 'km', 'marathon', 'semi'] keywords_sexe = ['femmes', 'hommes', 'masculin', 'féminin', 'féminines', 'masculines', 'F ', 'H ', 'SEF', 'SEM'] keywords_categorie = ['seniors', 'vétérans', 'juniors', 'cadets', 'minimes', 'benjamins', 'poussins', 'espoirs', 'u20', 'u18', 'u16', 'u14', 'u23', 'm0', 'm1', 'm2', 'm3', 'm4'] # Vérifier la présence de mots-clés has_distance = any(re.search(kw, text, re.IGNORECASE) for kw in keywords_distance) has_sexe = any(re.search(kw, text, re.IGNORECASE) for kw in keywords_sexe) has_categorie = any(re.search(kw, text, re.IGNORECASE) for kw in keywords_categorie) # C'est probablement une épreuve si elle a au moins un de ces éléments # et que le texte n'est pas trop court return len(text) > 10 and (has_distance or has_sexe or has_categorie) def _extract_distance(self, text): """Extraire la distance depuis le texte de l'épreuve""" # Distance en mètres match = re.search(r'(\d+)\s*m\b', text, re.IGNORECASE) if match: return f"{match.group(1)}m" # Distance en km match = re.search(r'(\d+(?:\.\d+)?)\s*km\b', text, re.IGNORECASE) if match: return f"{match.group(1)}km" # Marathon ou semi-marathon if 'marathon' in text.lower() and 'semi' not in text.lower(): return "42.195km" elif 'semi' in text.lower(): return "21.1km" return "" def _extract_categorie(self, text): """Extraire la catégorie depuis le texte de l'épreuve""" categories = ['Vétérans', 'Seniors', 'Juniors', 'Cadets', 'Minimes', 'Benjamins', 'Poussins', 'Espoirs', 'U20', 'U18', 'U16', 'U14', 'U23'] for cat in categories: if cat in text: return cat return "" def _extract_sexe(self, text): """Extraire le sexe depuis le texte de l'épreuve""" text_upper = text.upper() if any(kw in text_upper for kw in ['FEMMES', 'FÉMININ', 'FÉMININES', ' F ', 'SEF']): return 'F' elif any(kw in text_upper for kw in ['HOMMES', 'MASCULIN', 'MASCULINES', ' H ', 'SEM']): return 'M' else: return "" def _extract_type(self, text): """Extraire le type d'épreuve""" if any(kw in text.lower() for kw in ['relais', 'relay']): return 'Relais' elif any(kw in text.lower() for kw in ['cross']): return 'Cross' elif any(kw in text.lower() for kw in ['trail']): return 'Trail' elif any(kw in text.lower() for kw in ['marathon', 'semi']): return 'Route' elif any(kw in text.lower() for kw in ['saut', 'lancer', 'lancé']): return 'Épreuves combinées' else: return 'Piste' def append_to_csv(self, data, filename, headers): """Ajouter des données au fichier CSV""" if not data: return with open(filename, 'a', newline='', encoding='utf-8-sig') as f: writer = csv.DictWriter(f, fieldnames=headers) # data peut être un dict ou une liste if isinstance(data, dict): writer.writerow(data) else: writer.writerows(data) logger.info(f"{len(data) if isinstance(data, list) else 1} enregistrement(s) ajouté(s) à {filename}") def run(self, courses_file, limit=None, start_from=0): """Exécuter le scraping des détails pour toutes les courses""" logger.info(f"Début du scraping des détails depuis {courses_file}") # Charger les courses courses = self.get_courses_from_csv(courses_file) if not courses: logger.error("Aucune course trouvée") return # Limiter ou démarrer à un index spécifique if limit: courses = courses[:limit] logger.info(f"Limitation à {limit} courses") if start_from > 0: courses = courses[start_from:] logger.info(f"Démarrage à l'index {start_from}") # Initialiser le driver self._init_driver() total_details = 0 total_epreuves = 0 skipped = 0 # Scraper les détails pour chaque course with tqdm(courses, desc="Courses scrapées", unit="course") as pbar: for course in pbar: try: details, epreuves = self.parse_course_page(course['resultats_url'], course) if details: self.append_to_csv(details, self.details_file, self.details_headers) total_details += 1 if epreuves: self.append_to_csv(epreuves, self.epreuves_file, self.epreuves_headers) total_epreuves += len(epreuves) else: skipped += 1 pbar.set_postfix({ 'details': total_details, 'épreuves': total_epreuves, 'skipped': skipped }) # Pause entre les requêtes time.sleep(1) except Exception as e: logger.error(f"Erreur pour la course {course['nom']}: {e}") skipped += 1 continue # Fermer le driver if self.driver: self.driver.quit() logger.info(f"Scraping terminé!") logger.info(f"Details récupérés: {total_details}") logger.info(f"Epreuves récupérées: {total_epreuves}") logger.info(f"Courses sans épreuves: {skipped}") return { 'total_courses': len(courses), 'total_details': total_details, 'total_epreuves': total_epreuves, 'skipped': skipped } def main(): """Point d'entrée principal""" parser = argparse.ArgumentParser( description='Récupérer les détails des pages de courses FFA', formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" Exemples: # Scraper toutes les courses python scrape_course_details.py # Scraper seulement les 100 premières courses python scrape_course_details.py --limit 100 # Reprendre à partir de l'index 500 python scrape_course_details.py --start-from 500 # Utiliser un autre fichier de courses python scrape_course_details.py --courses-file data/courses_daily.csv """ ) parser.add_argument('--courses-file', default='data/courses_daily.csv', help='Fichier CSV des courses (défaut: data/courses_daily.csv)') parser.add_argument('--output-dir', default='data', help='Répertoire de sortie (défaut: data)') parser.add_argument('--limit', type=int, help='Limiter le nombre de courses à scraper') parser.add_argument('--start-from', type=int, default=0, help='Index de départ pour reprendre le scraping') args = parser.parse_args() print("=" * 70) print("🔄 Scraper des détails de courses FFA") print("=" * 70) print(f"Fichier des courses: {args.courses_file}") print(f"Répertoire de sortie: {args.output_dir}") if args.limit: print(f"Limitation: {args.limit} courses") if args.start_from > 0: print(f"Démarrage à l'index: {args.start_from}") print("=" * 70) scraper = CourseDetailsScraper(output_dir=args.output_dir) # Lancer le scraping stats = scraper.run( args.courses_file, limit=args.limit, start_from=args.start_from ) print("\n" + "=" * 70) print("✅ Scraping terminé!") print("=" * 70) print(f"Courses traitées: {stats['total_courses']}") print(f"Détails récupérés: {stats['total_details']}") print(f"Epreuves récupérées: {stats['total_epreuves']}") print(f"Courses sans épreuves: {stats['skipped']}") print("=" * 70) if __name__ == "__main__": main()