Spaces:
Sleeping
Sleeping
File size: 4,678 Bytes
79d167d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 | from fastapi import APIRouter, HTTPException
import pandas as pd
from database import fetch_data
from utils.data_processing import preprocess_data
from logger import logger
# Initialize router
router = APIRouter()
def get_filtered_data(columns):
"""Fetches and filters data based on required columns."""
try:
raw_data = fetch_data()
df = preprocess_data(pd.DataFrame(raw_data))
missing_cols = [col for col in columns if col not in df.columns]
if missing_cols:
raise HTTPException(status_code=400, detail=f"Missing columns: {missing_cols}")
return df[columns].dropna()
except Exception as e:
logger.error(f"Error fetching data: {e}")
raise HTTPException(status_code=500, detail=f"Internal server error: {e}")
@router.get("/satisfaction-analysis")
def satisfaction_analysis():
try:
data = get_filtered_data(['Satisfaction Score', 'DepartmentType'])
result = data.groupby("DepartmentType")["Satisfaction Score"].mean().reset_index()
return result.to_dict(orient="records")
except Exception as e:
raise HTTPException(status_code=500, detail=f"Internal server error: {e}")
@router.get("/department-performance")
def department_performance():
try:
data = get_filtered_data(["Performance Score", "Current Employee Rating", "DepartmentType"])
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=f"Internal server error: {e}")
@router.get("/training-analytics")
def training_analytics():
try:
data = get_filtered_data(['Training Outcome', 'Training Cost'])
completion_rates = data.groupby("Training Cost")['Training Outcome'].value_counts(normalize=True).unstack(fill_value=0)
return completion_rates.to_dict()
except Exception as e:
raise HTTPException(status_code=500, detail=f"Internal server error: {e}")
@router.get("/engagement-performance-correlation")
def engagement_performance_correlation():
try:
data = get_filtered_data(['Engagement Score', 'Performance Score'])
correlation = data[['Engagement Score', 'Performance Score']].corr().iloc[0, 1]
return {"correlation_coefficient": correlation}
except Exception as e:
raise HTTPException(status_code=500, detail=f"Internal server error: {e}")
@router.get("/cost-benefit-analysis")
def cost_benefit_analysis():
try:
data = get_filtered_data(['Training Cost', 'Performance Score', 'DepartmentType'])
result = data.groupby("DepartmentType").apply(lambda g: g["Performance Score"].mean() / g["Training Cost"].sum()).reset_index(name="ROI")
return result.to_dict(orient="records")
except Exception as e:
raise HTTPException(status_code=500, detail=f"Internal server error: {e}")
@router.get("/training-effectiveness")
def training_effectiveness():
try:
data = get_filtered_data(['Training Outcome', 'Performance Score', 'Training Program Name'])
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=f"Internal server error: {e}")
@router.get("/diversity-dashboard")
def diversity_dashboard():
try:
data = get_filtered_data(['GenderCode', 'RaceDesc', 'DepartmentType'])
diversity_metrics = data.groupby("DepartmentType")['GenderCode'].value_counts(normalize=True).unstack(fill_value=0)
return diversity_metrics.to_dict()
except Exception as e:
raise HTTPException(status_code=500, detail=f"Internal server error: {e}")
@router.get("/worklife-balance-impact")
def worklife_balance_impact():
try:
data = get_filtered_data(['Work-Life Balance Score', 'Performance Score'])
correlation = data[['Work-Life Balance Score', 'Performance Score']].corr().iloc[0, 1]
return {"correlation_coefficient": correlation}
except Exception as e:
raise HTTPException(status_code=500, detail=f"Internal server error: {e}")
@router.get("/career-development")
def career_development():
try:
data = get_filtered_data(['Employee ID', 'StartDate'])
career_progress = 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=f"Internal server error: {e}")
|