6de44556f4
- Add get_club_results.py: Extract all results for a club from results CSV - Fuzzy search for club names - List clubs functionality - Export results to CSV - Add scrape_course_details.py: Scrape detailed info from course pages - Extract competition ID (frmcompetition) - Extract event details (distance, category, sex) - Extract course website link - Extract location and organizer info - Update README.md with new scripts and usage examples - Update version to v1.3.0
494 lines
19 KiB
Python
Executable File
494 lines
19 KiB
Python
Executable File
#!/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()
|