app.py

import os
import gradio as gr
import pandas as pd
from sentence_transformers import SentenceTransformer, util
from PyPDF2 import PdfReader
import docx
import re
import google.generativeai as genai
import concurrent.futures
from fuzzywuzzy import fuzz
from typing import List, Dict, Tuple, Any
from dataclasses import dataclass
import logging
from pathlib import Path

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

@dataclass
class Config:
    MAX_RESUMES: int = 10
    MAX_LEADERSHIP_EXP: int = 10
    MAX_MANAGEMENT_EXP: int = 10
    MODEL_NAME: str = 'paraphrase-MiniLM-L6-v2'
    GEMINI_MODEL: str = 'gemini-1.5-flash'

class ResumeAnalyzer:
    def __init__(self):
        self.config = Config()
        self._initialize_models()
        self.required_skills = self._load_required_skills()
        self.role_hierarchy = self._load_role_hierarchy()

    def _initialize_models(self) -> None:
        """Initialize the required models and API configurations."""
        try:
            self.sentence_model = SentenceTransformer(self.config.MODEL_NAME)
            
            api_key = os.getenv('GOOGLE_API_KEY')
            if not api_key:
                raise ValueError("Google API key not found. Please set GOOGLE_API_KEY.")
            genai.configure(api_key=api_key)
            
        except Exception as e:
            logger.error(f"Failed to initialize models: {str(e)}")
            raise

    @staticmethod
    def _load_required_skills() -> List[str]:
        """Load the list of required leadership and management skills."""
        return [
            "strategic planning", "team management", "project management",
            "decision making", "communication", "leadership",
            "conflict resolution", "delegation", "performance management",
            "budget management", "resource allocation", "staff development",
            "change management", "risk management", "problem solving",
            "negotiation", "executive leadership", "organizational skills",
            "business development", "stakeholder management", "collaboration",
            "emotional intelligence", "coaching", "mentoring",
            "time management", "cross-functional team leadership", "innovation",
            "organizational culture", "team motivation", "employee engagement",
            "organizational design", "continuous improvement",
            "decision-making under pressure", "adaptability", "accountability",
            "team building", "succession planning", "strategic partnerships",
            "executive presence", "influencing", "visionary leadership"
        ]

    @staticmethod
    def _load_role_hierarchy() -> Dict[str, int]:
        """Load the role hierarchy for scoring."""
        return {
            "CEO": 5, "CIO": 5, "CFO": 5, "COO": 5,
            "Director": 4, "VP": 4, "Head": 4,
            "Manager": 3, "Senior": 3,
            "Team Lead": 2, "Lead": 2,
            "Junior": 1, "Associate": 1
        }

    def extract_text_from_file(self, file_path: str) -> str:
        """Extract text content from various file formats."""
        try:
            file_path = Path(file_path)
            if not file_path.exists():
                raise FileNotFoundError(f"File not found: {file_path}")

            ext = file_path.suffix.lower()
            if ext == ".txt":
                return file_path.read_text(encoding='utf-8')
            elif ext == ".pdf":
                with open(file_path, 'rb') as file:
                    reader = PdfReader(file)
                    return " ".join(page.extract_text() for page in reader.pages)
            elif ext == ".docx":
                doc = docx.Document(file_path)
                return " ".join(para.text for para in doc.paragraphs)
            else:
                raise ValueError(f"Unsupported file format: {ext}")
        except Exception as e:
            logger.error(f"Error extracting text from {file_path}: {str(e)}")
            return ""

    def analyze_with_gemini(self, resume_text: str, job_desc: str) -> str:
        """Analyze resume using Gemini model."""
        try:
            prompt = f"""
            Analyze the resume with respect to the job description.
            Resume: {resume_text}
            Job Description: {job_desc}
            
            Please provide a structured analysis with the following information:
            1. Candidate Name:
            2. Email Address:
            3. Contact Number:
            4. Relevant Skills:
            5. Educational Background:
            6. Team Leadership Experience (years):
            7. Management Experience (years):
            8. Management Skills:
            9. Match Percentage:
            
            Summary of Qualifications:
            • 
            • 
            • 
            • 
            • 
            """
            
            model = genai.GenerativeModel(self.config.GEMINI_MODEL)
            response = model.generate_content(prompt)
            return response.text.strip()
        except Exception as e:
            logger.error(f"Gemini analysis failed: {str(e)}")
            raise

    def extract_management_details(self, gemini_response: str) -> Tuple[int, int, str]:
        """Extract management experience details from Gemini response."""
        try:
            patterns = {
                'leadership': r"Team Leadership Experience \(years\):\s*(\d+)",
                'management': r"Management Experience \(years\):\s*(\d+)",
                'skills': r"Management Skills\s*[:\-]?\s*(.*?)(?=\n|$)"
            }
            
            matches = {
                key: re.search(pattern, gemini_response)
                for key, pattern in patterns.items()
            }
            
            leadership_years = int(matches['leadership'].group(1)) if matches['leadership'] else 0
            management_years = int(matches['management'].group(1)) if matches['management'] else 0
            skills = matches['skills'].group(1) if matches['skills'] else ""
            
            return leadership_years, management_years, skills
        except Exception as e:
            logger.error(f"Error extracting management details: {str(e)}")
            return 0, 0, ""

    def calculate_role_score(self, role_keywords: str) -> float:
        """Calculate seniority score based on role keywords."""
        try:
            seniority_score = 0
            for keyword, score in self.role_hierarchy.items():
                if fuzz.partial_ratio(keyword.lower(), role_keywords.lower()) > 80:
                    seniority_score = max(seniority_score, score)
            return seniority_score
        except Exception as e:
            logger.error(f"Error calculating role score: {str(e)}")
            return 0

    def calculate_advanced_match(self, leadership_years: int, management_years: int,
                               skills: str, role_keywords: str) -> float:
        """Calculate overall match percentage using weighted criteria."""
        try:
            weights = {
                'leadership': 0.35,
                'management': 0.35,
                'skills': 0.20,
                'role': 0.10
            }

            leadership_score = min(leadership_years / self.config.MAX_LEADERSHIP_EXP, 1.0) * 100
            management_score = min(management_years / self.config.MAX_MANAGEMENT_EXP, 1.0) * 100
            
            role_score = self.calculate_role_score(role_keywords) * 20  # Scale to 100
            
            skills_matched = sum(1 for skill in self.required_skills 
                               if fuzz.partial_ratio(skill.lower(), skills.lower()) > 80)
            skill_match_score = (skills_matched / len(self.required_skills)) * 100

            overall_match = sum([
                leadership_score * weights['leadership'],
                management_score * weights['management'],
                skill_match_score * weights['skills'],
                role_score * weights['role']
            ])
            
            return round(overall_match, 2)
        except Exception as e:
            logger.error(f"Error calculating advanced match: {str(e)}")
            return 0.0

    def process_resume(self, resume: Any, job_desc: str, 
                      progress_callback: callable) -> Dict[str, Any]:
        """Process a single resume and return analysis results."""
        try:
            resume_text = self.extract_text_from_file(resume.name)
            if not resume_text.strip():
                return self._create_error_result(resume.name, "Failed to extract text from resume")

            gemini_analysis = self.analyze_with_gemini(resume_text, job_desc)
            leadership_years, management_years, skills = self.extract_management_details(gemini_analysis)
            overall_match = self.calculate_advanced_match(
                leadership_years, management_years, skills, gemini_analysis.lower()
            )
            
            result = {
                "Resume": resume.name,
                "Candidate Name": self._extract_field(gemini_analysis, "Candidate Name"),
                "Email": self._extract_field(gemini_analysis, "Email Address"),
                "Contact": self._extract_field(gemini_analysis, "Contact Number"),
                "Overall Match Percentage": f"{overall_match}%",
                "Gemini Analysis": gemini_analysis
            }
            
            if progress_callback:
                progress_callback(1)
                
            return result
        except Exception as e:
            logger.error(f"Error processing resume {resume.name}: {str(e)}")
            return self._create_error_result(resume.name, str(e))

    @staticmethod
    def _extract_field(text: str, field: str) -> str:
        """Extract a specific field from the analysis text."""
        pattern = f"{field}\\s*[:\\-]?\\s*(.*?)(?=\\n|$)"
        match = re.search(pattern, text)
        return match.group(1) if match else "N/A"

    @staticmethod
    def _create_error_result(resume_name: str, error_message: str) -> Dict[str, str]:
        """Create a standardized error result."""
        return {
            "Resume": resume_name,
            "Candidate Name": "N/A",
            "Email": "N/A",
            "Contact": "N/A",
            "Overall Match Percentage": "0.0%",
            "Gemini Analysis": f"Analysis failed: {error_message}"
        }

    def analyze_resumes(self, resumes: List[Any], job_desc: str) -> pd.DataFrame:
        """Analyze multiple resumes in parallel."""
        if len(resumes) > self.config.MAX_RESUMES:
            return pd.DataFrame([{
                "Error": f"Cannot process more than {self.config.MAX_RESUMES} resumes at once."
            }])

        progress = gr.Progress()
        
        try:
            with concurrent.futures.ThreadPoolExecutor() as executor:
                futures = [
                    executor.submit(self.process_resume, resume, job_desc, progress.update)
                    for resume in resumes
                ]
                results = [future.result() for future in concurrent.futures.as_completed(futures)]
            
            return pd.DataFrame(results)
        except Exception as e:
            logger.error(f"Error in batch resume analysis: {str(e)}")
            return pd.DataFrame([{"Error": f"Analysis failed: {str(e)}"}])

# Create Gradio interface
def create_interface():
    analyzer = ResumeAnalyzer()
    
    iface = gr.Interface(
        fn=analyzer.analyze_resumes,
        inputs=[
            gr.File(
                label="Upload Resumes (max 10)",
                file_count="multiple"
            ),
            gr.Textbox(
                label="Enter Job Description",
                placeholder="Paste the job description here..."
            )
        ],
        outputs=[
            gr.DataFrame(label="Analysis Results")
        ],
        title="Resume Analysis Tool",
        description="Upload resumes and a job description to analyze candidates' leadership and management potential.",
        examples=[],
        cache_examples=False,
        theme="default"
    )
    
    return iface

if __name__ == "__main__":
    iface = create_interface()
    iface.launch(
        share=False,
        debug=True,
    )