"""Unified Schedule service — ADP v0.1.""" from __future__ import annotations from datetime import date, datetime, time, timedelta, timezone from typing import Any, Literal, Optional from psycopg2.extras import RealDictCursor from db import get_connection ScheduleKind = Literal["interval", "weekdays"] _SCHEDULE_COLUMNS = """ id, tenant_id, schedule_kind, interval_days, weekday_mask, pause_until, created_at, updated_at """ WEEKDAY_LABELS = ("Mo", "Di", "Mi", "Do", "Fr", "Sa", "So") def weekday_bit(weekday: int) -> int: """ISO weekday: Monday=0 … Sunday=6.""" return 1 << weekday def _serialize_row(row: dict[str, Any]) -> dict[str, Any]: result = dict(row) for key in ("id", "tenant_id"): if result.get(key): result[key] = str(result[key]) for ts_key in ("created_at", "updated_at"): if result.get(ts_key): result[ts_key] = result[ts_key].isoformat() if result.get("pause_until"): result["pause_until"] = result["pause_until"].isoformat() return result def validate_schedule_payload( *, schedule_kind: str, interval_days: Optional[int], weekday_mask: int, ) -> None: if schedule_kind not in ("interval", "weekdays"): raise ValueError("Ungültiger schedule_kind") if schedule_kind == "interval": days = int(interval_days or 1) if days <= 0: raise ValueError("interval_days muss positiv sein") elif weekday_mask <= 0 or weekday_mask > 127: raise ValueError("Mindestens ein Wochentag auswählen") def format_schedule_label(schedule: dict[str, Any]) -> str: kind = schedule.get("schedule_kind") or "interval" if kind == "weekdays": mask = int(schedule.get("weekday_mask") or 0) days = [WEEKDAY_LABELS[i] for i in range(7) if mask & weekday_bit(i)] label = ", ".join(days) if days else "Wochentage" else: days = max(int(schedule.get("interval_days") or 1), 1) if days == 1: label = "Täglich" elif days == 7: label = "Wöchentlich" else: label = f"Alle {days} Tage" pause = schedule.get("pause_until") if pause: label = f"{label} (Pause bis {pause})" return label def _parse_pause_until(value: Any) -> Optional[date]: if value is None: return None if isinstance(value, date) and not isinstance(value, datetime): return value if isinstance(value, datetime): return value.date() text = str(value).strip() if not text: return None return date.fromisoformat(text[:10]) def is_paused_on(schedule: dict[str, Any], on_date: date) -> bool: pause = _parse_pause_until(schedule.get("pause_until")) return pause is not None and on_date <= pause def _weekday_matches(schedule: dict[str, Any], on_date: date) -> bool: mask = int(schedule.get("weekday_mask") or 0) return bool(mask & weekday_bit(on_date.weekday())) def _as_utc_midnight(d: date) -> datetime: return datetime.combine(d, time.min, tzinfo=timezone.utc) def _next_weekday_on_or_after( start: date, *, weekday_mask: int, pause_until: Optional[date] ) -> date: for offset in range(366): candidate = start + timedelta(days=offset) if pause_until is not None and candidate <= pause_until: continue if weekday_mask & weekday_bit(candidate.weekday()): return candidate return start + timedelta(days=7) def initial_due_at(*, schedule: dict[str, Any], now: Optional[datetime] = None) -> datetime: now = now or datetime.now(timezone.utc) today = now.date() if is_paused_on(schedule, today): pause = _parse_pause_until(schedule.get("pause_until")) assert pause is not None start = pause + timedelta(days=1) else: start = today kind = schedule.get("schedule_kind") or "interval" if kind == "weekdays": mask = int(schedule.get("weekday_mask") or 0) pause = _parse_pause_until(schedule.get("pause_until")) due_date = _next_weekday_on_or_after(start, weekday_mask=mask, pause_until=pause) return _as_utc_midnight(due_date) return _as_utc_midnight(start) def next_due_after_completion( *, schedule: dict[str, Any], completed_at: Optional[datetime] = None ) -> datetime: now = completed_at or datetime.now(timezone.utc) today = now.date() kind = schedule.get("schedule_kind") or "interval" if kind == "weekdays": mask = int(schedule.get("weekday_mask") or 0) pause = _parse_pause_until(schedule.get("pause_until")) start = today + timedelta(days=1) due_date = _next_weekday_on_or_after(start, weekday_mask=mask, pause_until=pause) return _as_utc_midnight(due_date) days = max(int(schedule.get("interval_days") or 1), 1) next_date = today + timedelta(days=days) pause = _parse_pause_until(schedule.get("pause_until")) if pause is not None and next_date <= pause: next_date = pause + timedelta(days=1) return _as_utc_midnight(next_date) def create_schedule( *, tenant_id: str, schedule_kind: ScheduleKind = "interval", interval_days: Optional[int] = 1, weekday_mask: int = 0, pause_until: Optional[date] = None, ) -> dict[str, Any]: validate_schedule_payload( schedule_kind=schedule_kind, interval_days=interval_days, weekday_mask=weekday_mask, ) conn = get_connection() try: with conn.cursor(cursor_factory=RealDictCursor) as cur: cur.execute( f""" INSERT INTO schedules ( tenant_id, schedule_kind, interval_days, weekday_mask, pause_until ) VALUES (%s, %s, %s, %s, %s) RETURNING {_SCHEDULE_COLUMNS} """, ( tenant_id, schedule_kind, interval_days if schedule_kind == "interval" else None, weekday_mask, pause_until, ), ) row = _serialize_row(dict(cur.fetchone())) conn.commit() finally: conn.close() row["label"] = format_schedule_label(row) return row def get_schedule(*, tenant_id: str, schedule_id: str) -> Optional[dict[str, Any]]: conn = get_connection() try: with conn.cursor(cursor_factory=RealDictCursor) as cur: cur.execute( f""" SELECT {_SCHEDULE_COLUMNS} FROM schedules WHERE id = %s AND tenant_id = %s """, (schedule_id, tenant_id), ) row = cur.fetchone() if not row: return None result = _serialize_row(dict(row)) result["label"] = format_schedule_label(result) return result finally: conn.close() def update_schedule( *, tenant_id: str, schedule_id: str, schedule_kind: Optional[ScheduleKind] = None, interval_days: Optional[int] = None, weekday_mask: Optional[int] = None, pause_until: Optional[date] = None, clear_pause_until: bool = False, ) -> Optional[dict[str, Any]]: existing = get_schedule(tenant_id=tenant_id, schedule_id=schedule_id) if not existing: return None kind = schedule_kind or existing["schedule_kind"] interval = interval_days if interval_days is not None else existing.get("interval_days") mask = weekday_mask if weekday_mask is not None else int(existing.get("weekday_mask") or 0) validate_schedule_payload( schedule_kind=kind, interval_days=interval, weekday_mask=mask ) updates: list[str] = [] params: list[Any] = [] if schedule_kind is not None: updates.append("schedule_kind = %s") params.append(schedule_kind) if interval_days is not None or schedule_kind == "weekdays": updates.append("interval_days = %s") params.append(None if kind == "weekdays" else interval) if weekday_mask is not None: updates.append("weekday_mask = %s") params.append(mask) if clear_pause_until: updates.append("pause_until = NULL") elif pause_until is not None: updates.append("pause_until = %s") params.append(pause_until) if not updates: return existing updates.append("updated_at = NOW()") params.extend([schedule_id, tenant_id]) conn = get_connection() try: with conn.cursor(cursor_factory=RealDictCursor) as cur: cur.execute( f""" UPDATE schedules SET {", ".join(updates)} WHERE id = %s AND tenant_id = %s RETURNING {_SCHEDULE_COLUMNS} """, params, ) row = cur.fetchone() if not row: return None result = _serialize_row(dict(row)) conn.commit() finally: conn.close() result["label"] = format_schedule_label(result) return result def attach_schedules_to_items( items: list[dict[str, Any]], *, tenant_id: str, schedule_id_key: str = "schedule_id" ) -> list[dict[str, Any]]: ids = [ str(item[schedule_id_key]) for item in items if item.get(schedule_id_key) ] if not ids: return [dict(item, schedule=None) for item in items] conn = get_connection() try: with conn.cursor(cursor_factory=RealDictCursor) as cur: cur.execute( f""" SELECT {_SCHEDULE_COLUMNS} FROM schedules WHERE tenant_id = %s AND id = ANY(%s::uuid[]) """, (tenant_id, ids), ) by_id = {} for row in cur.fetchall(): sched = _serialize_row(dict(row)) sched["label"] = format_schedule_label(sched) by_id[str(row["id"])] = sched finally: conn.close() enriched = [] for item in items: copy = dict(item) sid = copy.get(schedule_id_key) copy["schedule"] = by_id.get(str(sid)) if sid else None enriched.append(copy) return enriched def resolve_schedule_for_recurring( recurring: dict[str, Any], *, tenant_id: str ) -> Optional[dict[str, Any]]: sid = recurring.get("schedule_id") if sid: return get_schedule(tenant_id=tenant_id, schedule_id=str(sid)) interval = recurring.get("interval_days") if interval is None: return None return { "schedule_kind": "interval", "interval_days": max(int(interval), 1), "weekday_mask": 0, "pause_until": None, "label": format_schedule_label( {"schedule_kind": "interval", "interval_days": interval, "weekday_mask": 0} ), }