Initial chiffon

.gitignore

@@ -0,0 +1,2 @@
+**.pyc
+/session_id.txt

pyproject.toml

@@ -0,0 +1,18 @@
+[project]
+name = "odoo-employee-schedule"
+version = "0.1.0"
+description = "Query Odoo for an employee's working days in a date range, including days off with no leave taken"
+requires-python = ">=3.11"
+dependencies = [
+    "requests>=2.31",
+]
+
+[project.scripts]
+odoo-employee-schedule = "odoo_employee_schedule.cli:main"
+
+[build-system]
+requires = ["hatchling"]
+build-backend = "hatchling.build"
+
+[tool.hatch.build.targets.wheel]
+packages = ["src/odoo_employee_schedule"]

src/odoo_employee_schedule/cli.py

@@ -0,0 +1,153 @@
+from __future__ import annotations
+
+import argparse
+import os
+import sys
+from datetime import date
+from pathlib import Path
+from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
+
+from .client import OdooClient, OdooError
+from .schedule import DAY_NAMES, get_schedules, resolve_employees
+
+
+def parse_date(value: str) -> date:
+    return date.fromisoformat(value)
+
+
+def _valid_tz(name: str | None) -> str | None:
+    if not name:
+        return None
+    try:
+        ZoneInfo(name)
+    except (ZoneInfoNotFoundError, ValueError):
+        return None
+    return name
+
+
+def detect_local_tz() -> str | None:
+    """Best-effort IANA zone name for the machine this script runs on (stdlib only, POSIX).
+
+    Checks $TZ first, then resolves the /etc/localtime symlink most Linux/macOS
+    systems point at their zoneinfo file. Returns None if neither works (e.g.
+    Windows, or a system that copies the zoneinfo file instead of symlinking).
+    """
+    tz = _valid_tz(os.environ.get("TZ"))
+    if tz:
+        return tz
+
+    localtime = Path("/etc/localtime")
+    try:
+        target = localtime.resolve()
+    except OSError:
+        return None
+    parts = target.parts
+    if "zoneinfo" in parts:
+        name = "/".join(parts[parts.index("zoneinfo") + 1 :])
+        return _valid_tz(name)
+    return None
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(
+        description=(
+            "Show when one or more employees are working in a date range, including "
+            "days off with no leave taken."
+        )
+    )
+    parser.add_argument(
+        "employees",
+        nargs="+",
+        help=(
+            "Employee ids and/or name tags (the parenthesized suffix on the real "
+            "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",
+    )
+    parser.add_argument(
+        "--tz",
+        default=None,
+        help="Timezone for day boundaries (default: detected from this machine's environment, "
+        "falling back to your Odoo session tz)",
+    )
+    parser.add_argument(
+        "--no-explain",
+        action="store_true",
+        help="Don't label why a day is off (leave / public holiday / weekly day off); skips the public-holiday lookup only, hr.leave is still read for correctness",
+    )
+    parser.add_argument(
+        "--include-weekends",
+        action="store_true",
+        help="Also print Saturdays and Sundays (skipped by default)",
+    )
+    parser.add_argument(
+        "--debug",
+        action="store_true",
+        help="Log each outgoing request (model/method/args, not the response) to stderr",
+    )
+    args = parser.parse_args()
+
+    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:
+        print(f"error: {exc}", file=sys.stderr)
+        return 1
+
+    tz_name = args.tz
+    if tz_name is None:
+        tz_name = detect_local_tz()
+    if tz_name is None:
+        try:
+            info = client.session_info()
+            tz_name = info.get("user_context", {}).get("tz") or "UTC"
+        except OdooError:
+            tz_name = "UTC"
+
+    try:
+        employees = resolve_employees(client, args.employees)
+    except (ValueError, OdooError) as exc:
+        print(f"error: {exc}", file=sys.stderr)
+        return 1
+
+    try:
+        schedules = get_schedules(
+            client,
+            employees,
+            args.date_from,
+            args.date_to,
+            tz_name,
+            explain=not args.no_explain,
+        )
+    except OdooError as exc:
+        print(f"error calling Odoo: {exc}", file=sys.stderr)
+        return 1
+
+    for i, emp in enumerate(employees):
+        if i:
+            print()
+        print(f"{emp.name} (id={emp.id}) -- {args.date_from} to {args.date_to} [{tz_name}]\n")
+        for d in schedules[emp.id]:
+            if not args.include_weekends and d.day.weekday() in (5, 6):
+                continue
+            weekday = DAY_NAMES[d.day.weekday()]
+            line = f"{d.day} {weekday:<9} {d.status}"
+            if d.notes:
+                line += f"  ({'; '.join(d.notes)})"
+            print(line)
+
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

src/odoo_employee_schedule/client.py

@@ -0,0 +1,64 @@
+"""Minimal JSON-RPC client for Odoo's /web/dataset/call_kw route, authenticated via a session_id cookie."""
+
+from __future__ import annotations
+
+import json
+import sys
+from pathlib import Path
+
+import requests
+
+
+class OdooError(RuntimeError):
+    """Raised when Odoo returns a JSON-RPC error (access rights, missing record, etc.)."""
+
+    def __init__(self, message: str, data: dict | None = None):
+        super().__init__(message)
+        self.data = data or {}
+
+
+class OdooClient:
+    def __init__(self, base_url: str, session_id: str, debug: bool = False):
+        self.base_url = base_url.rstrip("/")
+        self.session = requests.Session()
+        self.session.cookies.set("session_id", session_id)
+        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()
+        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)
+
+    def _jsonrpc(self, endpoint: str, params: dict) -> dict:
+        self._request_count += 1
+        if self.debug:
+            print(
+                f"[debug] request #{self._request_count}: POST {self.base_url}{endpoint} "
+                + json.dumps(params, default=str),
+                file=sys.stderr,
+            )
+        resp = self.session.post(
+            f"{self.base_url}{endpoint}",
+            json={"jsonrpc": "2.0", "method": "call", "params": params},
+            headers={"Content-Type": "application/json"},
+            timeout=30,
+        )
+        resp.raise_for_status()
+        payload = resp.json()
+        if "error" in payload:
+            error = payload["error"]
+            message = error.get("data", {}).get("message") or error.get("message")
+            raise OdooError(message, error.get("data"))
+        return payload.get("result")
+
+    def call_kw(self, model: str, method: str, args: list, kwargs: dict | None = None):
+        return self._jsonrpc(
+            "/web/dataset/call_kw",
+            {"model": model, "method": method, "args": args, "kwargs": kwargs or {}},
+        )
+
+    def session_info(self) -> dict:
+        return self._jsonrpc("/web/session/get_session_info", {})

src/odoo_employee_schedule/schedule.py

@@ -0,0 +1,357 @@
+"""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:
+
+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.
+"""
+
+from __future__ import annotations
+
+import re
+from collections import defaultdict
+from dataclasses import dataclass, field
+from datetime import date, datetime, time, timedelta
+from zoneinfo import ZoneInfo
+
+from .client import OdooClient, OdooError
+
+DAY_NAMES = ["Monday", "Tuesday", "Wednesday", "Thursday", "Friday", "Saturday", "Sunday"]
+
+UTC = ZoneInfo("UTC")
+
+# Employee display names on the real server end with a short identifier tag,
+# e.g. "Michel Bram (mibr)", "Alice Peintre (AlPe)".
+TAG_RE = re.compile(r"\(([^()]*)\)\s*$")
+
+
+def or_domain(atoms: list[tuple]) -> list:
+    """Combine domain atoms with OR, using Odoo's prefix notation."""
+    if not atoms:
+        return [("id", "=", False)]
+    if len(atoms) == 1:
+        return list(atoms)
+    return ["|"] * (len(atoms) - 1) + list(atoms)
+
+
+@dataclass
+class DaySchedule:
+    day: date
+    morning_working: bool
+    afternoon_working: bool
+    notes: list[str] = field(default_factory=list)
+
+    @property
+    def status(self) -> str:
+        if self.morning_working and self.afternoon_working:
+            return "working (full day)"
+        if self.morning_working:
+            return "working (morning only)"
+        if self.afternoon_working:
+            return "working (afternoon only)"
+        return "off"
+
+
+@dataclass
+class ResolvedEmployee:
+    id: int
+    name: str
+    requested_as: str
+
+
+def resolve_employees(client: OdooClient, identifiers: list[str]) -> list[ResolvedEmployee]:
+    """Resolve a list of employee ids and/or `(tag)` identifiers in a single request.
+
+    A token is treated as a numeric employee id if it's all digits, otherwise
+    as a tag to match against the parenthesized suffix of `hr.employee.name`
+    (e.g. "mibr" matches "Michel Bram (mibr)"), using Odoo's `=ilike` operator
+    with an explicit `%(tag)` pattern so it only matches names literally
+    ending in "(tag)" (case-insensitively) rather than any substring match.
+
+    Raises ValueError listing any tokens that matched zero or more than one
+    employee.
+    """
+    id_tokens = [t for t in identifiers if t.isdigit()]
+    tag_tokens = [t for t in identifiers if not t.isdigit()]
+
+    atoms = []
+    if id_tokens:
+        atoms.append(("id", "in", [int(t) for t in id_tokens]))
+    for tag in tag_tokens:
+        atoms.append(("name", "=ilike", f"%({tag})"))
+
+    rows = client.call_kw("hr.employee", "search_read", [or_domain(atoms), ["id", "name"]])
+
+    by_id = {r["id"]: r["name"] 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)
+
+    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))
+            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))
+        elif not matches:
+            problems.append(f"{token!r}: no employee name ends with '({token})'")
+        else:
+            options = ", ".join(f"{m['name']} (id={m['id']})" for m in matches)
+            problems.append(f"{token!r}: matches more than one employee: {options}")
+
+    if problems:
+        raise ValueError("could not resolve some employees:\n  " + "\n  ".join(problems))
+
+    return resolved
+
+
+def _local_midnight_to_utc_naive(d: date, tz: ZoneInfo) -> datetime:
+    local_dt = datetime.combine(d, time.min, tzinfo=tz)
+    return local_dt.astimezone(UTC).replace(tzinfo=None)
+
+
+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
+    stop: datetime
+    label: str
+    validated: bool
+
+
+def _fetch_employee_leaves_batch(
+    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."""
+    rows = client.call_kw(
+        "hr.leave",
+        "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"),
+            ],
+            ["employee_id", "date_from", "date_to", "work_entry_type_id", "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(
+            LeavePeriod(
+                start=_to_local(r["date_from"], tz),
+                stop=_to_local(r["date_to"], tz),
+                label=type_name,
+                validated=r["state"] == "validate",
+            )
+        )
+    return out
+
+
+def _fetch_public_holidays(
+    client: OdooClient, 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"],
+        ],
+    )
+    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.
+MORNING_ANCHOR_HOUR = 10
+AFTERNOON_ANCHOR_HOUR = 15
+
+
+def _point_covered(intervals: list[tuple[datetime, datetime]], point: datetime) -> bool:
+    return any(start <= point < stop for start, stop in intervals)
+
+
+def _overlaps(a_start: datetime, a_stop: datetime, b_start: datetime, b_stop: datetime) -> bool:
+    return a_start < b_stop and b_start < a_stop
+
+
+def _build_days(
+    date_from: date,
+    date_to: date,
+    tz: ZoneInfo,
+    pattern_intervals: list[tuple[datetime, datetime]],
+    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
+    while d <= date_to:
+        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)
+
+        notes = []
+        if explain:
+            if leave_fetch_failed:
+                notes.append("could not fetch this employee's leave requests (access denied) -- personal leave not reflected")
+
+            off_windows = []
+            if not morning_working:
+                off_windows.append(("morning", morning_start, noon))
+            if not afternoon_working:
+                off_windows.append(("afternoon", noon, day_end))
+
+            for label, w_start, w_stop in off_windows:
+                reason = None
+                for p in leave_periods:
+                    if p.validated and _overlaps(w_start, w_stop, p.start, p.stop):
+                        reason = f"leave: {p.label}"
+                        break
+                if reason is None:
+                    for h_start, h_stop, name in holidays:
+                        if _overlaps(w_start, w_stop, h_start, h_stop):
+                            reason = f"public holiday: {name}"
+                            break
+                if reason is None:
+                    reason = "not scheduled to work (weekly day off)"
+                prefix = "" if (not morning_working and not afternoon_working) else f"{label}: "
+                notes.append(prefix + reason)
+                if not morning_working and not afternoon_working:
+                    break  # same reason applies to the whole day, don't repeat it
+
+            for p in leave_periods:
+                if not p.validated and _overlaps(morning_start, day_end, p.start, p.stop):
+                    notes.append(f"pending leave request: {p.label} (awaiting approval)")
+
+        days.append(DaySchedule(d, morning_working, afternoon_working, notes))
+        d += timedelta(days=1)
+
+    return days
+
+
+def get_schedules(
+    client: OdooClient,
+    employees: list[ResolvedEmployee],
+    date_from: date,
+    date_to: date,
+    tz_name: str,
+    explain: bool = True,
+) -> dict[int, list[DaySchedule]]:
+    """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]
+
+    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
+
+    holidays: list[tuple[datetime, datetime, str]] = []
+    if explain:
+        try:
+            holidays = _fetch_public_holidays(client, date_from, date_to, tz)
+        except OdooError:
+            pass
+
+    return {
+        emp.id: _build_days(
+            date_from,
+            date_to,
+            tz,
+            pattern_by_employee.get(emp.id, []),
+            leaves_by_employee.get(emp.id, []),
+            holidays,
+            leave_fetch_failed,
+            explain,
+        )
+        for emp in employees
+    }