#!/usr/bin/env python3 """ Analyseur avancé des résultats de scraping Jobs.ch Génère des statistiques, détecte les tendances et produit des rapports """ import json import pandas as pd import re import argparse from datetime import datetime, timedelta from collections import Counter, defaultdict from pathlib import Path import matplotlib matplotlib.use('Agg') # Backend sans GUI import matplotlib.pyplot as plt import seaborn as sns class JobsResultsAnalyzer: def __init__(self, results_file=None): self.results_file = results_file self.data = None self.df = None self.analysis_results = {} def load_data(self, file_path=None): """Charger les données depuis un fichier JSON""" if file_path: self.results_file = file_path if not self.results_file: # Chercher le fichier le plus récent json_files = list(Path('.').glob('jobs_scraping_*.json')) if not json_files: raise FileNotFoundError("Aucun fichier de résultats trouvé") self.results_file = max(json_files) print(f"📂 Chargement du fichier: {self.results_file}") with open(self.results_file, 'r', encoding='utf-8') as f: self.data = json.load(f) # Déterminer le format du fichier if isinstance(self.data, dict) and 'jobs' in self.data: # Format détaillé jobs = self.data['jobs'] self.metadata = self.data.get('metadata', {}) self.statistics = self.data.get('statistics', {}) else: # Format legacy jobs = self.data if isinstance(self.data, list) else [] self.metadata = {} self.statistics = {} # Convertir en DataFrame self.df = pd.DataFrame(jobs) if self.df.empty: raise ValueError("Aucune donnée d'emploi trouvée") print(f"✅ {len(self.df)} emplois chargés") return self.df def clean_and_prepare_data(self): """Nettoyer et préparer les données pour l'analyse""" if self.df is None: raise ValueError("Données non chargées") print("🧹 Nettoyage des données...") # Nettoyer les titres self.df['titre_clean'] = self.df['titre'].astype(str).str.strip() self.df['titre_clean'] = self.df['titre_clean'].str.replace(r'\s+', ' ', regex=True) # Nettoyer les entreprises self.df['entreprise_clean'] = self.df['entreprise'].astype(str).str.strip() self.df['entreprise_clean'] = self.df['entreprise_clean'].str.replace(r'\s+', ' ', regex=True) # Nettoyer les lieux self.df['lieu_clean'] = self.df['lieu'].astype(str).str.strip() self.df['lieu_clean'] = self.df['lieu_clean'].str.replace(r'\s+', ' ', regex=True) # Parser les dates if 'date_scraping' in self.df.columns: self.df['date_scraping_parsed'] = pd.to_datetime( self.df['date_scraping'], errors='coerce' ) # Extraire les technologies du titre et de la description self.df['technologies'] = self.df.apply(self._extract_technologies, axis=1) # Extraire les niveaux d'expérience self.df['niveau_experience'] = self.df.apply(self._extract_experience_level, axis=1) # Nettoyer les salaires self.df['salaire_clean'] = self.df['salaire'].astype(str).apply(self._clean_salary) print("✅ Données nettoyées") def _extract_technologies(self, row): """Extraire les technologies mentionnées""" text = f"{row.get('titre', '')} {row.get('description', '')}".lower() technologies = { 'python': ['python', 'django', 'flask', 'fastapi', 'pandas', 'numpy'], 'javascript': ['javascript', 'js', 'node.js', 'nodejs', 'react', 'vue', 'angular'], 'java': ['java', 'spring', 'hibernate', 'maven', 'gradle'], 'php': ['php', 'laravel', 'symfony', 'codeigniter', 'wordpress'], 'dotnet': ['.net', 'c#', 'asp.net', 'entity framework'], 'database': ['sql', 'mysql', 'postgresql', 'mongodb', 'oracle', 'sqlite'], 'cloud': ['aws', 'azure', 'gcp', 'docker', 'kubernetes', 'terraform'], 'frontend': ['html', 'css', 'bootstrap', 'tailwind', 'sass', 'less'], 'mobile': ['android', 'ios', 'react native', 'flutter', 'swift', 'kotlin'], 'devops': ['devops', 'ci/cd', 'jenkins', 'gitlab', 'github actions'] } found_techs = [] for category, tech_list in technologies.items(): for tech in tech_list: if tech in text: found_techs.append(category) break return found_techs def _extract_experience_level(self, row): """Extraire le niveau d'expérience requis""" text = f"{row.get('titre', '')} {row.get('description', '')}".lower() patterns = { 'junior': ['junior', 'débutant', '0-2 ans', 'entry level'], 'senior': ['senior', 'expérimenté', '5+ ans', '7+ ans', '10+ ans'], 'lead': ['lead', 'chef', 'responsable', 'manager', 'directeur'], 'architect': ['architecte', 'architect', 'principal'], } for level, keywords in patterns.items(): if any(keyword in text for keyword in keywords): return level return 'mid-level' # Par défaut def _clean_salary(self, salary_str): """Nettoyer et normaliser les informations de salaire""" if pd.isna(salary_str) or salary_str == '': return None salary_str = str(salary_str).lower() # Extraire les montants numériques numbers = re.findall(r'(\d+(?:\.\d+)?)', salary_str.replace(' ', '').replace(',', '.')) if not numbers: return salary_str # Détecter le type de salaire if 'mensuel' in salary_str or '/mois' in salary_str: return f"{numbers[0]} CHF/mois" elif 'annuel' in salary_str or '/an' in salary_str: return f"{numbers[0]} CHF/an" elif 'heure' in salary_str or '/h' in salary_str: return f"{numbers[0]} CHF/h" else: return salary_str def analyze_companies(self): """Analyser les entreprises""" print("🏢 Analyse des entreprises...") # Top entreprises company_counts = self.df['entreprise_clean'].value_counts().head(20) # Filtrer les entreprises invalides valid_companies = company_counts[ ~company_counts.index.isin([ 'Entreprise non trouvée', 'Non spécifiée', '', 'nan' ]) ] self.analysis_results['companies'] = { 'top_companies': valid_companies.to_dict(), 'total_unique': len(self.df['entreprise_clean'].unique()), 'companies_with_multiple_jobs': len(company_counts[company_counts > 1]), } return self.analysis_results['companies'] def analyze_locations(self): """Analyser les localisations""" print("📍 Analyse des localisations...") # Nettoyer et grouper les lieux locations = self.df['lieu_clean'].str.split(',').str[0] # Prendre la ville principale location_counts = locations.value_counts().head(15) # Filtrer les lieux invalides valid_locations = location_counts[ ~location_counts.index.isin([ 'Lieu non trouvé', 'Non spécifié', '', 'nan' ]) ] self.analysis_results['locations'] = { 'top_locations': valid_locations.to_dict(), 'total_unique': len(locations.unique()), 'remote_jobs': len(self.df[ self.df['lieu_clean'].str.lower().str.contains('remote|télétravail|home', na=False) ]) } return self.analysis_results['locations'] def analyze_technologies(self): """Analyser les technologies demandées""" print("💻 Analyse des technologies...") all_technologies = [] for tech_list in self.df['technologies']: if isinstance(tech_list, list): all_technologies.extend(tech_list) tech_counts = Counter(all_technologies) self.analysis_results['technologies'] = { 'top_technologies': dict(tech_counts.most_common(15)), 'jobs_with_tech': len(self.df[self.df['technologies'].astype(str) != '[]']), 'avg_tech_per_job': len(all_technologies) / len(self.df) if len(self.df) > 0 else 0 } return self.analysis_results['technologies'] def analyze_experience_levels(self): """Analyser les niveaux d'expérience""" print("👥 Analyse des niveaux d'expérience...") exp_counts = self.df['niveau_experience'].value_counts() self.analysis_results['experience'] = { 'distribution': exp_counts.to_dict(), 'most_common': exp_counts.index[0] if not exp_counts.empty else None } return self.analysis_results['experience'] def analyze_salary_info(self): """Analyser les informations de salaire""" print("💰 Analyse des salaires...") salary_info = self.df[self.df['salaire_clean'].notna()] self.analysis_results['salary'] = { 'jobs_with_salary': len(salary_info), 'percentage_with_salary': (len(salary_info) / len(self.df)) * 100 if len(self.df) > 0 else 0, 'salary_types': salary_info['salaire_clean'].value_counts().head(10).to_dict() } return self.analysis_results['salary'] def analyze_posting_trends(self): """Analyser les tendances de publication""" print("📈 Analyse des tendances temporelles...") if 'date_scraping_parsed' not in self.df.columns: return {'error': 'Pas de données temporelles disponibles'} # Grouper par jour daily_counts = self.df['date_scraping_parsed'].dt.date.value_counts().sort_index() # Grouper par heure hourly_counts = self.df['date_scraping_parsed'].dt.hour.value_counts().sort_index() self.analysis_results['trends'] = { 'posts_by_day': daily_counts.to_dict(), 'posts_by_hour': hourly_counts.to_dict(), 'total_days': len(daily_counts), 'avg_posts_per_day': daily_counts.mean() if not daily_counts.empty else 0 } return self.analysis_results['trends'] def detect_patterns(self): """Détecter des patterns intéressants""" print("🔍 Détection de patterns...") patterns = {} # Entreprises qui recrutent le plus top_hiring_companies = self.df['entreprise_clean'].value_counts().head(5) patterns['top_hiring'] = top_hiring_companies.to_dict() # Corrélations technologie-lieu tech_location_pairs = [] for _, row in self.df.iterrows(): techs = row.get('technologies', []) location = row.get('lieu_clean', '') if isinstance(techs, list) and location: for tech in techs: tech_location_pairs.append((tech, location.split(',')[0])) # Mots-clés populaires dans les titres all_titles = ' '.join(self.df['titre_clean'].astype(str)).lower() words = re.findall(r'\b\w{4,}\b', all_titles) # Mots de 4+ caractères common_words = Counter(words).most_common(20) patterns['title_keywords'] = dict(common_words) self.analysis_results['patterns'] = patterns return patterns def generate_visualizations(self, output_dir='exports'): """Générer des visualisations""" print("📊 Génération des visualisations...") output_path = Path(output_dir) output_path.mkdir(exist_ok=True) plt.style.use('seaborn-v0_8') # Style moderne try: # 1. Top entreprises if 'companies' in self.analysis_results: companies = self.analysis_results['companies']['top_companies'] if companies: plt.figure(figsize=(12, 8)) companies_df = pd.Series(companies).head(10) companies_df.plot(kind='barh') plt.title('Top 10 des entreprises qui recrutent') plt.xlabel('Nombre d\'offres') plt.tight_layout() plt.savefig(output_path / 'top_companies.png', dpi=300, bbox_inches='tight') plt.close() # 2. Répartition géographique if 'locations' in self.analysis_results: locations = self.analysis_results['locations']['top_locations'] if locations: plt.figure(figsize=(10, 8)) locations_df = pd.Series(locations).head(10) locations_df.plot(kind='pie', autopct='%1.1f%%') plt.title('Répartition géographique des offres') plt.ylabel('') plt.tight_layout() plt.savefig(output_path / 'locations_distribution.png', dpi=300, bbox_inches='tight') plt.close() # 3. Technologies demandées if 'technologies' in self.analysis_results: technologies = self.analysis_results['technologies']['top_technologies'] if technologies: plt.figure(figsize=(12, 8)) tech_df = pd.Series(technologies) tech_df.plot(kind='bar', color='skyblue') plt.title('Technologies les plus demandées') plt.xlabel('Technologies') plt.ylabel('Nombre d\'offres') plt.xticks(rotation=45) plt.tight_layout() plt.savefig(output_path / 'technologies.png', dpi=300, bbox_inches='tight') plt.close() # 4. Niveaux d'expérience if 'experience' in self.analysis_results: experience = self.analysis_results['experience']['distribution'] if experience: plt.figure(figsize=(8, 6)) exp_df = pd.Series(experience) exp_df.plot(kind='bar', color='lightcoral') plt.title('Répartition par niveau d\'expérience') plt.xlabel('Niveau') plt.ylabel('Nombre d\'offres') plt.xticks(rotation=45) plt.tight_layout() plt.savefig(output_path / 'experience_levels.png', dpi=300, bbox_inches='tight') plt.close() print(f"✅ Visualisations sauvegardées dans {output_path}/") except Exception as e: print(f"⚠️ Erreur génération visualisations: {e}") def generate_report(self, output_file=None): """Générer un rapport complet""" if not output_file: timestamp = datetime.now().strftime('%Y%m%d_%H%M%S') output_file = f'exports/analysis_report_{timestamp}.json' Path(output_file).parent.mkdir(exist_ok=True) report = { 'report_info': { 'generated_at': datetime.now().isoformat(), 'source_file': str(self.results_file), 'total_jobs_analyzed': len(self.df) if self.df is not None else 0, 'analysis_version': '1.0' }, 'data_quality': { 'jobs_with_title': len(self.df[self.df['titre_clean'].notna()]) if self.df is not None else 0, 'jobs_with_company': len(self.df[self.df['entreprise_clean'].notna()]) if self.df is not None else 0, 'jobs_with_location': len(self.df[self.df['lieu_clean'].notna()]) if self.df is not None else 0, 'jobs_with_description': len(self.df[self.df['description'].notna()]) if self.df is not None else 0, }, 'analysis_results': self.analysis_results, 'metadata': self.metadata, 'recommendations': self._generate_recommendations() } with open(output_file, 'w', encoding='utf-8') as f: json.dump(report, f, indent=2, ensure_ascii=False) print(f"📄 Rapport sauvegardé: {output_file}") return report def _generate_recommendations(self): """Générer des recommandations basées sur l'analyse""" recommendations = [] if 'technologies' in self.analysis_results: top_tech = list(self.analysis_results['technologies']['top_technologies'].keys()) if top_tech: recommendations.append( f"Les technologies les plus demandées sont {', '.join(top_tech[:3])}. " "Considérez l'acquisition de ces compétences." ) if 'locations' in self.analysis_results: top_locations = list(self.analysis_results['locations']['top_locations'].keys()) if top_locations: recommendations.append( f"Les principales zones de recrutement sont {', '.join(top_locations[:3])}." ) if 'companies' in self.analysis_results: top_companies = list(self.analysis_results['companies']['top_companies'].keys()) if top_companies: recommendations.append( f"Les entreprises qui recrutent le plus sont {', '.join(top_companies[:3])}." ) return recommendations def run_full_analysis(self, generate_viz=True): """Exécuter l'analyse complète""" print("🚀 Démarrage de l'analyse complète...") # Charger et nettoyer les données self.load_data() self.clean_and_prepare_data() # Exécuter toutes les analyses self.analyze_companies() self.analyze_locations() self.analyze_technologies() self.analyze_experience_levels() self.analyze_salary_info() self.analyze_posting_trends() self.detect_patterns() # Générer les visualisations if generate_viz: self.generate_visualizations() # Générer le rapport report = self.generate_report() print("✅ Analyse terminée!") return report def main(): parser = argparse.ArgumentParser(description='Analyseur de résultats de scraping Jobs.ch') parser.add_argument('--file', '-f', help='Fichier JSON à analyser') parser.add_argument('--output', '-o', help='Répertoire de sortie', default='exports') parser.add_argument('--no-viz', action='store_true', help='Ne pas générer les visualisations') parser.add_argument('--report-only', action='store_true', help='Générer seulement le rapport JSON') args = parser.parse_args() try: analyzer = JobsResultsAnalyzer(args.file) if args.report_only: analyzer.load_data() analyzer.clean_and_prepare_data() report = analyzer.generate_report() print(json.dumps(report, indent=2, ensure_ascii=False)) else: analyzer.run_full_analysis(generate_viz=not args.no_viz) except Exception as e: print(f"❌ Erreur lors de l'analyse: {e}") return 1 return 0 if __name__ == "__main__": exit(main())