# -*- coding: utf-8 -*- # Part of Odoo. See LICENSE file for full copyright and licensing details. from collections import defaultdict from datetime import datetime, timedelta from operator import itemgetter import pytz from odoo import models, fields, api, exceptions, _ from odoo.tools import format_datetime from odoo.osv.expression import AND, OR from odoo.tools.float_utils import float_is_zero class HrAttendance(models.Model): _name = "hr.attendance" _description = "Attendance" _order = "check_in desc" def _default_employee(self): return self.env.user.employee_id employee_id = fields.Many2one('hr.employee', string="Employee", default=_default_employee, required=True, ondelete='cascade', index=True) department_id = fields.Many2one('hr.department', string="Department", related="employee_id.department_id", readonly=True) check_in = fields.Datetime(string="Check In", default=fields.Datetime.now, required=True) check_out = fields.Datetime(string="Check Out") worked_hours = fields.Float(string='Worked Hours', compute='_compute_worked_hours', store=True, readonly=True) def name_get(self): result = [] for attendance in self: if not attendance.check_out: result.append((attendance.id, _("%(empl_name)s from %(check_in)s") % { 'empl_name': attendance.employee_id.name, 'check_in': format_datetime(self.env, attendance.check_in, dt_format=False), })) else: result.append((attendance.id, _("%(empl_name)s from %(check_in)s to %(check_out)s") % { 'empl_name': attendance.employee_id.name, 'check_in': format_datetime(self.env, attendance.check_in, dt_format=False), 'check_out': format_datetime(self.env, attendance.check_out, dt_format=False), })) return result @api.depends('check_in', 'check_out') def _compute_worked_hours(self): for attendance in self: if attendance.check_out and attendance.check_in: delta = attendance.check_out - attendance.check_in attendance.worked_hours = delta.total_seconds() / 3600.0 else: attendance.worked_hours = False @api.constrains('check_in', 'check_out') def _check_validity_check_in_check_out(self): """ verifies if check_in is earlier than check_out. """ for attendance in self: if attendance.check_in and attendance.check_out: if attendance.check_out < attendance.check_in: raise exceptions.ValidationError(_('"Check Out" time cannot be earlier than "Check In" time.')) @api.constrains('check_in', 'check_out', 'employee_id') def _check_validity(self): """ Verifies the validity of the attendance record compared to the others from the same employee. For the same employee we must have : * maximum 1 "open" attendance record (without check_out) * no overlapping time slices with previous employee records """ for attendance in self: # we take the latest attendance before our check_in time and check it doesn't overlap with ours last_attendance_before_check_in = self.env['hr.attendance'].search([ ('employee_id', '=', attendance.employee_id.id), ('check_in', '<=', attendance.check_in), ('id', '!=', attendance.id), ], order='check_in desc', limit=1) if last_attendance_before_check_in and last_attendance_before_check_in.check_out and last_attendance_before_check_in.check_out > attendance.check_in: raise exceptions.ValidationError(_("Cannot create new attendance record for %(empl_name)s, the employee was already checked in on %(datetime)s") % { 'empl_name': attendance.employee_id.name, 'datetime': format_datetime(self.env, attendance.check_in, dt_format=False), }) if not attendance.check_out: # if our attendance is "open" (no check_out), we verify there is no other "open" attendance no_check_out_attendances = self.env['hr.attendance'].search([ ('employee_id', '=', attendance.employee_id.id), ('check_out', '=', False), ('id', '!=', attendance.id), ], order='check_in desc', limit=1) if no_check_out_attendances: raise exceptions.ValidationError(_("Cannot create new attendance record for %(empl_name)s, the employee hasn't checked out since %(datetime)s") % { 'empl_name': attendance.employee_id.name, 'datetime': format_datetime(self.env, no_check_out_attendances.check_in, dt_format=False), }) else: # we verify that the latest attendance with check_in time before our check_out time # is the same as the one before our check_in time computed before, otherwise it overlaps last_attendance_before_check_out = self.env['hr.attendance'].search([ ('employee_id', '=', attendance.employee_id.id), ('check_in', '<', attendance.check_out), ('id', '!=', attendance.id), ], order='check_in desc', limit=1) if last_attendance_before_check_out and last_attendance_before_check_in != last_attendance_before_check_out: raise exceptions.ValidationError(_("Cannot create new attendance record for %(empl_name)s, the employee was already checked in on %(datetime)s") % { 'empl_name': attendance.employee_id.name, 'datetime': format_datetime(self.env, last_attendance_before_check_out.check_in, dt_format=False), }) @api.model def _get_day_start_and_day(self, employee, dt): #Returns a tuple containing the datetime in naive UTC of the employee's start of the day # and the date it was for that employee if not dt.tzinfo: date_employee_tz = pytz.utc.localize(dt).astimezone(pytz.timezone(employee._get_tz())) else: date_employee_tz = dt start_day_employee_tz = date_employee_tz.replace(hour=0, minute=0, second=0) return (start_day_employee_tz.astimezone(pytz.utc).replace(tzinfo=None), start_day_employee_tz.date()) def _get_attendances_dates(self): # Returns a dictionnary {employee_id: set((datetimes, dates))} attendances_emp = defaultdict(set) for attendance in self.filtered(lambda a: a.employee_id.company_id.hr_attendance_overtime and a.check_in): check_in_day_start = attendance._get_day_start_and_day(attendance.employee_id, attendance.check_in) if check_in_day_start[0] < datetime.combine(attendance.employee_id.company_id.overtime_start_date, datetime.min.time()): continue attendances_emp[attendance.employee_id].add(check_in_day_start) if attendance.check_out: check_out_day_start = attendance._get_day_start_and_day(attendance.employee_id, attendance.check_out) attendances_emp[attendance.employee_id].add(check_out_day_start) return attendances_emp def _update_overtime(self, employee_attendance_dates=None): if employee_attendance_dates is None: employee_attendance_dates = self._get_attendances_dates() overtime_to_unlink = self.env['hr.attendance.overtime'] overtime_vals_list = [] for emp, attendance_dates in employee_attendance_dates.items(): # get_attendances_dates returns the date translated from the local timezone without tzinfo, # and contains all the date which we need to check for overtime attendance_domain = [] for attendance_date in attendance_dates: attendance_domain = OR([attendance_domain, [ ('check_in', '>=', attendance_date[0]), ('check_in', '<', attendance_date[0] + timedelta(hours=24)), ]]) attendance_domain = AND([[('employee_id', '=', emp.id)], attendance_domain]) # Attendances per LOCAL day attendances_per_day = defaultdict(lambda: self.env['hr.attendance']) all_attendances = self.env['hr.attendance'].search(attendance_domain) for attendance in all_attendances: check_in_day_start = attendance._get_day_start_and_day(attendance.employee_id, attendance.check_in) attendances_per_day[check_in_day_start[1]] += attendance # As _attendance_intervals_batch and _leave_intervals_batch both take localized dates we need to localize those date start = pytz.utc.localize(min(attendance_dates, key=itemgetter(0))[0]) stop = pytz.utc.localize(max(attendance_dates, key=itemgetter(0))[0] + timedelta(hours=24)) # Retrieve expected attendance intervals expected_attendances = emp.resource_calendar_id._attendance_intervals_batch( start, stop, emp.resource_id )[emp.resource_id.id] # Substract Global Leaves and Employee's Leaves leave_intervals = emp.resource_calendar_id._leave_intervals_batch(start, stop, emp.resource_id, domain=[]) expected_attendances -= leave_intervals[False] | leave_intervals[emp.resource_id.id] # working_times = {date: [(start, stop)]} working_times = defaultdict(lambda: []) for expected_attendance in expected_attendances: # Exclude resource.calendar.attendance working_times[expected_attendance[0].date()].append(expected_attendance[:2]) overtimes = self.env['hr.attendance.overtime'].sudo().search([ ('employee_id', '=', emp.id), ('date', 'in', [day_data[1] for day_data in attendance_dates]), ('adjustment', '=', False), ]) company_threshold = emp.company_id.overtime_company_threshold / 60.0 employee_threshold = emp.company_id.overtime_employee_threshold / 60.0 for day_data in attendance_dates: attendance_date = day_data[1] attendances = attendances_per_day.get(attendance_date, self.browse()) unfinished_shifts = attendances.filtered(lambda a: not a.check_out) overtime_duration = 0 overtime_duration_real = 0 # Overtime is not counted if any shift is not closed or if there are no attendances for that day, # this could happen when deleting attendances. if not unfinished_shifts and attendances: # The employee usually doesn't work on that day if not working_times[attendance_date]: # User does not have any resource_calendar_attendance for that day (week-end for example) overtime_duration = sum(attendances.mapped('worked_hours')) overtime_duration_real = overtime_duration # The employee usually work on that day else: # Compute start and end time for that day planned_start_dt, planned_end_dt = False, False planned_work_duration = 0 for calendar_attendance in working_times[attendance_date]: planned_start_dt = min(planned_start_dt, calendar_attendance[0]) if planned_start_dt else calendar_attendance[0] planned_end_dt = max(planned_end_dt, calendar_attendance[1]) if planned_end_dt else calendar_attendance[1] planned_work_duration += (calendar_attendance[1] - calendar_attendance[0]).total_seconds() / 3600.0 # Count time before, during and after 'working hours' pre_work_time, work_duration, post_work_time = 0, 0, 0 for attendance in attendances: # consider check_in as planned_start_dt if within threshold # if delta_in < 0: Checked in after supposed start of the day # if delta_in > 0: Checked in before supposed start of the day local_check_in = pytz.utc.localize(attendance.check_in) delta_in = (planned_start_dt - local_check_in).total_seconds() / 3600.0 # Started before or after planned date within the threshold interval if (delta_in > 0 and delta_in <= company_threshold) or\ (delta_in < 0 and abs(delta_in) <= employee_threshold): local_check_in = planned_start_dt local_check_out = pytz.utc.localize(attendance.check_out) # same for check_out as planned_end_dt delta_out = (local_check_out - planned_end_dt).total_seconds() / 3600.0 # if delta_out < 0: Checked out before supposed start of the day # if delta_out > 0: Checked out after supposed start of the day # Finised before or after planned date within the threshold interval if (delta_out > 0 and delta_out <= company_threshold) or\ (delta_out < 0 and abs(delta_out) <= employee_threshold): local_check_out = planned_end_dt # There is an overtime at the start of the day if local_check_in < planned_start_dt: pre_work_time += (min(planned_start_dt, local_check_out) - local_check_in).total_seconds() / 3600.0 # Interval inside the working hours -> Considered as working time if local_check_in <= planned_end_dt and local_check_out >= planned_start_dt: work_duration += (min(planned_end_dt, local_check_out) - max(planned_start_dt, local_check_in)).total_seconds() / 3600.0 # There is an overtime at the end of the day if local_check_out > planned_end_dt: post_work_time += (local_check_out - max(planned_end_dt, local_check_in)).total_seconds() / 3600.0 # Overtime within the planned work hours + overtime before/after work hours is > company threshold overtime_duration = work_duration - planned_work_duration if pre_work_time > company_threshold: overtime_duration += pre_work_time if post_work_time > company_threshold: overtime_duration += post_work_time # Global overtime including the thresholds overtime_duration_real = sum(attendances.mapped('worked_hours')) - planned_work_duration overtime = overtimes.filtered(lambda o: o.date == attendance_date) if not float_is_zero(overtime_duration, 2) or unfinished_shifts: # Do not create if any attendance doesn't have a check_out, update if exists if unfinished_shifts: overtime_duration = 0 if not overtime and overtime_duration: overtime_vals_list.append({ 'employee_id': emp.id, 'date': attendance_date, 'duration': overtime_duration, 'duration_real': overtime_duration_real, }) elif overtime: overtime.sudo().write({ 'duration': overtime_duration, 'duration_real': overtime_duration }) elif overtime: overtime_to_unlink |= overtime self.env['hr.attendance.overtime'].sudo().create(overtime_vals_list) overtime_to_unlink.sudo().unlink() @api.model_create_multi def create(self, vals_list): res = super().create(vals_list) res._update_overtime() return res def write(self, vals): attendances_dates = self._get_attendances_dates() result = super(HrAttendance, self).write(vals) if any(field in vals for field in ['employee_id', 'check_in', 'check_out']): # Merge attendance dates before and after write to recompute the # overtime if the attendances have been moved to another day for emp, dates in self._get_attendances_dates().items(): attendances_dates[emp] |= dates self._update_overtime(attendances_dates) return result def unlink(self): attendances_dates = self._get_attendances_dates() super(HrAttendance, self).unlink() self._update_overtime(attendances_dates) @api.returns('self', lambda value: value.id) def copy(self): raise exceptions.UserError(_('You cannot duplicate an attendance.'))