app.py

import gradio as gr
import pandas as pd
import datetime
import numpy as np
import docx
from PyPDF2 import PdfReader
from sentence_transformers import SentenceTransformer, util

class AIHRAgent:
    def __init__(self):
        # Advanced model for semantic similarity
        self.resume_model = SentenceTransformer("sentence-transformers/all-MiniLM-L6-v2")
        self.employee_records = pd.DataFrame(columns=["Name", "Position", "Start Date", "Attendance", "Performance", "Leaves"])
        self.company_policies = "Employees are entitled to 24 annual leaves and must adhere to company policies regarding attendance and punctuality."

    def extract_text_from_file(self, file_path):
        """Extract text from uploaded file (PDF or DOCX)."""
        try:
            if file_path.name.endswith(".pdf"):
                pdf_reader = PdfReader(file_path)
                text = " ".join(page.extract_text() for page in pdf_reader.pages if page.extract_text())
            elif file_path.name.endswith(".docx"):
                doc = docx.Document(file_path)
                text = " ".join(paragraph.text for paragraph in doc.paragraphs)
            else:
                raise ValueError("Unsupported file format. Please upload a PDF or DOCX file.")
            return text
        except Exception as e:
            return f"Error extracting text from file: {e}"

    def screen_resume(self, resume_text, job_description):
        """Advanced resume screening using sentence embeddings."""
        try:
            if not resume_text or not job_description:
                return "Please provide both the resume text and job description."

            # Semantic similarity scoring
            job_embedding = self.resume_model.encode(job_description, convert_to_tensor=True)
            resume_embedding = self.resume_model.encode(resume_text, convert_to_tensor=True)
            similarity = util.pytorch_cos_sim(job_embedding, resume_embedding).item()
            return f"Relevance Score: {similarity:.2f} for the position of {job_description}."
        except Exception as e:
            return f"Error during resume screening: {e}"

    def onboarding_guide(self, employee_name, position):
        """Automated onboarding guide generation."""
        return (f"Welcome {employee_name}!\n"
                f"As a {position}, your onboarding plan includes:\n"
                f"1. Orientation session.\n"
                f"2. Team introductions.\n"
                f"3. Work system setup.\n"
                f"4. Initial training and goal setting.")

    def add_employee(self, name, position, start_date):
        new_employee = {
            "Name": name,
            "Position": position,
            "Start Date": start_date,
            "Attendance": 0,
            "Performance": "Not Reviewed",
            "Leaves": 0
        }
        self.employee_records = self.employee_records.append(new_employee, ignore_index=True)
        return f"Employee {name} added successfully."

    def track_attendance(self, employee_name):
        if employee_name in self.employee_records["Name"].values:
            self.employee_records.loc[self.employee_records["Name"] == employee_name, "Attendance"] += 1
            return f"Attendance recorded for {employee_name}."
        return f"Employee {employee_name} not found."

    def process_payroll(self, employee_name, base_salary):
        if employee_name in self.employee_records["Name"].values:
            tax = base_salary * 0.1
            net_salary = base_salary - tax
            return f"Payroll Processed: Gross Salary = {base_salary}, Tax = {tax}, Net Salary = {net_salary}."
        return f"Employee {employee_name} not found."

    def pulse_survey(self):
        return "Pulse Survey: On a scale of 1-5, how satisfied are you with your current role?"

    def feedback_analysis(self, feedback_scores):
        avg_score = np.mean(feedback_scores)
        return f"Average Engagement Score: {avg_score:.2f}. Action Needed: {'Yes' if avg_score < 3 else 'No'}."

    def performance_review(self, employee_name, review_score):
        if employee_name in self.employee_records["Name"].values:
            self.employee_records.loc[self.employee_records["Name"] == employee_name, "Performance"] = review_score
            return f"Performance of {employee_name} updated to {review_score}."
        return f"Employee {employee_name} not found."

    def get_policy(self):
        return self.company_policies

    def exit_interview(self, employee_name, feedback):
        if employee_name in self.employee_records["Name"].values:
            self.employee_records = self.employee_records[self.employee_records["Name"] != employee_name]
            return f"Exit interview recorded for {employee_name}. Feedback: {feedback}"
        return f"Employee {employee_name} not found."

# AI HR Agent Instance
ai_hr = AIHRAgent()

# Gradio Interface
def gradio_interface():
    with gr.Blocks() as interface:
        gr.Markdown("# **Advanced AI HR Agent**")
        gr.Markdown("This AI automates all HR tasks and provides advanced features such as resume screening and policy management.")

        with gr.Tab("Recruitment and Onboarding"):
            with gr.Row():
                with gr.Column():
                    resume_upload = gr.File(label="Upload Resume (PDF/DOCX)")
                    job_description_input = gr.Textbox(label="Job Description")
                    resume_screen_output = gr.Textbox(label="Screening Result")
                    screen_button = gr.Button("Screen Resume")

                with gr.Column():
                    onboarding_name = gr.Textbox(label="Employee Name")
                    onboarding_position = gr.Textbox(label="Position")
                    onboarding_output = gr.Textbox(label="Onboarding Guide")
                    onboarding_button = gr.Button("Generate Onboarding Guide")

        with gr.Tab("Employee Management"):
            add_name = gr.Textbox(label="Employee Name")
            add_position = gr.Textbox(label="Position")
            add_start_date = gr.Textbox(label="Start Date (YYYY-MM-DD)")
            add_output = gr.Textbox(label="Add Employee Result")
            add_button = gr.Button("Add Employee")

            attendance_name = gr.Textbox(label="Employee Name for Attendance")
            attendance_output = gr.Textbox(label="Attendance Result")
            attendance_button = gr.Button("Record Attendance")

        with gr.Tab("Payroll Management"):
            payroll_name = gr.Textbox(label="Employee Name")
            payroll_salary = gr.Number(label="Base Salary")
            payroll_output = gr.Textbox(label="Payroll Result")
            payroll_button = gr.Button("Process Payroll")

        with gr.Tab("Exit Management"):
            exit_name = gr.Textbox(label="Employee Name")
            exit_feedback = gr.Textbox(label="Exit Feedback")
            exit_output = gr.Textbox(label="Exit Interview Result")
            exit_button = gr.Button("Record Exit Interview")

        # Button Actions
        screen_button.click(
            lambda file, job_desc: ai_hr.screen_resume(ai_hr.extract_text_from_file(file), job_desc) if file else "No resume file uploaded.",
            inputs=[resume_upload, job_description_input],
            outputs=resume_screen_output,
        )
        onboarding_button.click(ai_hr.onboarding_guide, inputs=[onboarding_name, onboarding_position], outputs=onboarding_output)
        add_button.click(ai_hr.add_employee, inputs=[add_name, add_position, add_start_date], outputs=add_output)
        attendance_button.click(ai_hr.track_attendance, inputs=attendance_name, outputs=attendance_output)
        payroll_button.click(ai_hr.process_payroll, inputs=[payroll_name, payroll_salary], outputs=payroll_output)
        exit_button.click(ai_hr.exit_interview, inputs=[exit_name, exit_feedback], outputs=exit_output)

    return interface

# Launch Interface
interface = gradio_interface()
interface.launch(share=True)