Update app.py

app.py

@@ -160,6 +160,207 @@ def classify_occupational_group_by_level(responsibilities: List[str]) -> dict:
         result['error'] = str(e)
     return result
 
+
+ 
+def classify_esco_by_hierarchical_level(responsibilities: List[str]) -> dict:
+    """
+    Classifies job responsibilities into occupational groups at 4 levels,
+    [European Skills, Competences, Qualifications, and Occupations (ESCO)](https://esco.ec.europa.eu/en)
+    returning codes, names, and descriptions for each level.
+    Args:
+        responsibilities: List of job responsibility strings
+    Returns:
+        Dictionary containing classification information or error message
+    """
+
+    esco_df = pd.read_csv(
+        "ISCOGroups_en.csv",
+        dtype={'code': str}  # Force 'code' to be read as string
+    )
+
+    esco_level5_df = pd.read_csv(
+        "occupations_en.csv",
+        dtype={'code': str, 'iscoGroup': str, }  # Force 'code' to be read as string
+    )
+
+    result = {}
+
+    ######################## Level 1 ###################
+    # Get all top-level codes (single character/digit)
+    top_level_codes = sorted({
+        code for code in esco_df['code']
+        if len(code) == 1 and code.isalnum()
+    })
+
+    level1_code = None
+    if top_level_codes:
+        level1_df = esco_df[esco_df['code'].isin(top_level_codes)]
+        job_occupation_list = "\n".join(f"- {row['code']}: {row['preferredLabel']} - {row['description']}"
+                                        for _, row in level1_df.iterrows())
+        list1_output = level1_df["code"].tolist()  # Convert Series to list
+        list1 = ", ".join(map(str, list1_output))  # Join elements with comma
+
+    user_prompt1 = f"""
+    Here is a list of job responsibilities:
+    {responsibilities}
+
+    Select the most relevant top-level code from these options:
+    {job_occupation_list}
+
+    Based on the responsibilities, suggest the most relevant level 1 Occupation code from within this list: {list1}.
+        **Important:**
+        - Return ONLY the code, nothing else.
+        - The code should be exactly as shown in the list.
+        - Do not include any additional text or explanation.
+    """
+    level1_code = gpt_call("Identify top-level occupational group", user_prompt1).strip()
+    level1_code = code_sanitize(level1_code, list1_output)
+    result.update(get_level_ESCO_info(level1_df, level1_code, 'Level_1'))
+
+    ######################## Level 2 ###################
+
+    level2_code = None
+    if level1_code:
+        level2_df = esco_df[
+            (esco_df['code'].str.startswith(level1_code)) & (esco_df['code'].str.len() == len(level1_code) + 1)
+        ]
+        if not level2_df.empty:
+            level2_options = "\n".join(f"- {row['code']}: {row['preferredLabel']} - {row['description']}"
+                                        for _, row in level2_df.iterrows())
+        list2_output = level2_df["code"].tolist()  # Convert Series to list
+        list2 = ", ".join(map(str, list2_output))  # Join elements with comma
+
+        user_prompt2 = f"""
+    Here is a list of job responsibilities:
+    {responsibilities}
+
+    Here is a list of level 2 Occupation classifications within {level1_code}:
+    {level2_options}
+
+    Based on the responsibilities, suggest the most relevant level 2 Occupation code from within this list: {list2}.
+        **Important:**
+        - Return ONLY the code, nothing else.
+        - The code should be exactly as shown in the list.
+        - Do not include any additional text or explanation.
+    """
+        level2_code = gpt_call("Identify second-level occupational group", user_prompt2).strip()
+        level2_code = code_sanitize(level2_code, list2_output)
+        result.update(get_level_ESCO_info(level2_df, level2_code, 'Level_2'))
+
+    ######################## Level 3 ###################
+    level3_code = None
+    if level2_code:
+        level3_df = esco_df[
+            (esco_df['code'].str.startswith(level2_code)) & (esco_df['code'].str.len() == len(level2_code) + 1)
+        ]
+        if not level3_df.empty:
+            level3_options = "\n".join(f"- {row['code']}: {row['preferredLabel']} - {row['description']}"
+                                        for _, row in level3_df.iterrows())
+        list3_output = level3_df["code"].tolist()  # Convert Series to list
+        list3 = ", ".join(map(str, list3_output))  # Join elements with comma
+
+        user_prompt3 = f"""
+    Here is a list of job responsibilities:
+    {responsibilities}
+
+    Here is a list of level 3 Occupation classifications within {level2_code}:
+    {level3_options}
+
+    Based on the responsibilities, suggest the most relevant level 3 Occupation code from within this list: {list3}.
+
+        **Important:**
+        - Return ONLY the code, nothing else.
+        - The code should be exactly as shown in the list.
+        - Do not include any additional text or explanation.
+
+    """
+        level3_code = gpt_call("Identify third-level occupational group", user_prompt3).strip()
+        level3_code = code_sanitize(level3_code, list3_output)
+        result.update(get_level_ESCO_info(level3_df, level3_code, 'Level_3'))
+
+    ######################## Level 4 ###################
+    level4_code = None
+    if level3_code:
+        level4_df = esco_df[
+            (esco_df['code'].str.startswith(level3_code)) & (esco_df['code'].str.len() == len(level3_code) + 1)
+        ]
+        if not level4_df.empty:
+            level4_options = "\n".join(f"- {row['code']}: {row['preferredLabel']} - {row['description']}"
+                                        for _, row in level4_df.iterrows())
+        list4_output = level4_df["code"].tolist()  # Convert Series to list
+        list4 = ", ".join(map(str, list4_output))  # Join elements with comma
+        user_prompt4 = f"""
+    Here is a list of job responsibilities:
+    {responsibilities}
+
+    Here is a list of level 4 Occupation classifications within {level3_code}:
+    {level4_options}
+
+    Based on the responsibilities, suggest the most relevant level 4 Occupation code from within this list: {list4}.
+        **Important:**
+        - Return ONLY the code, nothing else.
+        - The code should be exactly as shown in the list.
+        - Do not include any additional text or explanation.
+    """
+
+        level4_code = gpt_call("Identify fourth-level occupational group", user_prompt4).strip()
+        level4_code = code_sanitize(level4_code, list4_output)
+        result.update(get_level_ESCO_info(level4_df, level4_code, 'Level_4'))
+
+    ######################## Level 5 ###################
+    level5_code = None
+    if level4_code:
+        level5_df = esco_level5_df[
+            (esco_level5_df['iscoGroup'].str.startswith(level4_code))
+        ]
+        if not level5_df.empty:
+            level5_options = "\n".join(f"- {row['code']}: {row['preferredLabel']} - {row['description']}"
+                                        for _, row in level5_df.iterrows())
+
+        list5_output = level5_df["code"].tolist()  # Convert Series to list
+        list5 = ", ".join(map(str, list5_output))  # Join elements with comma
+        user_prompt5 = f"""
+    Here is a list of job responsibilities:
+    {responsibilities}
+
+    Here is a list of level 4 Occupation classifications within {level4_code}:
+    {level5_options}
+
+    Based on the responsibilities, suggest the most relevant level 4 Occupation code from within this list: {list5}.
+        **Important:**
+        - Return ONLY the code as stated in the provided list, nothing else.
+        - The code should be exactly as shown in the list.
+        - Do not include any additional text, occupation code or explanation.
+    """
+
+        level5_code = gpt_call("Identify fifth-level occupational group", user_prompt5).strip()
+        # Handle the case where the LLM might return just the code part
+        level5_code = code_sanitize(level5_code, list5_output)
+        result.update(get_level_ESCO_info(level5_df, level5_code, 'Level_5'))
+
+    ## Et voila!!
+    return result
+
+
+
+def get_level_ESCO_info(df, code, level_name):
+    """Helper function to get level info with error handling"""
+    matches = df[df['code'] == code]
+    if len(matches) == 0:
+        print(f"Warning: No {level_name} found for ESCO code {code}")
+        return {
+            f'{level_name}_ESCO_code': code,
+            f'{level_name}_ESCO_name': 'UNKNOWN',
+            f'{level_name}_ESCO_desc': 'No matching occupation found'
+        }
+    info = matches.iloc[0]
+    return {
+        f'{level_name}_ESCO_code': code,
+        f'{level_name}_ESCO_name': info['preferredLabel'],
+        f'{level_name}_ESCO_desc': info.get('description', '')
+    }
+
+
 def get_skills_info_esco(Level_5_code):
  
     try: