Upload app.py

app.py

@@ -0,0 +1,636 @@
+from typing import Literal, Tuple, Dict, Optional
+import os
+import time
+import json
+import requests
+import PyPDF2
+from datetime import datetime, timedelta
+import pytz
+import streamlit as st
+from phi.agent import Agent
+from phi.model.openai import OpenAIChat
+from phi.tools.email import EmailTools
+from phi.tools.zoom import ZoomTool
+from phi.utils.log import logger
+from streamlit_pdf_viewer import pdf_viewer
+
+
+
+class CustomZoomTool(ZoomTool):
+    def __init__(self, *, account_id: Optional[str] = None, client_id: Optional[str] = None, client_secret: Optional[str] = None, name: str = "zoom_tool"):
+        super().__init__(account_id=account_id, client_id=client_id, client_secret=client_secret, name=name)
+        self.token_url = "https://zoom.us/oauth/token"
+        self.access_token = None
+        self.token_expires_at = 0
+
+    def get_access_token(self) -> str:
+        if self.access_token and time.time() < self.token_expires_at:
+            return str(self.access_token)
+            
+        headers = {"Content-Type": "application/x-www-form-urlencoded"}
+        data = {"grant_type": "account_credentials", "account_id": self.account_id}
+
+        try:
+            response = requests.post(self.token_url, headers=headers, data=data, auth=(self.client_id, self.client_secret))
+            response.raise_for_status()
+
+            token_info = response.json()
+            self.access_token = token_info["access_token"]
+            expires_in = token_info["expires_in"]
+            self.token_expires_at = time.time() + expires_in - 60
+
+            self._set_parent_token(str(self.access_token))
+            return str(self.access_token)
+
+        except requests.RequestException as e:
+            logger.error(f"Error fetching access token: {e}")
+            return ""
+
+    def _set_parent_token(self, token: str) -> None:
+        """Helper method to set the token in the parent ZoomTool class"""
+        if token:
+            self._ZoomTool__access_token = token
+
+
+# Role requirements as a constant dictionary
+ROLE_REQUIREMENTS: Dict[str, str] = {
+    "ai_ml_engineer": """
+        Required Skills:
+        - Python, PyTorch/TensorFlow
+        - Machine Learning algorithms and frameworks
+        - Deep Learning and Neural Networks
+        - Data preprocessing and analysis
+        - MLOps and model deployment
+        - RAG, LLM, Finetuning and Prompt Engineering
+    """,
+
+    "frontend_engineer": """
+        Required Skills:
+        - React/Vue.js/Angular
+        - HTML5, CSS3, JavaScript/TypeScript
+        - Responsive design
+        - State management
+        - Frontend testing
+    """,
+
+    "backend_engineer": """
+        Required Skills:
+        - Python/Java/Node.js
+        - REST APIs
+        - Database design and management
+        - System architecture
+        - Cloud services (AWS/GCP/Azure)
+        - Kubernetes, Docker, CI/CD
+    """
+}
+
+
+def init_session_state() -> None:
+    """Initialize only necessary session state variables."""
+    defaults = {
+        'candidate_email': "", 'openai_api_key': "", 'resume_text': "", 'analysis_complete': False,
+        'is_selected': False, 'zoom_account_id': "", 'zoom_client_id': "", 'zoom_client_secret': "",
+        'email_sender': "", 'email_passkey': "", 'company_name': "", 'current_pdf': None
+    }
+    for key, value in defaults.items():
+        if key not in st.session_state:
+            st.session_state[key] = value
+
+
+def create_resume_analyzer() -> Agent:
+    """Creates and returns a resume analysis agent."""
+    if not st.session_state.openai_api_key:
+        st.error("Please enter your OpenAI API key first.")
+        return None
+
+    return Agent(
+        model=OpenAIChat(
+            id="gpt-4o",
+            api_key=st.session_state.openai_api_key
+        ),
+        description="You are an expert technical recruiter who analyzes resumes.",
+        instructions=[
+            "Analyze the resume against the provided job requirements",
+            "Be lenient with AI/ML candidates who show strong potential",
+            "Consider project experience as valid experience",
+            "Value hands-on experience with key technologies",
+            "Return a JSON response with selection decision and feedback"
+        ],
+        markdown=True
+    )
+
+def create_email_agent() -> Agent:
+    return Agent(
+        model=OpenAIChat(
+            id="gpt-4o",
+            api_key=st.session_state.openai_api_key
+        ),
+        tools=[EmailTools(
+            receiver_email=st.session_state.candidate_email,
+            sender_email=st.session_state.email_sender,
+            sender_name=st.session_state.company_name,
+            sender_passkey=st.session_state.email_passkey
+        )],
+        description="You are a professional recruitment coordinator handling email communications.",
+        instructions=[
+            "Draft and send professional recruitment emails",
+            "Act like a human writing an email",
+            "Maintain a friendly yet professional tone",
+            "Always end emails with exactly: 'Best,\nthe ClutchAi recruiting team'",
+            "Never include the sender's or receiver's name in the signature",
+            f"The name of the company is '{st.session_state.company_name}'"
+        ],
+        markdown=True,
+        show_tool_calls=True
+    )
+
+
+def create_scheduler_agent() -> Agent:
+    zoom_tools = CustomZoomTool(
+        account_id=st.session_state.zoom_account_id,
+        client_id=st.session_state.zoom_client_id,
+        client_secret=st.session_state.zoom_client_secret
+    )
+
+    return Agent(
+        name="Interview Scheduler",
+        model=OpenAIChat(
+            id="gpt-4o",
+            api_key=st.session_state.openai_api_key
+        ),
+        tools=[zoom_tools],
+        description="You are an interview scheduling coordinator.",
+        instructions=[
+            "You are an expert at scheduling technical interviews using Zoom.",
+            "Schedule interviews during business hours (9 AM - 5 PM EST)",
+            "Create meetings with proper titles and descriptions",
+            "Ensure all meeting details are included in responses",
+            "Use ISO 8601 format for dates",
+            "Handle scheduling errors gracefully"
+        ],
+        markdown=True,
+        show_tool_calls=True
+    )
+
+
+def extract_text_from_pdf(pdf_file) -> str:
+    try:
+        pdf_reader = PyPDF2.PdfReader(pdf_file)
+        text = ""
+        for page in pdf_reader.pages:
+            text += page.extract_text()
+        return text
+    except Exception as e:
+        st.error(f"Error extracting PDF text: {str(e)}")
+        return ""
+
+def extract_email_from_resume(resume_text: str, analyzer: Agent) -> Optional[str]:
+    """
+    Uses GPT-4 to extract an email address from the resume text.
+    """
+    try:
+        response = analyzer.run(
+            f"""Extract the email address from the following text:
+            Resume Text:
+            {resume_text}
+            
+            Return the email address only, or 'None' if not found."""
+        )
+        assistant_message = next((msg.content for msg in response.messages if msg.role == 'assistant'), None)
+        return assistant_message.strip() if assistant_message and assistant_message != 'None' else None
+
+    except Exception as e:
+        logger.error(f"Error extracting email from resume: {str(e)}")
+        return None
+
+
+def analyze_resume(
+    resume_text: str,
+    role: Literal["ai_ml_engineer", "frontend_engineer", "backend_engineer"],
+    analyzer: Agent
+) -> Tuple[bool, str,int]:
+    try:
+        response = analyzer.run(
+            f"""Act as a ATS tracking agent, and score the resume first and Please analyze this resume against the following requirements and provide your response in valid JSON format also provide a score out of 100.:
+            Role Requirements:
+            {ROLE_REQUIREMENTS[role]}
+            Resume Text:
+            {resume_text}
+            Your response must be a valid JSON object like this:
+            {{
+                "selected": true/false,
+                "feedback": "Detailed feedback explaining the decision",
+                "matching_skills": ["skill1", "skill2"],
+                "missing_skills": ["skill3", "skill4"],
+                "experience_level": "junior/mid/senior"
+            }}
+            Evaluation criteria:
+            1. Match at least 70% of required skills
+            2. Consider both theoretical knowledge and practical experience
+            3. Value project experience and real-world applications
+            4. Consider transferable skills from similar technologies
+            5. Look for evidence of continuous learning and adaptability
+            Important: Return ONLY the JSON object without any markdown formatting or backticks.
+            """
+        )
+
+        assistant_message = next((msg.content for msg in response.messages if msg.role == 'assistant'), None)
+        if not assistant_message:
+            raise ValueError("No assistant message found in response.")
+
+        result = json.loads(assistant_message.strip())
+        if not isinstance(result, dict) or not all(k in result for k in ["selected", "feedback"]):
+            raise ValueError("Invalid response format")
+
+        return result["selected"], result["feedback"]
+
+    except (json.JSONDecodeError, ValueError) as e:
+        st.error(f"Error processing response: {str(e)}")
+        return False, f"Error analyzing resume: {str(e)}"
+
+# The modified sections include analyze_resume_with_score and main logic.
+
+def analyze_resume_with_score(
+    resume_text: str,
+    role: Literal["ai_ml_engineer", "frontend_engineer", "backend_engineer"],
+    analyzer: Agent
+) -> Tuple[bool, str, int]:
+    try:
+        response = analyzer.run(
+            f"""Act as an ATS tracking agent. Please score this resume and analyze it against the following requirements:
+            Role Requirements:
+            {ROLE_REQUIREMENTS[role]}
+            Resume Text:
+            {resume_text}
+            Your response must be a valid JSON object like this:
+            {{
+                "selected": true/false,
+                "feedback": "Detailed feedback explaining the decision",
+                "ats_score": # ATS score as an integer percentage
+                "matching_skills": ["skill1", "skill2"],
+                "missing_skills": ["skill3", "skill4"],
+                "experience_level": "junior/mid/senior"
+            }}
+            Evaluation criteria:
+            1. Match at least 70% of required skills
+            2. Consider both theoretical knowledge and practical experience
+            3. Value project experience and real-world applications
+            4. Consider transferable skills from similar technologies
+            5. Look for evidence of continuous learning and adaptability
+            Important: Return ONLY the JSON object without any markdown formatting or backticks.
+            """
+        )
+
+        assistant_message = next((msg.content for msg in response.messages if msg.role == 'assistant'), None)
+        
+        # DEBUG: Print the raw response
+        print("DEBUG: Raw assistant message:", assistant_message)
+        
+        if not assistant_message:
+            raise ValueError("No assistant message found in response.")
+
+        result = json.loads(assistant_message.strip())
+        
+        # DEBUG: Print parsed JSON response
+        print("DEBUG: Parsed JSON response:", result)
+
+        if not isinstance(result, dict) or not all(k in result for k in ["selected", "feedback", "ats_score"]):
+            raise ValueError("Invalid response format")
+
+        return result["selected"], result["feedback"], result["ats_score"]
+
+    except (json.JSONDecodeError, ValueError) as e:
+        st.error(f"Error processing response: {str(e)}")
+        return False, f"Error analyzing resume: {str(e)}", 0
+
+
+
+def send_selection_email(email_agent: Agent, to_email: str, role: str) -> None:
+    email_agent.run(
+        f"""
+        Send an email to {to_email} regarding their selection for the {role} position.
+        The email should:
+        1. Congratulate them on being selected
+        2. Explain the next steps in the process
+        3. Mention that they will receive interview details shortly
+        4. The name of the company is 'ClutchAI Recruiting Team'
+        """
+    )
+
+
+def send_rejection_email(email_agent: Agent, to_email: str, role: str, feedback: str) -> None:
+    """
+    Send a rejection email with constructive feedback.
+    """
+    email_agent.run(
+        f"""
+        Send an email to {to_email} regarding their application for the {role} position.
+        Use this specific style:
+        2. be empathetic and human
+        3. mention specific feedback from: {feedback}
+        4. encourage them to upskill and try again
+        5. suggest some learning resources based on missing skills
+        6. end the email with exactly:
+           Best,
+           The Clutchai recruiting team
+        
+        Do not include any names in the signature.
+        The tone should be like a human writing a quick but thoughtful email.
+        """
+    )
+
+
+def schedule_interview(scheduler: Agent, candidate_email: str, email_agent: Agent, role: str) -> None:
+    """
+    Schedule interviews during business hours (9 AM - 5 PM IST).
+    """
+    try:
+        # Get current time in IST
+        ny_tz= pytz.timezone('US/Eastern')
+        current_time_ist = datetime.now(ny_tz)
+
+        tomorrow_ist = current_time_ist + timedelta(days=1)
+        interview_time = tomorrow_ist.replace(hour=11, minute=0, second=0, microsecond=0)
+        formatted_time = interview_time.strftime('%Y-%m-%dT%H:%M:%S')
+
+        meeting_response = scheduler.run(
+            f"""Schedule a 60-minute technical interview with these specifications:
+            - Title: '{role} Technical Interview'
+            - Date: {formatted_time}
+            - Timezone: EST (Eastern Standard Time)
+            - Attendee: {candidate_email}
+            
+            Important Notes:
+            - The meeting must be between 9 AM - 5 PM IST
+            - Use EST (UTC+5:30) timezone for all communications
+            - Include timezone information in the meeting details
+            """
+        )
+
+        email_agent.run(
+            f"""Send an interview confirmation email with these details:
+            - Role: {role} position
+            - Meeting Details: {meeting_response}
+            
+            Important:
+            - Clearly specify that the time is in EST (Eastern Standard Time)
+            - Ask the candidate to join 5 minutes early
+            - Include timezone conversion link if possible
+            - Ask him to be confident and not so nervous and prepare well for the interview
+            """
+        )
+        
+        st.success("Interview scheduled successfully!")
+        
+    except Exception as e:
+        logger.error(f"Error scheduling interview: {str(e)}")
+        st.error("Unable to schedule interview. Please try again.")
+
+
+def main() -> None:
+    st.set_page_config(page_title="AI Recruitment System", page_icon="🤖", layout="wide")
+    
+    # Inject custom CSS for dark blue background
+    st.markdown(
+        """
+        <style>
+        .stApp {
+            background-color: #001f3f; /* Dark blue */
+            color: white; /* White text for contrast */
+        }
+        </style>
+        """,
+        unsafe_allow_html=True
+    )
+
+    st.title("🌟 AI Recruitment System")
+    st.markdown(
+        """
+        Welcome to the AI Recruitment System! 
+        Streamline your hiring process with AI-powered resume analysis and interview scheduling.
+        """
+    )
+    st.markdown("---")
+
+
+    init_session_state()
+    with st.sidebar:
+        st.header("Configuration")
+        
+        # OpenAI Configuration
+        st.subheader("OpenAI Settings")
+        api_key = st.text_input("OpenAI API Key", type="password", value=st.session_state.openai_api_key, help="Get your API key from platform.openai.com")
+        if api_key: st.session_state.openai_api_key = api_key
+
+        st.subheader("Zoom Settings")
+        zoom_account_id = st.text_input("Zoom Account ID", type="password", value=st.session_state.zoom_account_id)
+        zoom_client_id = st.text_input("Zoom Client ID", type="password", value=st.session_state.zoom_client_id)
+        zoom_client_secret = st.text_input("Zoom Client Secret", type="password", value=st.session_state.zoom_client_secret)
+        
+        st.subheader("Email Settings")
+        email_sender = st.text_input("Sender Email", value=st.session_state.email_sender, help="Email address to send from")
+        email_passkey = st.text_input("Email App Password", type="password", value=st.session_state.email_passkey, help="App-specific password for email")
+        company_name = st.text_input("Company Name", value=st.session_state.company_name, help="Name to use in email communications")
+
+        if zoom_account_id: st.session_state.zoom_account_id = zoom_account_id
+        if zoom_client_id: st.session_state.zoom_client_id = zoom_client_id
+        if zoom_client_secret: st.session_state.zoom_client_secret = zoom_client_secret
+        if email_sender: st.session_state.email_sender = email_sender
+        if email_passkey: st.session_state.email_passkey = email_passkey
+        if company_name: st.session_state.company_name = company_name
+
+        required_configs = {'OpenAI API Key': st.session_state.openai_api_key, 'Zoom Account ID': st.session_state.zoom_account_id,
+                          'Zoom Client ID': st.session_state.zoom_client_id, 'Zoom Client Secret': st.session_state.zoom_client_secret,
+                          'Email Sender': st.session_state.email_sender, 'Email Password': st.session_state.email_passkey,
+                          'Company Name': st.session_state.company_name}
+
+    missing_configs = [k for k, v in required_configs.items() if not v]
+    if missing_configs:
+        st.warning(f"Please configure the following in the sidebar: {', '.join(missing_configs)}")
+        return
+
+    if not st.session_state.openai_api_key:
+        st.warning("Please enter your OpenAI API key in the sidebar to continue.")
+        return
+
+    role = st.selectbox("Select the role you're applying for:", ["ai_ml_engineer", "frontend_engineer", "backend_engineer"])
+    with st.expander("View Required Skills", expanded=True): st.markdown(ROLE_REQUIREMENTS[role])
+
+    # Add a "New Application" button before the resume upload
+    if st.button("📝 New Application"):
+        # Clear only the application-related states
+        keys_to_clear = ['resume_text', 'analysis_complete', 'is_selected', 'candidate_email', 'current_pdf']
+        for key in keys_to_clear:
+            if key in st.session_state:
+                st.session_state[key] = None if key == 'current_pdf' else ""
+        st.rerun()
+
+    resume_file = st.file_uploader("Upload your resume (PDF)", type=["pdf"], key="resume_uploader")
+    if resume_file is not None and resume_file != st.session_state.get('current_pdf'):
+        st.session_state.current_pdf = resume_file
+        st.session_state.resume_text = ""
+        st.session_state.analysis_complete = False
+        st.session_state.is_selected = False
+        st.rerun()
+
+    if resume_file:
+        st.subheader("Uploaded Resume")
+        col1, col2 = st.columns([4, 1])
+        
+        with col1:
+            import tempfile, os
+            with tempfile.NamedTemporaryFile(delete=False, suffix='.pdf') as tmp_file:
+                tmp_file.write(resume_file.read())
+                tmp_file_path = tmp_file.name
+            resume_file.seek(0)
+            try: pdf_viewer(tmp_file_path)
+            finally: os.unlink(tmp_file_path)
+        
+        with col2:
+            st.download_button(label="📥 Download", data=resume_file, file_name=resume_file.name, mime="application/pdf")
+        # Process the resume text
+        if not st.session_state.resume_text:
+            with st.spinner("Processing your resume..."):
+                resume_text = extract_text_from_pdf(resume_file)
+                if resume_text:
+                    st.session_state.resume_text = resume_text
+                    st.success("Resume processed successfully!")
+            
+                    # Create a resume analyzer to extract email
+                    resume_analyzer = create_resume_analyzer()
+                    if resume_analyzer:
+                        email_from_resume = extract_email_from_resume(resume_text, resume_analyzer)
+                        if email_from_resume:
+                            st.session_state.candidate_email = email_from_resume
+                            st.success(f"Email ID extracted: {email_from_resume}")
+                        else:
+                            st.warning("Could not extract an email ID from the resume.")
+                else:
+                    st.error("Could not process the PDF. Please try again.")
+
+
+    # Email input with session state
+    email = st.text_input(
+        "Candidate's email address",
+        value=st.session_state.candidate_email,
+        key="email_input"
+    )
+    st.session_state.candidate_email = email
+
+    # Analysis and next steps
+    if st.session_state.resume_text and email and not st.session_state.analysis_complete:
+        if st.button("Analyze Resume"):
+            with st.spinner("Analyzing your resume..."):
+                resume_analyzer = create_resume_analyzer()
+                email_agent = create_email_agent()  # Create email agent here
+                
+                if resume_analyzer and email_agent:
+                    print("DEBUG: Starting resume analysis")
+                    is_selected, feedback, ats_score = analyze_resume_with_score(
+                        st.session_state.resume_text,
+                        role,
+                        resume_analyzer
+                    )
+                    print(f"DEBUG: Analysis complete - Selected: {is_selected}, Feedback: {feedback}")
+
+                    # Display resume text
+                    st.subheader("Resume Contents")
+                    st.text_area("Extracted Resume Text", st.session_state.resume_text, height=300, disabled=True)
+
+                    # Display ATS score
+                    st.subheader("ATS Score")
+                    st.metric(label="ATS Match Score", value=f"{ats_score}%", help="Percentage match with the job requirements")    
+
+                    if is_selected:
+                        st.success("Great! Candiate's skills match our requirements.")
+                        st.session_state.analysis_complete = True
+                        st.session_state.is_selected = True
+                        st.rerun()
+                    else:
+                        st.warning("Unfortunately, your skills don't match our requirements.")
+                        st.write(f"Feedback: {feedback}")
+                        
+                        # Send rejection email
+                        with st.spinner("Sending feedback email..."):
+                            try:
+                                send_rejection_email(
+                                    email_agent=email_agent,
+                                    to_email=email,
+                                    role=role,
+                                    feedback=feedback
+                                )
+                                st.info("We've sent you an email with detailed feedback.")
+                            except Exception as e:
+                                logger.error(f"Error sending rejection email: {e}")
+                                st.error("Could not send feedback email. Please try again.")
+
+    if st.session_state.get('analysis_complete') and st.session_state.get('is_selected', False):
+        st.success("Congratulations! Your skills match our requirements.")
+        st.info("Click 'Proceed with Application' to continue with the interview process.")
+        
+        if st.button("Proceed with Application", key="proceed_button"):
+            print("DEBUG: Proceed button clicked")  # Debug
+            with st.spinner("🔄 Processing your application..."):
+                try:
+                    print("DEBUG: Creating email agent")  # Debug
+                    email_agent = create_email_agent()
+                    print(f"DEBUG: Email agent created: {email_agent}")  # Debug
+                    
+                    print("DEBUG: Creating scheduler agent")  # Debug
+                    scheduler_agent = create_scheduler_agent()
+                    print(f"DEBUG: Scheduler agent created: {scheduler_agent}")  # Debug
+
+                    # 3. Send selection email
+                    with st.status("📧 Sending confirmation email...", expanded=True) as status:
+                        print(f"DEBUG: Attempting to send email to {st.session_state.candidate_email}")  # Debug
+                        send_selection_email(
+                            email_agent,
+                            st.session_state.candidate_email,
+                            role
+                        )
+                        print("DEBUG: Email sent successfully")  # Debug
+                        status.update(label="✅ Confirmation email sent!")
+
+                    # 4. Schedule interview
+                    with st.status("📅 Scheduling interview...", expanded=True) as status:
+                        print("DEBUG: Attempting to schedule interview")  # Debug
+                        schedule_interview(
+                            scheduler_agent,
+                            st.session_state.candidate_email,
+                            email_agent,
+                            role
+                        )
+                        print("DEBUG: Interview scheduled successfully")  # Debug
+                        status.update(label="✅ Interview scheduled!")
+
+                    print("DEBUG: All processes completed successfully")  # Debug
+                    st.success("""
+                        🎉 Application Successfully Processed!
+                        
+                        Tasks Done:
+                        1. Selection confirmation ✅
+                        2. Interview details with Zoom link 🔗
+                        
+                        Next steps:
+                        1. Join the interview 5 minutes early
+                        2. Evaluate the candiate
+                        3. Report the result to the HR
+                    """)
+
+                except Exception as e:
+                    print(f"DEBUG: Error occurred: {str(e)}")  # Debug
+                    print(f"DEBUG: Error type: {type(e)}")  # Debug
+                    import traceback
+                    print(f"DEBUG: Full traceback: {traceback.format_exc()}")  # Debug
+                    st.error(f"An error occurred: {str(e)}")
+                    st.error("Please try again or contact support.")
+
+    # Reset button
+    if st.sidebar.button("Reset Application"):
+        for key in st.session_state.keys():
+            if key != 'openai_api_key':
+                del st.session_state[key]
+        st.rerun()
+
+if __name__ == "__main__":
+    main()