Odoo18-Base/addons/hr_holidays/models/hr_employee.py
2025-01-06 10:57:38 +07:00

347 lines
17 KiB
Python

# Part of Odoo. See LICENSE file for full copyright and licensing details.
from datetime import datetime, time
from collections import defaultdict
from dateutil.relativedelta import relativedelta
import pytz
from odoo import _, api, fields, models
class HrEmployee(models.Model):
_inherit = 'hr.employee'
current_leave_id = fields.Many2one('hr.leave.type', compute='_compute_current_leave', string="Current Time Off Type",
groups="hr.group_hr_user")
def _compute_current_leave(self):
self.current_leave_id = False
holidays = self.env['hr.leave'].sudo().search([
('employee_id', 'in', self.ids),
('date_from', '<=', fields.Datetime.now()),
('date_to', '>=', fields.Datetime.now()),
('state', '=', 'validate'),
])
for holiday in holidays:
employee = self.filtered(lambda e: e.id == holiday.employee_id.id)
employee.current_leave_id = holiday.holiday_status_id.id
def _get_user_m2o_to_empty_on_archived_employees(self):
return super()._get_user_m2o_to_empty_on_archived_employees() + ['leave_manager_id']
def action_time_off_dashboard(self):
return {
'name': _('Time Off Dashboard'),
'type': 'ir.actions.act_window',
'res_model': 'hr.leave',
'views': [[self.env.ref('hr_holidays.hr_leave_employee_view_dashboard').id, 'calendar']],
'domain': [('employee_id', 'in', self.ids)],
'context': {
'employee_id': self.ids,
},
}
def _is_leave_user(self):
return self == self.env.user.employee_id and self.env.user.has_group('hr_holidays.group_hr_holidays_user')
def get_mandatory_days(self, start_date, end_date):
all_days = {}
self = self or self.env.user.employee_id
mandatory_days = self._get_mandatory_days(start_date, end_date)
for mandatory_day in mandatory_days:
num_days = (mandatory_day.end_date - mandatory_day.start_date).days
for d in range(num_days + 1):
all_days[str(mandatory_day.start_date + relativedelta(days=d))] = mandatory_day.color
return all_days
@api.model
def get_special_days_data(self, date_start, date_end):
return {
'mandatoryDays': self.get_mandatory_days_data(date_start, date_end),
'bankHolidays': self.get_public_holidays_data(date_start, date_end),
}
@api.model
def get_public_holidays_data(self, date_start, date_end):
self = self._get_contextual_employee()
employee_tz = pytz.timezone(self._get_tz() if self else self.env.user.tz or 'utc')
public_holidays = self._get_public_holidays(date_start, date_end).sorted('date_from')
return list(map(lambda bh: {
'id': -bh.id,
'colorIndex': 0,
'end': datetime.combine(bh.date_to.astimezone(employee_tz), datetime.max.time()).isoformat(),
'endType': "datetime",
'isAllDay': True,
'start': datetime.combine(bh.date_from.astimezone(employee_tz), datetime.min.time()).isoformat(),
'startType': "datetime",
'title': bh.name,
}, public_holidays))
@api.model
def get_allocation_requests_amount(self):
employee = self._get_contextual_employee()
return self.env['hr.leave.allocation'].search_count([
('employee_id', '=', employee.id),
('state', '=', 'confirm'),
])
def _get_public_holidays(self, date_start, date_end):
domain = [
('resource_id', '=', False),
('company_id', 'in', self.env.companies.ids),
('date_from', '<=', date_end),
('date_to', '>=', date_start),
'|',
('calendar_id', '=', False),
('calendar_id', '=', self.resource_calendar_id.id),
]
return self.env['resource.calendar.leaves'].search(domain)
@api.model
def get_mandatory_days_data(self, date_start, date_end):
self = self._get_contextual_employee()
mandatory_days = self._get_mandatory_days(date_start, date_end).sorted('start_date')
return list(map(lambda sd: {
'id': -sd.id,
'colorIndex': sd.color,
'end': datetime.combine(sd.end_date, datetime.max.time()).isoformat(),
'endType': "datetime",
'isAllDay': True,
'start': datetime.combine(sd.start_date, datetime.min.time()).isoformat(),
'startType': "datetime",
'title': sd.name,
}, mandatory_days))
def _get_mandatory_days(self, start_date, end_date):
domain = [
('start_date', '<=', end_date),
('end_date', '>=', start_date),
('company_id', 'in', self.env.companies.ids),
'|',
('resource_calendar_id', '=', False),
('resource_calendar_id', '=', self.resource_calendar_id.id),
]
if self.department_id:
domain += [
'|',
('department_ids', '=', False),
('department_ids', 'parent_of', self.department_id.id),
]
else:
domain += [('department_ids', '=', False)]
return self.env['hr.leave.mandatory.day'].search(domain)
@api.model
def _get_contextual_employee(self):
ctx = self.env.context
if self.env.context.get('employee_id') is not None:
return self.browse(ctx.get('employee_id'))
if self.env.context.get('default_employee_id') is not None:
return self.browse(ctx.get('default_employee_id'))
return self.env.user.employee_id
def _get_consumed_leaves(self, leave_types, target_date=False, ignore_future=False):
employees = self or self._get_contextual_employee()
leaves_domain = [
('holiday_status_id', 'in', leave_types.ids),
('employee_id', 'in', employees.ids),
('state', 'in', ['confirm', 'validate1', 'validate']),
]
if self.env.context.get('ignored_leave_ids'):
leaves_domain.append(('id', 'not in', self.env.context.get('ignored_leave_ids')))
if not target_date:
target_date = fields.Date.today()
if ignore_future:
leaves_domain.append(('date_from', '<=', target_date))
leaves = self.env['hr.leave'].search(leaves_domain)
leaves_per_employee_type = defaultdict(lambda: defaultdict(lambda: self.env['hr.leave']))
for leave in leaves:
leaves_per_employee_type[leave.employee_id][leave.holiday_status_id] |= leave
allocations = self.env['hr.leave.allocation'].with_context(active_test=False).search([
('employee_id', 'in', employees.ids),
('holiday_status_id', 'in', leave_types.ids),
('state', '=', 'validate'),
])
allocations_per_employee_type = defaultdict(lambda: defaultdict(lambda: self.env['hr.leave.allocation']))
for allocation in allocations:
allocations_per_employee_type[allocation.employee_id][allocation.holiday_status_id] |= allocation
# _get_consumed_leaves returns a tuple of two dictionnaries.
# 1) The first is a dictionary to map the number of days/hours of leaves taken per allocation
# The structure is the following:
# - KEYS:
# allocation_leaves_consumed
# |--employee_id
# |--holiday_status_id
# |--allocation
# |--virtual_leaves_taken
# |--leaves_taken
# |--virtual_remaining_leaves
# |--remaining_leaves
# |--max_leaves
# |--accrual_bonus
# - VALUES:
# Integer representing the number of (virtual) remaining leaves, (virtual) leaves taken or max leaves
# for each allocation.
# leaves_taken and remaining_leaves only take into account validated leaves, while the "virtual" equivalent are
# also based on leaves in "confirm" or "validate1" state.
# Accrual bonus gives the amount of additional leaves that will have been granted at the given
# target_date in comparison to today.
# The unit is in hour or days depending on the leave type request unit
# 2) The second is a dictionary mapping the remaining days per employee and per leave type that are either
# not taken into account by the allocations, mainly because accruals don't take future leaves into account.
# This is used to warn the user if the leaves they takes bring them above their available limit.
# - KEYS:
# allocation_leaves_consumed
# |--employee_id
# |--holiday_status_id
# |--to_recheck_leaves
# |--excess_days
# |--exceeding_duration
# - VALUES:
# "to_recheck_leaves" stores every leave that is not yet taken into account by the "allocation_leaves_consumed" dictionary.
# "excess_days" represents the excess amount that somehow isn't taken into account by the first dictionary.
# "exceeding_duration" sum up the to_recheck_leaves duration and compares it to the maximum allocated for that time period.
allocations_leaves_consumed = defaultdict(lambda: defaultdict(lambda: defaultdict(lambda: defaultdict(lambda: 0))))
to_recheck_leaves_per_leave_type = defaultdict(lambda:
defaultdict(lambda: {
'excess_days': defaultdict(lambda: {
'amount': 0,
'is_virtual': True,
}),
'exceeding_duration': 0,
'to_recheck_leaves': self.env['hr.leave']
})
)
for allocation in allocations:
allocation_data = allocations_leaves_consumed[allocation.employee_id][allocation.holiday_status_id][allocation]
future_leaves = 0
if allocation.allocation_type == 'accrual':
future_leaves = allocation._get_future_leaves_on(target_date)
max_leaves = allocation.number_of_hours_display\
if allocation.type_request_unit in ['hour']\
else allocation.number_of_days_display
max_leaves += future_leaves
allocation_data.update({
'max_leaves': max_leaves,
'accrual_bonus': future_leaves,
'virtual_remaining_leaves': max_leaves,
'remaining_leaves': max_leaves,
'leaves_taken': 0,
'virtual_leaves_taken': 0,
})
for employee in employees:
for leave_type in leave_types:
allocations_with_date_to = self.env['hr.leave.allocation']
allocations_without_date_to = self.env['hr.leave.allocation']
for leave_allocation in allocations_per_employee_type[employee][leave_type]:
if leave_allocation.date_to:
allocations_with_date_to |= leave_allocation
else:
allocations_without_date_to |= leave_allocation
sorted_leave_allocations = allocations_with_date_to.sorted(key='date_to') + allocations_without_date_to
if leave_type.request_unit in ['day', 'half_day']:
leave_duration_field = 'number_of_days'
leave_unit = 'days'
else:
leave_duration_field = 'number_of_hours'
leave_unit = 'hours'
leave_type_data = allocations_leaves_consumed[employee][leave_type]
for leave in leaves_per_employee_type[employee][leave_type].sorted('date_from'):
leave_duration = leave[leave_duration_field]
skip_excess = False
if sorted_leave_allocations.filtered(lambda alloc: alloc.allocation_type == 'accrual') and leave.date_from.date() > target_date:
to_recheck_leaves_per_leave_type[employee][leave_type]['to_recheck_leaves'] |= leave
skip_excess = True
continue
if leave_type.requires_allocation == 'yes':
for allocation in sorted_leave_allocations:
# We don't want to include future leaves linked to accruals into the total count of available leaves.
# However, we'll need to check if those leaves take more than what will be accrued in total of those days
# to give a warning if the total exceeds what will be accrued.
if allocation.date_from > leave.date_to.date() or (allocation.date_to and allocation.date_to < leave.date_from.date()):
continue
interval_start = max(
leave.date_from,
datetime.combine(allocation.date_from, time.min)
)
interval_end = min(
leave.date_to,
datetime.combine(allocation.date_to, time.max)
if allocation.date_to else leave.date_to
)
duration = leave[leave_duration_field]
if leave.date_from != interval_start or leave.date_to != interval_end:
duration_info = employee._get_calendar_attendances(interval_start.replace(tzinfo=pytz.UTC), interval_end.replace(tzinfo=pytz.UTC))
duration = duration_info['hours' if leave_unit == 'hours' else 'days']
max_allowed_duration = min(
duration,
leave_type_data[allocation]['virtual_remaining_leaves']
)
if not max_allowed_duration:
continue
allocated_time = min(max_allowed_duration, leave_duration)
leave_type_data[allocation]['virtual_leaves_taken'] += allocated_time
leave_type_data[allocation]['virtual_remaining_leaves'] -= allocated_time
if leave.state == 'validate':
leave_type_data[allocation]['leaves_taken'] += allocated_time
leave_type_data[allocation]['remaining_leaves'] -= allocated_time
leave_duration -= allocated_time
if not leave_duration:
break
if round(leave_duration, 2) > 0 and not skip_excess:
to_recheck_leaves_per_leave_type[employee][leave_type]['excess_days'][leave.date_to.date()] = {
'amount': leave_duration,
'is_virtual': leave.state != 'validate',
'leave_id': leave.id,
}
else:
if leave_unit == 'hours':
allocated_time = leave.number_of_hours
else:
allocated_time = leave.number_of_days
leave_type_data[False]['virtual_leaves_taken'] += allocated_time
leave_type_data[False]['virtual_remaining_leaves'] = 0
leave_type_data[False]['remaining_leaves'] = 0
if leave.state == 'validate':
leave_type_data[False]['leaves_taken'] += allocated_time
for employee in to_recheck_leaves_per_leave_type:
for leave_type in to_recheck_leaves_per_leave_type[employee]:
content = to_recheck_leaves_per_leave_type[employee][leave_type]
consumed_content = allocations_leaves_consumed[employee][leave_type]
if content['to_recheck_leaves']:
date_to_simulate = max(content['to_recheck_leaves'].mapped('date_from')).date()
latest_accrual_bonus = 0
date_accrual_bonus = 0
virtual_remaining = 0
additional_leaves_duration = 0
for allocation in consumed_content:
latest_accrual_bonus += allocation and allocation._get_future_leaves_on(date_to_simulate)
date_accrual_bonus += consumed_content[allocation]['accrual_bonus']
virtual_remaining += consumed_content[allocation]['virtual_remaining_leaves']
for leave in content['to_recheck_leaves']:
additional_leaves_duration += leave.number_of_hours if leave_type.request_unit == 'hours' else leave.number_of_days
latest_remaining = virtual_remaining - date_accrual_bonus + latest_accrual_bonus
content['exceeding_duration'] = round(min(0, latest_remaining - additional_leaves_duration), 2)
return (allocations_leaves_consumed, to_recheck_leaves_per_leave_type)