170 lines
7.2 KiB
Python
170 lines
7.2 KiB
Python
import requests
|
|
import pandas as pd
|
|
from dotenv import load_dotenv
|
|
load_dotenv()
|
|
import os
|
|
|
|
DATA_KEY = os.getenv("AI_DATA_KEY")
|
|
# Constants for API requests
|
|
URL = "https://erpai.mkdlabs.com/v3/api/custom/erpai/common/get-data-ai"
|
|
HEADERS = {
|
|
"x-project": DATA_KEY # Replace with your actual key
|
|
}
|
|
|
|
# JSON bodies for API requests
|
|
def create_json_body(area_type, company_id):
|
|
return {
|
|
"type": area_type,
|
|
"options": {
|
|
"company_id": company_id
|
|
}
|
|
}
|
|
|
|
# Function to fetch data from the API
|
|
def fetch_data(json_body):
|
|
json_body["options"]["company_id"] = json_body["options"].get("company_id") # Ensure company_id is included
|
|
response = requests.post(URL, headers=HEADERS, json=json_body)
|
|
response.raise_for_status() # Raise an error for bad responses
|
|
return response.json()
|
|
|
|
def convert_assessment_data_to_dataframe(assessment_data):
|
|
if not assessment_data or "data" not in assessment_data or not assessment_data["data"]:
|
|
# Return empty DataFrame with expected columns
|
|
return pd.DataFrame(columns=[
|
|
"assessment_id", "assessment_name", "start_date", "open_items_overall",
|
|
"completed_items_overall", "total_assigned_items_overall", "user_name",
|
|
"user_total_assigned_items", "user_completed_items", "area", "red_flags"
|
|
])
|
|
|
|
df_assessment = []
|
|
for assessment in assessment_data.get("data", []):
|
|
assessment_id = assessment.get("assessment_id", "")
|
|
assessment_name = assessment.get("assessment_name", "")
|
|
start_date = assessment.get("start_date", "")
|
|
open_items = assessment.get("open_items", 0)
|
|
completed_items = assessment.get("completed_items", 0)
|
|
total_assigned_items = assessment.get("total_assigned_items", 0)
|
|
red_flags = assessment.get("red_flags", 0)
|
|
|
|
for user in assessment.get("user_details", []):
|
|
user_name = user.get("name", "")
|
|
user_total_items = user.get("total_assigned_items", 0)
|
|
user_completed_items = user.get("completed_items", 0)
|
|
|
|
for area in user.get("area_list", []):
|
|
df_assessment.append({
|
|
"assessment_id": assessment_id,
|
|
"assessment_name": assessment_name,
|
|
"start_date": start_date,
|
|
"open_items_overall": open_items,
|
|
"completed_items_overall": completed_items,
|
|
"total_assigned_items_overall": total_assigned_items,
|
|
"user_name": user_name,
|
|
"user_total_assigned_items": user_total_items,
|
|
"user_completed_items": user_completed_items,
|
|
"area": area,
|
|
"red_flags": red_flags
|
|
})
|
|
return pd.DataFrame(df_assessment)
|
|
|
|
def generate_summary_statistics(df):
|
|
if df.empty:
|
|
return {
|
|
"total_assessments": 0,
|
|
"avg_open_items_per_assessment": 0,
|
|
"avg_completed_items_per_assessment": 0,
|
|
"avg_total_assigned_items_per_assessment": 0,
|
|
"avg_red_flags": 0,
|
|
"total_users": 0,
|
|
"avg_user_total_assigned_items": 0,
|
|
"avg_user_completed_items": 0,
|
|
"completion_rate_per_user": 0,
|
|
"area_summary": {}
|
|
}
|
|
|
|
total_assessments = df['assessment_id'].nunique()
|
|
avg_open_items = df.groupby('assessment_id')['open_items_overall'].mean().mean()
|
|
avg_completed_items = df.groupby('assessment_id')['completed_items_overall'].mean().mean()
|
|
avg_total_assigned_items = df.groupby('assessment_id')['total_assigned_items_overall'].mean().mean()
|
|
avg_red_flags = df['red_flags'].mean()
|
|
|
|
total_users = df['user_name'].nunique()
|
|
avg_user_total_items = df.groupby('user_name')['user_total_assigned_items'].mean().mean()
|
|
avg_user_completed_items = df.groupby('user_name')['user_completed_items'].mean().mean()
|
|
completion_rate_per_user = (df['user_completed_items'].sum() / df['user_total_assigned_items'].sum()) * 100 if df['user_total_assigned_items'].sum() > 0 else 0
|
|
|
|
area_summary = df['area'].value_counts()
|
|
|
|
return {
|
|
"total_assessments": total_assessments,
|
|
"avg_open_items_per_assessment": avg_open_items,
|
|
"avg_completed_items_per_assessment": avg_completed_items,
|
|
"avg_total_assigned_items_per_assessment": avg_total_assigned_items,
|
|
"avg_red_flags": avg_red_flags,
|
|
"total_users": total_users,
|
|
"avg_user_total_assigned_items": avg_user_total_items,
|
|
"avg_user_completed_items": avg_user_completed_items,
|
|
"completion_rate_per_user": completion_rate_per_user,
|
|
"area_summary": area_summary.to_dict()
|
|
}
|
|
|
|
def generate_extended_statistics(df):
|
|
if df.empty:
|
|
return {
|
|
"top_5_efficient_users": {},
|
|
"bottom_5_least_efficient_users": {},
|
|
"areas_with_most_uncompleted_items": {}
|
|
}
|
|
|
|
df['user_completion_rate'] = (df['user_completed_items'] / df['user_total_assigned_items']).fillna(0) * 100
|
|
top_5_efficient_users = df.groupby('user_name')['user_completion_rate'].mean().nlargest(5).to_dict()
|
|
bottom_5_least_efficient_users = df.groupby('user_name')['user_completion_rate'].mean().nsmallest(5).to_dict()
|
|
|
|
df['uncompleted_items'] = df['user_total_assigned_items'] - df['user_completed_items']
|
|
areas_with_most_uncompleted_items = df.groupby('area')['uncompleted_items'].sum().nlargest(5).to_dict()
|
|
|
|
return {
|
|
"top_5_efficient_users": top_5_efficient_users,
|
|
"bottom_5_least_efficient_users": bottom_5_least_efficient_users,
|
|
"areas_with_most_uncompleted_items": areas_with_most_uncompleted_items
|
|
}
|
|
|
|
def generate_problematic_area_statistics(df):
|
|
if df.empty:
|
|
return pd.DataFrame(columns=["total_open_items", "total_red_flags"])
|
|
|
|
total_open_items = df.groupby('name')['open_items'].sum().sort_values(ascending=False)
|
|
total_red_flags = df.groupby('name')['red_flags'].sum().sort_values(ascending=False)
|
|
|
|
return pd.DataFrame({
|
|
"total_open_items": total_open_items,
|
|
"total_red_flags": total_red_flags
|
|
}).fillna(0)
|
|
|
|
def generate_summary_stats(assessment_data, area_data):
|
|
# Handle empty or invalid assessment data
|
|
assessment_df = convert_assessment_data_to_dataframe(assessment_data)
|
|
|
|
# Handle empty or invalid area data
|
|
problematic_area_df = pd.DataFrame(area_data.get("data", []) if area_data and "data" in area_data else [])
|
|
|
|
summary_stats = generate_summary_statistics(assessment_df)
|
|
extended_stats = generate_extended_statistics(assessment_df)
|
|
summary_stats["users(Workers) based stats"] = extended_stats
|
|
|
|
problematic_stats = generate_problematic_area_statistics(problematic_area_df)
|
|
summary_stats["Area based stats"] = problematic_stats.to_dict(orient='index')
|
|
|
|
return summary_stats
|
|
|
|
|
|
if __name__ == "__main__":
|
|
json_body_area = create_json_body("problematic-areas", 106)
|
|
json_body_assessment = create_json_body("user-stats-by-assessment", 106)
|
|
|
|
# Fetching data
|
|
problematic_areas_data = fetch_data(json_body_area)
|
|
assessment_data = fetch_data(json_body_assessment)
|
|
|
|
example_result = generate_summary_stats(assessment_data, problematic_areas_data)
|
|
print(example_result) |