Update app.py

app.py

@@ -6,238 +6,309 @@ from PyPDF2 import PdfReader
 import docx
 import re
 import google.generativeai as genai
-import time
 import concurrent.futures
 from fuzzywuzzy import fuzz
+from typing import List, Dict, Tuple, Any
+from dataclasses import dataclass
+import logging
+from pathlib import Path
 
-# Load pre-trained embedding model for basic analysis
-sentence_model = SentenceTransformer('paraphrase-MiniLM-L6-v2')
-
-# Configure Google API for Gemini 1.5 Flash
-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)
-
-# Maximum resumes to process
-MAX_RESUMES = 10
-
-# Define the key leadership and management skills you're looking for
-required_skills = [
-    "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"
-]
-
-# Helper Functions
-
-def extract_text_from_file(file_path):
-    ext = os.path.splitext(file_path)[1].lower()
-    if ext == ".txt":
-        with open(file_path, 'r', encoding='utf-8') as f:
-            return f.read()
-    elif ext == ".pdf":
-        reader = PdfReader(file_path)
-        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:
-        return ""
-
-def analyze_with_gemini(resume_text, job_desc):
-    prompt = f"""
-    Analyze the resume with respect to the job description.
-    Resume: {resume_text}
-    Job Description: {job_desc}
-    Extract:
-    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 (e.g. strategic planning, team management, project management, etc.)
-    9. Match Percentage (leadership and management focus)
-    Provide a summary of qualifications in 5 bullet points.
-    """
-    response = genai.GenerativeModel('gemini-1.5-flash').generate_content(prompt)
-    return response.text.strip()
-
-def extract_management_details(gemini_response):
-    leadership_exp_pattern = r"Team Leadership Experience \(years\):\s*(\d+)"
-    management_exp_pattern = r"Management Experience \(years\):\s*(\d+)"
-    management_skills_pattern = r"Management Skills\s*[:\-]?\s*(.*?)(?=\n|$)"
-
-    leadership_match = re.search(leadership_exp_pattern, gemini_response)
-    management_match = re.search(management_exp_pattern, gemini_response)
-    skills_match = re.search(management_skills_pattern, gemini_response)
-
-    leadership_years = int(leadership_match.group(1)) if leadership_match else 0
-    management_years = int(management_match.group(1)) if management_match else 0
-    skills = skills_match.group(1) if skills_match else ""
-
-    return leadership_years, management_years, skills
-
-def extract_candidate_details(gemini_response):
-    name_pattern = r"Candidate Name\s*[:\-]?\s*(.*?)(?=\n|$)"
-    email_pattern = r"Email Address\s*[:\-]?\s*(.*?)(?=\n|$)"
-    contact_pattern = r"Contact Number\s*[:\-]?\s*(.*?)(?=\n|$)"
-
-    name_match = re.search(name_pattern, gemini_response)
-    email_match = re.search(email_pattern, gemini_response)
-    contact_match = re.search(contact_pattern, gemini_response)
-
-    name = name_match.group(1) if name_match else "N/A"
-    email = email_match.group(1) if email_match else "N/A"
-    contact = contact_match.group(1) if contact_match else "N/A"
-
-    return name, email, contact
-
-def calculate_role_score(role_keywords):
-    seniority_score = 0
-    role_hierarchy = {
-        "CEO": 5,
-        "CIO": 5,
-        "Director": 4,
-        "VP": 4,
-        "Manager": 3,
-        "Team Lead": 2,
-        "Junior": 1
-    }
-
-    for keyword, score in role_hierarchy.items():
-        if fuzz.partial_ratio(keyword.lower(), role_keywords.lower()) > 80:
-            seniority_score = max(seniority_score, score)
-    
-    return seniority_score
+# 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'
 
-def calculate_advanced_match(leadership_years, management_years, skills, required_skills, role_keywords, max_leadership_exp=10, max_management_exp=10):
-    leadership_weight = 0.35
-    management_weight = 0.35
-    skills_weight = 0.2
-    role_weight = 0.1
+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()
 
-    leadership_score = min(leadership_years / max_leadership_exp, 1.0) * 100
-    management_score = min(management_years / max_management_exp, 1.0) * 100
+    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
 
-    role_score = calculate_role_score(role_keywords)
-    role_score = role_score * 100
+    @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"
+        ]
 
-    skills_matched = sum(1 for skill in required_skills if fuzz.partial_ratio(skill.lower(), skills.lower()) > 80)
-    total_skills = len(required_skills)
-    skill_match_score = (skills_matched / total_skills) * 100
+    @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
+        }
 
-    overall_match = (leadership_score * leadership_weight) + \
-                    (management_score * management_weight) + \
-                    (skill_match_score * skills_weight) + \
-                    (role_score * role_weight)
-    return round(overall_match, 2)
+    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}")
 
-def process_resume(resume, job_desc, progress_callback):
-    resume_text = extract_text_from_file(resume.name)
-    
-    if not resume_text.strip():
+            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,
+            "Resume": resume_name,
             "Candidate Name": "N/A",
             "Email": "N/A",
             "Contact": "N/A",
-            "Overall Match Percentage": 0.0,
-            "Gemini Analysis": "Failed to extract text from resume."
+            "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()
     
-    try:
-        gemini_analysis = analyze_with_gemini(resume_text, job_desc)
-        leadership_years, management_years, skills = extract_management_details(gemini_analysis)
-        role_keywords = gemini_analysis.lower()
-        overall_match = calculate_advanced_match(leadership_years, management_years, skills, required_skills, role_keywords)
-        name, email, contact = extract_candidate_details(gemini_analysis)
-    except Exception as e:
-        gemini_analysis = f"Gemini analysis failed: {str(e)}"
-        name, email, contact = "N/A", "N/A", "N/A"
-        overall_match = 0.0
-
-    progress_callback(1)  # Update progress for this resume
-
-    return {
-        "Resume": resume.name,
-        "Candidate Name": name,
-        "Email": email,
-        "Contact": contact,
-        "Overall Match Percentage": f"{overall_match}%",
-        "Gemini Analysis": gemini_analysis
-    }
-
-def analyze_resumes(resumes, job_desc):
-    progress = gr.Progress()
-    results = []
-    
-    if len(resumes) > MAX_RESUMES:
-        return "Error: Cannot upload more than 10 resumes."
+    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"
+    )
     
-    with concurrent.futures.ThreadPoolExecutor() as executor:
-        futures = []
-        for resume in resumes:
-            futures.append(executor.submit(process_resume, resume, job_desc, progress.update))
-        
-        for future in concurrent.futures.as_completed(futures):
-            results.append(future.result())
-
-    return pd.DataFrame(results)
-
-def download_results():
-    # You need to return a file path, here we use a CSV output
-    results_df = pd.DataFrame(results)  # Assume results are available globally or passed in another way
-    results_df.to_csv("/tmp/analysis_results.csv", index=False)
-    return "/tmp/analysis_results.csv"
-
-# Define Gradio Interface
-iface = gr.Interface(
-    fn=analyze_resumes,
-    inputs=[gr.File(label="Upload Resumes", file_count="multiple"), gr.Textbox(label="Enter Job Description")],
-    outputs=[gr.DataFrame(label="Analysis Results"), gr.Textbox(label="Resume Count Message"), gr.File(label="Download Results", file=download_results)],
-    flagging_mode="never",
-    live=True
-)
+    return iface
 
-iface.launch()
+if __name__ == "__main__":
+    iface = create_interface()
+    iface.launch(
+        share=False,
+        debug=True,
+        server_name="0.0.0.0",
+        server_port=7860
+    )