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        "report": task.report,
162        "schedule": serialize_task_schedule_state(task.schedule),
163        "schedule_enabled": task.schedule.enabled if task.schedule else True,
164    }
165
166
167def restore_task_state(task: BackgroundTask, state: Mapping[str, Any]) -> None:
168    """Restore persisted runtime state for a scheduled task."""
169    if task.schedule and isinstance(state.get("schedule"), Mapping):
170        task.schedule = merge_task_schedule_state(task.schedule, state["schedule"])
171    elif task.schedule and isinstance(state.get("schedule_enabled"), bool):
172        task.schedule.enabled = state["schedule_enabled"]
173
174    status_value = state.get("status")
175    if isinstance(status_value, str):
176        status = TaskStatus(status_value)
177        if status in (TaskStatus.PENDING, TaskStatus.RUNNING):
178            status = TaskStatus.IDLE
179        task.status = status
180
181    task.last_run = parse_utc_datetime(state.get("last_run"))
182    task.last_run_user_id = (
183        state.get("last_run_user_id") if isinstance(state.get("last_run_user_id"), str) else None
184    )
185    task.finished_at = parse_utc_datetime(state.get("finished_at"))
186    task.updated_at = parse_utc_datetime(state.get("updated_at")) or task.updated_at
187    task.last_error = state.get("last_error") if isinstance(state.get("last_error"), str) else None
188
189    failure_count = state.get("failure_count")
190    task.failure_count = (
191        failure_count if isinstance(failure_count, int) and failure_count > 0 else 0
192    )
193
194    failure_messages = state.get("failure_messages")
195    if isinstance(failure_messages, list):
196        task.failure_messages = [item for item in failure_messages if isinstance(item, str)]
197    else:
198        task.failure_messages = []
199
200    task.report = state.get("report") if isinstance(state.get("report"), str) else None
201
202
203def serialize_task_schedule_state(schedule: TaskSchedule | None) -> dict[str, Any] | None:
204    """Serialize only the editable schedule fields for persistence."""
205    if schedule is None:
206        return None
207    result: dict[str, Any] = {
208        "type": schedule.type.value,
209        "enabled": schedule.enabled,
210    }
211    if schedule.type in (TaskScheduleType.HOURLY, TaskScheduleType.DAILY):
212        result["every"] = schedule.every
213    if schedule.type == TaskScheduleType.DAILY:
214        result["hour"] = schedule.hour
215        result["minute"] = schedule.minute
216    elif schedule.type == TaskScheduleType.WEEKLY:
217        result["days_of_week"] = list(schedule.days_of_week or [])
218        result["hour"] = schedule.hour
219        result["minute"] = schedule.minute
220    return result
221
222
223def _build_schedule_from_state(
224    default_schedule: TaskSchedule,
225    schedule_type: TaskScheduleType,
226    state: Mapping[str, Any],
227    enabled: bool,
228) -> TaskSchedule:
229    """Build a schedule from persisted editable state."""
230    if schedule_type == TaskScheduleType.HOURLY:
231        return TaskSchedule(
232            type=TaskScheduleType.HOURLY,
233            enabled=enabled,
234            every=(
235                state["every"]
236                if isinstance(state.get("every"), int) and state["every"] > 0
237                else (
238                    default_schedule.every
239                    if default_schedule.type == TaskScheduleType.HOURLY
240                    else 1
241                )
242            ),
243        )
244    if schedule_type == TaskScheduleType.DAILY:
245        raw_hour = state.get("hour")
246        raw_minute = state.get("minute")
247        return TaskSchedule(
248            type=TaskScheduleType.DAILY,
249            enabled=enabled,
250            every=(
251                state["every"]
252                if isinstance(state.get("every"), int) and state["every"] > 0
253                else (
254                    default_schedule.every if default_schedule.type == TaskScheduleType.DAILY else 1
255                )
256            ),
257            hour=(
258                raw_hour
259                if isinstance(raw_hour, int)
260                else (
261                    default_schedule.hour
262                    if default_schedule.type == TaskScheduleType.DAILY
263                    and default_schedule.hour is not None
264                    else 0
265                )
266            ),
267            minute=(
268                raw_minute
269                if isinstance(raw_minute, int)
270                else (
271                    default_schedule.minute
272                    if default_schedule.type == TaskScheduleType.DAILY
273                    and default_schedule.minute is not None
274                    else 0
275                )
276            ),
277        )
278    if schedule_type == TaskScheduleType.WEEKLY:
279        raw_days_of_week = state.get("days_of_week")
280        raw_hour = state.get("hour")
281        raw_minute = state.get("minute")
282        return TaskSchedule(
283            type=TaskScheduleType.WEEKLY,
284            enabled=enabled,
285            days_of_week=(
286                raw_days_of_week
287                if isinstance(raw_days_of_week, list)
288                and all(isinstance(day, int) for day in raw_days_of_week)
289                else (
290                    default_schedule.days_of_week
291                    if default_schedule.type == TaskScheduleType.WEEKLY
292                    and default_schedule.days_of_week
293                    else [0]
294                )
295            ),
296            hour=(
297                raw_hour
298                if isinstance(raw_hour, int)
299                else (
300                    default_schedule.hour
301                    if default_schedule.type == TaskScheduleType.WEEKLY
302                    and default_schedule.hour is not None
303                    else 0
304                )
305            ),
306            minute=(
307                raw_minute
308                if isinstance(raw_minute, int)
309                else (
310                    default_schedule.minute
311                    if default_schedule.type == TaskScheduleType.WEEKLY
312                    and default_schedule.minute is not None
313                    else 0
314                )
315            ),
316        )
317    raise ValueError(f"Unsupported task schedule type: {schedule_type}")
318
319
320def merge_task_schedule_state(
321    default_schedule: TaskSchedule,
322    state: Mapping[str, Any],
323) -> TaskSchedule:
324    """Merge persisted/editable schedule state onto a default schedule definition."""
325    enabled = (
326        state["enabled"] if isinstance(state.get("enabled"), bool) else default_schedule.enabled
327    )
328
329    schedule_type = default_schedule.type
330    raw_type = state.get("type")
331    if isinstance(raw_type, str):
332        try:
333            parsed_type = TaskScheduleType(raw_type)
334        except ValueError:
335            parsed_type = default_schedule.type
336        if parsed_type != TaskScheduleType.UNKNOWN:
337            schedule_type = parsed_type
338
339    try:
340        return _build_schedule_from_state(default_schedule, schedule_type, state, enabled)
341    except ValueError:
342        return default_schedule
343
344
345def parse_utc_datetime(value: Any) -> datetime | None:
346    """Parse an ISO datetime string into an aware UTC datetime."""
347    if not isinstance(value, str):
348        return None
349    try:
350        parsed = datetime.fromisoformat(value)
351    except ValueError:
352        return None
353    if parsed.tzinfo is None:
354        return parsed.replace(tzinfo=UTC)
355    return parsed.astimezone(UTC)
356
357
358def format_task_log_line(
359    message: str,
360    *,
361    level: int,
362    logger_name: str,
363    created_at: datetime | None = None,
364) -> str:
365    """Format a synthetic task lifecycle log line like regular captured logs."""
366    record = logging.LogRecord(logger_name, level, "", 0, message, (), None)
367    if created_at is not None:
368        created_ts = created_at.timestamp()
369        record.created = created_ts
370        record.msecs = (created_ts - int(created_ts)) * 1000
371    return TASK_LOG_FORMATTER.format(record)
372
373
374class TaskLogHandler(logging.Handler):
375    """Logging handler that mirrors log lines into the active managed task."""
376
377    def __init__(
378        self,
379        mass: MusicAssistant,
380        append_log: Callable[[str, str], None],
381    ) -> None:
382        """Initialize the handler."""
383        super().__init__(logging.DEBUG)
384        self._mass = mass
385        self._append_log = append_log
386        self.setFormatter(TASK_LOG_FORMATTER)
387
388    def emit(self, record: logging.LogRecord) -> None:
389        """Forward the formatted log line to the active task buffer."""
390        if not (task_id := ACTIVE_TASK_ID.get()):
391            return
392        try:
393            line = self.format(record)
394        except Exception:  # pragma: no cover - logging internals
395            self.handleError(record)
396            return
397        self._mass.loop.call_soon_threadsafe(self._append_log, task_id, line)
398