commot

.env

@@ -0,0 +1,2 @@
+SUPABASE_URL=https://qxvpaoeakhddzabctekw.supabase.co
+SUPABASE_KEY=eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJpc3MiOiJzdXBhYmFzZSIsInJlZiI6InF4dnBhb2Vha2hkZHphYmN0ZWt3Iiwicm9sZSI6ImFub24iLCJpYXQiOjE3NDEwNjU2MzEsImV4cCI6MjA1NjY0MTYzMX0.I3GsBjFRfuBKw-KxmSJ7R5iKn2cgGegqIls2Bf32UpI

Dockerfile

@@ -0,0 +1,13 @@
+FROM python:3.9
+
+RUN useradd -m -u 1000 user
+USER user
+ENV PATH="/home/user/.local/bin:$PATH"
+
+WORKDIR /app/folder1
+
+COPY --chown=user ./requirements.txt requirements.txt
+RUN pip install --no-cache-dir --upgrade -r requirements.txt
+
+COPY --chown=user . /app
+CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "7860"]

config.py

@@ -0,0 +1,8 @@
+import os
+from dotenv import load_dotenv
+
+# Load environment variables
+load_dotenv()
+
+SUPABASE_URL = os.getenv("SUPABASE_URL")
+SUPABASE_KEY = os.getenv("SUPABASE_KEY")

database.py

@@ -0,0 +1,13 @@
+import os
+from dotenv import load_dotenv
+from supabase import create_client, Client
+
+# Load environment variables
+load_dotenv()
+
+# Read Supabase credentials
+SUPABASE_URL = os.getenv("SUPABASE_URL")
+SUPABASE_KEY = os.getenv("SUPABASE_KEY")
+
+# Initialize Supabase client
+supabase: Client = create_client(SUPABASE_URL, SUPABASE_KEY)

logger.py

@@ -0,0 +1,8 @@
+import logging
+
+logging.basicConfig(
+    filename="logs/app.log",
+    level=logging.INFO,
+    format="%(asctime)s - %(levelname)s - %(message)s",
+)
+logger = logging.getLogger(__name__)

main.py

@@ -0,0 +1,10 @@
+from fastapi import FastAPI
+from routes.analytics import router as analytics_router
+
+app = FastAPI()
+
+app.include_router(analytics_router)
+
+@app.get("/")
+def home():
+    return {"message": "HR Analytics API is running"}

models/schemas.py

@@ -0,0 +1,10 @@
+from pydantic import BaseModel
+from typing import Optional
+
+class HRAnalysis(BaseModel):
+    Employee_ID: str
+    DepartmentType: Optional[str]
+    Satisfaction_Score: Optional[float]
+    Performance_Score: Optional[int]
+    Training_Program_Name: Optional[str]
+    Training_Outcome: Optional[str]

requirements.txt

@@ -0,0 +1,5 @@
+fastapi
+uvicorn
+pandas
+supabase
+python-dotenv

routes/analytics.py

@@ -0,0 +1,141 @@
+from fastapi import APIRouter, HTTPException, Query
+import pandas as pd
+from database import supabase
+
+router = APIRouter()
+
+# Fetch data from Supabase
+try:
+    response = supabase.table("HR analysis").select("*").execute()
+    data = pd.DataFrame(response.data) if response.data else pd.DataFrame()
+except Exception as e:
+    print(f"Error fetching data: {e}")
+    data = pd.DataFrame()
+
+# Convert date columns
+for col in ['Survey Date', 'StartDate', 'DOB']:
+    if col in data.columns:
+        data[col] = pd.to_datetime(data[col], errors='coerce')
+
+# Calculate Age
+if 'DOB' in data.columns:
+    data['Age'] = (pd.to_datetime("today") - data['DOB']).dt.days // 365
+
+# Clean Performance Score
+score_map = {"Exceeds": 5, "Fully Meets": 4, "Needs Improvement": 3, "PIP": 2}
+if 'Performance Score' in data.columns:
+    data['Performance Score'] = data['Performance Score'].map(lambda x: score_map.get(str(x).strip(), None))
+    data['Performance Score'] = pd.to_numeric(data['Performance Score'], errors='coerce')
+
+@router.get("/satisfaction-analysis")
+def satisfaction_analysis(department: str = Query(None, description="Filter by department")):
+    try:
+        if "DepartmentType" not in data.columns or "Satisfaction Score" not in data.columns:
+            raise HTTPException(status_code=500, detail="Required columns missing in dataset")
+
+        filtered_data = data.copy()
+
+        if department:
+            department = department.strip().title()
+            filtered_data = filtered_data[filtered_data["DepartmentType"].str.strip().str.title() == department]
+
+        if filtered_data.empty:
+            return []
+
+        result = filtered_data.groupby("DepartmentType")["Satisfaction Score"].mean().reset_index()
+        return result.to_dict(orient="records")
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+
+@router.get("/department-performance")
+def department_performance():
+    try:
+        result = data.groupby("DepartmentType")[["Performance Score", "Current Employee Rating"]].mean().reset_index()
+        return result.to_dict(orient="records")
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+
+@router.get("/training-analytics")
+def training_analytics(program_name: str = Query(None, description="Filter by training program name")):
+    try:
+        filtered_data = data if program_name is None else data[data["Training Program Name"] == program_name]
+
+        if filtered_data.empty:
+            return []
+
+        result = filtered_data.groupby("Training Program Name")["Training Outcome"].value_counts(normalize=True).unstack(fill_value=0)
+        return result.reset_index().to_dict(orient="records")
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+
+
+
+@router.get("/engagement-performance")
+def engagement_performance():
+    try:
+        correlation = data[['Engagement Score', 'Performance Score']].corr().iloc[0, 1]
+        return {"correlation_coefficient": correlation}
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+
+@router.get("/cost-benefit-analysis")
+def cost_benefit_analysis():
+    try:
+        result = data.groupby("DepartmentType").apply(lambda x: x['Performance Score'].mean() / x['Training Cost'].sum()).reset_index(name="ROI")
+        return result.to_dict(orient="records")
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+
+@router.get("/training-effectiveness")
+def training_effectiveness():
+    try:
+        result = data.groupby("Training Program Name")["Performance Score"].mean().reset_index()
+        return result.to_dict(orient="records")
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+
+@router.get("/diversity-inclusion")
+def diversity_dashboard():
+    try:
+        if "DepartmentType" not in data.columns or "GenderCode" not in data.columns:
+            raise HTTPException(status_code=500, detail="Required columns missing in dataset")
+
+        # Compute gender distribution by department
+        diversity_metrics = data.groupby("DepartmentType")["GenderCode"].value_counts(normalize=True).unstack(fill_value=0).reset_index()
+
+        return diversity_metrics.to_dict(orient="records")
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+
+
+@router.get("/work-life-balance")
+def worklife_balance_impact():
+    try:
+        correlation = data[['Work-Life Balance Score', 'Performance Score']].corr().iloc[0, 1]
+        return {"correlation_coefficient": round(correlation, 3)}  # Return as a JSON object
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+
+
+@router.get("/career-development")
+def career_development(employee_id: str = Query(None, description="Filter by Employee ID")):
+    try:
+        if "Employee ID" not in data.columns or "StartDate" not in data.columns:
+            raise HTTPException(status_code=500, detail="Required columns missing in dataset")
+
+        # Print available Employee IDs for debugging
+        print("Available Employee IDs:", data["Employee ID"].unique())
+
+        filtered_data = data.copy()
+
+        if employee_id:
+            employee_id = employee_id.strip()  # Remove leading/trailing spaces
+            filtered_data = filtered_data[filtered_data["Employee ID"].astype(str) == employee_id]
+
+        if filtered_data.empty:
+            return []  # Return an empty list if no matching records
+
+        career_progress = filtered_data.groupby("Employee ID")["StartDate"].count().reset_index(name="Career Movements")
+        return career_progress.to_dict(orient="records")
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))

utils/data_processing.py

@@ -0,0 +1,19 @@
+import pandas as pd
+
+def convert_dates(df, columns):
+    for col in columns:
+        if col in df.columns:
+            df[col] = pd.to_datetime(df[col], errors='coerce')
+    return df
+
+def calculate_age(df, dob_col="DOB"):
+    if dob_col in df.columns:
+        df["Age"] = (pd.to_datetime("today") - df[dob_col]).dt.days // 365
+    return df
+
+def clean_performance_score(df, col="Performance Score"):
+    score_map = {"Exceeds": 5, "Fully Meets": 4, "Needs Improvement": 3, "PIP": 2}
+    if col in df.columns:
+        df[col] = df[col].map(lambda x: score_map.get(str(x).strip(), None))
+        df[col] = pd.to_numeric(df[col], errors='coerce')
+    return df