229 lines
8.6 KiB
Python
229 lines
8.6 KiB
Python
import logging
|
|
import datetime
|
|
from odoo import api, fields, models
|
|
from datetime import timedelta
|
|
from odoo.fields import Date
|
|
import pytz
|
|
|
|
_logger = logging.getLogger(__name__)
|
|
|
|
|
|
class HrEmployeeStats(models.Model):
|
|
_name = "hr.employee.stats"
|
|
_description = "Employee Stats"
|
|
_order = "date desc"
|
|
_inherit = ["mail.thread", "mail.activity.mixin"]
|
|
|
|
name = fields.Char("Name", compute="_compute_name", store=True)
|
|
dayofweek = fields.Integer("Day of Week", compute="_compute_dayofweek")
|
|
is_public_holiday = fields.Boolean("Public Holiday", compute="_compute_dayofweek")
|
|
employee_id = fields.Many2one("hr.employee", "Employee", required=True)
|
|
department_id = fields.Many2one("hr.department", "Department")
|
|
timesheet_line_ids = fields.One2many(
|
|
"account.analytic.line",
|
|
"employee_id",
|
|
"Timesheet lines",
|
|
compute="_compute_timesheet_line_ids",
|
|
)
|
|
date = fields.Date("Date", required=True)
|
|
company_id = fields.Many2one(
|
|
"res.company",
|
|
"Company",
|
|
default=lambda self: self.env.company,
|
|
required=True,
|
|
)
|
|
sheet_id = fields.Many2one("hr_timesheet.sheet", "Timesheet")
|
|
total_hours = fields.Float("Total Hours", compute="_compute_hours")
|
|
total_planned_hours = fields.Float("Total Planning Hours", compute="_compute_hours")
|
|
total_leave_hours = fields.Float("Total Leave Hours", compute="_compute_hours")
|
|
total_recovery_hours = fields.Float(
|
|
"Total Recovery Hours", compute="_compute_hours"
|
|
)
|
|
gap_hours = fields.Float("Gap Hours", compute="_compute_hours")
|
|
|
|
def _get_holiday_status_id(self):
|
|
recovery_type_id = self.env.company.recovery_type_id
|
|
if recovery_type_id:
|
|
return recovery_type_id.id
|
|
else:
|
|
return False
|
|
|
|
def _compute_timesheet_line_ids(self):
|
|
for stat in self:
|
|
stat.timesheet_line_ids = self.env["account.analytic.line"].search(
|
|
[
|
|
("employee_id", "=", stat.employee_id.id),
|
|
("date", "=", stat.date),
|
|
]
|
|
)
|
|
|
|
def _get_intersects(
|
|
self, datetime1_start, datetime1_end, datetime2_start, datetime2_end
|
|
):
|
|
latest_start = max(datetime1_start, datetime2_start)
|
|
earliest_end = min(datetime1_end, datetime2_end)
|
|
delta = (earliest_end - latest_start).total_seconds() / 3600
|
|
return max(0, delta)
|
|
|
|
def get_total_hours_domain(self):
|
|
return [
|
|
("employee_id", "=", self.employee_id.id),
|
|
("date", "=", self.date),
|
|
]
|
|
|
|
@api.depends("timesheet_line_ids")
|
|
def _get_total_hours(self):
|
|
self.ensure_one()
|
|
total_hours = 0
|
|
timesheet_line = self.env["account.analytic.line"]
|
|
if self.date and self.employee_id:
|
|
timesheet_line_ids = timesheet_line.search(
|
|
self.get_total_hours_domain()
|
|
)
|
|
total_hours = sum(timesheet_line_ids.mapped("unit_amount"))
|
|
return total_hours
|
|
|
|
def _get_total_planned_hours(self):
|
|
self.ensure_one()
|
|
total_planned_hours = 0
|
|
if self.employee_id and self.date and not self.is_public_holiday:
|
|
dayofweek = int(self.date.strftime("%u")) - 1
|
|
calendar_id = self.employee_id.resource_calendar_id
|
|
week_number = self.date.isocalendar()[1] % 2
|
|
if calendar_id.two_weeks_calendar:
|
|
hours = calendar_id.attendance_ids.search(
|
|
[
|
|
("dayofweek", "=", dayofweek),
|
|
("calendar_id", "=", calendar_id.id),
|
|
("week_type", "=", week_number),
|
|
]
|
|
)
|
|
else:
|
|
hours = calendar_id.attendance_ids.search(
|
|
[
|
|
("dayofweek", "=", dayofweek),
|
|
("calendar_id", "=", calendar_id.id),
|
|
]
|
|
)
|
|
total_planned_hours = sum(
|
|
hours.mapped(lambda r: r.hour_to - r.hour_from)
|
|
)
|
|
return total_planned_hours
|
|
|
|
def _get_total_recovery_hours(self):
|
|
self.ensure_one()
|
|
recovery = self.env["hr.leave"]
|
|
total_recovery_hours = 0
|
|
if self.date and self.employee_id and self._get_holiday_status_id():
|
|
recovery_ids = recovery.search(
|
|
[
|
|
("employee_id", "=", self.employee_id.id),
|
|
("holiday_status_id", "=", self._get_holiday_status_id()),
|
|
("request_date_from", "<=", self.date),
|
|
("request_date_to", ">=", self.date),
|
|
]
|
|
)
|
|
total_recovery_hours = sum(
|
|
recovery_ids.mapped("number_of_hours_display")
|
|
)
|
|
return total_recovery_hours
|
|
|
|
def _get_total_leave_hours(self):
|
|
self.ensure_one()
|
|
leave = self.env["hr.leave"]
|
|
total_leave_hours = 0
|
|
if self.date and self.employee_id:
|
|
leave_id = leave.search(
|
|
[
|
|
("employee_id", "=", self.employee_id.id),
|
|
("holiday_status_id", "!=", self._get_holiday_status_id()),
|
|
("request_date_from", "<=", self.date),
|
|
("request_date_to", ">=", self.date),
|
|
],
|
|
limit=1
|
|
)
|
|
if leave_id:
|
|
if leave_id.request_unit_hours:
|
|
total_leave_hours = leave_id.number_of_hours_display
|
|
elif leave_id.request_unit_half:
|
|
total_leave_hours = self._get_total_planned_hours() / 2
|
|
else:
|
|
total_leave_hours = self._get_total_planned_hours()
|
|
return total_leave_hours
|
|
|
|
@api.depends("employee_id", "date")
|
|
def _compute_name(self):
|
|
for stat in self:
|
|
stat.name = "%s - %s" % (stat.employee_id.name, stat.date)
|
|
|
|
@api.depends("date","employee_id")
|
|
def _compute_dayofweek(self):
|
|
for stat in self:
|
|
if not stat.date:
|
|
stat.dayofweek = None
|
|
stat.is_public_holiday = False
|
|
continue
|
|
stat.dayofweek = int(stat.date.strftime("%u")) - 1
|
|
stat.is_public_holiday = stat._is_public_holiday_accordig_to_employe_tz()
|
|
|
|
def _convert_to_employee_tz(self, date):
|
|
"""Convertit un datetime UTC en datetime dans le fuseau de l'employé."""
|
|
self.ensure_one()
|
|
if not date:
|
|
return None
|
|
employee_tz = pytz.timezone(self.sheet_id.employee_id.tz or "UTC")
|
|
if date.tzinfo is None:
|
|
dt = pytz.utc.localize(date)
|
|
return dt.astimezone(employee_tz) - timedelta(minutes=1)
|
|
|
|
def _is_public_holiday_accordig_to_employe_tz(self):
|
|
self.ensure_one()
|
|
if not self.date or not self.sheet_id or not self.sheet_id.employee_id:
|
|
return False
|
|
#get public holidays for the employee
|
|
public_holidays = self.sheet_id.employee_id._get_public_holidays(
|
|
self.date, self.date
|
|
)
|
|
if not public_holidays:
|
|
return False
|
|
ph = public_holidays[0]
|
|
# Convert public holiday to the employee timezone
|
|
ph_datetime_from_tz = self._convert_to_employee_tz(ph.date_from)
|
|
ph_datetime_to_tz = self._convert_to_employee_tz(ph.date_to)
|
|
# Convert datetime to date
|
|
ph_date_from = ph_datetime_from_tz.date()
|
|
ph_date_to = ph_datetime_to_tz.date()
|
|
# Check if the stat date falls within the public holiday range after conversion in employee tz
|
|
if ph_date_from <= self.date <= ph_date_to:
|
|
return True
|
|
else:
|
|
return False
|
|
|
|
def _get_gap_hours(
|
|
self, total_hours, total_recovery_hours, total_leave_hours, total_planned_hours
|
|
):
|
|
self.ensure_one()
|
|
balance = (
|
|
total_hours + total_recovery_hours + total_leave_hours - total_planned_hours
|
|
)
|
|
return balance
|
|
|
|
@api.depends(
|
|
"employee_id",
|
|
"date",
|
|
"total_hours",
|
|
"total_planned_hours",
|
|
"timesheet_line_ids",
|
|
)
|
|
def _compute_hours(self):
|
|
for stat in self:
|
|
total_hours = stat._get_total_hours()
|
|
total_planned_hours = stat._get_total_planned_hours()
|
|
total_recovery_hours = stat._get_total_recovery_hours()
|
|
total_leave_hours = stat._get_total_leave_hours()
|
|
stat.total_hours = total_hours
|
|
stat.total_planned_hours = total_planned_hours
|
|
stat.gap_hours = stat._get_gap_hours(total_hours, total_recovery_hours, total_leave_hours, total_planned_hours)
|
|
stat.total_recovery_hours = total_recovery_hours
|
|
stat.total_leave_hours = total_leave_hours
|