This commit is contained in:
Hubert Van De Walle
2026-08-28 13:05:55 +02:00
parent ee6cbe0c39
commit 12440a11f9
2 changed files with 86 additions and 67 deletions
+13 -4
View File
@@ -28,6 +28,7 @@ from .schedule import (
REASON_PUBLIC_HOLIDAY, REASON_PUBLIC_HOLIDAY,
REASON_WEEKLY_OFF, REASON_WEEKLY_OFF,
REASON_WORKING, REASON_WORKING,
TAG_RE,
DaySchedule, DaySchedule,
ResolvedEmployee, ResolvedEmployee,
) )
@@ -51,10 +52,18 @@ HALF_BLOCK = "▀" # ▀ -- foreground = top (morning), background = bottom (af
DAY_COL_WIDTH = 2 DAY_COL_WIDTH = 2
SEP_COL_WIDTH = 1 SEP_COL_WIDTH = 1
EMPLOYEE_COL_MIN_WIDTH = 14 EMPLOYEE_COL_MIN_WIDTH = len("Employee") # the header word itself is the floor, tags are usually shorter
EMPLOYEE_COL_MAX_WIDTH = 30 EMPLOYEE_COL_MAX_WIDTH = 30
def _display_label(name: str) -> str:
"""The parenthesized tag suffix (e.g. "huvw" from "Van de Walle Hubert (huvw)"),
or the full name if it doesn't have one.
"""
m = TAG_RE.search(name)
return m.group(1) if m else name
def _day_cell(day: DaySchedule) -> Text: def _day_cell(day: DaySchedule) -> Text:
top = STATUS_COLOR[day.morning_reason] top = STATUS_COLOR[day.morning_reason]
bottom = STATUS_COLOR[day.afternoon_reason] bottom = STATUS_COLOR[day.afternoon_reason]
@@ -196,7 +205,7 @@ def render_gantt(
return return
employee_col_width = max( employee_col_width = max(
EMPLOYEE_COL_MIN_WIDTH, min(EMPLOYEE_COL_MAX_WIDTH, max(len(e.name) for e in employees)) EMPLOYEE_COL_MIN_WIDTH, min(EMPLOYEE_COL_MAX_WIDTH, max(len(_display_label(e.name)) for e in employees))
) )
ranges = _paginate(console, reference_days, employee_col_width) ranges = _paginate(console, reference_days, employee_col_width)
@@ -207,7 +216,7 @@ def render_gantt(
slot_pattern = _with_week_separators(page_days) slot_pattern = _with_week_separators(page_days)
table = _build_page_table(slot_pattern, employee_col_width) table = _build_page_table(slot_pattern, employee_col_width)
for emp in employees: for emp in employees:
row = [emp.name] + _row_cells(slot_pattern, visible_by_emp[emp.id][start:end]) row = [_display_label(emp.name)] + _row_cells(slot_pattern, visible_by_emp[emp.id][start:end])
table.add_row(*row) table.add_row(*row)
console.print(table) console.print(table)
if page_num < len(ranges) - 1: if page_num < len(ranges) - 1:
@@ -249,7 +258,7 @@ def _print_footnotes(
emp_lines.append(f" {d.day}: {note}") emp_lines.append(f" {d.day}: {note}")
if emp_lines: if emp_lines:
any_notes = True any_notes = True
lines.append(f"{emp.name}:") lines.append(f"{_display_label(emp.name)}:")
lines.extend(emp_lines) lines.extend(emp_lines)
if any_notes: if any_notes:
+73 -63
View File
@@ -22,8 +22,17 @@ them separate:
calculation explicitly filters those out (see calculation explicitly filters those out (see
`hr_holidays_gantt/models/hr_leave.py::_gantt_unavailability`, which `hr_holidays_gantt/models/hr_leave.py::_gantt_unavailability`, which
restricts the leave domain to `resource_id = False`, i.e. company-wide restricts the leave domain to `resource_id = False`, i.e. company-wide
holidays only). We fetch these separately (again, one batched call for holidays only).
every employee) and merge them in ourselves.
We get these from the SAME `get_gantt_data` call's `records`, rather
than a separate `hr.leave.search_read` -- `hr.leave` itself is normally
row-restricted by an `ir.rule` to "your own leaves + leaves of people
you manage", and Odoo record rules filter silently rather than raising,
so a plain search_read for a colleague outside that scope just comes
back empty with no error to catch. `hr.leave.report.calendar` is a
SQL-view-backed model (`_auto = False`) without that per-employee row
restriction, so its rows show the same colleague leave bars the Gantt
view itself already displays to any employee.
Public holidays are company/date-range-wide, not employee-specific, so Public holidays are company/date-range-wide, not employee-specific, so
that lookup is naturally already a single shared call. that lookup is naturally already a single shared call.
@@ -157,33 +166,6 @@ def _to_local(naive_utc_str: str, tz: ZoneInfo) -> datetime:
return datetime.strptime(naive_utc_str, "%Y-%m-%d %H:%M:%S").replace(tzinfo=UTC).astimezone(tz) return datetime.strptime(naive_utc_str, "%Y-%m-%d %H:%M:%S").replace(tzinfo=UTC).astimezone(tz)
def _fetch_pattern_unavailability_batch(
client: OdooClient, employee_ids: list[int], date_from: date, date_to: date, tz: ZoneInfo
) -> dict[int, list[tuple[datetime, datetime]]]:
"""Weekly working-hours pattern + company-wide holidays, for every employee in one call."""
start_utc = _local_midnight_to_utc_naive(date_from, tz)
stop_utc = _local_midnight_to_utc_naive(date_to + timedelta(days=1), tz)
result = client.call_kw(
"hr.leave.report.calendar",
"get_gantt_data",
[[("employee_id", "in", employee_ids)], ["employee_id"], {"id": {}}],
{
"unavailability_fields": ["employee_id"],
"start_date": start_utc.strftime("%Y-%m-%d %H:%M:%S"),
"stop_date": stop_utc.strftime("%Y-%m-%d %H:%M:%S"),
"scale": "week",
},
)
raw = result["unavailabilities"].get("employee_id", {})
out = {}
for emp_id in employee_ids:
items = raw.get(str(emp_id), [])
out[emp_id] = [(_to_local(item["start"], tz), _to_local(item["stop"], tz)) for item in items]
return out
@dataclass @dataclass
class LeavePeriod: class LeavePeriod:
start: datetime start: datetime
@@ -192,35 +174,74 @@ class LeavePeriod:
validated: bool validated: bool
def _fetch_employee_leaves_batch( def _fetch_gantt_data_batch(
client: OdooClient, employee_ids: list[int], date_from: date, date_to: date, tz: ZoneInfo client: OdooClient, employee_ids: list[int], date_from: date, date_to: date, tz: ZoneInfo
) -> dict[int, list[LeavePeriod]]: ) -> tuple[dict[int, list[tuple[datetime, datetime]]], dict[int, list[LeavePeriod]]]:
"""Every employee's own leave requests overlapping the range, in one call.""" """Weekly working-hours pattern + company-wide holidays (from `unavailabilities`) AND
rows = client.call_kw( each employee's own leave requests (from `records`), both from one call. See the module
"hr.leave", docstring for why leave data comes from here rather than a plain `hr.leave.search_read`.
"search_read", """
start_utc = _local_midnight_to_utc_naive(date_from, tz)
stop_utc = _local_midnight_to_utc_naive(date_to + timedelta(days=1), tz)
result = client.call_kw(
"hr.leave.report.calendar",
"get_gantt_data",
[ [
[ [("employee_id", "in", employee_ids)],
("employee_id", "in", employee_ids), ["employee_id"],
("state", "in", ["confirm", "validate", "validate1"]), {
("date_from", "<=", f"{date_to} 23:59:59"), "id": {},
("date_to", ">=", f"{date_from} 00:00:00"), "employee_id": {},
], "start_datetime": {},
["employee_id", "date_from", "date_to", "work_entry_type_id", "state"], "stop_datetime": {},
"state": {},
},
], ],
{
"unavailability_fields": ["employee_id"],
"start_date": start_utc.strftime("%Y-%m-%d %H:%M:%S"),
"stop_date": stop_utc.strftime("%Y-%m-%d %H:%M:%S"),
"scale": "week",
},
) )
out: dict[int, list[LeavePeriod]] = defaultdict(list)
for r in rows: raw_unavail = result["unavailabilities"].get("employee_id", {})
type_name = r["work_entry_type_id"][1] if r["work_entry_type_id"] else "Time Off" pattern_by_employee = {}
out[r["employee_id"][0]].append( for emp_id in employee_ids:
items = raw_unavail.get(str(emp_id), [])
pattern_by_employee[emp_id] = [(_to_local(item["start"], tz), _to_local(item["stop"], tz)) for item in items]
# work_entry_type_id (and description, leave_id) require
# hr_holidays.group_hr_holidays_user -- unlike a plain read(), web_read
# (which get_gantt_data uses internally) raises AccessError if you
# request a field you don't have group access to, rather than silently
# omitting it, so those aren't in the read_specification above and
# every leave is labeled generically.
leaves_by_employee: dict[int, list[LeavePeriod]] = defaultdict(list)
for rec in result.get("records", []):
employee = rec.get("employee_id")
start = rec.get("start_datetime")
stop = rec.get("stop_datetime")
if not employee or not start or not stop:
continue
state = rec.get("state")
if state == "refuse":
continue
# employee_id normally comes back as [id, name], but resolving the
# name requires read access to hr.employee itself -- without that,
# web_read degrades it to a bare id instead of raising.
employee_id = employee[0] if isinstance(employee, (list, tuple)) else employee
leaves_by_employee[employee_id].append(
LeavePeriod( LeavePeriod(
start=_to_local(r["date_from"], tz), start=_to_local(start, tz),
stop=_to_local(r["date_to"], tz), stop=_to_local(stop, tz),
label=type_name, label="Time Off",
validated=r["state"] == "validate", validated=state == "validate",
) )
) )
return out
return pattern_by_employee, dict(leaves_by_employee)
def _fetch_public_holidays( def _fetch_public_holidays(
@@ -283,7 +304,6 @@ def _build_days(
pattern_intervals: list[tuple[datetime, datetime]], pattern_intervals: list[tuple[datetime, datetime]],
leave_periods: list[LeavePeriod], leave_periods: list[LeavePeriod],
holidays: list[tuple[datetime, datetime, str]], holidays: list[tuple[datetime, datetime, str]],
leave_fetch_failed: bool,
explain: bool, explain: bool,
) -> list[DaySchedule]: ) -> list[DaySchedule]:
validated_leave_intervals = [(p.start, p.stop) for p in leave_periods if p.validated] validated_leave_intervals = [(p.start, p.stop) for p in leave_periods if p.validated]
@@ -319,8 +339,6 @@ def _build_days(
notes = [] notes = []
if explain: if explain:
if leave_fetch_failed:
notes.append("could not fetch this employee's leave requests (access denied) -- personal leave not reflected")
def _describe(reason: str, label: str | None) -> str: def _describe(reason: str, label: str | None) -> str:
if reason == REASON_LEAVE: if reason == REASON_LEAVE:
@@ -361,14 +379,7 @@ def get_schedules(
tz = ZoneInfo(tz_name) tz = ZoneInfo(tz_name)
employee_ids = [e.id for e in employees] employee_ids = [e.id for e in employees]
pattern_by_employee = _fetch_pattern_unavailability_batch(client, employee_ids, date_from, date_to, tz) pattern_by_employee, leaves_by_employee = _fetch_gantt_data_batch(client, employee_ids, date_from, date_to, tz)
leaves_by_employee: dict[int, list[LeavePeriod]] = {}
leave_fetch_failed = False
try:
leaves_by_employee = _fetch_employee_leaves_batch(client, employee_ids, date_from, date_to, tz)
except OdooError:
leave_fetch_failed = True
holidays: list[tuple[datetime, datetime, str]] = [] holidays: list[tuple[datetime, datetime, str]] = []
if explain: if explain:
@@ -385,7 +396,6 @@ def get_schedules(
pattern_by_employee.get(emp.id, []), pattern_by_employee.get(emp.id, []),
leaves_by_employee.get(emp.id, []), leaves_by_employee.get(emp.id, []),
holidays, holidays,
leave_fetch_failed,
explain, explain,
) )
for emp in employees for emp in employees