Update src/agents/scoring_agent.py

src/agents/scoring_agent.py

@@ -6,22 +6,31 @@ from typing import Dict, List, Any
 
 logger = logging.getLogger(__name__)
 
-class ScoringAgent:
-    ALPHA = 0.5  
-    BETA = 0.3    
-    GAMMA = 0.2  
-
-    CONTEXT_VALUES = {
-        "formations": 0.3,
-        "projets": 0.6,
-        "experiences_professionnelles": 0.8,
-    }
-
+class SimpleScoringAgent:
+    
     def calculate_scores(self, candidat_data: Dict[str, Any]) -> Dict[str, List[Dict[str, Any]]]:
         if not candidat_data or not isinstance(candidat_data, dict):
             return {"analyse_competences": []}
             
         skills_data = candidat_data.get("compétences", {})
+        skills_list = self._extract_skills_list(skills_data)
+        
+        if not skills_list:
+            return {"analyse_competences": []}
+
+        skill_analysis = []
+        
+        for skill in skills_list:
+            level = self._determine_skill_level(skill, candidat_data)
+            skill_analysis.append({
+                "skill": skill,
+                "level": level
+            })
+        
+        return {"analyse_competences": skill_analysis}
+
+    def _extract_skills_list(self, skills_data: Dict[str, Any]) -> List[str]:
+        """Extrait la liste des compétences"""
         skills_list = []
         
         if isinstance(skills_data, dict):
@@ -30,36 +39,46 @@ class ScoringAgent:
         elif isinstance(skills_data, list):
             skills_list = [item.get("nom") for item in skills_data if item.get("nom")]
         
-        # Filtrer les skills valides
-        skills_list = [skill for skill in skills_list if skill and isinstance(skill, str) and skill.strip()]
-        
-        if not skills_list:
-            return {"analyse_competences": []}
-
-        skill_metrics = {
-            skill.lower(): {
-                "original_name": skill,
-                "contexts": set(),
-                "frequency": 0,
-                "max_duration": 0.0
-            }
-            for skill in skills_list if skill
-        }
+        return [skill for skill in skills_list if skill and isinstance(skill, str) and skill.strip()]
 
-        self._analyze_experiences(candidat_data, skill_metrics)
-        self._analyze_projects(candidat_data, skill_metrics)
-        self._analyze_formations(candidat_data, skill_metrics)
+    def _determine_skill_level(self, skill: str, candidat_data: Dict[str, Any]) -> str:
+        """Détermine le niveau d'une compétence selon des règles simples"""
+        
+        frequency = self._count_skill_mentions(skill, candidat_data)
+        max_duration = self._get_max_duration_for_skill(skill, candidat_data)
+        has_pro_experience = self._has_professional_experience(skill, candidat_data)
+        
+        # Règles simples de classification
+        if has_pro_experience and max_duration >= 3.0:
+            return "expert"
+        elif has_pro_experience and max_duration >= 1.0:
+            return "avance"
+        elif frequency >= 3 or max_duration >= 0.5:
+            return "intermediaire"
+        else:
+            return "debutant"
 
-        final_scores = self._calculate_final_scores(skill_metrics)
+    def _count_skill_mentions(self, skill: str, candidat_data: Dict[str, Any]) -> int:
+        """Compte le nombre de mentions de la compétence"""
+        skill_lower = skill.lower()
+        total_mentions = 0
         
-        return {"analyse_competences": final_scores}
+        # Recherche dans toutes les sections
+        all_text = self._get_all_text_content(candidat_data).lower()
+        total_mentions = all_text.count(skill_lower)
+        
+        return total_mentions
 
-    def _analyze_experiences(self, candidat_data: Dict[str, Any], skill_metrics: Dict[str, Any]):
+    def _get_max_duration_for_skill(self, skill: str, candidat_data: Dict[str, Any]) -> float:
+        """Trouve la durée maximum d'utilisation de la compétence"""
+        skill_lower = skill.lower()
+        max_duration = 0.0
+        
         experiences_key = "expériences" if "expériences" in candidat_data else "experiences_professionnelles"
         experiences = candidat_data.get(experiences_key, [])
         
         if not isinstance(experiences, list):
-            return
+            return 0.0
         
         for exp in experiences:
             if not isinstance(exp, dict):
@@ -67,85 +86,83 @@ class ScoringAgent:
                 
             exp_text = json.dumps(exp, ensure_ascii=False).lower()
             
-            # Récupération sécurisée des dates
-            start_date_str = exp.get("date_debut", exp.get("start_date", ""))
-            end_date_str = exp.get("date_fin", exp.get("end_date", ""))
-            
-            # S'assurer que ce sont des strings
-            if not isinstance(start_date_str, str):
-                start_date_str = str(start_date_str) if start_date_str else ""
-            if not isinstance(end_date_str, str):
-                end_date_str = str(end_date_str) if end_date_str else ""
-            
-            duration = self._calculate_duration_in_years(start_date_str, end_date_str)
-            
-            for skill in skill_metrics:
-                if skill in exp_text:
-                    skill_metrics[skill]["contexts"].add("experiences_professionnelles")
-                    skill_metrics[skill]["frequency"] += exp_text.count(skill)
-                    if duration > skill_metrics[skill]["max_duration"]:
-                        skill_metrics[skill]["max_duration"] = duration
-
-    def _analyze_projects(self, candidat_data: Dict[str, Any], skill_metrics: Dict[str, Any]):
-        projects_data = candidat_data.get("projets", {})
+            if skill_lower in exp_text:
+                duration = self._calculate_experience_duration(exp)
+                max_duration = max(max_duration, duration)
         
-        if isinstance(projects_data, dict):
-            for project_type in ["professional", "personal"]:
-                for project in projects_data.get(project_type, []):
-                    project_text = json.dumps(project, ensure_ascii=False).lower()
-                    for skill in skill_metrics:
-                        if skill in project_text:
-                            skill_metrics[skill]["contexts"].add("projets")
-                            skill_metrics[skill]["frequency"] += project_text.count(skill)
+        return max_duration
 
-    def _analyze_formations(self, candidat_data: Dict[str, Any], skill_metrics: Dict[str, Any]):
-        for formation in candidat_data.get("formations", []):
-            formation_text = json.dumps(formation, ensure_ascii=False).lower()
-            for skill in skill_metrics:
-                if skill in formation_text:
-                    skill_metrics[skill]["contexts"].add("formations")
-                    skill_metrics[skill]["frequency"] += formation_text.count(skill)
-
-    def _calculate_final_scores(self, skill_metrics: Dict[str, Any]) -> List[Dict[str, Any]]:
-        final_scores = []
+    def _has_professional_experience(self, skill: str, candidat_data: Dict[str, Any]) -> bool:
+        """Vérifie si la compétence a été utilisée en contexte professionnel"""
+        skill_lower = skill.lower()
+        
+        experiences_key = "expériences" if "expériences" in candidat_data else "experiences_professionnelles"
+        experiences = candidat_data.get(experiences_key, [])
+        
+        if not isinstance(experiences, list):
+            return False
         
-        for skill, metrics in skill_metrics.items():
-            if metrics["frequency"] == 0:
+        for exp in experiences:
+            if not isinstance(exp, dict):
                 continue
-            
-            context_score = max(
-                (self.CONTEXT_VALUES.get(c, 0) for c in metrics["contexts"]), 
-                default=0.1
-            )
-            
-            if len(metrics["contexts"]) > 1:
-                context_score = 1.0
-
-            normalized_frequency = self._normalize_score(metrics["frequency"])
-            normalized_depth = self._normalize_score(metrics["max_duration"])
-
-            final_score = (self.ALPHA * context_score) + \
-                          (self.BETA * normalized_frequency) + \
-                          (self.GAMMA * normalized_depth)
+                
+            exp_text = json.dumps(exp, ensure_ascii=False).lower()
+            if skill_lower in exp_text:
+                return True
+        
+        return False
 
-            final_scores.append({
-                "skill": metrics["original_name"],
-                "score": round(final_score, 2),
-                "details": {
-                    "context_score": round(context_score, 2),
-                    "contexts_found": list(metrics["contexts"]),
-                    "frequency": metrics["frequency"],
-                    "max_duration_years": round(metrics["max_duration"], 1)
-                }
-            })
+    def _get_all_text_content(self, candidat_data: Dict[str, Any]) -> str:
+        """Récupère tout le contenu textuel du CV"""
+        all_content = []
+        
+        # Expériences
+        experiences_key = "expériences" if "expériences" in candidat_data else "experiences_professionnelles"
+        for exp in candidat_data.get(experiences_key, []):
+            if isinstance(exp, dict):
+                all_content.append(json.dumps(exp, ensure_ascii=False))
+        
+        # Projets
+        projects = candidat_data.get("projets", {})
+        if isinstance(projects, dict):
+            for project_type in ["professional", "personal"]:
+                for project in projects.get(project_type, []):
+                    if isinstance(project, dict):
+                        all_content.append(json.dumps(project, ensure_ascii=False))
+        
+        # Formations
+        for formation in candidat_data.get("formations", []):
+            if isinstance(formation, dict):
+                all_content.append(json.dumps(formation, ensure_ascii=False))
+        
+        return " ".join(all_content)
 
-        final_scores.sort(key=lambda x: x["score"], reverse=True)
-        return final_scores
+    def _calculate_experience_duration(self, exp: Dict[str, Any]) -> float:
+        """Calcule la durée d'une expérience en années"""
+        start_date_str = exp.get("date_debut", exp.get("start_date", ""))
+        end_date_str = exp.get("date_fin", exp.get("end_date", ""))
+        
+        if not isinstance(start_date_str, str):
+            start_date_str = str(start_date_str) if start_date_str else ""
+        if not isinstance(end_date_str, str):
+            end_date_str = str(end_date_str) if end_date_str else ""
+        
+        return self._calculate_duration_in_years(start_date_str, end_date_str)
 
-    def _normalize_score(self, value: float) -> float:
-        return 1 - (1 / (1 + float(value)))
+    def _calculate_duration_in_years(self, start_date_str: str, end_date_str: str) -> float:
+        """Calcule la durée entre deux dates en années"""
+        start_date = self._parse_date(start_date_str)
+        end_date = self._parse_date(end_date_str)
+        
+        if start_date and end_date:
+            if end_date < start_date:
+                return 0.0
+            return (end_date - start_date).days / 365.25
+        
+        return 0.0
 
     def _parse_date(self, date_str: str) -> datetime:
+        """Parse une date de manière simple"""
         if not date_str or not isinstance(date_str, str):
             return None
             
@@ -153,15 +170,7 @@ class ScoringAgent:
         if date_str_lower in ["aujourd'hui", "maintenant", "en cours", "current", "présent", "actuellement"]:
             return datetime.now()
         
-        # Nettoyer la chaîne de date
-        date_str_clean = date_str.strip()
-        
-        # Validation préalable avant parsing
-        validated_date = self._validate_and_parse_date(date_str_clean)
-        if validated_date:
-            return validated_date
-        
-        # Tentative d'extraction de l'année seulement
+        # Extraction simple de l'année
         year_match = re.search(r'\b(20\d{2}|19\d{2})\b', date_str)
         if year_match:
             year = int(year_match.group(1))
@@ -169,45 +178,6 @@ class ScoringAgent:
         
         return None
 
-    def _validate_and_parse_date(self, date_str: str) -> datetime:
-        """Valide et parse une date en vérifiant le format avant parsing"""
-        
-        # Format YYYY
-        if re.match(r'^\d{4}$', date_str):
-            year = int(date_str)
-            if 1900 <= year <= 2030:
-                return datetime(year, 1, 1)
-        
-        # Format MM/YYYY
-        if re.match(r'^\d{1,2}/\d{4}$', date_str):
-            parts = date_str.split('/')
-            month, year = int(parts[0]), int(parts[1])
-            if 1 <= month <= 12 and 1900 <= year <= 2030:
-                return datetime(year, month, 1)
-        
-        # Format YYYY-MM
-        if re.match(r'^\d{4}-\d{1,2}$', date_str):
-            parts = date_str.split('-')
-            year, month = int(parts[0]), int(parts[1])
-            if 1 <= month <= 12 and 1900 <= year <= 2030:
-                return datetime(year, month, 1)
-        
-        # Format DD/MM/YYYY
-        if re.match(r'^\d{1,2}/\d{1,2}/\d{4}$', date_str):
-            parts = date_str.split('/')
-            day, month, year = int(parts[0]), int(parts[1]), int(parts[2])
-            if 1 <= day <= 31 and 1 <= month <= 12 and 1900 <= year <= 2030:
-                return datetime(year, month, day)
-        
-        return None
-
-    def _calculate_duration_in_years(self, start_date_str: str, end_date_str: str) -> float:
-        start_date = self._parse_date(start_date_str)
-        end_date = self._parse_date(end_date_str)
-        
-        if start_date and end_date:
-            if end_date < start_date:
-                return 0.0
-            return (end_date - start_date).days / 365.25
-        
-        return 0.0
+# Alias pour maintenir la compatibilité
+ScoringAgent = SimpleScoringAgent
+ImprovedScoringAgent = SimpleScoringAgent