/
/
/
1"""
2Shared recency engine for the Music controller.
3
4Reads the playlog once and exposes fast in-memory "was this heard recently?" tests, used by
5smart shuffle, the smart playlist dedup and (later) radio refills. Song recency is matched on
6provider/item-id pairs resolved across a track's provider mappings (the exact escape hatch),
7plus a fuzzy same-song match on (version-stripped title, artist name) keys so the same
8recording is recognized across different releases/providers with differing ids or artist
9credits; artist recency is matched by (lowercased) name, which is provider-agnostic (artist
10playlog rows are keyed by the library id).
11"""
12
13from __future__ import annotations
14
15import time
16from dataclasses import dataclass, field
17from typing import TYPE_CHECKING
18
19from music_assistant_models.enums import MediaType
20from music_assistant_models.helpers import create_safe_string
21
22from music_assistant.constants import DB_TABLE_PLAYLOG
23from music_assistant.controllers.webserver.helpers.auth_middleware import get_current_user
24from music_assistant.helpers.json import json_loads
25from music_assistant.helpers.util import parse_title_and_version
26
27if TYPE_CHECKING:
28 from collections.abc import Iterable
29
30 from music_assistant_models.media_items import MediaItemType
31
32 from music_assistant import MusicAssistant
33
34
35@dataclass(slots=True)
36class RecencyWindows:
37 """Lookback windows (in seconds) for the recency engine; None/0 disables that dimension."""
38
39 song_seconds: int | None = None
40 artist_seconds: int | None = None
41 duplicate_gap_seconds: int | None = None
42
43 @property
44 def song_lookback(self) -> int | None:
45 """Furthest-back we must read track rows: the larger of song window and duplicate gap."""
46 candidates = [value for value in (self.song_seconds, self.duplicate_gap_seconds) if value]
47 return max(candidates) if candidates else None
48
49
50@dataclass(slots=True)
51class RecencySnapshot:
52 """In-memory last-played lookup built from a single playlog query."""
53
54 now: int
55 song_ts: dict[tuple[str, str], int] = field(default_factory=dict)
56 song_key_ts: dict[tuple[str, str], int] = field(default_factory=dict)
57 artist_ts: dict[str, int] = field(default_factory=dict)
58
59 def track_recent(self, item: MediaItemType, within_seconds: int | None) -> bool:
60 """
61 Return True if the track was last played within the given window.
62
63 :param item: The track (or queue media item) to test.
64 :param within_seconds: The lookback window in seconds; None/0 disables the test.
65 """
66 if not within_seconds:
67 return False
68 timestamp = self.last_played(item)
69 return timestamp is not None and timestamp >= self.now - within_seconds
70
71 def last_played(self, item: MediaItemType) -> int | None:
72 """
73 Return the most recent play timestamp for the track, or None if it has no play recorded.
74
75 Matched across the track's own ``(provider, item_id)`` and all of its provider mappings
76 (exact), plus the fuzzy same-song keys so a different release/version of the same
77 recording is recognized too.
78
79 :param item: The track (or queue media item) to look up.
80 """
81 timestamps = []
82 timestamp = self.song_ts.get((item.provider, item.item_id))
83 if timestamp is not None:
84 timestamps.append(timestamp)
85 for mapping in getattr(item, "provider_mappings", None) or ():
86 timestamp = self.song_ts.get((mapping.provider_instance, mapping.item_id))
87 if timestamp is not None:
88 timestamps.append(timestamp)
89 for key in song_keys(item):
90 timestamp = self.song_key_ts.get(key)
91 if timestamp is not None:
92 timestamps.append(timestamp)
93 return max(timestamps) if timestamps else None
94
95 def artist_recent(self, name: str, within_seconds: int | None) -> bool:
96 """
97 Return True if the (named) artist was last played within the given window.
98
99 :param name: The artist name, matched case-insensitively.
100 :param within_seconds: The lookback window in seconds; None/0 disables the test.
101 """
102 if not within_seconds:
103 return False
104 timestamp = self.artist_ts.get(name.lower())
105 return timestamp is not None and timestamp >= self.now - within_seconds
106
107
108class RecencyEngine:
109 """Builds RecencySnapshots from the playlog for recency-aware features."""
110
111 def __init__(self, mass: MusicAssistant) -> None:
112 """Initialize the recency engine."""
113 self.mass = mass
114
115 async def snapshot(
116 self,
117 windows: RecencyWindows,
118 *,
119 userid: str | None = None,
120 include_partially_played: bool = False,
121 ) -> RecencySnapshot:
122 """
123 Build a recency snapshot from a single batched, user-scoped playlog query.
124
125 :param windows: The lookback windows to read.
126 :param userid: The user whose play history to read; falls back to the current user.
127 :param include_partially_played: Include tracks that were stopped before completion.
128 """
129 now = int(time.time())
130 snapshot = RecencySnapshot(now=now)
131 clauses: list[str] = []
132 params: dict[str, int | str] = {}
133 if song_lookback := windows.song_lookback:
134 clauses.append(
135 f"(media_type = '{MediaType.TRACK.value}' AND timestamp >= :song_cutoff)"
136 )
137 params["song_cutoff"] = now - song_lookback
138 if windows.artist_seconds:
139 clauses.append(
140 f"(media_type = '{MediaType.ARTIST.value}' AND timestamp >= :artist_cutoff)"
141 )
142 params["artist_cutoff"] = now - windows.artist_seconds
143 if not clauses:
144 return snapshot
145 where = f"({' OR '.join(clauses)})"
146 if not include_partially_played:
147 # Artist credit rows are always fully played, so this only affects track recency.
148 where = f"fully_played = 1 AND {where}"
149 if not userid and (user := get_current_user()):
150 userid = user.user_id
151 if userid:
152 where = f"{where} AND userid = :userid"
153 params["userid"] = userid
154 # one row per item is guaranteed when scoped to a user (unique playlog index); the
155 # MAX(timestamp) group keeps the most recent play when no user scope is applied.
156 query = (
157 f"SELECT item_id, provider, media_type, name, artists, MAX(timestamp) AS ts "
158 f"FROM {DB_TABLE_PLAYLOG} WHERE {where} "
159 f"GROUP BY item_id, provider, media_type"
160 )
161 for row in await self.mass.music.database.get_rows_from_query(
162 query, params=params, limit=0
163 ):
164 timestamp = int(row["ts"])
165 if row["media_type"] == MediaType.TRACK.value:
166 snapshot.song_ts[(row["provider"], row["item_id"])] = timestamp
167 for key in _song_keys(row["name"], _row_artist_names(row["artists"])):
168 if timestamp > snapshot.song_key_ts.get(key, 0):
169 snapshot.song_key_ts[key] = timestamp
170 elif name := row["name"]:
171 snapshot.artist_ts[name.lower()] = timestamp
172 return snapshot
173
174
175def song_keys(item: MediaItemType) -> set[tuple[str, str]]:
176 """
177 Return the fuzzy same-song identity keys for a track: (safe title, safe artist) per artist.
178
179 Version/featuring info is stripped from the title so different releases of the same recording
180 (remaster, single vs album edit, differing artist credits across providers) produce
181 overlapping keys. Returns an empty set for items without artists (non-track media).
182
183 :param item: The track (or queue media item) to build keys for.
184 """
185 artist_names = [artist.name for artist in getattr(item, "artists", None) or () if artist.name]
186 return _song_keys(item.name, artist_names)
187
188
189def _song_keys(name: str, artist_names: Iterable[str]) -> set[tuple[str, str]]:
190 """Build (safe title, safe artist) keys from a raw title and artist names."""
191 if not name:
192 return set()
193 title, _ = parse_title_and_version(name, strip_for_search=True)
194 if not (safe_title := create_safe_string(title)):
195 return set()
196 return {
197 (safe_title, safe_artist)
198 for artist_name in artist_names
199 if (safe_artist := create_safe_string(artist_name))
200 }
201
202
203def _row_artist_names(raw_artists: str | None) -> list[str]:
204 """Extract artist names from a playlog row's artists json column (None for legacy rows)."""
205 if not raw_artists:
206 return []
207 try:
208 return [name for artist in json_loads(raw_artists) if (name := artist.get("name"))]
209 except ValueError, TypeError, AttributeError:
210 return []
211