import logging from datetime import datetime, timedelta from django.utils import timezone from .models import WorkingTimePolicy from stamps.models import Stamp logger = logging.getLogger(__name__) class ComplianceChecker: """Real-time compliance checking to prevent violations before they occur""" def __init__(self, user_id, company_id=None): self.user_id = user_id self.company_id = company_id self.policy = None self._load_policy() def _load_policy(self): """Load working time policy for the user""" try: from user.models import User try: user = User.objects.get(id=self.user_id) # Use user's assigned policy if available if user.working_time_policy: self.policy = user.working_time_policy else: # Fallback: get latest policy for company if self.company_id: self.policy = WorkingTimePolicy.objects.filter(company_id=self.company_id).order_by('-created_at').first() else: # Get user's company if available company_id = getattr(user, 'company_id', None) if company_id: self.policy = WorkingTimePolicy.objects.filter(company_id=company_id).order_by('-created_at').first() else: self.policy = WorkingTimePolicy.objects.order_by('-created_at').first() except User.DoesNotExist: # Fallback if user doesn't exist if self.company_id: self.policy = WorkingTimePolicy.objects.filter(company_id=self.company_id).order_by('-created_at').first() else: self.policy = WorkingTimePolicy.objects.order_by('-created_at').first() except Exception as e: logger.error(f"Error loading policy: {e}") self.policy = None def check_compliance(self, current_time=None): """ Check current compliance status and return any approaching violations Returns: list: Array of compliance alerts with type, message, time_remaining """ if not self.policy: return [] alerts = [] current_time = current_time or timezone.now() today = current_time.date() # Get today's stamps today_stamps = Stamp.objects.filter( user_id=self.user_id, date=today ).order_by('time') if not today_stamps.exists(): return [] # Calculate work time today work_time = self._calculate_work_time(today_stamps) # Check daily limits daily_alert = self._check_daily_limit(work_time, current_time) if daily_alert: alerts.append(daily_alert) # Check break requirements break_alert = self._check_break_requirements(today_stamps, current_time) if break_alert: alerts.append(break_alert) # Check consecutive work days consecutive_alert = self._check_consecutive_work_days(current_time) if consecutive_alert: alerts.append(consecutive_alert) return alerts def _calculate_work_time(self, stamps): """Calculate total work time from stamps""" from datetime import timedelta total_seconds = 0 clock_ins = [] clock_outs = [] for stamp in stamps: # Get function type if hasattr(stamp, 'stamp_function'): from functions.models import Function try: func = Function.objects.get(id=stamp.stamp_function) question_type = func.question_type if question_type in ['clock_in', 'lunch_out']: clock_ins.append(stamp.time) elif question_type in ['clock_out', 'lunch_in']: clock_outs.append(stamp.time) except: pass # Match clock ins with clock outs for i, clock_in_time in enumerate(sorted(clock_ins)): if i < len(clock_outs): clock_out_time = sorted(clock_outs)[i] if clock_out_time > clock_in_time: duration = (clock_out_time - clock_in_time).total_seconds() total_seconds += duration return total_seconds / 3600 # Return in hours def _check_daily_limit(self, work_hours, current_time): """Check if approaching or at daily hour limit""" if not self.policy: return None # Get daily limit from policy daily_limit = getattr(self.policy, 'monday_friday_end', None) if not daily_limit: return None # Calculate expected daily hours start_time = getattr(self.policy, 'monday_friday_start', None) if not start_time: return None # Convert time to hours start_datetime = datetime.combine(current_time.date(), start_time) end_datetime = datetime.combine(current_time.date(), daily_limit) if end_datetime < start_datetime: end_datetime += timedelta(days=1) max_hours = (end_datetime - start_datetime).total_seconds() / 3600 remaining_hours = max_hours - work_hours if remaining_hours <= 0.5: # Less than 30 minutes remaining return { 'type': 'daily_limit_reached', 'severity': 'critical', 'message': f"You've reached your daily limit ({max_hours:.1f} hours). Please clock out.", 'time_remaining': 0 } elif remaining_hours <= 1: # Less than 1 hour remaining return { 'type': 'daily_limit_warning', 'severity': 'high', 'message': f"Approaching daily limit. {remaining_hours:.1f} hours remaining.", 'time_remaining': remaining_hours } return None def _check_break_requirements(self, stamps, current_time): """Check if break is required soon""" if not self.policy: return None # Get break policy min_break_duration = getattr(self.policy, 'minimum_total_break_time_minutes', 0) break_after_hours = getattr(self.policy, 'short_break_required_after_hours', 0) if not break_after_hours or not min_break_duration: return None # Get latest clock in last_clock_in = None total_work_minutes = 0 for stamp in stamps: if hasattr(stamp, 'stamp_function'): from functions.models import Function try: func = Function.objects.get(id=stamp.stamp_function) question_type = func.question_type if question_type in ['clock_in', 'lunch_in']: last_clock_in = stamp.time elif question_type in ['clock_out', 'lunch_out']: if last_clock_in: duration = (stamp.time - last_clock_in).total_seconds() / 60 total_work_minutes += duration last_clock_in = None except: pass # Check if currently clocked in if last_clock_in: work_duration = (current_time.time() - last_clock_in).total_seconds() / 60 total_work_minutes += work_duration # Check if break is required if total_work_minutes >= (break_after_hours * 60) - 15: # 15 minutes before requirement return { 'type': 'break_required_soon', 'severity': 'medium', 'message': f"Break required soon. You've worked for {(total_work_minutes/60):.1f} hours.", 'time_remaining': max(0, 15 - (total_work_minutes - break_after_hours * 60)) } return None def _check_consecutive_work_days(self, current_time): """Check consecutive work days""" if not self.policy: return None # Get work days in the past 7 days days_worked = 0 check_date = current_time.date() for i in range(7): date_to_check = check_date - timedelta(days=i) stamps = Stamp.objects.filter( user_id=self.user_id, date=date_to_check ) if stamps.exists(): # Check if there's a clock_in from functions.models import Function for stamp in stamps: try: func = Function.objects.get(id=stamp.stamp_function) if func.question_type == 'clock_in': days_worked += 1 break except: pass # Check if too many consecutive days if days_worked >= 5: return { 'type': 'consecutive_days_warning', 'severity': 'low', 'message': f"You've worked {days_worked} consecutive days. Consider taking a rest day.", 'time_remaining': None } return None def check_before_stamp_creation(self, stamp_type): """ Check if creating a stamp would cause a violation Returns: dict: Result with allowed (bool), alerts (list), warnings (list) """ if not self.policy: return {'allowed': True, 'alerts': [], 'warnings': []} # Get current compliance status alerts = self.check_compliance() # Filter critical alerts that would prevent stamp creation critical_alerts = [a for a in alerts if a.get('severity') == 'critical'] return { 'allowed': len(critical_alerts) == 0, 'alerts': alerts, 'warnings': [a for a in alerts if a.get('severity') in ['medium', 'low']] }