from django.shortcuts import render from rest_framework import viewsets, status from rest_framework.decorators import api_view, permission_classes, authentication_classes from rest_framework.permissions import AllowAny from rest_framework.authentication import SessionAuthentication from rest_framework.response import Response from django.db import transaction from django.db.models import Prefetch from .models import Stamp from .serializers import StampSerializer from .permissions import get_user_stamp_permissions, check_stamp_permission from ErrorLogs.utils import log_error from functions.models import Function from worktimeservice.models import WorkBalance from configurations.models import Configuration, ConfigurationTranslation from balancedetail.models import BalanceDetail from paycode.models import Paycode from company.models import WorkingTimePolicy, VacationPolicy from datetime import datetime, timedelta, date from django.utils import timezone import calendar from django.db.models import Q, Sum, Count, Avg, F from user.models import User from projects.models import Project from clients.models import Client from django.http import HttpResponse from openpyxl import Workbook from openpyxl.styles import Font, Alignment, PatternFill, Border, Side from openpyxl.utils import get_column_letter from reportlab.lib import colors from reportlab.lib.pagesizes import letter, A4 from reportlab.platypus import SimpleDocTemplate, Table, TableStyle, Paragraph, Spacer, PageBreak from reportlab.lib.styles import getSampleStyleSheet, ParagraphStyle from reportlab.lib.units import inch import io import uuid import re def get_configuration_data_for_functions(stamp_functions, user_language='en'): """ Helper function to consistently fetch configuration data (header, icon, color) for given stamp_functions, applying translations if needed. Returns a dictionary: {function_ref_id: {'header': ..., 'icon': ..., 'color': ...}} """ configurations_dict = {} if stamp_functions: configurations = Configuration.objects.filter(function_ref_id__in=stamp_functions) for config in configurations: header = config.header # Apply translation if language is not English if user_language != 'en': try: translation = ConfigurationTranslation.objects.filter( translation_type='function_header', language=user_language, reference_id=config.function_ref_id ).first() if translation: header = translation.translated_text except Exception: pass # If translation lookup fails, use original # Always include icon and color, even if they're empty strings # This ensures we return exactly what's in the database configurations_dict[config.function_ref_id] = { 'header': header, 'icon': str(config.icon) if config.icon is not None else '', # Return exactly as defined (empty string if empty) 'color': str(config.color) if config.color is not None else '' # Return exactly as defined (empty string if empty) } return configurations_dict # Helper function to convert seconds to HH:MM format def seconds_to_hhmm(seconds): """Convert seconds to HH:MM format""" if seconds is None or seconds == 0: return "00:00" hours = int(seconds // 3600) minutes = int((seconds % 3600) // 60) return f"{hours:02d}:{minutes:02d}" def calculate_expected_working_hours(start_date, end_date, working_policy, user_id=None): """ Calculate expected working hours for a date range based on working time policy. If working_policy is None, will fetch user's policy or company default. """ # If no policy provided, try to get it for the user if not working_policy and user_id: from worktimeservice.service import get_active_working_policy working_policy = get_active_working_policy(user_id) if not working_policy: # Fallback to default 8 hours per weekday total_days = (end_date - start_date).days + 1 weekdays = sum(1 for i in range(total_days) if (start_date + timedelta(days=i)).weekday() < 5) return weekdays * 8 * 3600 # Convert to seconds total_seconds = 0 current_date = start_date while current_date <= end_date: weekday = current_date.weekday() # 0=Monday, 6=Sunday if weekday < 5: # Monday to Friday start_time = working_policy.monday_friday_start end_time = working_policy.monday_friday_end elif weekday == 5: # Saturday start_time = working_policy.saturday_start end_time = working_policy.saturday_end else: # Sunday start_time = working_policy.sunday_start end_time = working_policy.sunday_end # Calculate working hours for this day start_datetime = datetime.combine(current_date, start_time) end_datetime = datetime.combine(current_date, end_time) # Handle case where end time is next day (e.g., night shift) if end_datetime <= start_datetime: end_datetime += timedelta(days=1) day_seconds = (end_datetime - start_datetime).total_seconds() total_seconds += day_seconds current_date += timedelta(days=1) return total_seconds # Create your views here. class StampViewSet(viewsets.ModelViewSet): serializer_class = StampSerializer def get_queryset(self): """Override queryset to include function names via join""" return Stamp.objects.select_related('user').all() def list(self, request, *args, **kwargs): try: # Get query parameters user_id = request.GET.get('user_id') date = request.GET.get('date') start_date = request.GET.get('start_date') end_date = request.GET.get('end_date') # Start with base queryset queryset = Stamp.objects.select_related('user').all() # Apply filters based on query parameters if user_id: queryset = queryset.filter(user_id=user_id) if date: queryset = queryset.filter(date=date) elif start_date and end_date: # Handle date range filter try: start_date_obj = datetime.strptime(start_date, '%Y-%m-%d').date() end_date_obj = datetime.strptime(end_date, '%Y-%m-%d').date() queryset = queryset.filter(date__range=[start_date_obj, end_date_obj]) except ValueError: # Invalid date format, ignore the filter pass # Get user's language preference (fetch fresh from database) # Priority: 1) user_id from query params, 2) request.user user_language = 'en' try: target_user_id = None # First, try to get language from user_id in query params (the user whose stamps we're fetching) if user_id: try: from user.models import User target_user = User.objects.filter(id=user_id).only('lang').first() if target_user and target_user.lang: user_language = target_user.lang target_user_id = user_id except Exception: pass # If we didn't get language from query params, try request.user if not target_user_id and request.user and hasattr(request.user, 'lang'): cached_lang = request.user.lang if cached_lang: # Try to fetch fresh from database, but don't fail if it doesn't work try: if hasattr(request.user, 'id') and request.user.id: from user.models import User user = User.objects.filter(id=request.user.id).only('lang').first() if user and user.lang: user_language = user.lang else: user_language = cached_lang or 'en' else: user_language = cached_lang or 'en' except Exception: # If database fetch fails, use cached value user_language = cached_lang or 'en' # If no cached value, try database fetch elif request.user and hasattr(request.user, 'id') and request.user.id: try: from user.models import User user = User.objects.filter(id=request.user.id).only('lang').first() if user and user.lang: user_language = user.lang except Exception: pass except Exception: # If anything goes wrong, just use default pass # Get all unique stamp_function values from filtered queryset stamp_functions = set(queryset.values_list('stamp_function', flat=True)) # Prefetch configuration data for all stamp_functions configurations_dict = get_configuration_data_for_functions(stamp_functions, user_language) # Create a custom serializer context with configuration headers serializer = self.get_serializer(queryset, many=True, context={'functions_dict': configurations_dict}) data = serializer.data return Response(data) except Exception as e: log_error('StampViewSet.list', e) return Response( {"error": "An error occurred while fetching stamps"}, status=status.HTTP_500_INTERNAL_SERVER_ERROR ) def create(self, request, *args, **kwargs): try: # Check permission to add stamps if request.user and hasattr(request.user, 'id') and request.user.id: if not check_stamp_permission(request.user.id, 'add'): return Response( {"error": "You do not have permission to add stamps"}, status=status.HTTP_403_FORBIDDEN ) with transaction.atomic(): return super().create(request, *args, **kwargs) except Exception as e: log_error('StampViewSet.create', e) return Response( {"error": "An error occurred while creating stamp"}, status=status.HTTP_500_INTERNAL_SERVER_ERROR ) def retrieve(self, request, *args, **kwargs): # CRITICAL: Validate pk BEFORE any database operations pk = kwargs.get('pk') if not pk: return Response( {"error": "Stamp ID is required"}, status=status.HTTP_400_BAD_REQUEST ) # Convert to string and normalize pk_str = str(pk).strip().lower() # Reject known non-UUID route names IMMEDIATELY non_uuid_routes = ['permissions', 'worktime-report', 'todays-progress', 'monthly-balance', 'balance-screen', 'approve-day', 'cancel-approval-day', 'excel', 'pdf'] if pk_str in non_uuid_routes: return Response( {"error": f"'{pk}' is not a valid stamp ID"}, status=status.HTTP_404_NOT_FOUND ) # Validate UUID format - must match UUID pattern exactly # UUID format: 8-4-4-4-12 hexadecimal characters uuid_pattern = r'^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$' if not re.match(uuid_pattern, pk_str, re.IGNORECASE): return Response( {"error": f"'{pk}' is not a valid UUID format"}, status=status.HTTP_404_NOT_FOUND ) # Now safe to proceed with database query try: instance = Stamp.objects.get(pk=pk) except Stamp.DoesNotExist: return Response( {"error": "Stamp not found"}, status=status.HTTP_404_NOT_FOUND ) try: # Get user's language preference (fetch fresh from database) user_language = 'en' try: # First try cached value (fast and reliable) if request.user and hasattr(request.user, 'lang'): cached_lang = request.user.lang if cached_lang: # Try to fetch fresh from database, but don't fail if it doesn't work try: if hasattr(request.user, 'id') and request.user.id: from user.models import User user = User.objects.filter(id=request.user.id).only('lang').first() if user and user.lang: user_language = user.lang else: user_language = cached_lang or 'en' else: user_language = cached_lang or 'en' except Exception: # If database fetch fails, use cached value user_language = cached_lang or 'en' # If no cached value, try database fetch elif request.user and hasattr(request.user, 'id') and request.user.id: try: from user.models import User user = User.objects.filter(id=request.user.id).only('lang').first() if user and user.lang: user_language = user.lang except Exception: pass except Exception: # If anything goes wrong, just use default pass # Get configuration data for this stamp (header, icon, color) - same way as header configurations_dict = {} if instance.stamp_function: # Use the same helper function to ensure consistency config_data = get_configuration_data_for_functions([instance.stamp_function], user_language) if config_data: configurations_dict = config_data else: configurations_dict = {instance.stamp_function: None} # Serialize with configuration header serializer = self.get_serializer(instance, context={'functions_dict': configurations_dict}) return Response(serializer.data) except Exception as e: log_error('StampViewSet.retrieve', e) return Response( {"error": "An error occurred while fetching stamp details"}, status=status.HTTP_500_INTERNAL_SERVER_ERROR ) def update(self, request, *args, **kwargs): try: # Check permission to edit stamps if request.user and hasattr(request.user, 'id') and request.user.id: if not check_stamp_permission(request.user.id, 'edit'): return Response( {"error": "You do not have permission to edit stamps"}, status=status.HTTP_403_FORBIDDEN ) with transaction.atomic(): return super().update(request, *args, **kwargs) except Exception as e: log_error('StampViewSet.update', e) return Response( {"error": "An error occurred while updating stamp"}, status=status.HTTP_500_INTERNAL_SERVER_ERROR ) def destroy(self, request, *args, **kwargs): try: # Check permission to delete stamps if request.user and hasattr(request.user, 'id') and request.user.id: if not check_stamp_permission(request.user.id, 'delete'): return Response( {"error": "You do not have permission to delete stamps"}, status=status.HTTP_403_FORBIDDEN ) with transaction.atomic(): # Get the stamp instance before deletion instance = self.get_object() user_id = instance.user_id stamp_date = instance.date print(f"Deleting stamp ID: {instance.id}, User: {user_id}, Date: {stamp_date}") # Delete the stamp response = super().destroy(request, *args, **kwargs) # Trigger immediate recalculation for the affected date if stamp_date: from worktimeservice.service import calculate_daily_balances_for_user_and_date calculate_daily_balances_for_user_and_date(user_id, stamp_date) return response except Exception as e: log_error('StampViewSet.destroy', e) print(f"Error in destroy: {e}") return Response( {"error": "An error occurred while deleting stamp"}, status=status.HTTP_500_INTERNAL_SERVER_ERROR ) @api_view(['POST']) @permission_classes([AllowAny]) def approve_day(request): """ Approve all stamps for a user on a specific date Request body: { user_id: string, date: string (YYYY-MM-DD) } """ try: user_id = request.data.get('user_id') date_str = request.data.get('date') if not user_id: return Response( {"error": "user_id is required"}, status=status.HTTP_400_BAD_REQUEST ) if not date_str: return Response( {"error": "date is required"}, status=status.HTTP_400_BAD_REQUEST ) # Parse the date try: stamp_date = datetime.strptime(date_str, '%Y-%m-%d').date() except ValueError: return Response( {"error": "Invalid date format. Use YYYY-MM-DD"}, status=status.HTTP_400_BAD_REQUEST ) # Find all stamps for this user on this date stamps = Stamp.objects.filter(user_id=user_id, date=stamp_date) if not stamps.exists(): return Response( {"error": "No stamps found for this user on this date"}, status=status.HTTP_404_NOT_FOUND ) # Update all stamps to status 'A' (Approved) updated_count = stamps.update(status='A') # Trigger balance recalculation for the affected date # If balance calculation fails, we still return success since stamps were approved balance_error = None if updated_count > 0: try: from worktimeservice.service import calculate_daily_balances_for_user_and_date calculate_daily_balances_for_user_and_date(user_id, stamp_date) except Exception as balance_e: # Log the balance calculation error but don't fail the approval log_error('approve_day balance_calculation', balance_e) balance_error = str(balance_e) response_data = { "message": f"Successfully approved {updated_count} stamp(s)", "user_id": user_id, "date": date_str, "updated_count": updated_count } if balance_error: response_data["warning"] = f"Balance recalculation failed: {balance_error}" return Response(response_data, status=status.HTTP_200_OK) except Exception as e: log_error('approve_day', e) import traceback error_details = str(e) traceback_str = traceback.format_exc() # Log full traceback for debugging print(f"Error in approve_day: {error_details}\n{traceback_str}") return Response( {"error": f"An error occurred while approving day: {error_details}"}, status=status.HTTP_500_INTERNAL_SERVER_ERROR ) @api_view(['POST']) @permission_classes([AllowAny]) def cancel_approval_day(request): """ Cancel approval for all stamps for a user on a specific date Request body: { user_id: string, date: string (YYYY-MM-DD) } """ try: user_id = request.data.get('user_id') date_str = request.data.get('date') if not user_id: return Response( {"error": "user_id is required"}, status=status.HTTP_400_BAD_REQUEST ) if not date_str: return Response( {"error": "date is required"}, status=status.HTTP_400_BAD_REQUEST ) # Parse the date try: stamp_date = datetime.strptime(date_str, '%Y-%m-%d').date() except ValueError: return Response( {"error": "Invalid date format. Use YYYY-MM-DD"}, status=status.HTTP_400_BAD_REQUEST ) # Find all stamps for this user on this date stamps = Stamp.objects.filter(user_id=user_id, date=stamp_date) if not stamps.exists(): return Response( {"error": "No stamps found for this user on this date"}, status=status.HTTP_404_NOT_FOUND ) # Update all stamps to status '' (Pending/Unapproved) updated_count = stamps.update(status='') # Trigger balance recalculation for the affected date if updated_count > 0: from worktimeservice.service import calculate_daily_balances_for_user_and_date calculate_daily_balances_for_user_and_date(user_id, stamp_date) return Response({ "message": f"Successfully canceled approval for {updated_count} stamp(s)", "user_id": user_id, "date": date_str, "updated_count": updated_count }, status=status.HTTP_200_OK) except Exception as e: log_error('cancel_approval_day', e) return Response( {"error": "An error occurred while canceling approval"}, status=status.HTTP_500_INTERNAL_SERVER_ERROR ) @api_view(['GET']) def todays_progress(request): """ Get today's progress data optimized for the progress card Query params: user_id """ try: user_id = request.query_params.get('user_id') if not user_id: return Response( {"error": "user_id is required"}, status=status.HTTP_400_BAD_REQUEST ) today = timezone.now().date() # Get today's work balance todays_balance = WorkBalance.objects.filter( user_id=str(user_id), date=today ).first() # Calculate current week range (Monday to Sunday) start_of_week = today - timedelta(days=today.weekday()) end_of_week = start_of_week + timedelta(days=6) # Get weekly work balances weekly_balances = WorkBalance.objects.filter( user_id=str(user_id), date__range=[start_of_week, end_of_week] ).order_by('date') # Calculate totals total_weekly_seconds = sum(wb.total_work_seconds for wb in weekly_balances) todays_seconds = todays_balance.total_work_seconds if todays_balance else 0 # Convert to decimal hours for easier frontend consumption todays_hours = todays_seconds / 3600 weekly_hours = total_weekly_seconds / 3600 expected_weekly_hours = 40.0 # Standard work week # Calculate progress percentage progress_percentage = min((weekly_hours / expected_weekly_hours) * 100, 100) response_data = { 'user_id': user_id, 'today': { 'date': today.strftime('%Y-%m-%d'), 'hours': round(todays_hours, 1), 'total_work': round(todays_hours, 1), # Explicit total work hours 'seconds': todays_seconds, 'formatted_time': seconds_to_hhmm(todays_seconds) }, 'week': { 'start_date': start_of_week.strftime('%Y-%m-%d'), 'end_date': end_of_week.strftime('%Y-%m-%d'), 'worked_hours': round(weekly_hours, 1), 'total_work': round(weekly_hours, 1), # Explicit total work hours 'expected_hours': expected_weekly_hours, 'progress_percentage': round(progress_percentage, 1), 'remaining_hours': max(0, round(expected_weekly_hours - weekly_hours, 1)) }, 'summary': { 'is_week_complete': weekly_hours >= expected_weekly_hours, 'days_worked': weekly_balances.count(), 'average_daily_hours': round(weekly_hours / max(weekly_balances.count(), 1), 1) } } return Response(response_data) except Exception as e: log_error('todays_progress', e) return Response( {"error": "An error occurred while fetching today's progress"}, status=status.HTTP_500_INTERNAL_SERVER_ERROR ) @api_view(['GET']) def monthly_balance(request): """ Get user monthly balance data with daily breakdown Query params: user_id, start_date, end_date """ try: user_id = request.query_params.get('user_id') start_date = request.query_params.get('start_date') end_date = request.query_params.get('end_date') if not all([user_id, start_date, end_date]): return Response( {"error": "user_id, start_date, and end_date are required"}, status=status.HTTP_400_BAD_REQUEST ) # Parse dates try: start_date_obj = datetime.strptime(start_date, '%Y-%m-%d').date() end_date_obj = datetime.strptime(end_date, '%Y-%m-%d').date() except ValueError: return Response( {"error": "Invalid date format. Use YYYY-MM-DD"}, status=status.HTTP_400_BAD_REQUEST ) # Get work balances for the user in date range work_balances = WorkBalance.objects.filter( user_id=str(user_id), date__range=[start_date_obj, end_date_obj] ).order_by('date') # Get stamps for the user in date range stamps = Stamp.objects.filter( user_id=user_id, date__range=[start_date_obj, end_date_obj] ).order_by('date', 'time') # Get user's language preference (fetch fresh from database) user_language = 'en' try: # First try cached value (fast and reliable) if request.user and hasattr(request.user, 'lang'): cached_lang = request.user.lang if cached_lang: # Try to fetch fresh from database, but don't fail if it doesn't work try: if hasattr(request.user, 'id') and request.user.id: from user.models import User user = User.objects.filter(id=request.user.id).only('lang').first() if user and user.lang: user_language = user.lang else: user_language = cached_lang or 'en' else: user_language = cached_lang or 'en' except Exception: # If database fetch fails, use cached value user_language = cached_lang or 'en' # If no cached value, try database fetch elif request.user and hasattr(request.user, 'id') and request.user.id: try: from user.models import User user = User.objects.filter(id=request.user.id).only('lang').first() if user and user.lang: user_language = user.lang except Exception: pass except Exception: # If anything goes wrong, just use default pass # Get configuration data for all stamp_functions (with icon and color exactly as defined) if stamps.exists(): stamp_functions = set(stamps.values_list('stamp_function', flat=True)) configurations_dict = get_configuration_data_for_functions(stamp_functions, user_language) else: configurations_dict = {} # Use the global seconds_to_hhmm function # Calculate total month regular work total_month_regular = sum(wb.regular_work_seconds for wb in work_balances) # Group stamps by date stamps_by_date = {} for stamp in stamps: date_str = stamp.date.strftime('%Y-%m-%d') if date_str not in stamps_by_date: stamps_by_date[date_str] = [] stamps_by_date[date_str].append(stamp) # Create date data array date_data = [] current_date = start_date_obj while current_date <= end_date_obj: date_str = current_date.strftime('%Y-%m-%d') formatted_date = current_date.strftime('%d-%m-%Y') # Find work balance for this date work_balance = next((wb for wb in work_balances if wb.date == current_date), None) # Get stamps for this date day_stamps = stamps_by_date.get(date_str, []) # Format time entries for this day time_entries = [] for stamp in day_stamps: # Get configuration data for this stamp function config_data = configurations_dict.get(stamp.stamp_function) if isinstance(config_data, dict): function_name = config_data.get('header') # Get icon and color exactly as defined in configuration (including empty strings) function_icon = config_data.get('icon', '') # Default to empty string if key missing function_color = config_data.get('color', '') # Default to empty string if key missing else: function_name = config_data if config_data else None function_icon = '' # Return empty string if config doesn't exist function_color = '' # Return empty string if config doesn't exist time_entries.append({ 'id': str(stamp.id), 'stamp_function': stamp.stamp_function, 'function_name': function_name, 'function_icon': function_icon, # Return exactly as defined in configuration 'function_color': function_color, # Return exactly as defined in configuration 'description': stamp.description, 'time': stamp.time.strftime('%H:%M:%S'), 'date': stamp.date.strftime('%Y-%m-%d'), 'start_date': stamp.start_date.isoformat(), 'return_date': stamp.return_date.isoformat() if stamp.return_date else None, 'work_id': str(stamp.work_id) if stamp.work_id else None, 'status': stamp.status or '', 'project': str(stamp.project.id) if stamp.project else None, 'project_id': str(stamp.project.id) if stamp.project else None, 'project_name': stamp.project.name if stamp.project else None, 'task': str(stamp.task.id) if stamp.task else None, 'task_id': str(stamp.task.id) if stamp.task else None, 'task_name': stamp.task.name if stamp.task else None, 'duration': stamp.duration or None, 'source': stamp.source or None, }) # Create day data day_data = { 'date': formatted_date, 'regular': seconds_to_hhmm(work_balance.regular_work_seconds) if work_balance else "00:00", 'total_work': seconds_to_hhmm(work_balance.total_work_seconds) if work_balance else "00:00", 'daily_break': seconds_to_hhmm(work_balance.daily_break_seconds) if work_balance else "00:00", 'net_work': seconds_to_hhmm(work_balance.net_work_seconds) if work_balance else "00:00", 'overtime': seconds_to_hhmm(work_balance.overtime_seconds) if work_balance else "00:00", 'flex': seconds_to_hhmm(work_balance.flex_seconds) if work_balance else "00:00", 'bank_credited': seconds_to_hhmm(work_balance.bank_credited_seconds) if work_balance else "00:00", 'bank_debited': seconds_to_hhmm(work_balance.bank_debited_seconds) if work_balance else "00:00", 'overtime_balance': seconds_to_hhmm(work_balance.overtime_balance_seconds) if work_balance else "00:00", 'toil': seconds_to_hhmm(work_balance.toil_seconds) if work_balance else "00:00", 'vacation_days_accrued': f"{work_balance.vacation_days_accrued:.2f}" if work_balance else "0.00", 'stamp_count': len(day_stamps), 'time_entries': time_entries } date_data.append(day_data) current_date += timedelta(days=1) # Prepare response response_data = { 'user_id': user_id, 'total_month_regular': seconds_to_hhmm(total_month_regular), 'date_data': date_data, 'summary': { 'total_days': len(date_data), 'days_with_stamps': len([d for d in date_data if d['stamp_count'] > 0]), 'total_stamps': stamps.count(), 'total_month_regular_hours': seconds_to_hhmm(total_month_regular), } } return Response(response_data) except Exception as e: log_error('monthly_balance', e) return Response( {"error": "An error occurred while fetching monthly balance"}, status=status.HTTP_500_INTERNAL_SERVER_ERROR ) def get_configuration_visibility(self, queryset, configurations_dict): """ Return only what's needed: header and visibility flag """ configurations = Configuration.objects.all() current_states = self.analyze_current_states(queryset, configurations_dict) configuration_visibility = {} for config in configurations: try: function = Function.objects.get(function_ref_id=config.function_ref_id) # Determine function type based on function properties if function.out: function_type = 'out' elif function.break_flag: function_type = 'break' elif function.pre_function: function_type = 'pre_function' else: function_type = 'in' except Function.DoesNotExist: function_type = 'neutral' current_state = current_states.get(function_type, {'is_active': False}) is_visible = self.should_configuration_be_visible(config, current_state['is_active']) configuration_visibility[config.function_ref_id] = { 'header': config.header, 'is_visible': is_visible } return configuration_visibility def analyze_current_states(self, queryset, configurations_dict): stamps_with_functions = [] for stamp in queryset: try: function = Function.objects.get(function_ref_id=stamp.stamp_function) # Determine function type based on function properties if function.out: function_type = 'out' elif function.break_flag: function_type = 'break' elif function.pre_function: function_type = 'pre_function' else: function_type = 'in' stamps_with_functions.append({ 'function_type': function_type, 'time': stamp.time, 'created_at': stamp.created_at }) except Function.DoesNotExist: continue # Sort by time stamps_with_functions.sort(key=lambda x: (x['time'], x['created_at'])) # Group by function type type_groups = {} for stamp in stamps_with_functions: function_type = stamp['function_type'] if function_type not in type_groups: type_groups[function_type] = [] type_groups[function_type].append(stamp) # Determine current state current_states = {} for function_type, stamps in type_groups.items(): if stamps: if function_type == 'in': current_states[function_type] = {'is_active': True} elif function_type == 'out': current_states[function_type] = {'is_active': False} else: current_states[function_type] = {'is_active': False} return current_states def should_configuration_be_visible(self, config, is_active): if config.visibility_rule == 'always_show': return True elif config.visibility_rule == 'hide_when_in': return not is_active elif config.visibility_rule == 'hide_when_out': return is_active elif config.visibility_rule == 'show_when_in': return is_active elif config.visibility_rule == 'show_when_out': return not is_active else: return True @api_view(['GET']) def balance_screen_data(request): """ Get comprehensive balance data for the balance screen Combines WorkBalance totals with BalanceDetail breakdowns by paycode Query params: user_id, start_date, end_date, period_type (week/month) """ try: user_id = request.query_params.get('user_id') start_date = request.query_params.get('start_date') end_date = request.query_params.get('end_date') period_type = request.query_params.get('period_type', 'week') # week or month if not all([user_id, start_date, end_date]): return Response( {"error": "user_id, start_date, and end_date are required"}, status=status.HTTP_400_BAD_REQUEST ) # Parse dates try: start_date_obj = datetime.strptime(start_date, '%Y-%m-%d').date() end_date_obj = datetime.strptime(end_date, '%Y-%m-%d').date() except ValueError: return Response( {"error": "Invalid date format. Use YYYY-MM-DD"}, status=status.HTTP_400_BAD_REQUEST ) # Get work balances for the user in date range work_balances = WorkBalance.objects.filter( user_id=str(user_id), date__range=[start_date_obj, end_date_obj] ).order_by('date') # Get cumulative work balances from beginning up to the selected period cumulative_work_balances = WorkBalance.objects.filter( user_id=str(user_id), date__lte=end_date_obj # All balances up to and including the selected period ).order_by('date') # Get working time policy for the user (falls back to company default) from worktimeservice.service import get_active_working_policy working_policy = get_active_working_policy(user_id) # Get vacation policy to check if sick leave, parental leave, and unpaid leave are tracked vacation_policy = VacationPolicy.objects.order_by('-created_at').first() # Calculate expected working hours based on policy expected_hours_seconds = calculate_expected_working_hours( start_date_obj, end_date_obj, working_policy, user_id ) # Get balance details with paycode information for the selected period balance_details = BalanceDetail.objects.filter( user_id=user_id, date__range=[start_date_obj, end_date_obj] ).select_related('paycode', 'work_balance').order_by('date', 'paycode__name') # Get cumulative balance details from beginning up to the selected period cumulative_balance_details = BalanceDetail.objects.filter( user_id=user_id, date__lte=end_date_obj # All balance details up to and including the selected period ).select_related('paycode', 'work_balance').order_by('date', 'paycode__name') # Get stamps for detailed transaction history (only for selected period) stamps = Stamp.objects.filter( user_id=user_id, date__range=[start_date_obj, end_date_obj] ).order_by('date', 'time') # Get user's language preference user_language = 'en' if request.user and hasattr(request.user, 'lang'): user_language = request.user.lang or 'en' # Get configuration headers for stamps # Get configuration data for all stamp_functions if stamps.exists(): stamp_functions = set(stamps.values_list('stamp_function', flat=True)) configurations_dict = get_configuration_data_for_functions(stamp_functions, user_language) else: configurations_dict = {} # Calculate period totals total_regular = sum(wb.regular_work_seconds for wb in work_balances) total_overtime = sum(wb.overtime_seconds for wb in work_balances) total_flex = sum(wb.flex_seconds for wb in work_balances) total_bank_credited = sum(wb.bank_credited_seconds for wb in work_balances) total_bank_debited = sum(wb.bank_debited_seconds for wb in work_balances) total_toil = sum(wb.toil_seconds for wb in work_balances) total_vacation_accrued = sum(float(wb.vacation_days_accrued) for wb in work_balances) # Calculate work time bank balance (credited - debited) work_time_bank_balance = total_bank_credited - total_bank_debited # Calculate cumulative totals from beginning up to selected period cumulative_total_regular = sum(wb.regular_work_seconds for wb in cumulative_work_balances) cumulative_total_overtime = sum(wb.overtime_seconds for wb in cumulative_work_balances) cumulative_total_flex = sum(wb.flex_seconds for wb in cumulative_work_balances) cumulative_total_bank_credited = sum(wb.bank_credited_seconds for wb in cumulative_work_balances) cumulative_total_bank_debited = sum(wb.bank_debited_seconds for wb in cumulative_work_balances) cumulative_total_toil = sum(wb.toil_seconds for wb in cumulative_work_balances) cumulative_total_vacation_accrued = sum(float(wb.vacation_days_accrued) for wb in cumulative_work_balances) # Calculate cumulative work time bank balance cumulative_work_time_bank_balance = cumulative_total_bank_credited - cumulative_total_bank_debited # Group balance details by balance type for transaction history # Check if sick leave is tracked based on vacation policy track_sick_leave = vacation_policy and vacation_policy.self_certification_days > 0 track_parental_leave = False # Not implemented in policy track_unpaid_leave = False # Not implemented in policy balance_type_transactions = { 'Flex Hours': [], 'Overtime': [], 'Vacation': [], 'Sick Leave': [] if track_sick_leave else [], 'Parental Leave': [], 'Unpaid Leave': [], 'Comp Time': [], 'Work Time Bank': [] } # Function to calculate cumulative balance up to a specific date def get_cumulative_balance_up_to_date(target_date, balance_type): """Calculate cumulative balance up to target_date for specific balance type""" if balance_type == 'Flex Hours': return sum(wb.flex_seconds for wb in cumulative_work_balances if wb.date <= target_date) elif balance_type == 'Overtime': return sum(wb.overtime_seconds for wb in cumulative_work_balances if wb.date <= target_date) elif balance_type == 'Comp Time': return sum(wb.toil_seconds for wb in cumulative_work_balances if wb.date <= target_date) elif balance_type == 'Work Time Bank': credited = sum(wb.bank_credited_seconds for wb in cumulative_work_balances if wb.date <= target_date) debited = sum(wb.bank_debited_seconds for wb in cumulative_work_balances if wb.date <= target_date) return credited - debited return 0 for detail in balance_details: # Get ALL stamps for this specific date (not filtered by paycode) day_stamps = stamps.filter(date=detail.date).order_by('time') # Create stamp details for this transaction stamp_details = [] for stamp in day_stamps: stamp_details.append({ 'time': stamp.time.strftime('%H:%M:%S'), 'function_name': configurations_dict.get(stamp.stamp_function, {}).get('header') if isinstance(configurations_dict.get(stamp.stamp_function), dict) else configurations_dict.get(stamp.stamp_function, stamp.stamp_function), 'function_icon': configurations_dict.get(stamp.stamp_function, {}).get('icon') if isinstance(configurations_dict.get(stamp.stamp_function), dict) else None, 'function_color': configurations_dict.get(stamp.stamp_function, {}).get('color') if isinstance(configurations_dict.get(stamp.stamp_function), dict) else None, 'description': stamp.description or '', 'stamp_function': stamp.stamp_function, }) # Add flex hours transaction if detail.flex_seconds != 0: # Use policy-based expected hours (correct approach) scheduled_work_seconds = 8 * 3600 # default if working_policy: weekday = detail.date.weekday() # Mon=0..Sun=6 if weekday <= 4 and getattr(working_policy, "monday_friday_start", None) and getattr(working_policy, "monday_friday_end", None): start = working_policy.monday_friday_start end = working_policy.monday_friday_end scheduled_work_seconds = int((datetime.combine(detail.date, end) - datetime.combine(detail.date, start)).total_seconds()) elif weekday == 5 and getattr(working_policy, "saturday_start", None) and getattr(working_policy, "saturday_end", None): start = working_policy.saturday_start end = working_policy.saturday_end scheduled_work_seconds = int((datetime.combine(detail.date, end) - datetime.combine(detail.date, start)).total_seconds()) elif weekday == 6 and getattr(working_policy, "sunday_start", None) and getattr(working_policy, "sunday_end", None): start = working_policy.sunday_start end = working_policy.sunday_end scheduled_work_seconds = int((datetime.combine(detail.date, end) - datetime.combine(detail.date, start)).total_seconds()) else: scheduled_work_seconds = getattr(working_policy, "default_daily_seconds", 8 * 3600) # Use correct values for calculation display total_work_hours = detail.total_work_seconds / 3600 # Use total work, not regular work expected_hours = scheduled_work_seconds / 3600 # Use policy-based expected hours flex_hours = detail.flex_seconds / 3600 # Calculate cumulative balance up to this date cumulative_flex_seconds = get_cumulative_balance_up_to_date(detail.date, 'Flex Hours') balance_type_transactions['Flex Hours'].append({ 'date': detail.date.strftime('%Y-%m-%d'), 'day': detail.date.strftime('%A'), 'hours': round(detail.flex_seconds / 3600, 2), 'type': 'Earned' if detail.flex_seconds > 0 else 'Used', 'description': '', 'paycode': detail.paycode.name, 'regular_hours': round(detail.regular_work_seconds / 3600, 2), 'overtime_hours': round(detail.overtime_seconds / 3600, 2), 'flex_hours': round(detail.flex_seconds / 3600, 2), 'cumulativeBalance': round(cumulative_flex_seconds / 3600, 2), 'calculation': { 'formula': 'Total Work - Expected Hours', 'total_work': f"{total_work_hours:.3f}h", 'expected_hours': f"{expected_hours:.3f}h", 'result': f"{flex_hours:.3f}h", 'explanation': f"{total_work_hours:.3f}h - {expected_hours:.3f}h = {flex_hours:.3f}h" }, 'stamp_details': stamp_details, }) # Add overtime transaction if detail.overtime_seconds != 0: # Calculate cumulative balance up to this date cumulative_overtime_seconds = get_cumulative_balance_up_to_date(detail.date, 'Overtime') # Use policy-based expected hours for calculation scheduled_work_seconds = 8 * 3600 # default if working_policy: weekday = detail.date.weekday() # Mon=0..Sun=6 if weekday <= 4 and getattr(working_policy, "monday_friday_start", None) and getattr(working_policy, "monday_friday_end", None): start = working_policy.monday_friday_start end = working_policy.monday_friday_end scheduled_work_seconds = int((datetime.combine(detail.date, end) - datetime.combine(detail.date, start)).total_seconds()) elif weekday == 5 and getattr(working_policy, "saturday_start", None) and getattr(working_policy, "saturday_end", None): start = working_policy.saturday_start end = working_policy.saturday_end scheduled_work_seconds = int((datetime.combine(detail.date, end) - datetime.combine(detail.date, start)).total_seconds()) elif weekday == 6 and getattr(working_policy, "sunday_start", None) and getattr(working_policy, "sunday_end", None): start = working_policy.sunday_start end = working_policy.sunday_end scheduled_work_seconds = int((datetime.combine(detail.date, end) - datetime.combine(detail.date, start)).total_seconds()) else: scheduled_work_seconds = getattr(working_policy, "default_daily_seconds", 8 * 3600) # Use correct values for calculation display total_work_hours = detail.total_work_seconds / 3600 expected_hours = scheduled_work_seconds / 3600 overtime_hours = detail.overtime_seconds / 3600 balance_type_transactions['Overtime'].append({ 'date': detail.date.strftime('%Y-%m-%d'), 'day': detail.date.strftime('%A'), 'hours': round(detail.overtime_seconds / 3600, 2), 'type': 'Earned' if detail.overtime_seconds > 0 else 'Used', 'description': '', 'paycode': detail.paycode.name, 'regular_hours': round(detail.regular_work_seconds / 3600, 2), 'overtime_hours': round(detail.overtime_seconds / 3600, 2), 'flex_hours': round(detail.flex_seconds / 3600, 2), 'cumulativeBalance': round(cumulative_overtime_seconds / 3600, 2), 'calculation': { 'formula': 'Total Work - Expected Hours', 'total_work': f"{total_work_hours:.3f}h", 'expected_hours': f"{expected_hours:.3f}h", 'result': f"{overtime_hours:.3f}h", 'explanation': f"{total_work_hours:.3f}h - {expected_hours:.3f}h = {overtime_hours:.3f}h" }, 'stamp_details': stamp_details, }) # Add comp time transaction if detail.toil_seconds != 0: # Calculate cumulative balance up to this date cumulative_toil_seconds = get_cumulative_balance_up_to_date(detail.date, 'Comp Time') # Use policy-based expected hours for calculation scheduled_work_seconds = 8 * 3600 # default if working_policy: weekday = detail.date.weekday() # Mon=0..Sun=6 if weekday <= 4 and getattr(working_policy, "monday_friday_start", None) and getattr(working_policy, "monday_friday_end", None): start = working_policy.monday_friday_start end = working_policy.monday_friday_end scheduled_work_seconds = int((datetime.combine(detail.date, end) - datetime.combine(detail.date, start)).total_seconds()) elif weekday == 5 and getattr(working_policy, "saturday_start", None) and getattr(working_policy, "saturday_end", None): start = working_policy.saturday_start end = working_policy.saturday_end scheduled_work_seconds = int((datetime.combine(detail.date, end) - datetime.combine(detail.date, start)).total_seconds()) elif weekday == 6 and getattr(working_policy, "sunday_start", None) and getattr(working_policy, "sunday_end", None): start = working_policy.sunday_start end = working_policy.sunday_end scheduled_work_seconds = int((datetime.combine(detail.date, end) - datetime.combine(detail.date, start)).total_seconds()) else: scheduled_work_seconds = getattr(working_policy, "default_daily_seconds", 8 * 3600) # Use correct values for calculation display total_work_hours = detail.total_work_seconds / 3600 expected_hours = scheduled_work_seconds / 3600 toil_hours = detail.toil_seconds / 3600 balance_type_transactions['Comp Time'].append({ 'date': detail.date.strftime('%Y-%m-%d'), 'day': detail.date.strftime('%A'), 'hours': round(detail.toil_seconds / 3600, 2), 'type': 'Earned' if detail.toil_seconds > 0 else 'Used', 'description': '', 'paycode': detail.paycode.name, 'regular_hours': round(detail.regular_work_seconds / 3600, 2), 'overtime_hours': round(detail.overtime_seconds / 3600, 2), 'flex_hours': round(detail.flex_seconds / 3600, 2), 'cumulativeBalance': round(cumulative_toil_seconds / 3600, 2), 'calculation': { 'formula': 'Total Work - Expected Hours', 'total_work': f"{total_work_hours:.3f}h", 'expected_hours': f"{expected_hours:.3f}h", 'result': f"{toil_hours:.3f}h", 'explanation': f"{total_work_hours:.3f}h - {expected_hours:.3f}h = {toil_hours:.3f}h" }, 'stamp_details': stamp_details, }) # Add vacation accrual transaction # Note: Vacation is tracked in WorkBalance, not BalanceDetail # We'll handle it separately after the balance_details loop if False and detail.vacation_days_accrued != 0: # Calculate cumulative balance up to this date cumulative_vacation_days = sum(float(wb.vacation_days_accrued) for wb in cumulative_work_balances if wb.date <= detail.date) balance_type_transactions['Vacation'].append({ 'date': detail.date.strftime('%Y-%m-%d'), 'day': detail.date.strftime('%A'), 'hours': round(float(detail.vacation_days_accrued), 2), 'type': 'Accrued', 'description': '', 'paycode': detail.paycode.name, 'regular_hours': round(detail.regular_work_seconds / 3600, 2), 'overtime_hours': round(detail.overtime_seconds / 3600, 2), 'flex_hours': round(detail.flex_seconds / 3600, 2), 'cumulativeBalance': round(cumulative_vacation_days, 2), 'calculation': { 'formula': 'Vacation accrual rate', 'accrued_days': f"{detail.vacation_days_accrued:.2f} days", 'source': f"Accrued from {detail.paycode.name}", 'explanation': f"Vacation accrual: {detail.vacation_days_accrued:.2f} days from {detail.paycode.name}" }, 'stamp_details': stamp_details, }) # Add vacation accrual transactions from WorkBalance (not BalanceDetail) for wb in work_balances: if float(wb.vacation_days_accrued) != 0: # Get ALL stamps for this specific date day_stamps = stamps.filter(date=wb.date).order_by('time') # Create stamp details for this transaction stamp_details = [] for stamp in day_stamps: stamp_details.append({ 'time': stamp.time.strftime('%H:%M:%S'), 'function_name': configurations_dict.get(stamp.stamp_function, {}).get('header') if isinstance(configurations_dict.get(stamp.stamp_function), dict) else configurations_dict.get(stamp.stamp_function, stamp.stamp_function), 'function_icon': configurations_dict.get(stamp.stamp_function, {}).get('icon') if isinstance(configurations_dict.get(stamp.stamp_function), dict) else None, 'function_color': configurations_dict.get(stamp.stamp_function, {}).get('color') if isinstance(configurations_dict.get(stamp.stamp_function), dict) else None, 'description': stamp.description or '', 'stamp_function': stamp.stamp_function, }) # Calculate cumulative balance up to this date cumulative_vacation_days = sum(float(wbalance.vacation_days_accrued) for wbalance in cumulative_work_balances if wbalance.date <= wb.date) # Format vacation in days (not hours) vacation_days = round(float(wb.vacation_days_accrued), 2) cumulative_balance_days = round(cumulative_vacation_days, 2) balance_type_transactions['Vacation'].append({ 'date': wb.date.strftime('%Y-%m-%d'), 'day': wb.date.strftime('%A'), 'hours': vacation_days, # Keep as 'hours' for consistency with frontend 'days': vacation_days, # Add explicit days field 'type': 'Accrued', 'description': '', 'paycode': 'Work', 'regular_hours': round(wb.regular_work_seconds / 3600, 2), 'overtime_hours': round(wb.overtime_seconds / 3600, 2), 'flex_hours': round(wb.flex_seconds / 3600, 2), 'cumulativeBalance': cumulative_balance_days, 'cumulativeBalanceDays': cumulative_balance_days, # Add explicit days field 'calculation': { 'formula': 'Vacation accrual rate', 'accrued_days': f"{vacation_days:.2f} days", 'source': "Vacation accrual", 'explanation': f"Vacation accrual: {vacation_days:.2f} days" }, 'stamp_details': stamp_details, }) # Add work time bank transactions for detail in balance_details: if detail.bank_credited_seconds > 0: # Get ALL stamps for this specific date (not filtered by paycode) day_stamps = stamps.filter(date=detail.date).order_by('time') # Create stamp details for this transaction stamp_details = [] for stamp in day_stamps: stamp_details.append({ 'time': stamp.time.strftime('%H:%M:%S'), 'function_name': configurations_dict.get(stamp.stamp_function, {}).get('header') if isinstance(configurations_dict.get(stamp.stamp_function), dict) else configurations_dict.get(stamp.stamp_function, stamp.stamp_function), 'function_icon': configurations_dict.get(stamp.stamp_function, {}).get('icon') if isinstance(configurations_dict.get(stamp.stamp_function), dict) else None, 'function_color': configurations_dict.get(stamp.stamp_function, {}).get('color') if isinstance(configurations_dict.get(stamp.stamp_function), dict) else None, 'description': stamp.description or '', 'stamp_function': stamp.stamp_function, }) # Use policy-based expected hours for calculation scheduled_work_seconds = 8 * 3600 # default if working_policy: weekday = detail.date.weekday() # Mon=0..Sun=6 if weekday <= 4 and getattr(working_policy, "monday_friday_start", None) and getattr(working_policy, "monday_friday_end", None): start = working_policy.monday_friday_start end = working_policy.monday_friday_end scheduled_work_seconds = int((datetime.combine(detail.date, end) - datetime.combine(detail.date, start)).total_seconds()) elif weekday == 5 and getattr(working_policy, "saturday_start", None) and getattr(working_policy, "saturday_end", None): start = working_policy.saturday_start end = working_policy.saturday_end scheduled_work_seconds = int((datetime.combine(detail.date, end) - datetime.combine(detail.date, start)).total_seconds()) elif weekday == 6 and getattr(working_policy, "sunday_start", None) and getattr(working_policy, "sunday_end", None): start = working_policy.sunday_start end = working_policy.sunday_end scheduled_work_seconds = int((datetime.combine(detail.date, end) - datetime.combine(detail.date, start)).total_seconds()) else: scheduled_work_seconds = getattr(working_policy, "default_daily_seconds", 8 * 3600) # Calculate cumulative balance up to this date cumulative_bank_seconds = get_cumulative_balance_up_to_date(detail.date, 'Work Time Bank') # Use correct values for calculation display bank_credited_hours = detail.bank_credited_seconds / 3600 balance_type_transactions['Work Time Bank'].append({ 'date': detail.date.strftime('%Y-%m-%d'), 'day': detail.date.strftime('%A'), 'hours': round(detail.bank_credited_seconds / 3600, 2), 'type': 'Deposited', 'description': '', 'paycode': detail.paycode.name, 'cumulativeBalance': round(cumulative_bank_seconds / 3600, 2), 'calculation': { 'formula': 'Work time bank deposit', 'deposited_hours': f"{bank_credited_hours:.3f}h", 'explanation': f"Work time bank: +{bank_credited_hours:.3f}h" }, 'stamp_details': stamp_details, }) if detail.bank_debited_seconds > 0: # Get ALL stamps for this specific date (not filtered by paycode) day_stamps = stamps.filter(date=detail.date).order_by('time') # Create stamp details for this transaction stamp_details = [] for stamp in day_stamps: stamp_details.append({ 'time': stamp.time.strftime('%H:%M:%S'), 'function_name': configurations_dict.get(stamp.stamp_function, {}).get('header') if isinstance(configurations_dict.get(stamp.stamp_function), dict) else configurations_dict.get(stamp.stamp_function, stamp.stamp_function), 'function_icon': configurations_dict.get(stamp.stamp_function, {}).get('icon') if isinstance(configurations_dict.get(stamp.stamp_function), dict) else None, 'function_color': configurations_dict.get(stamp.stamp_function, {}).get('color') if isinstance(configurations_dict.get(stamp.stamp_function), dict) else None, 'description': stamp.description or '', 'stamp_function': stamp.stamp_function, }) # Calculate cumulative balance up to this date cumulative_bank_seconds = get_cumulative_balance_up_to_date(detail.date, 'Work Time Bank') bank_debited_hours = detail.bank_debited_seconds / 3600 balance_type_transactions['Work Time Bank'].append({ 'date': detail.date.strftime('%Y-%m-%d'), 'day': detail.date.strftime('%A'), 'hours': -round(detail.bank_debited_seconds / 3600, 2), 'type': 'Withdrawn', 'description': f'Work time withdrawn via {detail.paycode.name}', 'paycode': detail.paycode.name, 'cumulativeBalance': round(cumulative_bank_seconds / 3600, 2), 'calculation': { 'formula': 'Work time bank withdrawal', 'withdrawn_hours': f"{bank_debited_hours:.3f}h", 'explanation': f"Work time bank: -{bank_debited_hours:.3f}h" }, 'stamp_details': stamp_details, }) # Sort all transactions by date (newest first) for balance_type in balance_type_transactions: balance_type_transactions[balance_type].sort(key=lambda x: x['date'], reverse=True) # Create balance summary using cumulative values balances = [ { 'type': 'Flex Hours', 'balance': round(cumulative_total_flex / 3600, 2), 'target': 0, # Flex time doesn't have a target 'worked': round(total_flex / 3600, 2), # Flex time earned/used this period 'color': 'green' }, { 'type': 'Overtime', 'balance': round(cumulative_total_overtime / 3600, 2), 'target': 0, 'worked': round(total_overtime / 3600, 2), 'color': 'slate' }, { 'type': 'Vacation', 'balance': round(cumulative_total_vacation_accrued, 2), 'target': 0, 'worked': 0, 'color': 'gray' }, { 'type': 'Sick Leave', 'balance': 0, # Not tracked in current model 'target': 0, 'worked': 0, 'color': 'stone' }, { 'type': 'Parental Leave', 'balance': 0, # Not tracked in current model 'target': 0, 'worked': 0, 'color': 'zinc' }, { 'type': 'Unpaid Leave', 'balance': 0, # Not tracked in current model 'target': 0, 'worked': 0, 'color': 'neutral' }, { 'type': 'Comp Time', 'balance': round(cumulative_total_toil / 3600, 2), 'target': 0, 'worked': round(total_toil / 3600, 2), 'color': 'green' }, { 'type': 'Work Time Bank', 'balance': round(cumulative_work_time_bank_balance / 3600, 2), 'target': 0, 'worked': 0, 'color': 'blue' } ] # Prepare response response_data = { 'user_id': user_id, 'period_type': period_type, 'start_date': start_date, 'end_date': end_date, 'balances': balances, 'transaction_history': balance_type_transactions, 'summary': { 'total_days': (end_date_obj - start_date_obj).days + 1, 'days_with_data': work_balances.count(), 'total_regular_hours': round(total_regular / 3600, 2), 'total_overtime_hours': round(total_overtime / 3600, 2), 'total_flex_hours': round(total_flex / 3600, 2), 'work_time_bank_balance': round(work_time_bank_balance / 3600, 2), 'expected_hours': round(expected_hours_seconds / 3600, 2), 'working_policy_applied': working_policy is not None, } } return Response(response_data) except Exception as e: return Response( {"error": f"Failed to fetch balance data: {str(e)}"}, status=status.HTTP_500_INTERNAL_SERVER_ERROR ) @api_view(['GET']) def worktime_report(request): """ Generate worktime report with all stamp details Query params: start_date, end_date, user_id, project_id, client_id, status """ try: # Get query parameters start_date = request.query_params.get('start_date') end_date = request.query_params.get('end_date') user_id = request.query_params.get('user_id') project_id = request.query_params.get('project_id') client_id = request.query_params.get('client_id') status_filter = request.query_params.get('status') # Start with base queryset queryset = Stamp.objects.select_related('user', 'project', 'task').all() # Apply filters if start_date: try: start_date_obj = datetime.strptime(start_date, '%Y-%m-%d').date() queryset = queryset.filter(date__gte=start_date_obj) except ValueError: pass if end_date: try: end_date_obj = datetime.strptime(end_date, '%Y-%m-%d').date() queryset = queryset.filter(date__lte=end_date_obj) except ValueError: pass if user_id: queryset = queryset.filter(user_id=user_id) if project_id: queryset = queryset.filter(project_id=project_id) if client_id: # Filter by client through project # Only filter stamps that have projects matching the client # Stamps without projects are excluded when client filter is applied queryset = queryset.filter( Q(project__customer_id=client_id) | Q(project__client=client_id) ) # Note: When no client_id filter is applied, all stamps are included (with or without projects) if status_filter: queryset = queryset.filter(status=status_filter) # Get stamps with related data - order by date descending, time ascending (earliest first) stamps = queryset.order_by('-date', 'time') # Get configuration data for functions stamp_functions = set(stamps.values_list('stamp_function', flat=True)) configurations_dict = get_configuration_data_for_functions(stamp_functions, 'en') # Serialize stamps with context serializer = StampSerializer(stamps, many=True, context={'functions_dict': configurations_dict}) stamp_data = serializer.data # Add user information to each stamp for i, stamp in enumerate(stamps): if stamp.user: stamp_data[i]['user'] = { 'id': str(stamp.user.id), 'firstname': stamp.user.firstname, 'lastname': stamp.user.lastname, 'email': stamp.user.email } # Group stamps by date and calculate daily totals using WorkBalance (same as timestamp view) from collections import defaultdict stamps_by_date = defaultdict(list) for i, stamp in enumerate(stamps): date_key = stamp.date.strftime('%Y-%m-%d') stamps_by_date[date_key].append({ 'stamp': stamp, 'stamp_data': stamp_data[i] }) # Get WorkBalance data for all dates to calculate accurate daily totals date_list = list(stamps_by_date.keys()) if date_list: date_objects = [datetime.strptime(d, '%Y-%m-%d').date() for d in date_list] min_date = min(date_objects) max_date = max(date_objects) # Get WorkBalance for all users and dates in the range work_balances = WorkBalance.objects.filter( date__range=[min_date, max_date] ) # Create a map of (user_id, date) -> WorkBalance work_balance_map = {} for wb in work_balances: key = (str(wb.user_id), wb.date.strftime('%Y-%m-%d')) work_balance_map[key] = wb else: work_balance_map = {} # Group stamps by date, then by employee grouped_by_date = [] total_stamps = 0 total_work_seconds = 0 for date_key in sorted(stamps_by_date.keys(), reverse=True): # Newest first date_obj = datetime.strptime(date_key, '%Y-%m-%d').date() day_stamps_info = stamps_by_date[date_key] day_stamps = [info['stamp'] for info in day_stamps_info] day_stamp_data = [info['stamp_data'] for info in day_stamps_info] # Group stamps by employee for this date employee_groups = defaultdict(list) for i, stamp in enumerate(day_stamps): user_id = str(stamp.user_id) if stamp.user_id else 'unknown' employee_name = f"{stamp.user.firstname or ''} {stamp.user.lastname or ''}".strip() if stamp.user else 'Unknown' if not employee_name: employee_name = stamp.user.email if stamp.user else 'Unknown' employee_key = (user_id, employee_name) employee_groups[employee_key].append({ 'stamp': stamp, 'stamp_data': day_stamp_data[i] }) # Build employee groups for this date employees_data = [] day_total_seconds = 0 for (user_id, employee_name), emp_stamps_info in sorted(employee_groups.items()): emp_stamps = [info['stamp'] for info in emp_stamps_info] emp_stamp_data = [info['stamp_data'] for info in emp_stamps_info] # Calculate total work for this employee on this day emp_total_work = "00:00" emp_total_seconds = 0 # Try to get from WorkBalance first (most accurate) wb_key = (user_id, date_key) if wb_key in work_balance_map: wb = work_balance_map[wb_key] emp_total_seconds = wb.total_work_seconds emp_total_work = seconds_to_hhmm(wb.total_work_seconds) else: # No WorkBalance - calculate from duration for stamp in emp_stamps: if stamp.duration: duration_str = str(stamp.duration).replace('hrs', '').strip() if ':' in duration_str: parts = duration_str.split(':') hours = int(parts[0]) if parts[0].isdigit() else 0 minutes = int(parts[1]) if len(parts) > 1 and parts[1].isdigit() else 0 emp_total_seconds += (hours * 3600) + (minutes * 60) else: try: decimal_hours = float(duration_str) emp_total_seconds += int(decimal_hours * 3600) except ValueError: pass emp_total_work = seconds_to_hhmm(emp_total_seconds) day_total_seconds += emp_total_seconds employees_data.append({ 'user_id': user_id, 'employee_name': employee_name, 'total_work': emp_total_work, 'stamp_count': len(emp_stamps), 'stamps': emp_stamp_data }) # Calculate day total day_total_work = seconds_to_hhmm(day_total_seconds) total_work_seconds += day_total_seconds total_stamps += len(day_stamps) grouped_by_date.append({ 'date': date_key, 'total_work': day_total_work, 'stamp_count': len(day_stamps), 'employees': employees_data }) # Calculate overall summary total_hours_formatted = seconds_to_hhmm(total_work_seconds) total_hours = total_work_seconds // 3600 total_minutes = (total_work_seconds % 3600) // 60 # Get unique users, projects, and clients unique_users = stamps.values('user_id', 'user__firstname', 'user__lastname', 'user__email').distinct() # Include stamps with or without projects unique_projects = stamps.filter(project__isnull=False).values('project_id', 'project__name', 'project__code').distinct() unique_clients = stamps.filter( Q(project__customer__isnull=False) | Q(project__client__isnull=False) ).values( 'project__customer_id', 'project__customer__name', 'project__client' ).distinct() # Group by status status_counts = stamps.values('status').annotate(count=Count('id')) status_summary = { 'approved': 0, 'pending': 0, 'suspended': 0 } for item in status_counts: if item['status'] == 'A': status_summary['approved'] = item['count'] elif item['status'] == 'S': status_summary['suspended'] = item['count'] else: status_summary['pending'] = item['count'] return Response({ 'stamps': stamp_data, # Keep flat list for backward compatibility 'grouped_by_date': grouped_by_date, # New grouped structure 'summary': { 'total_stamps': total_stamps, 'total_hours': total_hours, 'total_minutes': total_minutes, 'total_hours_formatted': total_hours_formatted, 'status_summary': status_summary, 'unique_users_count': unique_users.count(), 'unique_projects_count': unique_projects.count(), 'unique_clients_count': unique_clients.count() }, 'filters': { 'start_date': start_date, 'end_date': end_date, 'user_id': user_id, 'project_id': project_id, 'client_id': client_id, 'status': status_filter } }) except Exception as e: log_error('worktime_report', e) return Response( {"error": "An error occurred while generating worktime report"}, status=status.HTTP_500_INTERNAL_SERVER_ERROR ) @api_view(['GET']) @authentication_classes([]) # No authentication required @permission_classes([AllowAny]) def worktime_report_excel(request): """ Generate worktime report as Excel file Query params: start_date, end_date, user_id, project_id, client_id, status """ try: # Get query parameters (same as worktime_report) start_date = request.query_params.get('start_date') end_date = request.query_params.get('end_date') user_id = request.query_params.get('user_id') project_id = request.query_params.get('project_id') client_id = request.query_params.get('client_id') status_filter = request.query_params.get('status') # Get the report data (reuse the same logic) # Start with base queryset queryset = Stamp.objects.select_related('user', 'project', 'task').all() # Apply filters if start_date: try: start_date_obj = datetime.strptime(start_date, '%Y-%m-%d').date() queryset = queryset.filter(date__gte=start_date_obj) except ValueError: pass if end_date: try: end_date_obj = datetime.strptime(end_date, '%Y-%m-%d').date() queryset = queryset.filter(date__lte=end_date_obj) except ValueError: pass if user_id: queryset = queryset.filter(user_id=user_id) if project_id: queryset = queryset.filter(project_id=project_id) if client_id: queryset = queryset.filter( Q(project__customer_id=client_id) | Q(project__client=client_id) ) if status_filter: queryset = queryset.filter(status=status_filter) stamps = queryset.order_by('-date', 'time') # Get configuration data for functions (to get function names) stamp_functions = set(stamps.values_list('stamp_function', flat=True)) configurations_dict = get_configuration_data_for_functions(stamp_functions, 'en') # Create Excel workbook wb = Workbook() ws = wb.active ws.title = "Worktime Report" # Header style header_fill = PatternFill(start_color="366092", end_color="366092", fill_type="solid") header_font = Font(bold=True, color="FFFFFF", size=12) border = Border( left=Side(style='thin'), right=Side(style='thin'), top=Side(style='thin'), bottom=Side(style='thin') ) # Title ws.merge_cells('A1:I1') title_cell = ws['A1'] title_cell.value = "Worktime Report" title_cell.font = Font(bold=True, size=16) title_cell.alignment = Alignment(horizontal='center', vertical='center') # Filters info row = 3 if start_date or end_date: ws[f'A{row}'] = f"Date Range: {start_date or 'N/A'} to {end_date or 'N/A'}" row += 1 row += 1 # Headers headers = ['Date', 'Time', 'Employee', 'Function', 'Project', 'Task', 'Duration', 'Description', 'Status'] for col, header in enumerate(headers, 1): cell = ws.cell(row=row, column=col) cell.value = header cell.fill = header_fill cell.font = header_font cell.border = border cell.alignment = Alignment(horizontal='center', vertical='center') row += 1 # Group by date and employee stamps_by_date = {} for stamp in stamps: date_key = stamp.date.strftime('%Y-%m-%d') if date_key not in stamps_by_date: stamps_by_date[date_key] = {} user_id_str = str(stamp.user_id) if stamp.user_id else 'unknown' employee_name = f"{stamp.user.firstname or ''} {stamp.user.lastname or ''}".strip() if stamp.user else 'Unknown' if not employee_name: employee_name = stamp.user.email if stamp.user else 'Unknown' if user_id_str not in stamps_by_date[date_key]: stamps_by_date[date_key][user_id_str] = { 'employee_name': employee_name, 'stamps': [] } stamps_by_date[date_key][user_id_str]['stamps'].append(stamp) # Write data for date_key in sorted(stamps_by_date.keys(), reverse=True): date_obj = datetime.strptime(date_key, '%Y-%m-%d').date() formatted_date = date_obj.strftime('%Y-%m-%d') for user_id_str, emp_data in sorted(stamps_by_date[date_key].items()): # Employee header ws.merge_cells(f'A{row}:I{row}') emp_header = ws[f'A{row}'] emp_header.value = f"{formatted_date} - {emp_data['employee_name']}" emp_header.font = Font(bold=True, size=11) emp_header.fill = PatternFill(start_color="D3D3D3", end_color="D3D3D3", fill_type="solid") row += 1 # Stamps for this employee for stamp in emp_data['stamps']: status_text = 'Approved' if stamp.status == 'A' else ('Suspended' if stamp.status == 'S' else 'Pending') # Get function name from configuration function_name = stamp.stamp_function or 'N/A' if stamp.stamp_function and stamp.stamp_function in configurations_dict: func_data = configurations_dict[stamp.stamp_function] if isinstance(func_data, dict): function_name = func_data.get('header', stamp.stamp_function) else: function_name = func_data ws.cell(row=row, column=1).value = formatted_date ws.cell(row=row, column=2).value = stamp.time.strftime('%H:%M:%S') if stamp.time else 'N/A' ws.cell(row=row, column=3).value = emp_data['employee_name'] ws.cell(row=row, column=4).value = function_name ws.cell(row=row, column=5).value = stamp.project.name if stamp.project else 'No Project' ws.cell(row=row, column=6).value = stamp.task.name if stamp.task else 'No Task' ws.cell(row=row, column=7).value = stamp.duration or 'N/A' ws.cell(row=row, column=8).value = stamp.description or 'N/A' ws.cell(row=row, column=9).value = status_text # Apply borders for col in range(1, 10): ws.cell(row=row, column=col).border = border row += 1 row += 1 # Space between employees # Auto-adjust column widths for col in range(1, 10): max_length = 0 column = get_column_letter(col) for cell in ws[column]: try: if len(str(cell.value)) > max_length: max_length = len(str(cell.value)) except: pass adjusted_width = min(max_length + 2, 50) ws.column_dimensions[column].width = adjusted_width # Create response response = HttpResponse( content_type='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet' ) filename = f"worktime_report_{start_date or 'all'}_{end_date or 'all'}.xlsx" response['Content-Disposition'] = f'attachment; filename="{filename}"' wb.save(response) return response except Exception as e: log_error('worktime_report_excel', e) return Response( {"error": "An error occurred while generating Excel report"}, status=status.HTTP_500_INTERNAL_SERVER_ERROR ) @api_view(['GET']) @authentication_classes([]) # No authentication required @permission_classes([AllowAny]) def worktime_report_pdf(request): """ Generate worktime report as PDF file Query params: start_date, end_date, user_id, project_id, client_id, status """ try: # Get query parameters (same as worktime_report) start_date = request.query_params.get('start_date') end_date = request.query_params.get('end_date') user_id = request.query_params.get('user_id') project_id = request.query_params.get('project_id') client_id = request.query_params.get('client_id') status_filter = request.query_params.get('status') # Get the report data (reuse the same logic) queryset = Stamp.objects.select_related('user', 'project', 'task').all() # Apply filters if start_date: try: start_date_obj = datetime.strptime(start_date, '%Y-%m-%d').date() queryset = queryset.filter(date__gte=start_date_obj) except ValueError: pass if end_date: try: end_date_obj = datetime.strptime(end_date, '%Y-%m-%d').date() queryset = queryset.filter(date__lte=end_date_obj) except ValueError: pass if user_id: queryset = queryset.filter(user_id=user_id) if project_id: queryset = queryset.filter(project_id=project_id) if client_id: queryset = queryset.filter( Q(project__customer_id=client_id) | Q(project__client=client_id) ) if status_filter: queryset = queryset.filter(status=status_filter) stamps = queryset.order_by('-date', 'time') # Get configuration data for functions (to get function names) stamp_functions = set(stamps.values_list('stamp_function', flat=True)) configurations_dict = get_configuration_data_for_functions(stamp_functions, 'en') # Create PDF buffer = io.BytesIO() doc = SimpleDocTemplate(buffer, pagesize=A4, topMargin=0.5*inch, bottomMargin=0.5*inch) elements = [] styles = getSampleStyleSheet() title_style = ParagraphStyle( 'CustomTitle', parent=styles['Heading1'], fontSize=18, textColor=colors.HexColor('#366092'), spaceAfter=30, alignment=1 # Center ) # Title title = Paragraph("Worktime Report", title_style) elements.append(title) elements.append(Spacer(1, 0.2*inch)) # Filters info if start_date or end_date: filter_text = f"Date Range: {start_date or 'N/A'} to {end_date or 'N/A'}" elements.append(Paragraph(filter_text, styles['Normal'])) elements.append(Spacer(1, 0.1*inch)) # Group by date and employee stamps_by_date = {} for stamp in stamps: date_key = stamp.date.strftime('%Y-%m-%d') if date_key not in stamps_by_date: stamps_by_date[date_key] = {} user_id_str = str(stamp.user_id) if stamp.user_id else 'unknown' employee_name = f"{stamp.user.firstname or ''} {stamp.user.lastname or ''}".strip() if stamp.user else 'Unknown' if not employee_name: employee_name = stamp.user.email if stamp.user else 'Unknown' if user_id_str not in stamps_by_date[date_key]: stamps_by_date[date_key][user_id_str] = { 'employee_name': employee_name, 'stamps': [] } stamps_by_date[date_key][user_id_str]['stamps'].append(stamp) # Build PDF content for date_key in sorted(stamps_by_date.keys(), reverse=True): date_obj = datetime.strptime(date_key, '%Y-%m-%d').date() formatted_date = date_obj.strftime('%B %d, %Y') # Date header date_style = ParagraphStyle( 'DateHeader', parent=styles['Heading2'], fontSize=14, textColor=colors.HexColor('#366092'), spaceAfter=10, spaceBefore=20 ) elements.append(Paragraph(formatted_date, date_style)) for user_id_str, emp_data in sorted(stamps_by_date[date_key].items()): # Employee header emp_style = ParagraphStyle( 'EmployeeHeader', parent=styles['Heading3'], fontSize=12, textColor=colors.HexColor('#666666'), spaceAfter=5, spaceBefore=10 ) elements.append(Paragraph(f"Employee: {emp_data['employee_name']}", emp_style)) # Table data table_data = [['Time', 'Function', 'Project', 'Task', 'Duration', 'Status']] for stamp in emp_data['stamps']: status_text = 'Approved' if stamp.status == 'A' else ('Suspended' if stamp.status == 'S' else 'Pending') # Get function name from configuration function_name = stamp.stamp_function or 'N/A' if stamp.stamp_function and stamp.stamp_function in configurations_dict: func_data = configurations_dict[stamp.stamp_function] if isinstance(func_data, dict): function_name = func_data.get('header', stamp.stamp_function) else: function_name = func_data table_data.append([ stamp.time.strftime('%H:%M:%S') if stamp.time else 'N/A', function_name, stamp.project.name if stamp.project else 'No Project', stamp.task.name if stamp.task else 'No Task', stamp.duration or 'N/A', status_text ]) # Create table table = Table(table_data, colWidths=[1*inch, 1.2*inch, 1.5*inch, 1.2*inch, 1*inch, 0.8*inch]) table.setStyle(TableStyle([ ('BACKGROUND', (0, 0), (-1, 0), colors.HexColor('#366092')), ('TEXTCOLOR', (0, 0), (-1, 0), colors.whitesmoke), ('ALIGN', (0, 0), (-1, -1), 'LEFT'), ('FONTNAME', (0, 0), (-1, 0), 'Helvetica-Bold'), ('FONTSIZE', (0, 0), (-1, 0), 10), ('BOTTOMPADDING', (0, 0), (-1, 0), 12), ('BACKGROUND', (0, 1), (-1, -1), colors.beige), ('GRID', (0, 0), (-1, -1), 1, colors.grey), ('FONTSIZE', (0, 1), (-1, -1), 9), ])) elements.append(table) elements.append(Spacer(1, 0.2*inch)) elements.append(PageBreak()) # Build PDF doc.build(elements) # Create response response = HttpResponse(buffer.getvalue(), content_type='application/pdf') filename = f"worktime_report_{start_date or 'all'}_{end_date or 'all'}.pdf" response['Content-Disposition'] = f'attachment; filename="{filename}"' return response except Exception as e: log_error('worktime_report_pdf', e) return Response( {"error": "An error occurred while generating PDF report"}, status=status.HTTP_500_INTERNAL_SERVER_ERROR ) @api_view(['GET']) def get_user_stamp_permissions_view(request): """ Get stamp permissions for the currently authenticated user. Returns permissions based on the user's supervisor_code. """ try: if not request.user or not hasattr(request.user, 'id') or not request.user.id: return Response( {"error": "User not authenticated"}, status=status.HTTP_401_UNAUTHORIZED ) permissions = get_user_stamp_permissions(request.user.id) return Response(permissions, status=status.HTTP_200_OK) except Exception as e: log_error('get_user_stamp_permissions_view', e) return Response( {"error": "An error occurred while fetching permissions"}, status=status.HTTP_500_INTERNAL_SERVER_ERROR )