app.py

import gradio as gr
import pdfplumber
import pandas as pd
import re
import warnings
import logging
import os
from dotenv import load_dotenv
import json
from concurrent.futures import ThreadPoolExecutor
from typing import List, Dict, Optional
import traceback
import time

# Load environment variables
load_dotenv()

# Configure logging for pdfminer
logging.getLogger('pdfminer').setLevel(logging.ERROR)

# Suppress specific warnings
warnings.filterwarnings("ignore", category=UserWarning, message="CropBox.*")

# Initialize OpenAI client
def initialize_openai_client():
    try:
        client = openai.AzureOpenAI(
            api_key=os.getenv("AZURE_OPENAI_API_KEY"),
            azure_endpoint=os.getenv("AZURE_OPENAI_ENDPOINT"),
            api_version=os.getenv("OPENAI_API_VERSION"),
        )
        return client
    except Exception as e:
        raise Exception(f"Failed to initialize OpenAI client: {e}")

client = initialize_openai_client()

def gpt_call(system_prompt: str, user_prompt: str) -> str:
    try:
        response = client.chat.completions.create(
            model=os.getenv("AZURE_DEPLOYMENT_NAME"),
            messages=[
                {"role": "system", "content": system_prompt},
                {"role": "user", "content": user_prompt}
            ],
            temperature=0.3
        )
        return response.choices[0].message.content.strip()
    except Exception as e:
        return f"ERROR: {e}"

def extract_text_from_pdf(pdf_path: str) -> str:
    text = ""
    with pdfplumber.open(pdf_path) as pdf:
        for page in pdf.pages:
            page_text = page.extract_text()
            if page_text:
                text += page_text + "\n"
            for table in page.extract_tables():
                for row in table:
                    for cell in row:
                        if isinstance(cell, str):
                            text += cell + " "
                    text += "\n"
    return text

def extract_section_from_pdf(full_text: str, section_title: str) -> str:
    user_prompt = f"""
    Carefully evaluate the provided position description (PD) document and extract the content of the section titled "{section_title}" from the following text.
    Return only the content of the section, without the title.
    If the section cannot be found or explicitly mentioned in the text, use "N/A" as the default value.
    Do not repeat in the extracted text the name of the section.
    Extract precisely all the related text.
    Text of the position description:
    {full_text}
    Section to identify: "{section_title}":
    """
    return gpt_call("You are an HR expert working for IOM.", user_prompt)

def classify_job_family(responsibilities: List[str]) -> str:
    job_families_df = pd.read_csv("job_families1.csv")
    job_family_list = "\n".join(f"- {row['Job_family']}: {row['Job_subfamily']}" for _, row in job_families_df.iterrows())
    user_prompt = f"""
    Here is a list of job responsibilities:
    {responsibilities}
    Here is a list of Job families:
    {job_family_list}
    Based on the responsibilities, suggest the most relevant job family and subfamily from the list above.
    **Important:**
    - Return ONLY the job family, nothing else.
    - The job family should be exactly as shown in the list.
    - Do not include any additional text or explanation.
    """
    return gpt_call("Suggest job family and subfamily based on responsibilities.", user_prompt)

def get_level_CCOG_info(df, code, level_name):
    matches = df[df['code'] == code]
    if len(matches) == 0:
        print(f"Warning: No {level_name} found for CCOG code {code}")
        return {
            f'{level_name}_CCOG_code': code,
            f'{level_name}_CCOG_name': 'UNKNOWN',
            f'{level_name}_CCOG_desc': 'No matching occupation found'
        }
    info = matches.iloc[0]
    return {
        f'{level_name}_CCOG_code': code,
        f'{level_name}_CCOG_name': info['occupation'],
        f'{level_name}_CCOG_desc': info.get('occupation_description', '')
    }

def code_sanitize(input_string, valid_codes):
    for code in valid_codes:
        if code in input_string:
            return code
    return None

def classify_occupational_group_by_level(responsibilities: List[str]) -> dict:
    occupational_groups_df = pd.read_csv("occupational_groups.csv")
    result = {}
    try:
        for level in range(1, 5):
            level_df = occupational_groups_df[occupational_groups_df['level'] == f"Level {level}"]
            if level > 1:
                prev_level_code = result[f'Level_{level-1}_CCOG_code']
                level_df = level_df[level_df['code'].str.startswith(prev_level_code)]
            job_occupation_list = "\n".join(f"- {row['code']}: {row['occupation']} - {row.get('occupation_description', '')}" for _, row in level_df.iterrows())
            list_output = level_df["code"].tolist()
            user_prompt = f"""
            Here is a list of job responsibilities:
            {responsibilities}
            Here is a list of level {level} Occupation classifications:
            {job_occupation_list}
            Based on the responsibilities, suggest the most relevant level {level} Occupation code from within this list: {', '.join(map(str, list_output))}.
            **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.
            """
            level_code = gpt_call(f"Identify level {level} occupational group", user_prompt).strip()
            level_code = code_sanitize(level_code, list_output)
            result.update(get_level_CCOG_info(level_df, level_code, f'Level_{level}'))
    except Exception as e:
        print(f"Error during classification: {str(e)}")
        result['error'] = str(e)
    return result

def get_skills_info_esco(Level_5_code):
    esco_level5_df = pd.read_csv("occupations_en.csv", dtype={'code': str, 'iscoGroup': str})
    matches = esco_level5_df[esco_level5_df['code'] == Level_5_code]
    conceptUris = matches['conceptUri'].values.tolist()
    esco_skill_map_df = pd.read_csv("occupationSkillRelations_en.csv")
    skills = esco_skill_map_df[esco_skill_map_df['occupationUri'].isin(conceptUris)]
    skillUris = skills['skillUri'].values.tolist()
    esco_skill_df = pd.read_csv("skills_en.csv")
    thisskillslist = esco_skill_df[esco_skill_df['conceptUri'].isin(skillUris)]
    result = thisskillslist[['preferredLabel', 'conceptUri', 'description']].drop_duplicates()
    result = result.rename(columns={'preferredLabel': 'skill_name', 'description': 'skill_description', 'conceptUri': 'skill_code'})
    return result

def review_skills(Level_5_code: str, top_n: int = 10) -> List[Dict[str, str]]:
    matches = esco_level5_df[esco_level5_df['code'] == Level_5_code]
    esco_occup = matches['preferredLabel'].values.tolist()
    skill_filtered = get_skills_info_esco(Level_5_code)
    skill_filtered_options = "\n".join(f"- {row['skill_code']}: {row['skill_name']} - {row['skill_description']}" for _, row in skill_filtered.iterrows())
    prompt = f"""
    Here is a list of skills:
    {skill_filtered_options}
    Filter the skills that are relevant in the context of the work of the International Organisation for Migration.
    Ensure that skills are relevant in the context of a {esco_occup} working for a non-profit public organization.
    Required JSON structure:
    {{
        "skills": [
            {{
                "skill_name": "string",
                "skill_description": "string",
                "skill_code": "string"
            }}
        ]
    }}
    **Important:**
    - Do not duplicate any records of skills
    - Keep only the 10 most relevant skills
    - Return ONLY the JSON object with no other text
    - Use double quotes for all strings
    - No trailing commas in arrays/objects
    - No markdown formatting (no ```json)
    - No text before or after the JSON
    - Escape all special characters in strings
    - Ensure all brackets are properly closed
    - No trailing commas in arrays/objects, especially before closing brackets
    """
    raw = gpt_call("You are an HR expert working for the International Organisation for Migration and with in-depth knowledge of the European Skills, Competences, Qualifications and Occupations. Extract skills required for this position.", prompt)
    json_text = _extract_json(raw)
    if not json_text:
        return []
    try:
        result = json.loads(json_text)
        skills = result.get("skills", [])
    except json.JSONDecodeError as e:
        print(f"❌ JSON parsing error: {e}")
        print(f"🔍 Problematic JSON: {json_text}")
        return []
    validated_skills = []
    for skill in skills:
        try:
            validated = {
                "skill_name": str(skill["skill_name"]).strip(),
                "skill_description": str(skill["skill_description"]).strip(),
                "skill_code": str(skill["skill_code"]).strip()
            }
            validated_skills.append(validated)
        except (KeyError, TypeError) as e:
            print(f"⚠️ Skipping invalid skill: {skill}. Error: {e}")
            continue
    return validated_skills[\:top_n]

def extract_skills(responsibilities: List[str], top_n: int = 10) -> List[Dict[str, str]]:
    prompt = f"""
    Here is a list of job responsibilities:
    {responsibilities}
    List the required skills and knowledge as bullet points (without numbers) using ESCO-style terms.
    For each Skill:
    1. skill_name: precise skills name as used in ESCO framework
    2. skill_description: add the long description as mentioned in ESCO framework
    3. skill_code: include the detailed corresponding ESCO code for that skill.
    Required JSON structure:
    {{
        "skills": [
            {{
                "skill_name": "string",
                "skill_description": "string",
                "skill_code": "string"
            }}
        ]
    }}
    **Important:**
    - Return ONLY the JSON object with no other text
    - Use double quotes for all strings
    - No trailing commas in arrays/objects
    - No markdown formatting (no ```json)
    - No text before or after the JSON
    - Escape all special characters in strings
    - Ensure all brackets are properly closed
    """
    raw = gpt_call("You are an HR expert working for the International Organisation for Migration and with in-depth knowledge of the European Skills, Competences, Qualifications and Occupations. Extract skills required for this position.", prompt)
    json_text = _extract_json(raw)
    if not json_text:
        return []
    try:
        result = json.loads(json_text)
        skills = result.get("skills", [])
    except json.JSONDecodeError as e:
        print(f"❌ JSON parsing error: {e}")
        print(f"🔍 Problematic JSON: {json_text}")
        return []
    validated_skills = []
    for skill in skills:
        try:
            validated = {
                "skill_name": str(skill["skill_name"]).strip(),
                "skill_description": str(skill["skill_description"]).strip(),
                "skill_code": str(skill["skill_code"]).strip()
            }
            validated_skills.append(validated)
        except (KeyError, TypeError) as e:
            print(f"⚠️ Skipping invalid skill: {skill}. Error: {e}")
            continue
    return validated_skills[:top_n]

def map_proficiency_and_assessment(skills: List[str], responsibilities: List[str]) -> List[Dict]:
    prompt = f"""
    Here is a list of job responsibilities: {responsibilities} that have been associated with the following skills: {skills}
    For each skill, accounting for the context defined within the responsibilities, return a JSON object with:
        - skill_name: the name of the skill
        - importance: essential or optional
        - type: "skill/competence" or "knowledge"
        - proficiency_level: Basic, Intermediate, or Advanced
        - distinctive_elements: what specific and distinctive elements are required at this defined proficiency level?
        - resume_signals: what to look for in a resume to assess this skill?
        - assessment_method: what is the preferred assessment method to accurately assess this skill?
    Respond ONLY with a list of dictionaries in valid JSON.
    Use double quotes for all strings. No markdown, no commentary, no trailing commas.
    """
    raw = gpt_call("Define proficiency level and assessment for each skill.", prompt)
    json_text = _extract_json_array(raw)
    if not json_text:
        return []
    try:
        results = json.loads(json_text)
    except json.JSONDecodeError as e:
        print(f"❌ JSON parsing error: {e}")
        print(f"🔍 Problematic JSON: {json_text}")
        return []
    validated = []
    for item in results:
        try:
            validated.append({
                "skill_name": str(item["skill_name"]).strip(),
                "importance": item["importance"].strip().lower(),
                "type": item["type"].strip().lower(),
                "proficiency_level": item["proficiency_level"].strip().capitalize(),
                "distinctive_elements": item["distinctive_elements"].strip(),
                "resume_signals": item["resume_signals"].strip(),
                "assessment_method": item["assessment_method"].strip()
            })
        except (KeyError, TypeError) as e:
            print(f"⚠️ Skipping invalid item: {item}. Error: {e}")
            continue
    return validated

def _extract_json_array(raw: str) -> str:
    json_start = raw.find('[')
    json_end = raw.rfind(']') + 1
    if json_start == -1 or json_end == 0:
        print(f"❌ No JSON array found in response: {raw}")
        return ""
    json_text = raw[json_start:json_end]
    json_text = re.sub(r',\s*([}\]])', r'\1', json_text)
    json_text = re.sub(r'[\n\r\t]', ' ', json_text)
    json_text = re.sub(r'(?<!\\)"', '"', json_text)
    return json_text

def extract_qualification(responsibilities: List[str]) -> List[str]:
    prompt = f"""
    Here is a list of job responsibilities: {responsibilities}
    Infer the required level within the European Qualifications Framework (EQF) to implement them.
    Identify the potential diplomas to testify such qualification
    """
    raw = gpt_call("You are an HR expert that excel in developing competency-based interview questions.", prompt)
    return [line.strip("-• ").strip() for line in raw.splitlines() if line.strip()]

def build_interview(responsibilities: List[str], skill_assess: List[str]) -> List[str]:
    prompt = f"""
    Here is a list of job responsibilities: {responsibilities} and related skills: {skill_assess}
    Output: A structured 40-minute interview with:
        Opening questions (5 min)
        Core competency-based questions (30 min, 5-6 questions)
        Closing & candidate questions (5 min)
    """
    raw = gpt_call("You are an HR expert that excel in developing competency-based interview questions.", prompt)
    return [line.strip("-• ").strip() for line in raw.splitlines() if line.strip()]

def _extract_json(raw: str) -> str:
    json_start = raw.find('{')
    json_end = raw.rfind('}') + 1
    if json_start == -1 or json_end == 0:
        print(f"❌ No JSON found in response: {raw}")
        return ""
    json_text = raw[json_start:json_end]
    json_text = re.sub(r',\s*([}\]])', r'\1', json_text)
    json_text = re.sub(r'[\n\r\t]', ' ', json_text)
    json_text = re.sub(r'\s{2,}', ' ', json_text)
    json_text = re.sub(r'\\(?!["\\/bfnrtu])', r'\\\\', json_text)
    json_text = json_text.strip()
    return json_text

def process_pdf(file):
    if file is None:
        return "Please upload a PDF file."
    try:
        extracted_text = extract_text_from_pdf(file.name)
        responsibilities = extract_section_from_pdf(extracted_text, section_title="Responsibilities and Accountabilities")
        if not responsibilities:
            print(f"Skipping {os.path.basename(file.name)} - no responsibilities section found")
            return None
        job_family = classify_job_family(responsibilities)
        occ_group = classify_occupational_group_by_level(responsibilities)
        esco_occ = classify_esco_by_hierarchical_level(responsibilities)
        qualification = extract_qualification(responsibilities)
        skills = extract_skills(responsibilities)
        skill_map = map_proficiency_and_assessment(skills, responsibilities)
        has_esco = esco_occ.get("Level_5_ESCO_code") is not None
        skill_esco_extract = []
        skill_esco_map = []
        if has_esco:
            Level_5_code = esco_occ["Level_5_ESCO_code"]
            skill_esco_extract = review_skills(Level_5_code)
            skill_esco_map = map_proficiency_and_assessment(skill_esco_extract, responsibilities)
        else:
            print(f"No Level 5 ESCO code found for {os.path.basename(file.name)}, skipping ESCO skills mapping")
        time.sleep(6)
        assessment_lookup = {item['skill_name']: item for item in skill_map}
        joined_skills = [
            {
                "skill_name": skill["skill_name"],
                "skill_description": skill["skill_description"],
                "skill_code": skill["skill_code"],
                "importance": assessment_lookup.get(skill["skill_name"], {}).get("importance"),
                "type": assessment_lookup.get(skill["skill_name"], {}).get("type"),
                "proficiency_level": assessment_lookup.get(skill["skill_name"], {}).get("proficiency_level"),
                "distinctive_elements": assessment_lookup.get(skill["skill_name"], {}).get("distinctive_elements"),
                "resume_signals": assessment_lookup.get(skill["skill_name"], {}).get("resume_signals"),
                "assessment_method": assessment_lookup.get(skill["skill_name"], {}).get("assessment_method")
            }
            for skill in skills
        ]
        joined_skills_esco = []
        if has_esco and skill_esco_extract:
            assessment_esco_lookup = {item['skill_name']: item for item in skill_esco_map}
            joined_skills_esco = [
                {
                    "skill_name": skill["skill_name"],
                    "skill_description": skill["skill_description"],
                    "skill_code": skill["skill_code"],
                    **assessment_esco_lookup.get(skill["skill_name"], {})
                }
                for skill in skill_esco_extract
            ]
        interview = build_interview(responsibilities, skills)
        result = {
            "file": os.path.basename(file.name),
            "responsibilities": responsibilities,
            "job_family": job_fam1['Job_family'].values[0],
            "job_subfamily": job_fam1['Job_subfamily'].values[0],
            "classified_job_family": job_family,
            **{f"Level_{i}_CCOG_{field}": occ_group.get(f"Level_{i}_CCOG_{field}")
               for i in range(1, 5) for field in ["code", "name", "desc"]},
            "qualification": qualification,
            "interview": interview,
            "skills": {
                "file": os.path.basename(file.name),
                "job_family": job_fam1['Job_family'].values[0],
                "job_subfamily": job_fam1['Job_subfamily'].values[0],
                "skills": joined_skills
            }
        }
        if has_esco:
            result.update({
                **{f"Level_{i}_ESCO_{field}": esco_occ.get(f"Level_{i}_ESCO_{field}")
                   for i in range(1, 6) for field in ["code", "name", "desc"]},
                "skills_esco": {
                    "file": os.path.basename(file.name),
                    "job_family": job_fam1['Job_family'].values[0],
                    "job_subfamily": job_fam1['Job_subfamily'].values[0],
                    "skills": joined_skills_esco
                }
            })
        else:
            result.update({
                **{f"Level_{i}_ESCO_{field}": None
                   for i in range(1, 6) for field in ["code", "name", "desc"]},
                "skills_esco": None
            })
        return result
    except Exception as e:
        return f"Error processing PDF: {str(e)}"

with gr.Blocks() as demo:
    gr.Markdown("# Standardize Job Description!")
    gr.Markdown("Identify Job Family, Occupation, Qualification, match Skills and suggest interview questions.")
    with gr.Row():
        with gr.Column():
            file_input = gr.File(label="Upload a Job Description PDF file", file_types=[".pdf"])
            submit_btn = gr.Button("Extract Text")
        with gr.Column():
            text_output = gr.Textbox(label="Extracted Text", lines=30, max_lines=50, interactive=False)
    submit_btn.click(
        fn=process_pdf,
        inputs=file_input,
        outputs=text_output
    )

if __name__ == "__main__":
    demo.launch()