124 lines
3.5 KiB
Python
124 lines
3.5 KiB
Python
"""Helpers for building EUSRR -> LogStorm attendance requests."""
|
|
|
|
from datetime import date
|
|
from typing import Any, Optional
|
|
|
|
from common.logstorm_client import LogStormClient
|
|
from employees.models import Employee
|
|
|
|
DEFAULT_WORKDAYS = [
|
|
"Monday",
|
|
"Tuesday",
|
|
"Wednesday",
|
|
"Thursday",
|
|
"Friday",
|
|
]
|
|
|
|
|
|
def build_logstorm_attendance_payload(
|
|
*,
|
|
employee: Employee,
|
|
period_start: date,
|
|
period_end: date,
|
|
schedule: Optional[dict[str, Any]] = None,
|
|
date_overrides: Optional[list[dict[str, Any]]] = None,
|
|
aliases: Optional[list[Any]] = None,
|
|
) -> dict[str, Any]:
|
|
"""
|
|
Build a LogStorm attendance analysis payload.
|
|
|
|
If schedule is omitted, the payload intentionally omits it so LogStorm can
|
|
apply its configured default schedule or reject the request by policy.
|
|
"""
|
|
payload = {
|
|
"employee_id": str(employee.id),
|
|
"display_name": _employee_display_name(employee),
|
|
"period_start": period_start.isoformat(),
|
|
"period_end": period_end.isoformat(),
|
|
}
|
|
|
|
if schedule is not None:
|
|
payload["schedule"] = normalize_logstorm_schedule(
|
|
schedule,
|
|
date_overrides=date_overrides,
|
|
)
|
|
|
|
normalized_aliases = _normalize_aliases(aliases)
|
|
if normalized_aliases:
|
|
payload["aliases"] = normalized_aliases
|
|
|
|
return payload
|
|
|
|
|
|
def normalize_logstorm_schedule(
|
|
schedule: dict[str, Any],
|
|
*,
|
|
date_overrides: Optional[list[dict[str, Any]]] = None,
|
|
) -> dict[str, Any]:
|
|
return {
|
|
"start_time": schedule["start_time"],
|
|
"end_time": schedule["end_time"],
|
|
"expected_hours": schedule["expected_hours"],
|
|
"workdays": schedule.get("workdays", DEFAULT_WORKDAYS),
|
|
"date_overrides": [
|
|
_normalize_date_override(override)
|
|
for override in (
|
|
date_overrides
|
|
if date_overrides is not None
|
|
else schedule.get("date_overrides", [])
|
|
)
|
|
],
|
|
}
|
|
|
|
|
|
def analyze_employee_attendance(
|
|
*,
|
|
employee: Employee,
|
|
period_start: date,
|
|
period_end: date,
|
|
schedule: Optional[dict[str, Any]] = None,
|
|
date_overrides: Optional[list[dict[str, Any]]] = None,
|
|
aliases: Optional[list[Any]] = None,
|
|
client: Optional[LogStormClient] = None,
|
|
) -> dict[str, Any]:
|
|
payload = build_logstorm_attendance_payload(
|
|
employee=employee,
|
|
period_start=period_start,
|
|
period_end=period_end,
|
|
schedule=schedule,
|
|
date_overrides=date_overrides,
|
|
aliases=aliases,
|
|
)
|
|
return (client or LogStormClient()).analyze_attendance(payload)
|
|
|
|
|
|
def _normalize_date_override(override: dict[str, Any]) -> dict[str, Any]:
|
|
result = {
|
|
"date": _format_date(override["date"]),
|
|
"is_workday": bool(override["is_workday"]),
|
|
}
|
|
for key in ("reason", "start_time", "end_time", "expected_hours"):
|
|
if key in override and override[key] is not None:
|
|
result[key] = override[key]
|
|
return result
|
|
|
|
|
|
def _format_date(value: Any) -> str:
|
|
if isinstance(value, date):
|
|
return value.isoformat()
|
|
return str(value)
|
|
|
|
|
|
def _normalize_aliases(aliases: Optional[list[Any]]) -> list[str]:
|
|
result: list[str] = []
|
|
for alias in aliases or []:
|
|
alias_value = str(alias).strip()
|
|
if alias_value and alias_value not in result:
|
|
result.append(alias_value)
|
|
return result
|
|
|
|
|
|
def _employee_display_name(employee: Employee) -> str:
|
|
full_name = employee.get_full_name().strip()
|
|
return full_name or str(employee)
|