#!/usr/bin/env python3 """ Version ligne de commande du scraper Jobs.ch Compatible avec le contrôleur PHP - Version améliorée avec lecture CSV """ import argparse import json import time import signal import sys import os import csv import traceback 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, WebDriverException from bs4 import BeautifulSoup import logging class JobsChScraperCLI: def __init__(self, args): self.args = args self.driver = None self.results = [] self.processed_links = [] self.failed_links = [] self.status_file = 'scraper_status.json' self.should_stop = False self.captcha_detected_count = 0 self.total_links_to_process = 0 self.current_link_index = 0 # Configuration du logging self.setup_logging() # 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 setup_logging(self): """Configuration du système de logs""" log_format = '%(asctime)s - %(levelname)s - %(message)s' logging.basicConfig( level=logging.DEBUG if self.args.debug else logging.INFO, format=log_format, handlers=[ logging.FileHandler('scraper_debug.log', encoding='utf-8'), logging.StreamHandler() ] ) self.logger = logging.getLogger(__name__) self.logger.info("🔧 Système de logs initialisé") def signal_handler(self, signum, frame): """Gestionnaire pour arrêt propre du script""" self.logger.warning(f"🛑 Signal {signum} reçu, arrêt du scraping...") self.should_stop = True if self.driver: self.driver.quit() self.save_progress() sys.exit(0) def setup_driver(self): """Configuration du driver Firefox avec options anti-détection""" firefox_options = Options() if self.args.headless: firefox_options.add_argument("--headless") self.logger.info("🔧 Mode headless activé") else: self.logger.info("🔧 Mode graphique activé - Firefox sera visible") # ⚠️ Supprimer les arguments Chrome-only # firefox_options.add_argument("--disable-blink-features=AutomationControlled") # Préférences Firefox pour limiter la détection firefox_options.set_preference("dom.webdriver.enabled", False) firefox_options.set_preference("useAutomationExtension", False) firefox_options.set_preference( "general.useragent.override", "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:109.0) Gecko/20100101 Firefox/117.0" ) firefox_options.set_preference("dom.disable_beforeunload", True) firefox_options.set_preference("browser.tabs.warnOnClose", False) firefox_options.set_preference("dom.popup_maximum", 0) try: self.driver = webdriver.Firefox(options=firefox_options) self.driver.execute_script("Object.defineProperty(navigator, 'webdriver', {get: () => undefined})") self.driver.set_page_load_timeout(30) self.driver.implicitly_wait(5) self.logger.info("✅ Driver Firefox initialisé avec succès") self.update_status("Driver initialisé") except Exception as e: self.logger.error(f"❌ Erreur initialisation driver: {e}") self.logger.debug(traceback.format_exc()) sys.exit(1) def load_links_from_csv(self, csv_file_path): """Charge les liens depuis le fichier CSV""" links = [] if not os.path.exists(csv_file_path): self.logger.error(f"❌ Fichier CSV non trouvé: {csv_file_path}") return links try: with open(csv_file_path, 'r', encoding='utf-8') as csvfile: # Détecter le délimiteur sample = csvfile.read(1024) csvfile.seek(0) sniffer = csv.Sniffer() delimiter = sniffer.sniff(sample).delimiter reader = csv.DictReader(csvfile, delimiter=delimiter) self.logger.info(f"📊 Colonnes détectées dans le CSV: {reader.fieldnames}") for row_num, row in enumerate(reader, 1): # Chercher une colonne qui contient des URLs link_found = False for key, value in row.items(): if value and ('http' in str(value) or 'jobs.ch' in str(value)): # Nettoyer le lien clean_link = str(value).strip() if clean_link.startswith('http'): links.append({ 'url': clean_link, 'source_row': row_num, 'original_data': row }) link_found = True break if not link_found: self.logger.debug(f"🔍 Ligne {row_num}: Aucun lien détecté - {row}") self.logger.info(f"📋 {len(links)} liens chargés depuis {csv_file_path}") self.logger.debug(f"🔗 Exemple de liens: {links[:3] if links else 'Aucun'}") except Exception as e: self.logger.error(f"❌ Erreur lors de la lecture du CSV: {e}") self.logger.debug(traceback.format_exc()) return links def update_status(self, status, **kwargs): """Met à jour le fichier de statut avec plus de détails""" status_data = { 'status': status, 'timestamp': datetime.now().isoformat(), 'jobs_found': len(self.results), 'links_processed': len(self.processed_links), 'links_failed': len(self.failed_links), 'total_links': self.total_links_to_process, 'current_link': self.current_link_index, 'captcha_count': self.captcha_detected_count, 'progress_percentage': round((self.current_link_index / max(self.total_links_to_process, 1)) * 100, 2), **kwargs } # Lire le statut existant if os.path.exists(self.status_file): try: with open(self.status_file, 'r', encoding='utf-8') as f: existing_data = json.load(f) status_data.update(existing_data) except Exception as e: self.logger.debug(f"Erreur lecture statut existant: {e}") # Écrire le nouveau statut try: with open(self.status_file, 'w', encoding='utf-8') as f: json.dump(status_data, f, indent=2, ensure_ascii=False) except Exception as e: self.logger.error(f"❌ Erreur écriture statut: {e}") def detect_captcha_advanced(self): """Détection avancée des captchas et blocages""" self.logger.debug("🔍 Vérification captcha/blocage avancée...") # Vérifications multiples checks = [ { 'name': 'Cloudflare Challenge', 'selectors': ['.cf-browser-verification', '#challenge-form', '.cf-checking-browser', '[id*="challenge"]'], 'text_patterns': ['checking your browser', 'cloudflare', 'ddos protection'] }, { 'name': 'reCAPTCHA', 'selectors': ['iframe[src*="recaptcha"]', '.recaptcha-checkbox', '#recaptcha', '.g-recaptcha'], 'text_patterns': ['recaptcha', 'not a robot'] }, { 'name': 'hCaptcha', 'selectors': ['iframe[src*="hcaptcha"]', '.h-captcha', '#hcaptcha'], 'text_patterns': ['hcaptcha', 'human verification'] }, { 'name': 'Page bloquée', 'selectors': ['.blocked', '.access-denied', '.error-page'], 'text_patterns': ['access denied', 'blocked', 'forbidden', '403', '429'] }, { 'name': 'Jobs.ch Protection', 'selectors': ['[data-cy*="protection"]', '.protection-notice'], 'text_patterns': ['protection', 'veuillez patienter', 'please wait'] } ] page_source = self.driver.page_source.lower() page_title = self.driver.title.lower() for check in checks: self.logger.debug(f"🔍 Vérification: {check['name']}") # Vérifier les sélecteurs CSS for selector in check['selectors']: try: elements = self.driver.find_elements(By.CSS_SELECTOR, selector) if elements: self.logger.warning(f"⚠️ {check['name']} détecté via sélecteur: {selector}") return check['name'] except Exception as e: self.logger.debug(f"Erreur sélecteur {selector}: {e}") # Vérifier les patterns de texte for pattern in check['text_patterns']: if pattern in page_source or pattern in page_title: self.logger.warning(f"⚠️ {check['name']} détecté via pattern: {pattern}") return check['name'] # Vérifications supplémentaires current_url = self.driver.current_url.lower() if 'challenge' in current_url or 'captcha' in current_url: self.logger.warning(f"⚠️ Captcha détecté dans l'URL: {current_url}") return "URL Captcha" self.logger.debug("✅ Aucun captcha/blocage détecté") return None def wait_for_user_input(self, message, issue_type=None): """Attendre une action utilisateur (pour captchas) avec gestion avancée""" if self.args.headless: self.logger.error(f"❌ Mode headless: impossible de gérer {message}") if self.args.auto_retry: self.logger.info("🔄 Tentative de contournement automatique...") time.sleep(10) return True return False self.captcha_detected_count += 1 self.logger.warning(f"🔴 CAPTCHA/BLOCAGE DÉTECTÉ: {message}") self.logger.info(f"🔴 URL actuelle: {self.driver.current_url}") self.logger.info(f"🔴 Titre de la page: {self.driver.title}") print(f"\n🔴 {message}") print(f"🌐 URL: {self.driver.current_url}") print(f"📄 Titre: {self.driver.title}") print("="*60) print("INSTRUCTIONS:") print("1. Résolvez le captcha/blocage dans Firefox") print("2. Tapez ENTER pour continuer") print("3. Tapez 'r' pour recharger la page") print("4. Tapez 's' pour passer ce lien") print("5. Tapez 'q' pour quitter") print("="*60) self.update_status(f"PAUSE: {message}", issue_type=issue_type) try: user_input = input("Votre choix: ").strip().lower() if user_input == 'q': self.logger.info("👋 Arrêt demandé par l'utilisateur") return False elif user_input == 'r': self.logger.info("🔄 Rechargement de la page...") self.driver.refresh() time.sleep(3) return self.detect_captcha_advanced() is None elif user_input == 's': self.logger.info("⏭️ Passage au lien suivant") return 'skip' else: # Vérifier si le captcha est résolu if self.detect_captcha_advanced() is None: self.logger.info("✅ Captcha résolu avec succès") return True else: self.logger.warning("❌ Captcha toujours présent") return self.wait_for_user_input(f"Captcha toujours présent pour {message}", issue_type) except KeyboardInterrupt: self.logger.info("👋 Interruption par l'utilisateur") return False def scrape_from_links(self): """Lance le scraping depuis les liens du CSV""" csv_path = self.args.csv_file or "/root/liens/1jobs_ch_complete_extraction.csv" self.logger.info(f"📁 Chargement des liens depuis: {csv_path}") links = self.load_links_from_csv(csv_path) if not links: self.logger.error("❌ Aucun lien trouvé dans le fichier CSV") return self.total_links_to_process = len(links) self.logger.info(f"🎯 {self.total_links_to_process} liens à traiter") # Limiter le nombre de liens si spécifié if self.args.max_links and self.args.max_links < len(links): links = links[:self.args.max_links] self.total_links_to_process = len(links) self.logger.info(f"✂️ Limitation à {self.total_links_to_process} liens") self.update_status("Démarrage du scraping depuis CSV", csv_file=csv_path, total_links=self.total_links_to_process) for i, link_data in enumerate(links): if self.should_stop: break self.current_link_index = i + 1 url = link_data['url'] self.logger.info(f"\n🔗 [{self.current_link_index}/{self.total_links_to_process}] Traitement: {url}") self.update_status(f"Traitement lien {self.current_link_index}/{self.total_links_to_process}") try: success = self.process_single_link(url, link_data) if success == 'skip': self.logger.info("⏭️ Lien ignoré") continue elif success: self.processed_links.append(link_data) self.logger.info("✅ Lien traité avec succès") else: self.failed_links.append(link_data) self.logger.warning("❌ Échec du traitement du lien") except Exception as e: self.logger.error(f"❌ Erreur traitement lien {url}: {e}") self.logger.debug(traceback.format_exc()) self.failed_links.append(link_data) # Délai entre les liens if i < len(links) - 1: delay = self.args.delay self.logger.debug(f"⏳ Pause de {delay}s avant le prochain lien") time.sleep(delay) # Sauvegarder les résultats finaux self.save_results() self.save_progress() def process_single_link(self, url, link_data): """Traite un seul lien d'emploi""" try: self.logger.debug(f"🌐 Navigation vers: {url}") self.driver.get(url) # Attendre le chargement time.sleep(2) # Vérifier les captchas issue = self.detect_captcha_advanced() if issue: result = self.wait_for_user_input(f"{issue} détecté sur {url}", issue) if result == 'skip': return 'skip' elif not result: return False # Attendre que la page soit chargée try: WebDriverWait(self.driver, 10).until( lambda d: d.execute_script("return document.readyState") == "complete" ) except TimeoutException: self.logger.warning("⏱️ Timeout chargement page") # Extraire les données de l'emploi job_data = self.extract_job_details() if job_data: job_data['source_url'] = url job_data['source_row'] = link_data.get('source_row', '') job_data['original_csv_data'] = link_data.get('original_data', {}) self.results.append(job_data) return True else: self.logger.warning("❌ Aucune donnée extraite") return False except TimeoutException: self.logger.error(f"⏱️ Timeout lors du chargement de {url}") return False except WebDriverException as e: self.logger.error(f"🌐 Erreur WebDriver pour {url}: {e}") return False except Exception as e: self.logger.error(f"❌ Erreur générale pour {url}: {e}") self.logger.debug(traceback.format_exc()) return False def extract_job_details(self): """Extrait les détails d'une offre d'emploi""" try: html = self.driver.page_source soup = BeautifulSoup(html, 'html.parser') self.logger.debug("🔍 Extraction des détails de l'emploi...") # Titre title_selectors = [ 'h1[data-cy="vacancy-title"]', 'h1.job-title', 'h1', '.vacancy-title', '[data-testid="job-title"]' ] title = self.find_text_by_selectors(soup, title_selectors) # Entreprise company_selectors = [ '[data-cy="vacancy-company-name"]', '.company-name', '.employer-name', 'h2 a', '.company' ] company = self.find_text_by_selectors(soup, company_selectors) # Lieu location_selectors = [ '[data-cy="vacancy-location"]', '.job-location', '.location', '.workplace' ] location = self.find_text_by_selectors(soup, location_selectors) # Description description_selectors = [ '[data-cy="vacancy-description"]', '.job-description', '.vacancy-description', '.description' ] description = self.find_text_by_selectors(soup, description_selectors) # Salaire salary_selectors = [ '[data-cy="vacancy-salary"]', '.salary', '.wage', '.compensation' ] salary = self.find_text_by_selectors(soup, salary_selectors) # Type de contrat contract_selectors = [ '[data-cy="vacancy-employment-type"]', '.employment-type', '.contract-type' ] contract_type = self.find_text_by_selectors(soup, contract_selectors) # Date de publication date_selectors = [ '[data-cy="vacancy-publication-date"]', '.publication-date', 'time', '.posted-date' ] pub_date = self.find_text_by_selectors(soup, date_selectors) job_data = { 'titre': title or "Titre non trouvé", 'entreprise': company or "Entreprise non trouvée", 'lieu': location or "Lieu non trouvé", 'description': description[:500] + "..." if description and len(description) > 500 else description or "", 'salaire': salary or "", 'type_contrat': contract_type or "", 'date_publication': pub_date or "", 'date_scraping': datetime.now().strftime("%Y-%m-%d %H:%M:%S"), 'url_courante': self.driver.current_url } self.logger.debug(f"✅ Données extraites: {job_data['titre']} - {job_data['entreprise']}") return job_data except Exception as e: self.logger.error(f"❌ Erreur extraction détails: {e}") self.logger.debug(traceback.format_exc()) return None def find_text_by_selectors(self, soup, selectors): """Trouve du texte en essayant plusieurs sélecteurs""" for selector in selectors: try: element = soup.select_one(selector) if element: text = element.get_text(strip=True) if text: return text except Exception as e: self.logger.debug(f"Erreur sélecteur {selector}: {e}") return "" def save_results(self): """Sauvegarde les résultats avec plus de détails""" timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") filename = f"jobs_scraping_detailed_{timestamp}.json" results_data = { 'metadata': { 'scraping_date': datetime.now().isoformat(), 'total_jobs_found': len(self.results), 'total_links_processed': len(self.processed_links), 'total_links_failed': len(self.failed_links), 'total_links_intended': self.total_links_to_process, 'captcha_encounters': self.captcha_detected_count, 'csv_source': self.args.csv_file or "/root/liens/1jobs_ch_complete_extraction.csv" }, 'jobs': self.results, 'failed_links': self.failed_links[:10], # Garder seulement les 10 premiers échecs 'statistics': { 'success_rate': round((len(self.processed_links) / max(self.total_links_to_process, 1)) * 100, 2), 'extraction_rate': round((len(self.results) / max(len(self.processed_links), 1)) * 100, 2) } } try: with open(filename, 'w', encoding='utf-8') as f: json.dump(results_data, f, indent=2, ensure_ascii=False) self.logger.info(f"💾 {len(self.results)} emplois sauvegardés dans: {filename}") self.logger.info(f"📊 Statistiques: {len(self.processed_links)}/{self.total_links_to_process} liens traités") self.update_status("Terminé avec succès", filename=filename) except Exception as e: self.logger.error(f"❌ Erreur sauvegarde: {e}") self.update_status(f"Erreur sauvegarde: {str(e)}") def save_progress(self): """Sauvegarde le progrès pour reprendre plus tard""" progress_file = 'scraping_progress.json' progress_data = { 'processed_links': self.processed_links, 'failed_links': self.failed_links, 'current_index': self.current_link_index, 'timestamp': datetime.now().isoformat() } try: with open(progress_file, 'w', encoding='utf-8') as f: json.dump(progress_data, f, indent=2, ensure_ascii=False) self.logger.info(f"💾 Progrès sauvegardé dans {progress_file}") except Exception as e: self.logger.error(f"❌ Erreur sauvegarde progrès: {e}") def cleanup(self): """Nettoyage final avec logs détaillés""" if self.driver: self.driver.quit() self.logger.info("🔚 Navigateur fermé") # Résumé final self.logger.info("="*60) self.logger.info("📊 RÉSUMÉ FINAL:") self.logger.info(f"✅ Emplois trouvés: {len(self.results)}") self.logger.info(f"🔗 Liens traités: {len(self.processed_links)}") self.logger.info(f"❌ Liens échoués: {len(self.failed_links)}") self.logger.info(f"🛡️ Captchas rencontrés: {self.captcha_detected_count}") self.logger.info("="*60) def main(): parser = argparse.ArgumentParser(description='Scraper Jobs.ch avec lecture CSV et gestion avancée des captchas') parser.add_argument('--csv-file', '-c', help='Chemin vers le fichier CSV contenant les liens (défaut: /root/liens/1jobs_ch_complete_extraction.csv)') parser.add_argument('--max-links', type=int, help='Nombre maximum de liens à traiter') parser.add_argument('--delay', type=int, default=3, help='Délai entre les liens en secondes (défaut: 3)') parser.add_argument('--headless', action='store_true', help='Mode headless (sans interface graphique)') parser.add_argument('--debug', action='store_true', help='Mode debug avec logs détaillés') parser.add_argument('--auto-retry', action='store_true', help='Tentative de contournement automatique des captchas en mode headless') parser.add_argument('--output', '-o', help='Fichier de sortie (optionnel)') args = parser.parse_args() print("🚀 Jobs.ch Advanced CSV Scraper") print("=" * 50) scraper = None try: scraper = JobsChScraperCLI(args) scraper.scrape_from_links() except KeyboardInterrupt: print("\n🛑 Interruption par l'utilisateur") except Exception as e: print(f"❌ Erreur fatale: {e}") if scraper: scraper.logger.error(f"Erreur fatale: {e}") scraper.logger.debug(traceback.format_exc()) finally: if scraper: scraper.cleanup() if __name__ == "__main__": main()