#!/usr/bin/env python3 """ Script pour télécharger des pages web depuis un fichier CSV et les sauvegarder localement """ import csv import requests import os import time import urllib.parse from pathlib import Path import hashlib import logging class WebPageDownloader: def __init__(self, output_dir="html_pages", delay=1): """ Initialise le téléchargeur Args: output_dir (str): Répertoire de sortie pour les fichiers HTML delay (int): Délai en secondes entre chaque téléchargement """ self.output_dir = Path(output_dir) self.delay = delay # Créer le répertoire de sortie self.output_dir.mkdir(exist_ok=True) # Configuration du logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler('download.log'), logging.StreamHandler() ] ) self.logger = logging.getLogger(__name__) # Headers pour simuler un navigateur réel self.headers = { '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', 'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8', 'Accept-Language': 'fr-FR,fr;q=0.9,en;q=0.8', 'Accept-Encoding': 'gzip, deflate, br', 'Connection': 'keep-alive', 'Upgrade-Insecure-Requests': '1' } # Session pour réutiliser les connexions self.session = requests.Session() self.session.headers.update(self.headers) def get_safe_filename(self, url, max_length=200): """ Génère un nom de fichier sécurisé basé sur l'URL Args: url (str): URL à convertir max_length (int): Longueur maximale du nom de fichier Returns: str: Nom de fichier sécurisé """ # Extraire le domaine et le chemin parsed = urllib.parse.urlparse(url) domain = parsed.netloc.replace('www.', '') path = parsed.path.strip('/') # Créer un nom de base if path: base_name = f"{domain}_{path.replace('/', '_')}" else: base_name = domain # Nettoyer les caractères non autorisés safe_chars = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789-_." safe_name = ''.join(c if c in safe_chars else '_' for c in base_name) # Limiter la longueur if len(safe_name) > max_length: # Utiliser un hash pour raccourcir url_hash = hashlib.md5(url.encode()).hexdigest()[:8] safe_name = safe_name[:max_length-9] + '_' + url_hash return safe_name + '.html' def download_page(self, url, filename=None): """ Télécharge une page web et la sauvegarde Args: url (str): URL à télécharger filename (str): Nom de fichier personnalisé (optionnel) Returns: bool: True si le téléchargement a réussi, False sinon """ try: self.logger.info(f"Téléchargement de: {url}") # Télécharger la page response = self.session.get(url, timeout=30) response.raise_for_status() # Déterminer le nom de fichier if not filename: filename = self.get_safe_filename(url) filepath = self.output_dir / filename # Sauvegarder le contenu HTML with open(filepath, 'w', encoding='utf-8') as f: f.write(response.text) self.logger.info(f"Sauvegardé: {filepath}") # Créer un fichier de métadonnées meta_file = filepath.with_suffix('.meta') with open(meta_file, 'w', encoding='utf-8') as f: f.write(f"URL: {url}\n") f.write(f"Status: {response.status_code}\n") f.write(f"Content-Type: {response.headers.get('content-type', 'unknown')}\n") f.write(f"Date: {time.strftime('%Y-%m-%d %H:%M:%S')}\n") return True except requests.exceptions.RequestException as e: self.logger.error(f"Erreur lors du téléchargement de {url}: {e}") return False except Exception as e: self.logger.error(f"Erreur inattendue pour {url}: {e}") return False def download_from_csv(self, csv_file, url_column='url', filename_column=None): """ Télécharge toutes les pages listées dans un fichier CSV Args: csv_file (str): Chemin vers le fichier CSV url_column (str): Nom de la colonne contenant les URLs filename_column (str): Nom de la colonne pour les noms de fichier (optionnel) """ success_count = 0 error_count = 0 self.logger.info(f"Lecture du fichier CSV: {csv_file}") try: with open(csv_file, 'r', encoding='utf-8') as f: reader = csv.DictReader(f) for row_num, row in enumerate(reader, start=1): if url_column not in row: self.logger.error(f"Colonne '{url_column}' non trouvée dans le CSV") break url = row[url_column].strip() if not url: continue # Nom de fichier personnalisé si spécifié custom_filename = None if filename_column and filename_column in row: custom_filename = row[filename_column].strip() if custom_filename and not custom_filename.endswith('.html'): custom_filename += '.html' # Télécharger la page if self.download_page(url, custom_filename): success_count += 1 else: error_count += 1 # Attendre avant le prochain téléchargement if self.delay > 0: time.sleep(self.delay) self.logger.info(f"Progression: {row_num} pages traitées") except FileNotFoundError: self.logger.error(f"Fichier CSV non trouvé: {csv_file}") except Exception as e: self.logger.error(f"Erreur lors de la lecture du CSV: {e}") self.logger.info(f"Téléchargement terminé - Succès: {success_count}, Erreurs: {error_count}") return success_count, error_count def download_from_csv_jobs(self, csv_file): """ Télécharge toutes les pages listées dans un fichier CSV avec structure Jobs Format: Lien, Type, ID_Job Args: csv_file (str): Chemin vers le fichier CSV """ success_count = 0 error_count = 0 self.logger.info(f"Lecture du fichier CSV Jobs: {csv_file}") try: with open(csv_file, 'r', encoding='utf-8') as f: reader = csv.DictReader(f) for row_num, row in enumerate(reader, start=1): # Vérifier que les colonnes nécessaires existent if 'Lien' not in row: self.logger.error("Colonne 'Lien' non trouvée dans le CSV") break url = row['Lien'].strip() if not url: continue # Créer un nom de fichier basé sur ID_Job et Type custom_filename = None if 'ID_Job' in row and row['ID_Job'].strip(): job_id = row['ID_Job'].strip() job_type = row.get('Type', 'job').strip() or 'job' custom_filename = f"{job_type}_{job_id}.html" self.logger.info(f"Traitement ligne {row_num}: ID={row.get('ID_Job', 'N/A')}, Type={row.get('Type', 'N/A')}") # Télécharger la page if self.download_page(url, custom_filename): success_count += 1 else: error_count += 1 # Attendre avant le prochain téléchargement if self.delay > 0: time.sleep(self.delay) self.logger.info(f"Progression: {row_num} pages traitées") except FileNotFoundError: self.logger.error(f"Fichier CSV non trouvé: {csv_file}") except Exception as e: self.logger.error(f"Erreur lors de la lecture du CSV: {e}") self.logger.info(f"Téléchargement terminé - Succès: {success_count}, Erreurs: {error_count}") return success_count, error_count def main(): """Fonction principale""" # Configuration adaptée à votre structure CSV CSV_FILE = "jobs_ch_all_links_20250808_135526.csv" # Nom du fichier CSV OUTPUT_DIR = "html_pages" # Répertoire de sortie DELAY = 2 # Délai entre téléchargements en secondes URL_COLUMN = "Lien" # Colonne contenant les URLs # Créer le téléchargeur downloader = WebPageDownloader(output_dir=OUTPUT_DIR, delay=DELAY) # Vérifier si le fichier CSV existe if not os.path.exists(CSV_FILE): print(f"Création d'un fichier CSV d'exemple: {CSV_FILE}") with open(CSV_FILE, 'w', newline='', encoding='utf-8') as f: writer = csv.writer(f) writer.writerow(['Lien', 'Type', 'ID_Job']) # En-tête adapté writer.writerow(['https://www.jobs.ch/fr/offres-d-emploi/detail/12345', 'emploi', 'JOB_001']) writer.writerow(['https://www.jobs.ch/fr/offres-d-emploi/detail/67890', 'emploi', 'JOB_002']) print(f"Veuillez modifier le fichier {CSV_FILE} avec vos URLs et relancer le script.") return # Démarrer le téléchargement avec votre structure CSV success, errors = downloader.download_from_csv_jobs( csv_file=CSV_FILE ) print(f"\nRésultat final:") print(f"Pages téléchargées avec succès: {success}") print(f"Erreurs: {errors}") print(f"Fichiers sauvegardés dans: {OUTPUT_DIR}") if __name__ == "__main__": main()