#!/usr/bin/env python3 """ Version ligne de commande du scraper Jobs.ch Compatible avec le contrôleur PHP """ import argparse import json import time import signal import sys import os from datetime import datetime from selenium import webdriver from selenium.webdriver.firefox.options import Options from selenium.webdriver.common.by import By from selenium.webdriver.support.ui import WebDriverWait from selenium.webdriver.support import expected_conditions as EC from selenium.common.exceptions import TimeoutException, NoSuchElementException from bs4 import BeautifulSoup class JobsChScraperCLI: def __init__(self, args): self.args = args self.driver = None self.results = [] self.status_file = 'scraper_status.json' self.should_stop = False # Gestionnaire de signal pour arrêt propre signal.signal(signal.SIGTERM, self.signal_handler) signal.signal(signal.SIGINT, self.signal_handler) self.setup_driver() def signal_handler(self, signum, frame): """Gestionnaire pour arrêt propre du script""" print(f"\n🛑 Signal {signum} reçu, arrêt du scraping...") self.should_stop = True if self.driver: self.driver.quit() sys.exit(0) def setup_driver(self): """Configuration du driver Firefox""" firefox_options = Options() if self.args.headless: firefox_options.add_argument("--headless") print("🔧 Mode headless activé") else: print("🔧 Mode graphique activé - Firefox sera visible") # Options anti-détection firefox_options.add_argument("--disable-blink-features=AutomationControlled") firefox_options.set_preference("dom.webdriver.enabled", False) firefox_options.set_preference("useAutomationExtension", False) # Profil pour éviter les popups firefox_options.set_preference("dom.disable_beforeunload", True) firefox_options.set_preference("browser.tabs.warnOnClose", False) try: self.driver = webdriver.Firefox(options=firefox_options) self.driver.execute_script("Object.defineProperty(navigator, 'webdriver', {get: () => undefined})") print("✅ Driver Firefox initialisé") self.update_status("Driver initialisé") except Exception as e: print(f"❌ Erreur initialisation driver: {e}") sys.exit(1) def update_status(self, status, **kwargs): """Met à jour le fichier de statut""" status_data = { 'status': status, 'timestamp': datetime.now().isoformat(), 'jobs_found': len(self.results), **kwargs } # Lire le statut existant if os.path.exists(self.status_file): try: with open(self.status_file, 'r') as f: existing_data = json.load(f) status_data.update(existing_data) except: pass # Écrire le nouveau statut with open(self.status_file, 'w') as f: json.dump(status_data, f, indent=2, ensure_ascii=False) def wait_for_user_input(self, message): """Attendre une action utilisateur (pour captchas)""" if self.args.headless: print(f"⚠️ Mode headless: impossible de gérer {message}") return False print(f"\n🔴 {message}") print("Résolvez le problème dans Firefox puis tapez ENTER pour continuer (ou 'q' pour quitter):") self.update_status(f"PAUSE: {message}") try: user_input = input().strip().lower() if user_input == 'q': return False return True except KeyboardInterrupt: return False def check_for_issues(self): """Vérifie les problèmes courants (captcha, etc.)""" issue_selectors = [ ("captcha", "iframe[src*='captcha'], .captcha, #captcha"), ("cloudflare", ".cf-browser-verification, #challenge-form"), ("blocked", "h1:contains('Access Denied'), .blocked"), ("error", ".error-page, .error-message") ] for issue_type, selector in issue_selectors: try: if issue_type == "blocked": # Vérification spéciale pour les pages bloquées page_source = self.driver.page_source.lower() if "access denied" in page_source or "blocked" in page_source: return issue_type else: self.driver.find_element(By.CSS_SELECTOR, selector) return issue_type except NoSuchElementException: continue return None def scrape_jobs(self): """Lance le scraping principal""" print(f"🔍 Recherche: '{self.args.search_terms}'") print(f"📄 Pages à scraper: {self.args.pages}") print(f"⏱️ Délai entre pages: {self.args.delay}s") self.update_status("Démarrage du scraping", search_terms=self.args.search_terms, total_pages=self.args.pages) try: # URL de recherche base_url = "https://www.jobs.ch/fr/emplois/" search_url = f"{base_url}?term={'+'.join(self.args.search_terms.split())}" print(f"🌐 Navigation vers: {search_url}") self.driver.get(search_url) time.sleep(3) # Vérifier les problèmes issue = self.check_for_issues() if issue: if not self.wait_for_user_input(f"{issue.title()} détecté"): return # Scraper les pages for page in range(1, self.args.pages + 1): if self.should_stop: break print(f"\n📄 Scraping page {page}/{self.args.pages}") self.update_status(f"Page {page}/{self.args.pages}", current_page=page) # Attendre le chargement des résultats try: WebDriverWait(self.driver, 10).until( EC.presence_of_element_located((By.CSS_SELECTOR, "[data-cy='serp-jobad'], .job-item, article")) ) except TimeoutException: print("⚠️ Timeout - vérification nécessaire") if not self.wait_for_user_input("Page non chargée correctement"): break # Extraire les emplois page_jobs = self.extract_jobs_from_page() self.results.extend(page_jobs) jobs_count = len(page_jobs) total_jobs = len(self.results) print(f"✅ {jobs_count} emplois trouvés (total: {total_jobs})") # Page suivante if page < self.args.pages and not self.should_stop: if not self.navigate_to_next_page(): print("🔚 Impossible d'aller à la page suivante") break time.sleep(self.args.delay) except Exception as e: print(f"❌ Erreur pendant le scraping: {e}") self.update_status(f"Erreur: {str(e)}") finally: # Sauvegarder les résultats self.save_results() def extract_jobs_from_page(self): """Extrait les emplois de la page courante""" jobs = [] try: html = self.driver.page_source soup = BeautifulSoup(html, 'html.parser') # Différents sélecteurs possibles pour jobs.ch selectors = [ "[data-cy='serp-jobad']", ".job-item", "article[data-jobad-id]", ".jobListItem" ] job_elements = [] for selector in selectors: job_elements = soup.select(selector) if job_elements: print(f"🎯 Utilisation du sélecteur: {selector}") break if not job_elements: print("⚠️ Aucun emploi trouvé avec les sélecteurs standard") # Fallback: chercher tous les articles ou divs qui pourraient contenir des emplois job_elements = soup.find_all(['article', 'div'], class_=lambda x: x and 'job' in str(x).lower()) for element in job_elements: try: job_data = self.extract_single_job(element) if job_data and job_data.get('titre', '').strip(): jobs.append(job_data) except Exception as e: print(f"⚠️ Erreur extraction emploi: {e}") continue except Exception as e: print(f"❌ Erreur extraction page: {e}") return jobs def extract_single_job(self, element): """Extrait les données d'un emploi""" try: # Titre (priorité aux liens puis aux headings) title_selectors = ['h2 a', 'h3 a', 'h4 a', 'a[data-cy*="title"]', 'h2', 'h3', 'h4', '.title'] title = self.find_text_by_selectors(element, title_selectors) # Entreprise company_selectors = ['.company', '[data-cy*="company"]', '.employer', '.jobCompany'] company = self.find_text_by_selectors(element, company_selectors) # Lieu location_selectors = ['.location', '[data-cy*="location"]', '.jobLocation', '.place'] location = self.find_text_by_selectors(element, location_selectors) # Lien link_element = element.find('a', href=True) link = "" if link_element: link = link_element.get('href', '') if link and not link.startswith('http'): link = f"https://www.jobs.ch{link}" # Date date_selectors = ['time', '.date', '[data-cy*="date"]', '.published'] pub_date = self.find_text_by_selectors(element, date_selectors) return { 'titre': title or "Titre non trouvé", 'entreprise': company or "Entreprise non trouvée", 'lieu': location or "Lieu non trouvé", 'lien': link, 'date_publication': pub_date or "Date non trouvée", 'date_scraping': datetime.now().strftime("%Y-%m-%d %H:%M:%S") } except Exception as e: print(f"⚠️ Erreur extraction données: {e}") return None def find_text_by_selectors(self, element, selectors): """Trouve du texte en essayant plusieurs sélecteurs""" for selector in selectors: try: found = element.select_one(selector) if found: text = found.get_text(strip=True) if text: return text except: continue return "" def navigate_to_next_page(self): """Navigue vers la page suivante""" next_selectors = [ "a[data-cy*='next']", "a[aria-label*='suivant']", "a[title*='Next']", ".pagination a:last-child", "a:contains('›')", "a:contains('Suivant')" ] for selector in next_selectors: try: next_button = self.driver.find_element(By.CSS_SELECTOR, selector) if next_button.is_enabled() and next_button.is_displayed(): self.driver.execute_script("arguments[0].click();", next_button) print("➡️ Navigation vers page suivante") return True except NoSuchElementException: continue return False def save_results(self): """Sauvegarde les résultats""" if not self.results: print("⚠️ Aucun résultat à sauvegarder") return timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") filename = f"jobs_scraping_{timestamp}.json" try: with open(filename, 'w', encoding='utf-8') as f: json.dump(self.results, f, indent=2, ensure_ascii=False) print(f"💾 {len(self.results)} emplois sauvegardés dans: {filename}") self.update_status("Terminé avec succès", filename=filename) except Exception as e: print(f"❌ Erreur sauvegarde: {e}") self.update_status(f"Erreur sauvegarde: {str(e)}") def cleanup(self): """Nettoyage final""" if self.driver: self.driver.quit() print("🔚 Navigateur fermé") def main(): parser = argparse.ArgumentParser(description='Scraper Jobs.ch avec gestion des captchas') parser.add_argument('--search', '--search-terms', dest='search_terms', default='python developer', help='Termes de recherche (défaut: python developer)') parser.add_argument('--pages', type=int, default=5, help='Nombre de pages à scraper (défaut: 5)') parser.add_argument('--delay', type=int, default=3, help='Délai entre les pages en secondes (défaut: 3)') parser.add_argument('--headless', action='store_true', help='Mode headless (sans interface graphique)') parser.add_argument('--output', '-o', help='Fichier de sortie (optionnel)') args = parser.parse_args() print("🚀 Jobs.ch Scraper CLI") print("=" * 40) scraper = None try: scraper = JobsChScraperCLI(args) scraper.scrape_jobs() except KeyboardInterrupt: print("\n🛑 Interruption par l'utilisateur") except Exception as e: print(f"❌ Erreur fatale: {e}") finally: if scraper: scraper.cleanup() if __name__ == "__main__": main()