113 lines
4.1 KiB
Python
113 lines
4.1 KiB
Python
"""
|
|
Data formatting functions for employee workstation activity tracking.
|
|
|
|
This module contains functions for formatting database query results into API responses.
|
|
"""
|
|
from flask import current_app
|
|
|
|
def format_report_data(results, time_period):
|
|
"""
|
|
Formats the raw database results into a list of dictionaries for the API.
|
|
|
|
Args:
|
|
results (list): List of database result rows (SQLAlchemy RowMapping objects)
|
|
time_period (str): Time period of the report ('daily', 'weekly', or 'monthly')
|
|
|
|
Returns:
|
|
list: List of formatted dictionaries for API response
|
|
"""
|
|
current_app.logger.debug(f"Formatting report data for period: {time_period}. Input rows: {len(results)}")
|
|
period_key_map = {
|
|
'daily': 'day',
|
|
'weekly': 'week_start',
|
|
'monthly': 'month_start'
|
|
}
|
|
period_key = period_key_map.get(time_period, 'period_start') # Default if unexpected period
|
|
|
|
# First convert raw rows to dictionaries
|
|
raw_data = []
|
|
for row in results:
|
|
# Ensure period_start is converted to string if it's a date/datetime object
|
|
period_value = row['period_start']
|
|
if hasattr(period_value, 'isoformat'):
|
|
period_value = period_value.isoformat()
|
|
|
|
# Format first_login_time
|
|
first_login_time = row['first_login_time']
|
|
if hasattr(first_login_time, 'isoformat'):
|
|
first_login_time = first_login_time.isoformat()
|
|
|
|
# Ensure duration_hours is a float, not a string or Decimal
|
|
duration_hours = row['total_hours']
|
|
if duration_hours is None:
|
|
duration_hours = 0.0
|
|
else:
|
|
# Convert to float explicitly to ensure it's JSON serializable as a number
|
|
duration_hours = float(duration_hours)
|
|
|
|
raw_data.append({
|
|
'user': row['user'],
|
|
period_key: period_value,
|
|
'duration_hours': duration_hours,
|
|
'first_login_time': first_login_time
|
|
})
|
|
|
|
# Additional preprocessing to consolidate any duplicate user entries
|
|
user_period_map = {}
|
|
for entry in raw_data:
|
|
user = entry['user']
|
|
period = entry[period_key]
|
|
key = f"{user}_{period}"
|
|
|
|
if key in user_period_map:
|
|
# Aggregate duration for existing user+period
|
|
user_period_map[key]['duration_hours'] += entry['duration_hours']
|
|
|
|
# Use the earliest first_login_time
|
|
existing_time = user_period_map[key]['first_login_time']
|
|
new_time = entry['first_login_time']
|
|
|
|
if existing_time and new_time:
|
|
if new_time < existing_time:
|
|
user_period_map[key]['first_login_time'] = new_time
|
|
else:
|
|
# New user+period combination
|
|
user_period_map[key] = entry
|
|
|
|
# Convert consolidated map back to list
|
|
formatted_data = list(user_period_map.values())
|
|
|
|
current_app.logger.debug(f"Formatted report data created. Output rows: {len(formatted_data)}")
|
|
return formatted_data
|
|
|
|
def format_user_activity(results):
|
|
"""
|
|
Formats the raw user activity results into a list of dictionaries.
|
|
|
|
Args:
|
|
results (list): List of database result rows (SQLAlchemy RowMapping objects)
|
|
|
|
Returns:
|
|
list: List of formatted user activity dictionaries
|
|
"""
|
|
formatted_data = []
|
|
for row in results:
|
|
start_time = row['start_time']
|
|
end_time = row['end_time']
|
|
|
|
# Format timestamps for display
|
|
if hasattr(start_time, 'isoformat'):
|
|
start_time = start_time.isoformat()
|
|
if hasattr(end_time, 'isoformat'):
|
|
end_time = end_time.isoformat()
|
|
|
|
# Format duration as float
|
|
duration = float(row['session_duration_hours']) if row['session_duration_hours'] is not None else 0.0
|
|
|
|
formatted_data.append({
|
|
'date': row['work_date'].isoformat() if hasattr(row['work_date'], 'isoformat') else str(row['work_date']),
|
|
'start_time': start_time,
|
|
'end_time': end_time,
|
|
'duration_hours': round(duration, 2)
|
|
})
|
|
return formatted_data |