Compare commits

..

3 Commits

Author SHA1 Message Date
Hubert Van De Walle 35337afdf0 Fix bugs + use actions auth session get/set for sessions 2026-09-16 15:43:14 +02:00
Hubert Van De Walle 12440a11f9 fixes 2026-08-28 13:05:55 +02:00
Hubert Van De Walle ee6cbe0c39 Auto start date 2026-08-28 12:23:00 +02:00
4 changed files with 217 additions and 134 deletions
+21 -9
View File
@@ -13,6 +13,8 @@ from .client import OdooClient, OdooError
from .render import render_gantt
from .schedule import DAY_NAMES, get_schedules, resolve_employees
BASE_URL = "https://www.odoo.com"
def parse_date(value: str) -> date:
return date.fromisoformat(value)
@@ -66,14 +68,19 @@ def main() -> int:
"server's employee names, e.g. 'mibr' for 'Michel Bram (mibr)'), space-separated"
),
)
parser.add_argument("date_from", type=parse_date, help="Start date, YYYY-MM-DD")
parser.add_argument("date_to", type=parse_date, help="End date, YYYY-MM-DD (inclusive)")
parser.add_argument("--url", default="http://localhost:1930", help="Odoo base URL")
parser.add_argument(
"--session-file",
default=Path(__file__).resolve().parents[2] / "session_id.txt",
type=Path,
help="Path to a file containing the session_id cookie value",
"--date-from",
dest="date_from",
type=parse_date,
default=None,
help="Start date, YYYY-MM-DD (default: today)",
)
parser.add_argument(
"--date-to",
dest="date_to",
type=parse_date,
default=None,
help="End date, YYYY-MM-DD, inclusive (default: same as --date-from)",
)
parser.add_argument(
"--tz",
@@ -109,12 +116,17 @@ def main() -> int:
)
args = parser.parse_args()
if args.date_from is None:
args.date_from = date.today()
if args.date_to is None:
args.date_to = args.date_from
if args.date_from > args.date_to:
parser.error("date_from must be before or equal to date_to")
try:
client = OdooClient.from_session_file(args.url, args.session_file, debug=args.debug)
except (FileNotFoundError, ValueError) as exc:
client = OdooClient.from_actions(BASE_URL, debug=args.debug)
except ValueError as exc:
print(f"error: {exc}", file=sys.stderr)
return 1
+38 -7
View File
@@ -1,10 +1,17 @@
"""Minimal JSON-RPC client for Odoo's /web/dataset/call_kw route, authenticated via a session_id cookie."""
"""Minimal JSON-RPC client for Odoo's /web/dataset/call_kw route.
The session_id cookie is managed by the `actions` CLI's stored client profiles
(`actions auth session get/set <profile>`) rather than a local file: we read the
initial cookie from there, and if the server ever rotates it (a new Set-Cookie for
session_id on any response), we persist the new value back to the same profile so
the next run picks it up.
"""
from __future__ import annotations
import json
import subprocess
import sys
from pathlib import Path
import requests
@@ -17,20 +24,31 @@ class OdooError(RuntimeError):
self.data = data or {}
def _run_actions(*args: str) -> str:
try:
result = subprocess.run(["actions", *args], capture_output=True, text=True)
except FileNotFoundError as exc:
raise ValueError("the `actions` command isn't available on PATH") from exc
if result.returncode != 0:
raise ValueError((result.stderr or result.stdout).strip() or f"`actions {' '.join(args)}` failed")
return result.stdout.strip()
class OdooClient:
def __init__(self, base_url: str, session_id: str, debug: bool = False):
def __init__(self, base_url: str, session_id: str, profile: str, debug: bool = False):
self.base_url = base_url.rstrip("/")
self.session = requests.Session()
self.session.cookies.set("session_id", session_id)
self.profile = profile
self.debug = debug
self._request_count = 0
@classmethod
def from_session_file(cls, base_url: str, session_file: Path, debug: bool = False) -> "OdooClient":
session_id = session_file.read_text().strip()
def from_actions(cls, base_url: str, profile: str = "odoo", debug: bool = False) -> "OdooClient":
session_id = _run_actions("auth", "session", "get", profile)
if not session_id:
raise ValueError(f"{session_file} is empty; put a session_id cookie value in it")
return cls(base_url, session_id, debug=debug)
raise ValueError(f"`actions auth session get {profile}` returned an empty session_id")
return cls(base_url, session_id, profile, debug=debug)
def _jsonrpc(self, endpoint: str, params: dict) -> dict:
self._request_count += 1
@@ -40,6 +58,7 @@ class OdooClient:
+ json.dumps(params, default=str),
file=sys.stderr,
)
previous_session_id = self.session.cookies.get("session_id")
resp = self.session.post(
f"{self.base_url}{endpoint}",
json={"jsonrpc": "2.0", "method": "call", "params": params},
@@ -47,6 +66,7 @@ class OdooClient:
timeout=30,
)
resp.raise_for_status()
self._persist_rotated_session_id(previous_session_id)
payload = resp.json()
if "error" in payload:
error = payload["error"]
@@ -54,6 +74,17 @@ class OdooClient:
raise OdooError(message, error.get("data"))
return payload.get("result")
def _persist_rotated_session_id(self, previous_session_id: str | None) -> None:
new_session_id = self.session.cookies.get("session_id")
if not new_session_id or new_session_id == previous_session_id:
return
if self.debug:
print(f"[debug] session_id rotated, persisting to actions profile {self.profile!r}", file=sys.stderr)
try:
_run_actions("auth", "session", "set", self.profile, new_session_id)
except ValueError as exc:
print(f"warning: failed to persist rotated session_id: {exc}", file=sys.stderr)
def call_kw(self, model: str, method: str, args: list, kwargs: dict | None = None):
return self._jsonrpc(
"/web/dataset/call_kw",
+13 -4
View File
@@ -28,6 +28,7 @@ from .schedule import (
REASON_PUBLIC_HOLIDAY,
REASON_WEEKLY_OFF,
REASON_WORKING,
TAG_RE,
DaySchedule,
ResolvedEmployee,
)
@@ -51,10 +52,18 @@ HALF_BLOCK = "▀" # ▀ -- foreground = top (morning), background = bottom (af
DAY_COL_WIDTH = 2
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
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:
top = STATUS_COLOR[day.morning_reason]
bottom = STATUS_COLOR[day.afternoon_reason]
@@ -196,7 +205,7 @@ def render_gantt(
return
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)
@@ -207,7 +216,7 @@ def render_gantt(
slot_pattern = _with_week_separators(page_days)
table = _build_page_table(slot_pattern, employee_col_width)
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)
console.print(table)
if page_num < len(ranges) - 1:
@@ -249,7 +258,7 @@ def _print_footnotes(
emp_lines.append(f" {d.day}: {note}")
if emp_lines:
any_notes = True
lines.append(f"{emp.name}:")
lines.append(f"{_display_label(emp.name)}:")
lines.extend(emp_lines)
if any_notes:
+145 -114
View File
@@ -1,36 +1,33 @@
"""Compute several employees' per-day working schedules over a date range, in a
constant, small number of requests regardless of how many employees are asked for.
Two things have to be combined to get a correct answer, because Odoo keeps
them separate:
The weekly working-hours pattern is read directly from `resource.calendar.attendance`
(the raw dayofweek/day_period/week_type rows), not from the Gantt view's
`get_gantt_data`/`_gantt_unavailability`. That method branches on whether an employee
has an `hr.version` with a `contract_date_start` set: with one, it correctly computes
the pattern via `resource.calendar._unavailable_intervals_batch`; without one (e.g.
independent/freelance employees who are tracked as `hr.employee` but have no formal
contract), it falls back to `resource.resource._get_unavailable_intervals`, which for
such employees reports every day as unavailable. Reading the attendance rows ourselves
sidesteps that branch entirely and works the same way regardless of contract status.
1. The "unavailability" data behind the Gantt view
(`hr.leave.report.calendar.get_gantt_data`, called with a domain/groupby
that is *only* `employee_id` -- that specific shape triggers a
`hr_holidays_gantt` code path, `_get_gantt_data_groupby_employee`, that
fills in non-working time from the employee's working-hours calendar
even when they have zero `hr.leave` records in range). This gives the
weekly working-hours pattern (e.g. "doesn't work Fridays") merged with
company-wide/public holidays. The domain accepts `employee_id in [...]`
just as well as `employee_id = X`, and the result is keyed per employee,
so one call covers the whole list of employees.
2. The employees' own *validated* `hr.leave` records. These are
deliberately NOT included in (1): in the Gantt view a validated leave
is shown as its own colored bar on the employee's row, not as part of
the gray "unavailable" background, so the server-side unavailability
calculation explicitly filters those out (see
`hr_holidays_gantt/models/hr_leave.py::_gantt_unavailability`, which
restricts the leave domain to `resource_id = False`, i.e. company-wide
holidays only). We fetch these separately (again, one batched call for
every employee) and merge them in ourselves.
Public holidays are company/date-range-wide, not employee-specific, so
that lookup is naturally already a single shared call.
Three other things get merged in:
- Each employee's own leave requests, from a plain `hr.leave.report.calendar.search_read`
(not `hr.leave.search_read`: `hr.leave` is normally row-restricted by an `ir.rule` to
"your own leaves + leaves of people you manage", and record rules filter silently
rather than raising, so a colleague outside that scope just comes back empty with no
error to catch; `hr.leave.report.calendar` is a SQL-view-backed model without that
restriction).
- Company-wide (or per-calendar) public holidays, from `resource.calendar.leaves`.
- `work_entry_type_id` (the specific leave type name) needs `hr_holidays.group_hr_holidays_user`,
and unlike a plain `read()`, `web_read` raises `AccessError` if you request a field you
don't have group access to rather than silently omitting it, so it's never requested and
every leave is labeled generically as "Time Off".
"""
from __future__ import annotations
import math
import re
from collections import defaultdict
from dataclasses import dataclass, field
@@ -90,6 +87,7 @@ class ResolvedEmployee:
id: int
name: str
requested_as: str
calendar_id: int | None
def resolve_employees(client: OdooClient, identifiers: list[str]) -> list[ResolvedEmployee]:
@@ -113,29 +111,35 @@ def resolve_employees(client: OdooClient, identifiers: list[str]) -> list[Resolv
for tag in tag_tokens:
atoms.append(("name", "=ilike", f"%({tag})"))
rows = client.call_kw("hr.employee", "search_read", [or_domain(atoms), ["id", "name"]])
rows = client.call_kw(
"hr.employee", "search_read", [or_domain(atoms), ["id", "name", "resource_calendar_id"]]
)
by_id = {r["id"]: r["name"] for r in rows}
by_id = {r["id"]: r for r in rows}
by_tag: dict[str, list[dict]] = defaultdict(list)
for r in rows:
m = TAG_RE.search(r["name"])
if m:
by_tag[m.group(1).lower()].append(r)
def _make(row: dict, token: str) -> ResolvedEmployee:
calendar = row["resource_calendar_id"]
return ResolvedEmployee(row["id"], row["name"], token, calendar[0] if calendar else None)
resolved = []
problems = []
for token in identifiers:
if token.isdigit():
emp_id = int(token)
if emp_id in by_id:
resolved.append(ResolvedEmployee(emp_id, by_id[emp_id], token))
resolved.append(_make(by_id[emp_id], token))
else:
problems.append(f"{token!r}: no employee with that id")
continue
matches = by_tag.get(token.lower(), [])
if len(matches) == 1:
resolved.append(ResolvedEmployee(matches[0]["id"], matches[0]["name"], token))
resolved.append(_make(matches[0], token))
elif not matches:
problems.append(f"{token!r}: no employee name ends with '({token})'")
else:
@@ -157,33 +161,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)
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
class LeavePeriod:
start: datetime
@@ -192,31 +169,37 @@ class LeavePeriod:
validated: bool
def _fetch_employee_leaves_batch(
def _fetch_leaves(
client: OdooClient, employee_ids: list[int], date_from: date, date_to: date, tz: ZoneInfo
) -> dict[int, list[LeavePeriod]]:
"""Every employee's own leave requests overlapping the range, 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)
rows = client.call_kw(
"hr.leave",
"hr.leave.report.calendar",
"search_read",
[
[
("employee_id", "in", employee_ids),
("state", "in", ["confirm", "validate", "validate1"]),
("date_from", "<=", f"{date_to} 23:59:59"),
("date_to", ">=", f"{date_from} 00:00:00"),
("state", "not in", ("cancel", "refuse")),
("start_datetime", "<", stop_utc.strftime("%Y-%m-%d %H:%M:%S")),
("stop_datetime", ">=", start_utc.strftime("%Y-%m-%d %H:%M:%S")),
],
["employee_id", "date_from", "date_to", "work_entry_type_id", "state"],
["employee_id", "start_datetime", "stop_datetime", "state"],
],
)
out: dict[int, list[LeavePeriod]] = defaultdict(list)
for r in rows:
type_name = r["work_entry_type_id"][1] if r["work_entry_type_id"] else "Time Off"
out[r["employee_id"][0]].append(
employee = r["employee_id"]
# employee_id normally comes back as [id, name], but resolving the
# name requires read access to hr.employee itself -- without that,
# it degrades to a bare id instead of raising.
employee_id = employee[0] if isinstance(employee, (list, tuple)) else employee
out[employee_id].append(
LeavePeriod(
start=_to_local(r["date_from"], tz),
stop=_to_local(r["date_to"], tz),
label=type_name,
start=_to_local(r["start_datetime"], tz),
stop=_to_local(r["stop_datetime"], tz),
label="Time Off",
validated=r["state"] == "validate",
)
)
@@ -224,34 +207,95 @@ def _fetch_employee_leaves_batch(
def _fetch_public_holidays(
client: OdooClient, date_from: date, date_to: date, tz: ZoneInfo
client: OdooClient, calendar_ids: list[int], date_from: date, date_to: date, tz: ZoneInfo
) -> list[tuple[datetime, datetime, str]]:
"""Company-wide (non-employee-specific) holidays, for labeling purposes only. Already shared across employees."""
rows = client.call_kw(
"resource.calendar.leaves",
"search_read",
[
[
("resource_id", "=", False),
("date_from", "<=", f"{date_to} 23:59:59"),
("date_to", ">=", f"{date_from} 00:00:00"),
],
["name", "date_from", "date_to"],
],
)
"""Company-wide or per-calendar (non-employee-specific) holidays, for labeling
purposes only. Already shared across employees using the same calendars.
"""
start_utc = _local_midnight_to_utc_naive(date_from, tz)
stop_utc = _local_midnight_to_utc_naive(date_to + timedelta(days=1), tz)
domain = or_domain([("calendar_id", "=", False), ("calendar_id", "in", calendar_ids)]) + [
("resource_id", "=", False),
("date_from", "<", stop_utc.strftime("%Y-%m-%d %H:%M:%S")),
("date_to", ">=", start_utc.strftime("%Y-%m-%d %H:%M:%S")),
]
rows = client.call_kw("resource.calendar.leaves", "search_read", [domain, ["name", "date_from", "date_to"]])
return [(_to_local(r["date_from"], tz), _to_local(r["date_to"], tz), r["name"] or "Public Holiday") for r in rows]
# A half-day leave's date_from/date_to is the employee's actual half-day work
# block (Odoo computes it from their real attendance hours), which is small
# next to a generic 00:00-12:00/12:00-24:00 window -- measuring "how much of
# the window is covered" dilutes it away. Checking a fixed anchor point per
# half-day instead sidesteps that: does the leave/pattern-off time actually
# cover the point in time this half of the day is centered on.
@dataclass
class CalendarInfo:
two_weeks: bool
# (weekday 0=Monday..6=Sunday, day_period, week_type '0'/'1' or None)
rows: list[tuple[int, str, str | None]]
def _fetch_attendance_patterns(client: OdooClient, calendar_ids: list[int]) -> dict[int, CalendarInfo]:
"""Each calendar's raw weekly attendance pattern, for every calendar in use, in one call."""
if not calendar_ids:
return {}
rows = client.call_kw(
"resource.calendar.attendance",
"search_read",
[
[("calendar_id", "in", calendar_ids)],
["calendar_id", "dayofweek", "day_period", "week_type", "two_weeks_calendar"],
],
)
by_calendar: dict[int, CalendarInfo] = {}
for r in rows:
cal_id = r["calendar_id"][0]
info = by_calendar.setdefault(cal_id, CalendarInfo(two_weeks=r["two_weeks_calendar"], rows=[]))
info.rows.append((int(r["dayofweek"]), r["day_period"], r["week_type"] or None))
return by_calendar
def _week_type(day: date) -> str:
"""Same formula as `resource.calendar.attendance.get_week_type`: parity of the
number of weeks since January 1 of year 1, so an even week always follows an odd
one even in a 53-week year.
"""
return str(int(math.floor((day.toordinal() - 1) / 7) % 2))
def _pattern_for_day(calendar: CalendarInfo | None, day: date) -> tuple[bool, bool]:
"""(morning_working, afternoon_working) from the calendar's raw attendance rows.
An employee with no calendar at all has no data to say otherwise, so defaults to
available both halves rather than defaulting to always off.
"""
if calendar is None:
return True, True
weekday = day.weekday()
week_type = _week_type(day) if calendar.two_weeks else None
morning = afternoon = False
for dow, period, wt in calendar.rows:
if dow != weekday:
continue
if calendar.two_weeks and wt != week_type:
continue
if period == "full_day":
morning = afternoon = True
elif period == "morning":
morning = True
elif period == "afternoon":
afternoon = True
return morning, afternoon
# A half-day leave's start/stop is the employee's actual half-day work block, which
# is small next to a generic 00:00-12:00/12:00-24:00 window -- measuring "how much of
# the window is covered" dilutes it away. Checking a fixed anchor point per half-day
# instead sidesteps that.
MORNING_ANCHOR_HOUR = 10
AFTERNOON_ANCHOR_HOUR = 15
def _anchor(d: date, tz: ZoneInfo, hour: int) -> datetime:
return datetime.combine(d, time(hour, 0), tzinfo=tz)
def _point_covered(intervals: list[tuple[datetime, datetime]], point: datetime) -> bool:
return any(start <= point < stop for start, stop in intervals)
@@ -280,14 +324,12 @@ def _build_days(
date_from: date,
date_to: date,
tz: ZoneInfo,
pattern_intervals: list[tuple[datetime, datetime]],
calendar: CalendarInfo | None,
leave_periods: list[LeavePeriod],
holidays: list[tuple[datetime, datetime, str]],
leave_fetch_failed: bool,
explain: bool,
) -> list[DaySchedule]:
validated_leave_intervals = [(p.start, p.stop) for p in leave_periods if p.validated]
unavailable_intervals = pattern_intervals + validated_leave_intervals
days = []
d = date_from
@@ -295,19 +337,16 @@ def _build_days(
morning_start = datetime.combine(d, time(0, 0), tzinfo=tz)
noon = datetime.combine(d, time(12, 0), tzinfo=tz)
day_end = datetime.combine(d, time(23, 59, 59), tzinfo=tz)
morning_anchor = datetime.combine(d, time(MORNING_ANCHOR_HOUR, 0), tzinfo=tz)
afternoon_anchor = datetime.combine(d, time(AFTERNOON_ANCHOR_HOUR, 0), tzinfo=tz)
morning_working = not _point_covered(unavailable_intervals, morning_anchor)
afternoon_working = not _point_covered(unavailable_intervals, afternoon_anchor)
pattern_morning, pattern_afternoon = _pattern_for_day(calendar, d)
morning_working = pattern_morning and not _point_covered(
validated_leave_intervals, _anchor(d, tz, MORNING_ANCHOR_HOUR)
)
afternoon_working = pattern_afternoon and not _point_covered(
validated_leave_intervals, _anchor(d, tz, AFTERNOON_ANCHOR_HOUR)
)
# Reason categories are computed regardless of `explain` -- the gantt
# grid's colors need them even when text notes are suppressed. Only
# the specific name (leave type / holiday name) and free-text notes
# are gated behind `explain`, since the public-holiday lookup itself
# is skipped to save a request when `explain` is off (see
# get_schedules): in that case a public holiday can't be told apart
# from a plain weekly day off and falls back to that category.
morning_reason = REASON_WORKING
afternoon_reason = REASON_WORKING
morning_label = None
@@ -319,8 +358,6 @@ def _build_days(
notes = []
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:
if reason == REASON_LEAVE:
@@ -360,20 +397,15 @@ def get_schedules(
"""Compute the per-day schedule for every employee, in a constant number of requests."""
tz = ZoneInfo(tz_name)
employee_ids = [e.id for e in employees]
calendar_ids = sorted({e.calendar_id for e in employees if e.calendar_id})
pattern_by_employee = _fetch_pattern_unavailability_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
calendars_by_id = _fetch_attendance_patterns(client, calendar_ids)
leaves_by_employee = _fetch_leaves(client, employee_ids, date_from, date_to, tz)
holidays: list[tuple[datetime, datetime, str]] = []
if explain:
try:
holidays = _fetch_public_holidays(client, date_from, date_to, tz)
holidays = _fetch_public_holidays(client, calendar_ids, date_from, date_to, tz)
except OdooError:
pass
@@ -382,10 +414,9 @@ def get_schedules(
date_from,
date_to,
tz,
pattern_by_employee.get(emp.id, []),
calendars_by_id.get(emp.calendar_id) if emp.calendar_id else None,
leaves_by_employee.get(emp.id, []),
holidays,
leave_fetch_failed,
explain,
)
for emp in employees