Elabore/hr-tools
10
0
Fork 0
Code Issues Pull Requests 3 Actions Packages Releases Activity

[WIP]hr_holidays_timeoff_analysis
Some checks failed
pre-commit / pre-commit (pull_request) Has been cancelled
Details

Browse Source
This commit is contained in:
Laetitia Da Costa
2026-07-31 12:46:29 +02:00
parent e5d56d9d1a
commit bf27d0938c
3 changed files with 216 additions and 56 deletions
Download Patch File Download Diff File Expand all files Collapse all files

161
hr_holidays_timeoff_analysis/models/hr_leave_timeoff_day.py
View File

@@ -137,12 +137,18 @@ class TimeOffDay(models.Model):
@api.model
def cron_manage_timeoff_days(self):
self.cron_create_timeoff_days()
self.cron_delete_timeoff_days()
"""Reconcile hr.leave.timeoff.day records day by day.
def cron_create_timeoff_days(self):
# Browse all validated leaves
leaves = self.env["hr.leave"].search(
On each day within the relevant date range, compute which timeoff.days
*should* exist (validated leaves covering the day, employee scheduled
to work, not a public holiday) and reconcile against existing records:
- Stale records (leave dates changed, state changed, force-cancelled,
orphan, calendar/holiday change) are deleted.
- Missing records are created.
- Existing records with an outdated duration are updated in place.
"""
# Load all currently validated leaves that have dates and an employee.
valid_leaves = self.env["hr.leave"].search(
[
("state", "=", "validate"),
("request_date_from", "!=", False),
@@ -150,59 +156,104 @@ class TimeOffDay(models.Model):
("employee_id", "!=", False),
]
)
for leave in leaves:
current_date = leave.request_date_from
employee = leave.employee_id
while current_date <= leave.request_date_to:
# If no validated leaves exist, all existing timeoff.days are stale/orphan.
if not valid_leaves:
self.search([]).unlink()
return
# Determine the full date span to iterate over: from the earliest
# relevant date (oldest leave or oldest existing timeoff.day) to the
# latest relevant date (newest leave or newest existing timeoff.day).
#first, get the oldest and latest leave dates
leave_from_dates = valid_leaves.mapped("request_date_from")
leave_to_dates = valid_leaves.mapped("request_date_to")
start_date = min(leave_from_dates)
end_date = max(leave_to_dates)
#then, get oldest timeoff.day and latest timeoff.day
#and keep as start_date and end_date the min and max of both
oldest_td = self.search([], order="date asc", limit=1)
if oldest_td and oldest_td.date and oldest_td.date < start_date:
start_date = oldest_td.date
latest_td = self.search([], order="date desc", limit=1)
if latest_td and latest_td.date and latest_td.date > end_date:
end_date = latest_td.date
existing_tds = self.search(
[("date", ">=", start_date), ("date", "<=", end_date)]
)
# Build in-memory indexes for fast lookup per day.
# leaves_by_date: date -> list of validated leaves covering that day
leaves_by_date = {}
for leave in valid_leaves:
d = leave.request_date_from
while d <= leave.request_date_to:
if d not in leaves_by_date:
leaves_by_date[d] = []
leaves_by_date[d].append(leave)
d += timedelta(days=1)
# existing_by_date: date -> list of existing timeoff.day records
existing_by_date = {}
for td in existing_tds:
if td.date not in existing_by_date:
existing_by_date[td.date] = []
existing_by_date[td.date].append(td)
# Collect operations and apply them at the end to minimise DB round trips.
all_to_create = []
all_to_delete_ids = []
current_date = start_date
while current_date <= end_date:
covering_leaves = leaves_by_date.get(current_date, [])
# Compute the expected timeoff.days for this date.
# Key: (employee_id, leave_id) -> expected leave_duration_by_day
expected = {}
for leave in covering_leaves:
employee = leave.employee_id
if self.employee_is_scheduled_to_work_this_day(
current_date, employee, leave
) and not self._is_public_holiday_according_to_employe_tz(
current_date, employee
):
# The employee is scheduled to work this day
# according his calendar and it's not a
# public holiday,
# so create a timeoff day record if it does not already exist
if not self.search(
[
("date", "=", current_date),
("employee_id", "=", employee.id),
("hr_leave_id", "=", leave.id),
],
limit=1,
):
self.create(
{
"date": current_date,
"employee_id": employee.id,
"hr_leave_id": leave.id,
"leave_duration_by_day": (
self.compute_leave_duration_by_day(leave)
),
}
)
current_date += timedelta(days=1)
expected[(employee.id, leave.id)] = (
self.compute_leave_duration_by_day(leave)
)
def cron_delete_timeoff_days(self):
# Browse all unvalidated or canceled leaves
leaves = self.env["hr.leave"].search(
[
("state", "!=", "validate"),
("request_date_from", "!=", False),
("request_date_to", "!=", False),
("employee_id", "!=", False),
]
)
# Delete timeoff days for leaves that are no longer validated
for leave in leaves:
self.search(
[
("hr_leave_id", "=", leave.id),
]
).unlink()
# Delete timeoff days that are not linked to any leave
self.search(
[
("hr_leave_id", "=", False),
]
).unlink()
# Reconcile existing timeoff.days for this date.
existing_today = existing_by_date.get(current_date, [])
for td in existing_today:
key = (td.employee_id.id, td.hr_leave_id.id)
if key in expected:
# Matched: keep the record, update duration if it changed.
expected_duration = expected.pop(key)
if td.leave_duration_by_day != expected_duration:
td.leave_duration_by_day = expected_duration
else:
# No longer valid: leave dates changed, state changed,
# force-cancelled, orphan (hr_leave_id=False), calendar
# changed, or the day became a public holiday.
all_to_delete_ids.append(td.id)
# Remaining entries in expected are missing timeoff.days -> create.
for (employee_id, leave_id), duration in expected.items():
all_to_create.append(
{
"date": current_date,
"employee_id": employee_id,
"hr_leave_id": leave_id,
"leave_duration_by_day": duration,
}
)
current_date += timedelta(days=1)
# Apply batched operations.
if all_to_delete_ids:
self.browse(all_to_delete_ids).unlink()
if all_to_create:
self.create(all_to_create)