# worktime/services.py #from datetime import timedelta, datetime, time as dtime, timezone from django.db import transaction from django.db.models import Q from django.utils import timezone from datetime import datetime, time as dtime, timedelta from functions.models import Function #from .models import Stamps, WorkingTimePolicy, VacationPolicy, WorkBalance from stamps.models import Stamp as Stamps from company.models import WorkingTimePolicy,VacationPolicy from .models import WorkBalance from typing import List, Tuple, Optional # Classification sets - you can extend these if you have more stamp names IN_FUNCTIONS = {'clock_in', 'lunch_in', 'pre_entry_in', 'in'} OUT_FUNCTIONS = {'clock_out', 'lunch_out', 'post_exit_out', 'out'} REST_FUNCTIONS = {'pre_entry_time', 'rest', 'rest_start', 'rest_end'} BREAK_STARTS = {'break_start', 'lunch_start'} BREAK_ENDS = {'break_end', 'lunch_end'} # Helper: read the active policy for a user def get_active_working_policy(user_id=None): """ Get working time policy for a user. If user_id is provided, returns the user's assigned policy. Falls back to the company's default policy (first policy for the user's company) if user has no policy assigned. If no user_id, returns the first available policy (for backward compatibility). """ if user_id: from user.models import User try: user = User.objects.get(id=user_id) if user.working_time_policy: return user.working_time_policy # Fallback: get company's default policy (first policy for user's company) if hasattr(user, 'company_id') and user.company_id: company_policy = WorkingTimePolicy.objects.filter(company_id=user.company_id).order_by('created_at').first() if company_policy: return company_policy except User.DoesNotExist: pass # Final fallback: return first available policy (for backward compatibility or when no user_id) return WorkingTimePolicy.objects.order_by('created_at').first() def get_active_vacation_policy(): return VacationPolicy.objects.order_by('-created_at').first() """def seconds_between(a, b): if a.tzinfo is None: a = a.replace(tzinfo=timezone.utc) if b.tzinfo is None: b = b.replace(tzinfo=timezone.utc) return max(0, int((b - a).total_seconds())) def seconds_between(a, b): print("A date",a) print("B date", b) return max(0, int((b - a).total_seconds()))""" """def seconds_between(start, end): return int((end - start).total_seconds())""" """def _combine_datetime_from_date_and_time(d, t): #Return timezone-naive datetime combining date and time (used only if start_date absent). # create naive datetime; stamps.start_date is authoritative if present return datetime.combine(d, t)""" def _combine_datetime_from_date_and_time(d, t): return datetime.combine(d, t) def ensure_aware(dt: datetime) -> datetime | None: """ Ensure datetime is timezone-aware. If dt.tzinfo is None, make it aware using Django's current timezone (usually settings.TIME_ZONE). If dt already aware, normalize to the same timezone object (preserve tz). """ if dt is None: return None if dt.tzinfo is None: return timezone.make_aware(dt, timezone.get_current_timezone()) # ensure a consistent tzinfo object (use astimezone to normalize) return dt.astimezone(timezone.get_current_timezone()) def seconds_between(a: datetime, b: datetime) -> int: """ Return positive seconds between two datetimes. Both datetimes will be made timezone-aware (current timezone) before subtraction. Returns 0 if b <= a. """ a_aware = ensure_aware(a) b_aware = ensure_aware(b) diff = (b_aware - a_aware).total_seconds() return max(0, int(diff)) def _stamp_timestamp(s: Stamps, func_obj=None) -> datetime: """ Return the effective timestamp for a stamp (combine date + time), using current timezone and ensuring timezone-awareness. Special handling for with_reason stamps: - If out=true and with_reason=true: Use return_date.date() + time column - Otherwise: Use normal date + time """ # Check if this is a with_reason stamp if func_obj: is_out = getattr(func_obj, "out", False) with_reason = getattr(func_obj, "with_reason", False) if with_reason and is_out and s.return_date: # out=true + with_reason=true: Use return_date's date + time column return_date_only = s.return_date.date() if hasattr(s.return_date, 'date') else s.return_date ts = datetime.combine(return_date_only, s.time) return ensure_aware(ts) # Normal case: combine date + time ts = datetime.combine(s.date, s.time) return ensure_aware(ts) # ------------------------- # Rest checks helpers # ------------------------- def check_daily_rest_violation(user_id: str, for_date: datetime.date, work_intervals: List[Tuple[datetime, datetime]], function_map: dict) -> bool: """ Check if gap between previous day's last work end and today's first start is less than 11 hours. """ prev_date = for_date - timedelta(days=1) prev_stamps = Stamps.objects.filter(user_id=user_id, date=prev_date).order_by('start_date', 'created_at') prev_last_end: Optional[datetime] = None # search for latest end marker on prev_date for s in reversed(list(prev_stamps)): func_obj = function_map.get(s.stamp_function) with_reason = getattr(func_obj, "with_reason", False) if func_obj else False # Handle with_reason stamps: out=true + with_reason=true uses return_date.date() + time if with_reason and func_obj and bool(getattr(func_obj, "out", False)) and s.return_date: # For with_reason out stamps, use return_date.date() + time return_date_only = s.return_date.date() if hasattr(s.return_date, 'date') else s.return_date prev_last_end = ensure_aware(datetime.combine(return_date_only, s.time)) break # if pre-time entry with return_date use that elif s.return_date: prev_last_end = ensure_aware(s.return_date) break # otherwise if function marks an OUT -> use the normal stamp timestamp elif func_obj and bool(getattr(func_obj, "out", False)): prev_last_end = _stamp_timestamp(s, func_obj) break if prev_last_end and work_intervals: first_start = work_intervals[0][0] gap_secs = seconds_between(prev_last_end, first_start) return gap_secs < (11 * 3600) return False def check_weekly_rest_violation(user_id: str, for_date: datetime.date, function_map: dict) -> bool: """ Compute longest continuous rest in the calendar week (Mon-Sun). Violation if longest_rest < 35 hours (126000 seconds). """ week_start = for_date - timedelta(days=for_date.weekday()) # monday week_end = week_start + timedelta(days=6) # collect all stamps in the week (both normal and pre-time entries) week_qs = Stamps.objects.filter( user_id=user_id ).filter( Q(date__range=(week_start, week_end)) | Q(start_date__date__range=(week_start, week_end)) | Q(start_date__lte=timezone.make_aware(datetime.combine(week_end, dtime.max)), return_date__gte=timezone.make_aware(datetime.combine(week_start, dtime.min))) ).order_by('start_date', 'created_at') w_events: List[Tuple[datetime, object, Stamps]] = [] for s in week_qs: func_obj = function_map.get(s.stamp_function) # Check if this is a with_reason stamp with_reason = getattr(func_obj, "with_reason", False) if func_obj else False if with_reason: # with_reason stamps: use special timestamp logic w_events.append((_stamp_timestamp(s, func_obj), func_obj, s)) elif s.start_date and s.return_date: # pre-time entry with explicit start/return w_events.append((ensure_aware(s.start_date), func_obj, s)) w_events.append((ensure_aware(s.return_date), 'return_marker', s)) else: # Normal stamp w_events.append((_stamp_timestamp(s, func_obj), func_obj, s)) w_events.sort(key=lambda x: x[0]) week_intervals: List[Tuple[datetime, datetime]] = [] w_in: Optional[datetime] = None # Special handling for with_reason stamps in weekly calculation with_reason_out_timestamp = None waiting_for_with_reason_in = False for ts, func_obj, s in w_events: # Skip return_marker events if func_obj == 'return_marker': continue if func_obj: is_out = bool(getattr(func_obj, "out", False)) with_reason = getattr(func_obj, "with_reason", False) is_in = not is_out and func_obj not in (None, 'return_marker') else: is_out = "out" in (s.stamp_function or "").lower() is_in = not is_out with_reason = False # Handle IN stamps (normal or with_reason) if is_in: # Any IN stamp (with_reason or not) can start a work interval if w_in is None: w_in = ts # If we were waiting for a with_reason IN and this is it, handle the pair if with_reason and waiting_for_with_reason_in and with_reason_out_timestamp is not None: # This is the matching IN for a previous OUT with_reason # Add work interval from the OUT with_reason to this IN week_intervals.append((with_reason_out_timestamp, ts)) # Reset for next pair with_reason_out_timestamp = None waiting_for_with_reason_in = False # Note: w_in is already set above, so this IN can also be used for future OUTs # Handle OUT stamps (normal or with_reason) elif is_out: # If we have an active IN (w_in), calculate work time from IN to this OUT if w_in is not None: week_intervals.append((w_in, ts)) w_in = None # Close the interval # Also track OUT with_reason separately (for case where OUT comes before IN) if with_reason: with_reason_out_timestamp = ts waiting_for_with_reason_in = True # compute maximal rest block between intervals (from week_start..week_end) if week_intervals: prev_end = ensure_aware(datetime.combine(week_start, dtime.min)) longest_rest = 0 for a, b in week_intervals: gap = seconds_between(prev_end, a) if gap > longest_rest: longest_rest = gap prev_end = b # final gap to week_end final_gap = seconds_between(prev_end, ensure_aware(datetime.combine(week_end, dtime.max))) if final_gap > longest_rest: longest_rest = final_gap else: # no work in week -> full week rest longest_rest = seconds_between(ensure_aware(datetime.combine(week_start, dtime.min)), ensure_aware(datetime.combine(week_end, dtime.max))) return longest_rest < (35 * 3600) # ------------------------- # Main exported function # ------------------------- @transaction.atomic def calculate_daily_balances_for_user_and_date(user_id: str, for_date): """ Compute WorkBalance for a user & date. Rules / assumptions: - Normal stamps -> timestamp = combine(date, time) - Pre-time entries (start_date & return_date) -> used for multi-hour breaks / pre-entries - Function model fields expected: function_ref_id, out (bool), pre_function (bool), break_flag (bool) - Policy provides monday_friday_start, monday_friday_end, saturday_start/end, sunday_start/end, lunch_break_required_after (hours), lunch_break_duration (minutes), lunch_break_paid (bool), use_time_banking (bool), max_bank_balance (hours), workers_self_register_overtime (bool) """ # load policies policy = get_active_working_policy(user_id) vacation_policy = get_active_vacation_policy() # day boundaries (aware) tz = timezone.get_current_timezone() day_start = ensure_aware(datetime.combine(for_date, dtime.min)) day_end = ensure_aware(datetime.combine(for_date, dtime.max)) # fetch stamps that either are on the date or overlap via start/return (pre-time entries) stamps_qs = Stamps.objects.filter(user_id=user_id).filter( Q(date=for_date) | Q(start_date__lte=day_end, return_date__gte=day_start) | Q(start_date__date=for_date) ).order_by('start_date', 'created_at') # If no stamps exist for this date, delete old WorkBalance and BalanceDetails and return if not stamps_qs.exists(): # Delete WorkBalance for this date WorkBalance.objects.filter(user_id=user_id, date=for_date).delete() # Delete all BalanceDetails for this date from balancedetail.models import BalanceDetail BalanceDetail.objects.filter(user_id=user_id, date=for_date).delete() return None # function map (function_ref_id -> Function instance) function_map = {f.function_ref_id: f for f in Function.objects.all()} # build chronological events: # - normal stamps: (timestamp, func, stamp) # - pre-time entries: both start_date and return_date events # - with_reason stamps: special timestamp handling events: List[Tuple[datetime, object, Stamps]] = [] for s in stamps_qs: func_obj = function_map.get(s.stamp_function) # Check if this is a with_reason stamp with_reason = getattr(func_obj, "with_reason", False) if func_obj else False if with_reason: # with_reason stamps: use special timestamp logic # out=true + with_reason=true: return_date.date() + time # out=false + with_reason=true: date + time (normal) events.append((_stamp_timestamp(s, func_obj), func_obj, s)) elif s.start_date and s.return_date: # treat as pre-time entry only if both start_date and return_date present # (and not a with_reason stamp) events.append((ensure_aware(s.start_date), func_obj, s)) events.append((ensure_aware(s.return_date), 'return_marker', s)) else: # Normal stamp events.append((_stamp_timestamp(s, func_obj), func_obj, s)) events.sort(key=lambda x: x[0]) # Counters total_work_seconds = 0 total_break_seconds = 0 in_stack: Optional[datetime] = None break_start: Optional[datetime] = None work_intervals: List[Tuple[datetime, datetime]] = [] # Progressive IN/OUT handling in_stack = None # current open interval start last_out_time = None # last closed OUT timestamp # Special handling for with_reason stamps with_reason_out_timestamp = None # Track out=true + with_reason=true timestamp waiting_for_with_reason_in = False # Flag to indicate we're waiting for matching in for ts, func_obj, s in events: # Skip return_marker events (they're handled separately for pre-function entries) if func_obj == 'return_marker': continue # Determine type if func_obj: is_out = getattr(func_obj, "out", False) is_break = getattr(func_obj, 'break_flag', False) is_pre_function = getattr(func_obj, 'pre_function', False) with_reason = getattr(func_obj, "with_reason", False) is_in = not is_out # break_flag should not interfere with IN/OUT determination else: func_lower = (s.stamp_function or "").lower() is_in = 'in' in func_lower is_out = 'out' in func_lower is_break = False is_pre_function = False with_reason = False # Handle IN stamps (normal or with_reason) if is_in: # Any IN stamp (with_reason or not) can start a work interval if in_stack is None: in_stack = ts # If we were waiting for a with_reason IN and this is it, handle the pair if with_reason and waiting_for_with_reason_in and with_reason_out_timestamp is not None: # This is the matching IN for a previous OUT with_reason # Calculate work time from the OUT with_reason to this IN interval_secs = seconds_between(with_reason_out_timestamp, ts) total_work_seconds += interval_secs work_intervals.append((with_reason_out_timestamp, ts)) last_out_time = ts # Reset for next pair with_reason_out_timestamp = None waiting_for_with_reason_in = False # Note: in_stack is already set above, so this IN can also be used for future OUTs # Handle OUT stamps (normal or with_reason) elif is_out: # If we have an active IN (in_stack), calculate work time from IN to this OUT if in_stack is not None: interval_secs = seconds_between(in_stack, ts) total_work_seconds += interval_secs work_intervals.append((in_stack, ts)) last_out_time = ts in_stack = None # Close the interval # Also track OUT with_reason separately (for case where OUT comes before IN) if with_reason: with_reason_out_timestamp = ts waiting_for_with_reason_in = True # Handle Breaks - break_flag tracks time spent out as break # out=True + break_flag=True: start counting break seconds # out=False + break_flag=True: stop counting break seconds if is_break: if is_out: # Break OUT: start tracking break time if break_start is None: break_start = ts elif is_in: # Break IN: end tracking break time if break_start is not None: total_break_seconds += seconds_between(break_start, ts) break_start = None # Pre-function / Rest if is_pre_function and s.return_date: # Check if this is a paid absence - if so, treat as work, not break is_paid_absence = False if s.paycode: try: from paycode.models import Paycode import uuid # Try to find paycode by paycode string first paycode_obj = Paycode.objects.filter(paycode=s.paycode).first() # If not found, try to find by UUID (in case paycode field stores UUID) if not paycode_obj: try: paycode_uuid = uuid.UUID(str(s.paycode)) paycode_obj = Paycode.objects.filter(id=paycode_uuid).first() except (ValueError, AttributeError): pass if paycode_obj and paycode_obj.isAbsence and paycode_obj.isPaid: is_paid_absence = True except Exception: pass if is_paid_absence: # Treat paid absence as work hours - each day is calculated separately work_seconds = seconds_between(s.start_date, s.return_date) total_work_seconds += work_seconds work_intervals.append((ensure_aware(s.start_date), ensure_aware(s.return_date))) else: # Treat as break (original behavior for non-paid absence) total_break_seconds += seconds_between(s.start_date, s.return_date) # ------------------------- # Scheduled work seconds based on policy (supports Mon-Fri, Sat, Sun) # ------------------------- scheduled_work_seconds = 8 * 3600 # default if policy: weekday = for_date.weekday() # Mon=0..Sun=6 if weekday <= 4 and getattr(policy, "monday_friday_start", None) and getattr(policy, "monday_friday_end", None): start = policy.monday_friday_start end = policy.monday_friday_end scheduled_work_seconds = int((datetime.combine(for_date, end) - datetime.combine(for_date, start)).total_seconds()) elif weekday == 5 and getattr(policy, "saturday_start", None) and getattr(policy, "saturday_end", None): start = policy.saturday_start end = policy.saturday_end scheduled_work_seconds = int((datetime.combine(for_date, end) - datetime.combine(for_date, start)).total_seconds()) elif weekday == 6 and getattr(policy, "sunday_start", None) and getattr(policy, "sunday_end", None): start = policy.sunday_start end = policy.sunday_end scheduled_work_seconds = int((datetime.combine(for_date, end) - datetime.combine(for_date, start)).total_seconds()) else: # fallback default if policy doesn't define that day scheduled_work_seconds = getattr(policy, "default_daily_seconds", 8 * 3600) print("him", scheduled_work_seconds) # ------------------------- # Break rules (Lunch and Short breaks) - using all policy settings # ------------------------- lunch_required_after_seconds = (policy.lunch_break_required_after * 3600) if policy and getattr(policy, "lunch_break_required_after", None) is not None else (6 * 3600) lunch_break_duration_seconds = (policy.lunch_break_duration * 60) if policy and getattr(policy, "lunch_break_duration", None) is not None else (30 * 60) # Short break policy settings (from policy) short_break_required = policy and bool(getattr(policy, "short_break_required", True)) if policy else True short_break_duration_seconds = (policy.short_break_duration_minutes * 60) if policy and getattr(policy, "short_break_duration_minutes", None) is not None else (15 * 60) short_break_required_after_seconds = (policy.short_break_required_after_hours * 3600) if policy and getattr(policy, "short_break_required_after_hours", None) is not None else (4 * 3600) minimum_total_break_seconds = (policy.minimum_total_break_time_minutes * 60) if policy and getattr(policy, "minimum_total_break_time_minutes", None) is not None else (45 * 60) lunch_missing = False # Check if lunch break stamps exist (lunch_in and lunch_out) has_lunch_stamps = any( getattr(function_map.get(s.stamp_function), 'break_flag', False) or 'lunch' in (s.stamp_function or "").lower() for s in stamps_qs ) # Check lunch break requirement if total_work_seconds >= lunch_required_after_seconds and total_break_seconds < lunch_break_duration_seconds: lunch_missing = True # Only deduct unpaid lunch break if: # 1. Policy exists # 2. Lunch is explicitly unpaid # 3. BUT only if NO lunch stamps were recorded (user didn't break their lunch) if policy and not bool(getattr(policy, "lunch_break_paid", True)) and not has_lunch_stamps: missing = lunch_break_duration_seconds - total_break_seconds total_work_seconds = max(0, total_work_seconds - missing) # Check short break requirement (if enabled in policy) # Note: This could be used for violation monitoring if policy.enable_violation_monitoring is True if short_break_required and total_work_seconds >= short_break_required_after_seconds: # Ensure minimum total break time is met (could flag as violation) if total_break_seconds < minimum_total_break_seconds: # Could be used for violation monitoring pass # ------------------------- # Net / overtime / flex / bank logic - considering all policy sections # ------------------------- net_work_seconds = total_work_seconds overtime_seconds = int(max(0, net_work_seconds - scheduled_work_seconds)) # Calculate flex seconds considering flextime policy settings flex_seconds = int(net_work_seconds - scheduled_work_seconds) if policy and bool(getattr(policy, "use_flextime", False)): # Apply flextime limits if enabled (for tracking, actual enforcement would be elsewhere) max_flex_negative_secs = int(getattr(policy, "max_flex_negative", -10) * 3600) max_flex_positive_secs = int(getattr(policy, "max_flex_positive", 40) * 3600) # Note: flex_seconds can be negative or positive, limits are for tracking purposes bank_credited_seconds = 0 bank_debited_seconds = 0 if policy and bool(getattr(policy, "use_time_banking", False)): max_bank_secs = int(getattr(policy, "max_bank_balance", 0) * 3600) # Use overtime conversion rates if specified (for future enhancement) overtime_conversion_daily = getattr(policy, "overtime_conversion_daily", 8) overtime_conversion_weekly = getattr(policy, "overtime_conversion_weekly", 40) # Credit overtime up to max_bank_balance bank_credited_seconds = min(overtime_seconds, max_bank_secs) toil_seconds = 0 if policy and not bool(getattr(policy, "workers_self_register_overtime", False)): if getattr(getattr(vacation_policy, 'track_toil', None), '__bool__', None): toil_seconds = overtime_seconds # ------------------------- # Rest violation checks # ------------------------- daily_rest_violation = check_daily_rest_violation(user_id, for_date, work_intervals, function_map) weekly_rest_violation = check_weekly_rest_violation(user_id, for_date, function_map) # ------------------------- # Vacation accrual (simple) # ------------------------- vacation_days_accrued = 0.0 if vacation_policy: if getattr(vacation_policy, "vacation_accrual_type", None) == 'standard_finnish': vacation_days_accrued = round(2.0 / 30.0, 4) elif getattr(vacation_policy, "vacation_accrual_type", None) == 'custom' and getattr(vacation_policy, "custom_accrual_rate", None): vacation_days_accrued = float(vacation_policy.custom_accrual_rate) / 30.0 # ------------------------- # Check for pre-function stamps with return_date that span this date # If it's a paid absence, use scheduled_work_seconds for regular and total work # ------------------------- is_paid_absence_date = False for s in stamps_qs: func_obj = function_map.get(s.stamp_function) is_pre_function = bool(getattr(func_obj, 'pre_function', False)) if func_obj else False #print(f"DEBUG: Stamp {s.id}, function={s.stamp_function}, is_pre_function={is_pre_function}, has_start={bool(s.start_date)}, has_return={bool(s.return_date)}, paycode={s.paycode}") if is_pre_function and s.start_date and s.return_date: print("yes its a pre stamp",is_pre_function) # Check if this date falls within the stamp's date range stamp_start_date = s.start_date.date() if hasattr(s.start_date, 'date') else s.start_date stamp_return_date = s.return_date.date() if hasattr(s.return_date, 'date') else s.return_date #print(f"DEBUG: Date range check - for_date={for_date}, stamp_start={stamp_start_date}, stamp_return={stamp_return_date}, in_range={stamp_start_date <= for_date <= stamp_return_date}") # Check if for_date is within the range (inclusive) if stamp_start_date <= for_date <= stamp_return_date: # Check if it's a paid absence if s.paycode: try: from paycode.models import Paycode import uuid # Try to find paycode by paycode string first paycode_obj = Paycode.objects.filter(paycode=s.paycode).first() #print(f"DEBUG: Paycode lookup by string '{s.paycode}' - found: {paycode_obj is not None}") # If not found, try to find by UUID (in case paycode field stores UUID) if not paycode_obj: try: paycode_uuid = uuid.UUID(str(s.paycode)) paycode_obj = Paycode.objects.filter(id=paycode_uuid).first() #print(f"DEBUG: Paycode lookup by UUID '{paycode_uuid}' - found: {paycode_obj is not None}") except (ValueError, AttributeError) as e: #print(f"DEBUG: UUID conversion failed: {e}") pass if paycode_obj: #print(f"DEBUG: Paycode found - isAbsence={paycode_obj.isAbsence}, isPaid={paycode_obj.isPaid}") if paycode_obj.isAbsence and paycode_obj.isPaid: is_paid_absence_date = True #print(f"DEBUG: Setting is_paid_absence_date = True") break # Found a paid absence stamp for this date else: pass #print(f"DEBUG: Paycode not found for '{s.paycode}'") except Exception as e: print(f"DEBUG: Exception checking paycode: {e}") pass else: print(f"DEBUG: Stamp has no paycode field") else: print(f"DEBUG: Stamp does not meet pre-function criteria - is_pre_function={is_pre_function}, has_start={bool(s.start_date)}, has_return={bool(s.return_date)}") # ------------------------- # Persist WorkBalance # ------------------------- if is_paid_absence_date: print("is absence date") # For paid absence dates, use scheduled_work_seconds for regular and total work # All other fields use defaults wb, created = WorkBalance.objects.update_or_create( user_id=user_id, date=for_date, defaults={ 'regular_work_seconds': int(scheduled_work_seconds), 'total_work_seconds': int(scheduled_work_seconds), 'daily_break_seconds': 0, 'net_work_seconds': int(scheduled_work_seconds), 'overtime_seconds': 0, 'flex_seconds': 0, 'bank_credited_seconds': 0, 'bank_debited_seconds': 0, 'toil_seconds': 0, 'vacation_days_accrued': vacation_days_accrued, # Keep vacation accrual 'lunch_break_missing_flag': False, 'daily_rest_violation_flag': False, 'weekly_rest_violation_flag': False, } ) else: print("not is absence date") # Normal calculation for non-paid absence dates wb, created = WorkBalance.objects.update_or_create( user_id=user_id, date=for_date, defaults={ 'regular_work_seconds': min(total_work_seconds, scheduled_work_seconds), 'total_work_seconds': int(total_work_seconds), 'daily_break_seconds': int(total_break_seconds), 'net_work_seconds': int(net_work_seconds), 'overtime_seconds': int(overtime_seconds), 'flex_seconds': int(flex_seconds), 'bank_credited_seconds': int(bank_credited_seconds), 'bank_debited_seconds': int(bank_debited_seconds), 'toil_seconds': int(toil_seconds), 'vacation_days_accrued': vacation_days_accrued, 'lunch_break_missing_flag': lunch_missing, 'daily_rest_violation_flag': daily_rest_violation, 'weekly_rest_violation_flag': weekly_rest_violation, } ) # ------------------------- # Calculate and save balance details by paycode # ------------------------- calculate_balance_details_by_paycode(user_id, wb, for_date, stamps_qs, function_map, work_intervals, policy) return wb def calculate_balance_details_by_paycode(user_id: str, work_balance: WorkBalance, for_date, stamps_qs, function_map: dict, work_intervals: List[Tuple[datetime, datetime]], policy=None): """ Calculate detailed balance breakdown by paycode using the same logic as WorkBalance but grouped by paycode - this ensures natural tally without forced distribution """ # Lazy import to avoid circular dependency from balancedetail.models import BalanceDetail # Delete all old BalanceDetails for this date to ensure clean recalculation BalanceDetail.objects.filter(user_id=user_id, date=for_date).delete() # Get all functions with their paycode relationships functions_with_paycode = {func.function_ref_id: func for func in Function.objects.select_related('paycode').all() if func.paycode_id} # Group stamps by paycode paycode_stamps = {} for stamp in stamps_qs: if stamp.stamp_function in functions_with_paycode: function = functions_with_paycode[stamp.stamp_function] paycode = function.paycode if paycode.id not in paycode_stamps: paycode_stamps[paycode.id] = { 'paycode': paycode, 'stamps': [] } paycode_stamps[paycode.id]['stamps'].append(stamp) # Calculate balance for each paycode using the same logic as WorkBalance for paycode_id, paycode_data in paycode_stamps.items(): paycode = paycode_data['paycode'] paycode_stamp_list = paycode_data['stamps'] # Check if any stamp for this paycode is a paid absence pre-function with return_date is_paid_absence_paycode = False for s in paycode_stamp_list: func_obj = function_map.get(s.stamp_function) is_pre_function = bool(getattr(func_obj, 'pre_function', False)) if func_obj else False if is_pre_function and s.start_date and s.return_date: # Check if this date falls within the stamp's date range stamp_start_date = s.start_date.date() if hasattr(s.start_date, 'date') else s.start_date stamp_return_date = s.return_date.date() if hasattr(s.return_date, 'date') else s.return_date # Check if for_date is within the range (inclusive) if stamp_start_date <= for_date <= stamp_return_date: # Check if it's a paid absence if s.paycode: try: from paycode.models import Paycode import uuid # Try to find paycode by paycode string first paycode_obj = Paycode.objects.filter(paycode=s.paycode).first() # If not found, try to find by UUID (in case paycode field stores UUID) if not paycode_obj: try: paycode_uuid = uuid.UUID(str(s.paycode)) paycode_obj = Paycode.objects.filter(id=paycode_uuid).first() except (ValueError, AttributeError): pass if paycode_obj and paycode_obj.isAbsence and paycode_obj.isPaid: is_paid_absence_paycode = True break # Found a paid absence stamp for this paycode and date except Exception: pass # Calculate hours for this paycode using the same logic as WorkBalance paycode_hours = calculate_hours_for_paycode(paycode_stamp_list, for_date, function_map, user_id, policy) # If this is a paid absence date, override with scheduled_work_seconds if is_paid_absence_paycode: # Get scheduled work seconds for this date using user's policy policy = get_active_working_policy(user_id) scheduled_work_seconds = 8 * 3600 # default if policy: weekday = for_date.weekday() # Mon=0..Sun=6 if weekday <= 4 and getattr(policy, "monday_friday_start", None) and getattr(policy, "monday_friday_end", None): start = policy.monday_friday_start end = policy.monday_friday_end scheduled_work_seconds = int((datetime.combine(for_date, end) - datetime.combine(for_date, start)).total_seconds()) elif weekday == 5 and getattr(policy, "saturday_start", None) and getattr(policy, "saturday_end", None): start = policy.saturday_start end = policy.saturday_end scheduled_work_seconds = int((datetime.combine(for_date, end) - datetime.combine(for_date, start)).total_seconds()) elif weekday == 6 and getattr(policy, "sunday_start", None) and getattr(policy, "sunday_end", None): start = policy.sunday_start end = policy.sunday_end scheduled_work_seconds = int((datetime.combine(for_date, end) - datetime.combine(for_date, start)).total_seconds()) else: scheduled_work_seconds = getattr(policy, "default_daily_seconds", 8 * 3600) # Override paycode_hours with scheduled_work_seconds for paid absence paycode_hours = { 'regular_work_seconds': int(scheduled_work_seconds), 'total_work_seconds': int(scheduled_work_seconds), 'daily_break_seconds': 0, 'net_work_seconds': int(scheduled_work_seconds), 'overtime_seconds': 0, 'flex_seconds': 0, 'bank_credited_seconds': 0, 'bank_debited_seconds': 0, 'overtime_balance_seconds': 0, 'toil_seconds': 0, 'vacation_days_accrued': 0 } # Create or update BalanceDetail record balance_detail, created = BalanceDetail.objects.get_or_create( user_id=user_id, work_balance=work_balance, paycode=paycode, defaults={ 'date': for_date, **paycode_hours } ) if not created: # Update existing record with new calculated values for field, value in paycode_hours.items(): setattr(balance_detail, field, value) balance_detail.save() def calculate_hours_for_paycode(stamps: List[Stamps], for_date, function_map: dict, user_id=None, policy=None): """ Calculate hours for a specific paycode using the same logic as WorkBalance calculation """ # Use the same calculation logic as the main function but only for stamps with this specific paycode # Day boundaries (aware) tz = timezone.get_current_timezone() day_start = ensure_aware(datetime.combine(for_date, dtime.min)) day_end = ensure_aware(datetime.combine(for_date, dtime.max)) # Build chronological events for this paycode only events: List[Tuple[datetime, object, Stamps]] = [] for s in stamps: func_obj = function_map.get(s.stamp_function) # Check if this is a with_reason stamp with_reason = getattr(func_obj, "with_reason", False) if func_obj else False if with_reason: # with_reason stamps: use special timestamp logic events.append((_stamp_timestamp(s, func_obj), func_obj, s)) elif s.start_date and s.return_date: # Treat as pre-time entry only if both start_date and return_date present # (and not a with_reason stamp) events.append((ensure_aware(s.start_date), func_obj, s)) events.append((ensure_aware(s.return_date), 'return_marker', s)) else: # Normal stamp events.append((_stamp_timestamp(s, func_obj), func_obj, s)) events.sort(key=lambda x: x[0]) # Counters for this paycode total_work_seconds = 0 total_break_seconds = 0 in_stack: Optional[datetime] = None break_start: Optional[datetime] = None work_intervals: List[Tuple[datetime, datetime]] = [] # Progressive IN/OUT handling (same logic as main function) in_stack = None last_out_time = None # Special handling for with_reason stamps with_reason_out_timestamp = None waiting_for_with_reason_in = False for ts, func_obj, s in events: # Skip return_marker events if func_obj == 'return_marker': continue # Determine type (same logic as main function) if func_obj: is_out = getattr(func_obj, "out", False) is_break = getattr(func_obj, 'break_flag', False) is_pre_function = getattr(func_obj, 'pre_function', False) with_reason = getattr(func_obj, "with_reason", False) is_in = not is_out # break_flag should not interfere with IN/OUT determination else: func_lower = (s.stamp_function or "").lower() is_in = 'in' in func_lower is_out = 'out' in func_lower is_break = False is_pre_function = False with_reason = False # Handle IN stamps (normal or with_reason) if is_in: # Any IN stamp (with_reason or not) can start a work interval if in_stack is None: in_stack = ts # If we were waiting for a with_reason IN and this is it, handle the pair if with_reason and waiting_for_with_reason_in and with_reason_out_timestamp is not None: # This is the matching IN for a previous OUT with_reason # Calculate work time from the OUT with_reason to this IN interval_secs = seconds_between(with_reason_out_timestamp, ts) total_work_seconds += interval_secs work_intervals.append((with_reason_out_timestamp, ts)) last_out_time = ts # Reset for next pair with_reason_out_timestamp = None waiting_for_with_reason_in = False # Note: in_stack is already set above, so this IN can also be used for future OUTs # Handle OUT stamps (normal or with_reason) elif is_out: # If we have an active IN (in_stack), calculate work time from IN to this OUT if in_stack is not None: interval_secs = seconds_between(in_stack, ts) total_work_seconds += interval_secs work_intervals.append((in_stack, ts)) last_out_time = ts in_stack = None # Close the interval # Also track OUT with_reason separately (for case where OUT comes before IN) if with_reason: with_reason_out_timestamp = ts waiting_for_with_reason_in = True # Handle Breaks - break_flag tracks time spent out as break # out=True + break_flag=True: start counting break seconds # out=False + break_flag=True: stop counting break seconds if is_break: if is_out: # Break OUT: start tracking break time if break_start is None: break_start = ts elif is_in: # Break IN: end tracking break time if break_start is not None: total_break_seconds += seconds_between(break_start, ts) break_start = None # Pre-function / Rest if is_pre_function and s.return_date: # Check if this is a paid absence - if so, treat as work, not break is_paid_absence = False if s.paycode: try: from paycode.models import Paycode import uuid paycode_obj = Paycode.objects.filter(paycode=s.paycode).first() if not paycode_obj: try: paycode_uuid = uuid.UUID(str(s.paycode)) paycode_obj = Paycode.objects.filter(id=paycode_uuid).first() except (ValueError, AttributeError): pass if paycode_obj and paycode_obj.isAbsence and paycode_obj.isPaid: is_paid_absence = True except Exception: pass if is_paid_absence: # Treat paid absence as work hours work_seconds = seconds_between(s.start_date, s.return_date) total_work_seconds += work_seconds work_intervals.append((ensure_aware(s.start_date), ensure_aware(s.return_date))) else: # Treat as break (original behavior) total_break_seconds += seconds_between(s.start_date, s.return_date) # Get policy for scheduled work calculation (use provided policy or fetch for user) if not policy: policy = get_active_working_policy(user_id) scheduled_work_seconds = 8 * 3600 # default if policy: weekday = for_date.weekday() # Mon=0..Sun=6 if weekday <= 4 and getattr(policy, "monday_friday_start", None) and getattr(policy, "monday_friday_end", None): start = policy.monday_friday_start end = policy.monday_friday_end scheduled_work_seconds = int((datetime.combine(for_date, end) - datetime.combine(for_date, start)).total_seconds()) elif weekday == 5 and getattr(policy, "saturday_start", None) and getattr(policy, "saturday_end", None): start = policy.saturday_start end = policy.saturday_end scheduled_work_seconds = int((datetime.combine(for_date, end) - datetime.combine(for_date, start)).total_seconds()) elif weekday == 6 and getattr(policy, "sunday_start", None) and getattr(policy, "sunday_end", None): start = policy.sunday_start end = policy.sunday_end scheduled_work_seconds = int((datetime.combine(for_date, end) - datetime.combine(for_date, start)).total_seconds()) else: scheduled_work_seconds = getattr(policy, "default_daily_seconds", 8 * 3600) # Calculate net work, overtime, etc. for this paycode net_work_seconds = total_work_seconds overtime_seconds = int(max(0, net_work_seconds - scheduled_work_seconds)) flex_seconds = int(net_work_seconds - scheduled_work_seconds) # Store values in seconds (same as WorkBalance) regular_work_seconds = min(total_work_seconds, scheduled_work_seconds) return { 'regular_work_seconds': regular_work_seconds, 'total_work_seconds': total_work_seconds, 'daily_break_seconds': total_break_seconds, 'net_work_seconds': net_work_seconds, 'overtime_seconds': overtime_seconds, 'flex_seconds': flex_seconds, 'bank_credited_seconds': 0, # Could be calculated based on policy 'bank_debited_seconds': 0, # Could be calculated based on policy 'overtime_balance_seconds': 0, # Could be calculated based on policy 'toil_seconds': 0, # Could be calculated based on policy 'vacation_days_accrued': 0 # Could be calculated based on policy }