music-assistant-server

1"""Helper utilities for the background tasks controller."""
2
3from __future__ import annotations
4
5import logging
6from collections.abc import Callable, Iterable, Mapping
7from datetime import UTC, datetime, time, timedelta
8from typing import TYPE_CHECKING, Any
9
10from music_assistant_models.auth import Scope, User
11from music_assistant_models.background_task import BackgroundTask, TaskSchedule
12from music_assistant_models.enums import TaskScheduleType, TaskStatus
13
14from music_assistant.controllers.webserver.helpers.auth_middleware import has_scope
15from music_assistant.helpers.datetime import utc
16
17from .constants import ACTIVE_TASK_ID
18from .models import ManagedTask
19
20if TYPE_CHECKING:
21    from music_assistant import MusicAssistant
22
23TASK_LOG_FORMAT = "%(asctime)s %(levelname)s [%(name)s] %(message)s"
24TASK_LOG_DATE_FORMAT = "%Y-%m-%d %H:%M:%S"
25TASK_LOG_FORMATTER = logging.Formatter(TASK_LOG_FORMAT, datefmt=TASK_LOG_DATE_FORMAT)
26
27
28def utcnow() -> datetime:
29    """Return current UTC datetime."""
30    return utc()
31
32
33def get_task_timer_id(task_id: str) -> str:
34    """Return timer id for a managed task."""
35    return f"managed_task_timer_{task_id}"
36
37
38def task_sort_key(managed: ManagedTask) -> tuple[int, float]:
39    """Sort active tasks first, then recently updated tasks."""
40    task_info = managed.task_info
41    if task_info.status == TaskStatus.RUNNING:
42        priority = 0
43    elif task_info.status == TaskStatus.PENDING:
44        priority = 1
45    elif managed.is_scheduled:
46        priority = 2
47    else:
48        priority = 3
49    return (priority, -task_info.updated_at.timestamp())
50
51
52def is_task_visible_to_user(task: ManagedTask, user: User | None) -> bool:
53    """Return if a task should be visible to the given user."""
54    if user is None or has_scope(user, Scope.SYSTEM_MANAGE):
55        return True
56    return task.task_info.user_id == user.user_id
57
58
59def get_visible_tasks(tasks: Iterable[ManagedTask], user: User | None) -> list[ManagedTask]:
60    """Return tasks visible to the given user."""
61    visible_tasks = list(tasks)
62    if user is not None and not has_scope(user, Scope.SYSTEM_MANAGE):
63        visible_tasks = [task for task in visible_tasks if is_task_visible_to_user(task, user)]
64    visible_tasks.sort(key=task_sort_key)
65    return visible_tasks
66
67
68def trim_finished_history(tasks: dict[str, ManagedTask], max_history: int) -> None:
69    """Keep finished ad-hoc task history bounded."""
70    finished = [managed for managed in tasks.values() if managed.can_remove]
71    if len(finished) <= max_history:
72        return
73    finished.sort(key=lambda managed: managed.task_info.updated_at.timestamp())
74    for managed in finished[: len(finished) - max_history]:
75        tasks.pop(managed.task_info.id, None)
76
77
78def get_task_schedule_next_run(
79    schedule: TaskSchedule,
80    now: datetime | None = None,
81) -> datetime:
82    """Calculate the next run moment in UTC for a task schedule."""
83    current_time = now or utcnow()
84    if schedule.type == TaskScheduleType.HOURLY:
85        if schedule.every is None:
86            raise ValueError("Hourly schedule requires every")
87        return current_time + timedelta(hours=schedule.every)
88
89    local_now = current_time.astimezone(UTC)
90    scheduled_time = time(hour=schedule.hour or 0, minute=schedule.minute or 0)
91
92    if schedule.type == TaskScheduleType.DAILY:
93        candidate = datetime.combine(local_now.date(), scheduled_time, UTC)
94        if candidate <= local_now:
95            candidate += timedelta(days=schedule.every or 1)
96        return candidate
97
98    if schedule.type == TaskScheduleType.WEEKLY:
99        if not schedule.days_of_week:
100            raise ValueError("Weekly schedule requires days_of_week")
101        next_candidate: datetime | None = None
102        for day_of_week in schedule.days_of_week:
103            days_ahead = (day_of_week - local_now.weekday()) % 7
104            candidate_date = local_now.date() + timedelta(days=days_ahead)
105            candidate = datetime.combine(candidate_date, scheduled_time, UTC)
106            if candidate <= local_now:
107                candidate += timedelta(days=7)
108            if next_candidate is None or candidate < next_candidate:
109                next_candidate = candidate
110        if next_candidate is None:
111            raise ValueError("Unable to calculate next run for weekly schedule")
112        return next_candidate
113
114    raise ValueError(f"Unsupported task schedule type: {schedule.type}")
115
116
117def get_task_schedule_delay(
118    schedule: TaskSchedule,
119    now: datetime | None = None,
120    last_run: datetime | None = None,
121    fallback_delay: float | None = None,
122) -> tuple[float, datetime]:
123    """Calculate the delay and next run moment for a task schedule."""
124    current_time = now or utcnow()
125    if schedule.type == TaskScheduleType.HOURLY and schedule.every is not None:
126        if last_run is not None:
127            next_run = last_run + timedelta(hours=schedule.every)
128            if next_run > current_time:
129                return (next_run - current_time).total_seconds(), next_run
130        if fallback_delay is not None:
131            next_run = current_time + timedelta(seconds=fallback_delay)
132            return fallback_delay, next_run
133    elif schedule.type == TaskScheduleType.DAILY:
134        scheduled_time = time(hour=schedule.hour or 0, minute=schedule.minute or 0)
135        if last_run is not None:
136            next_run = datetime.combine(last_run.astimezone(UTC).date(), scheduled_time, UTC)
137            if next_run <= last_run:
138                next_run += timedelta(days=schedule.every or 1)
139            while next_run <= current_time:
140                next_run += timedelta(days=schedule.every or 1)
141            return max((next_run - current_time).total_seconds(), 0.0), next_run
142    next_run = get_task_schedule_next_run(schedule, current_time)
143    if last_run is None and fallback_delay is not None:
144        next_run = current_time + timedelta(seconds=fallback_delay)
145        return fallback_delay, next_run
146    delay = max((next_run - current_time).total_seconds(), 0.0)
147    return delay, next_run
148
149
150def serialize_task_state(task: BackgroundTask) -> dict[str, Any]:
151    """Serialize the persisted runtime state for a scheduled task."""
152    return {
153        "status": task.status.value,
154        "last_run": task.last_run.isoformat() if task.last_run else None,
155        "last_run_user_id": task.last_run_user_id,
156        "finished_at": task.finished_at.isoformat() if task.finished_at else None,
157        "updated_at": task.updated_at.isoformat() if task.updated_at else None,
158        "last_error": task.last_error,
159        "failure_count": task.failure_count,
160        "failure_messages": list(task.failure_messages),
161        "schedule": serialize_task_schedule_state(task.schedule),
162        "schedule_enabled": task.schedule.enabled if task.schedule else True,
163    }
164
165
166def restore_task_state(task: BackgroundTask, state: Mapping[str, Any]) -> None:
167    """Restore persisted runtime state for a scheduled task."""
168    if task.schedule and isinstance(state.get("schedule"), Mapping):
169        task.schedule = merge_task_schedule_state(task.schedule, state["schedule"])
170    elif task.schedule and isinstance(state.get("schedule_enabled"), bool):
171        task.schedule.enabled = state["schedule_enabled"]
172
173    status_value = state.get("status")
174    if isinstance(status_value, str):
175        status = TaskStatus(status_value)
176        if status in (TaskStatus.PENDING, TaskStatus.RUNNING):
177            status = TaskStatus.IDLE
178        task.status = status
179
180    task.last_run = parse_utc_datetime(state.get("last_run"))
181    task.last_run_user_id = (
182        state.get("last_run_user_id") if isinstance(state.get("last_run_user_id"), str) else None
183    )
184    task.finished_at = parse_utc_datetime(state.get("finished_at"))
185    task.updated_at = parse_utc_datetime(state.get("updated_at")) or task.updated_at
186    task.last_error = state.get("last_error") if isinstance(state.get("last_error"), str) else None
187
188    failure_count = state.get("failure_count")
189    task.failure_count = (
190        failure_count if isinstance(failure_count, int) and failure_count > 0 else 0
191    )
192
193    failure_messages = state.get("failure_messages")
194    if isinstance(failure_messages, list):
195        task.failure_messages = [item for item in failure_messages if isinstance(item, str)]
196    else:
197        task.failure_messages = []
198
199
200def serialize_task_schedule_state(schedule: TaskSchedule | None) -> dict[str, Any] | None:
201    """Serialize only the editable schedule fields for persistence."""
202    if schedule is None:
203        return None
204    result: dict[str, Any] = {
205        "type": schedule.type.value,
206        "enabled": schedule.enabled,
207    }
208    if schedule.type in (TaskScheduleType.HOURLY, TaskScheduleType.DAILY):
209        result["every"] = schedule.every
210    if schedule.type == TaskScheduleType.DAILY:
211        result["hour"] = schedule.hour
212        result["minute"] = schedule.minute
213    elif schedule.type == TaskScheduleType.WEEKLY:
214        result["days_of_week"] = list(schedule.days_of_week or [])
215        result["hour"] = schedule.hour
216        result["minute"] = schedule.minute
217    return result
218
219
220def _build_schedule_from_state(
221    default_schedule: TaskSchedule,
222    schedule_type: TaskScheduleType,
223    state: Mapping[str, Any],
224    enabled: bool,
225) -> TaskSchedule:
226    """Build a schedule from persisted editable state."""
227    if schedule_type == TaskScheduleType.HOURLY:
228        return TaskSchedule(
229            type=TaskScheduleType.HOURLY,
230            enabled=enabled,
231            every=(
232                state["every"]
233                if isinstance(state.get("every"), int) and state["every"] > 0
234                else (
235                    default_schedule.every
236                    if default_schedule.type == TaskScheduleType.HOURLY
237                    else 1
238                )
239            ),
240        )
241    if schedule_type == TaskScheduleType.DAILY:
242        raw_hour = state.get("hour")
243        raw_minute = state.get("minute")
244        return TaskSchedule(
245            type=TaskScheduleType.DAILY,
246            enabled=enabled,
247            every=(
248                state["every"]
249                if isinstance(state.get("every"), int) and state["every"] > 0
250                else (
251                    default_schedule.every if default_schedule.type == TaskScheduleType.DAILY else 1
252                )
253            ),
254            hour=(
255                raw_hour
256                if isinstance(raw_hour, int)
257                else (
258                    default_schedule.hour
259                    if default_schedule.type == TaskScheduleType.DAILY
260                    and default_schedule.hour is not None
261                    else 0
262                )
263            ),
264            minute=(
265                raw_minute
266                if isinstance(raw_minute, int)
267                else (
268                    default_schedule.minute
269                    if default_schedule.type == TaskScheduleType.DAILY
270                    and default_schedule.minute is not None
271                    else 0
272                )
273            ),
274        )
275    if schedule_type == TaskScheduleType.WEEKLY:
276        raw_days_of_week = state.get("days_of_week")
277        raw_hour = state.get("hour")
278        raw_minute = state.get("minute")
279        return TaskSchedule(
280            type=TaskScheduleType.WEEKLY,
281            enabled=enabled,
282            days_of_week=(
283                raw_days_of_week
284                if isinstance(raw_days_of_week, list)
285                and all(isinstance(day, int) for day in raw_days_of_week)
286                else (
287                    default_schedule.days_of_week
288                    if default_schedule.type == TaskScheduleType.WEEKLY
289                    and default_schedule.days_of_week
290                    else [0]
291                )
292            ),
293            hour=(
294                raw_hour
295                if isinstance(raw_hour, int)
296                else (
297                    default_schedule.hour
298                    if default_schedule.type == TaskScheduleType.WEEKLY
299                    and default_schedule.hour is not None
300                    else 0
301                )
302            ),
303            minute=(
304                raw_minute
305                if isinstance(raw_minute, int)
306                else (
307                    default_schedule.minute
308                    if default_schedule.type == TaskScheduleType.WEEKLY
309                    and default_schedule.minute is not None
310                    else 0
311                )
312            ),
313        )
314    raise ValueError(f"Unsupported task schedule type: {schedule_type}")
315
316
317def merge_task_schedule_state(
318    default_schedule: TaskSchedule,
319    state: Mapping[str, Any],
320) -> TaskSchedule:
321    """Merge persisted/editable schedule state onto a default schedule definition."""
322    enabled = (
323        state["enabled"] if isinstance(state.get("enabled"), bool) else default_schedule.enabled
324    )
325
326    schedule_type = default_schedule.type
327    raw_type = state.get("type")
328    if isinstance(raw_type, str):
329        try:
330            parsed_type = TaskScheduleType(raw_type)
331        except ValueError:
332            parsed_type = default_schedule.type
333        if parsed_type != TaskScheduleType.UNKNOWN:
334            schedule_type = parsed_type
335
336    try:
337        return _build_schedule_from_state(default_schedule, schedule_type, state, enabled)
338    except ValueError:
339        return default_schedule
340
341
342def parse_utc_datetime(value: Any) -> datetime | None:
343    """Parse an ISO datetime string into an aware UTC datetime."""
344    if not isinstance(value, str):
345        return None
346    try:
347        parsed = datetime.fromisoformat(value)
348    except ValueError:
349        return None
350    if parsed.tzinfo is None:
351        return parsed.replace(tzinfo=UTC)
352    return parsed.astimezone(UTC)
353
354
355def format_task_log_line(
356    message: str,
357    *,
358    level: int,
359    logger_name: str,
360    created_at: datetime | None = None,
361) -> str:
362    """Format a synthetic task lifecycle log line like regular captured logs."""
363    record = logging.LogRecord(logger_name, level, "", 0, message, (), None)
364    if created_at is not None:
365        created_ts = created_at.timestamp()
366        record.created = created_ts
367        record.msecs = (created_ts - int(created_ts)) * 1000
368    return TASK_LOG_FORMATTER.format(record)
369
370
371class TaskLogHandler(logging.Handler):
372    """Logging handler that mirrors log lines into the active managed task."""
373
374    def __init__(
375        self,
376        mass: MusicAssistant,
377        append_log: Callable[[str, str], None],
378    ) -> None:
379        """Initialize the handler."""
380        super().__init__(logging.DEBUG)
381        self._mass = mass
382        self._append_log = append_log
383        self.setFormatter(TASK_LOG_FORMATTER)
384
385    def emit(self, record: logging.LogRecord) -> None:
386        """Forward the formatted log line to the active task buffer."""
387        if not (task_id := ACTIVE_TASK_ID.get()):
388            return
389        try:
390            line = self.format(record)
391        except Exception:  # pragma: no cover - logging internals
392            self.handleError(record)
393            return
394        self._mass.loop.call_soon_threadsafe(self._append_log, task_id, line)
395