import streamlit as st import os from dotenv import load_dotenv import PyPDF2 from docx import Document import io from typing import Dict, Any, List from pydantic import BaseModel, Field import plotly.graph_objects as go import json import re from docx.shared import Inches from docx.enum.text import WD_ALIGN_PARAGRAPH import plotly.io as pio from mh_aspects import agent as aspects_agent from mh_classification import agent as clarification_agent from mh_evaluation import MHEvaluationAgent as mh_eval_agent # Load environment variables load_dotenv() # Get model from environment OPENAI_MODEL = os.getenv('OPENAI_MODEL', 'gpt-3.5-turbo') # Initialize evaluation agent mh_eval_agent = mh_eval_agent() def test_api_connection(): """Test if the OpenAI API is working""" try: # Create a test job description test_jd = """Test Job Description Position: Software Engineer Requirements: - 3+ years of Python experience - Bachelor's degree in Computer Science - Experience with web development """ # Try to get a response from the aspects agent response = aspects_agent.run(input=f"Analyze this job description and generate key must-have aspects only:\n\n{test_jd}") if response: st.success("✅ API connection successful!") return True else: st.error("❌ API connection failed: No response received") return False except Exception as e: st.error(f"❌ API connection failed: {str(e)}") return False # Pydantic model for must-have requirements class MustHaveAnalysis(BaseModel): category: str = Field(..., description="Category (1: No must-haves mentioned, 2: Meets Requirements, 3: Does Not Meet)") evidence: List[str] = Field(default_factory=list, description="Evidence supporting the categorization") confidence: float = Field(default=0.8, description="Confidence score between 0 and 1") # Set page config st.set_page_config( page_title="JD & Resume Analyzer", page_icon="📄", layout="wide" ) # Initialize session state if 'analysis_result' not in st.session_state: st.session_state.analysis_result = None if 'aspects' not in st.session_state: st.session_state.aspects = None if 'clarifications' not in st.session_state: st.session_state.clarifications = None def create_gauge_chart(value, title): fig = go.Figure(go.Indicator( mode="gauge+number", value=value, domain={'x': [0, 1], 'y': [0, 1]}, title={'text': title}, gauge={ 'axis': {'range': [0, 100]}, 'bar': {'color': "rgb(50, 168, 82)"}, 'steps': [ {'range': [0, 33], 'color': "lightgray"}, {'range': [33, 66], 'color': "gray"}, {'range': [66, 100], 'color': "darkgray"} ], 'threshold': { 'line': {'color': "red", 'width': 4}, 'thickness': 0.75, 'value': 80 } } )) fig.update_layout( height=250, margin=dict(l=10, r=10, t=50, b=10), paper_bgcolor="rgba(0,0,0,0)", font={'color': "#31333F"} ) return fig def extract_text_from_pdf(file): try: pdf_reader = PyPDF2.PdfReader(file) text = "" for page in pdf_reader.pages: text += page.extract_text() + "\n" return text.strip() except Exception as e: st.error(f"Error reading PDF: {str(e)}") return None def extract_text_from_docx(file): try: doc = Document(io.BytesIO(file.read())) text = "" for paragraph in doc.paragraphs: text += paragraph.text + "\n" return text.strip() except Exception as e: st.error(f"Error reading DOCX: {str(e)}") return None def read_file_content(file): if file is None: return None file_extension = file.name.split('.')[-1].lower() try: if file_extension == 'pdf': file_copy = io.BytesIO(file.read()) file.seek(0) return extract_text_from_pdf(file_copy) elif file_extension == 'docx': return extract_text_from_docx(file) elif file_extension == 'txt': return file.read().decode('utf-8').strip() else: raise ValueError(f"Unsupported file type: {file_extension}") except Exception as e: st.error(f"Error reading file {file.name}: {str(e)}") return None def analyze_must_haves(jd_text: str, resume_text: str) -> Dict: """Analyze must-have requirements using the three-step process""" try: # Step 1: Generate must-have aspects from JD aspects = aspects_agent.run(input=f"Analyze this job description and generate key must-have aspects only:\n\n{jd_text}") st.session_state.aspects = aspects # Step 2: Generate clarifications from resume input_text = f"""Checkpoints: {aspects} Resume: {resume_text}""" clarifications = clarification_agent.run(input=input_text) st.session_state.clarifications = clarifications # Step 3: Final evaluation evaluation = mh_eval_agent.forward( job_description=jd_text, profile=resume_text, checkpoints=aspects, answer_script=clarifications ) return { 'aspects': aspects, 'clarifications': clarifications, 'evaluation': evaluation } except Exception as e: st.error(f"Error in analysis pipeline: {str(e)}") return None def display_analysis_result(result: Dict): if not result: st.error("Analysis failed") return st.title("Must-Have Requirements Analysis") # Display aspects with st.expander("🎯 Must-Have Requirements", expanded=True): st.write(result['aspects']) # Display clarifications with st.expander("🔍 Clarifications", expanded=True): st.write(result['clarifications']) # Display evaluation st.header("📊 Final Evaluation") evaluation = result['evaluation'] # Display the evaluation in the requested format st.write(evaluation) def main(): st.title("📄 JD & Resume Must-Have Requirements Analyzer") # Test API connection when the page loads if not test_api_connection(): st.warning("⚠️ Please check your API key and model configuration in the .env file") return st.write("Upload a job description and resume to analyze if the candidate meets the must-have requirements.") # Display the model being used st.sidebar.info(f"Using model: {OPENAI_MODEL}") # File uploaders col1, col2 = st.columns(2) with col1: jd_file = st.file_uploader("Upload Job Description (PDF, DOCX, or TXT)", type=['pdf', 'docx', 'txt']) if jd_file: st.text_area("Job Description Content", read_file_content(jd_file), height=300) with col2: resume_file = st.file_uploader("Upload Resume (PDF, DOCX, or TXT)", type=['pdf', 'docx', 'txt']) if resume_file: st.text_area("Resume Content", read_file_content(resume_file), height=300) # Process button if st.button("Analyze Must-Have Requirements"): if jd_file and resume_file: with st.spinner("Analyzing documents..."): try: jd_text = read_file_content(jd_file) resume_text = read_file_content(resume_file) if jd_text and resume_text: analysis = analyze_must_haves(jd_text, resume_text) st.session_state.analysis_result = analysis display_analysis_result(analysis) else: st.error("Failed to extract text from one or both files.") except Exception as e: st.error(f"An error occurred: {str(e)}") else: st.warning("Please upload both a job description and resume.") # Display previous results if available if st.session_state.analysis_result and not (jd_file and resume_file): display_analysis_result(st.session_state.analysis_result) if __name__ == "__main__": main()