/
/
/
1"""MusicBrainz-based recommendations for Music Assistant."""
2
3from __future__ import annotations
4
5from datetime import datetime, time, timedelta
6from typing import TYPE_CHECKING
7
8from music_assistant_models.enums import ArtistEntityType, ExternalID, RecommendationFolderType
9from music_assistant_models.media_items import (
10 Artist,
11 RecommendationFolder,
12 UniqueList,
13)
14
15from music_assistant.helpers.datetime import utc as datetime_utc
16
17if TYPE_CHECKING:
18 from music_assistant_models.media_items import BrowseFolder, ItemMapping, MediaItemType
19
20 from .provider import MusicbrainzProvider
21
22# Cache key for the precomputed matches (namespaced per provider instance).
23RECOMMENDATIONS_CACHE_KEY = "artist_timeline_recommendations_v2"
24TIMELINE_FOLDER_ID = "musicbrainz_timeline"
25
26
27class MusicBrainzRecommendationManager:
28 """Manages MusicBrainz-based recommendations (birthdays and memorials)."""
29
30 def __init__(self, provider: MusicbrainzProvider) -> None:
31 """Initialize recommendation manager."""
32 self.provider = provider
33 self.logger = provider.logger
34 self.mass = provider.mass
35 self._refresh_task_id = f"{provider.instance_id}_recommendations_refresh"
36
37 async def get_recommendations(self) -> list[RecommendationFolder]:
38 """
39 Return recommendation folder metadata without items.
40
41 Fast cache lookup that returns only the folder structure. Items are fetched
42 separately via get_recommendation_items.
43 """
44 artist_dicts = await self.mass.cache.get(
45 RECOMMENDATIONS_CACHE_KEY, provider=self.provider.instance_id, default=None
46 )
47 if artist_dicts is not None:
48 return [self._build_folder_metadata()] if artist_dicts else []
49 # Nothing fresh: refresh in the background and serve stale data if we have any.
50 self.schedule_refresh()
51 stale = await self.mass.cache.get(
52 RECOMMENDATIONS_CACHE_KEY,
53 provider=self.provider.instance_id,
54 allow_expired_cache=True,
55 default=None,
56 )
57 if stale is not None:
58 return [self._build_folder_metadata()] if stale else []
59 return []
60
61 async def get_recommendation_items(
62 self, item_id: str
63 ) -> UniqueList[MediaItemType | ItemMapping | BrowseFolder]:
64 """
65 Return artists for the musicbrainz_timeline recommendation folder.
66
67 :param item_id: The folder ID (expected to be TIMELINE_FOLDER_ID).
68 """
69 if item_id != TIMELINE_FOLDER_ID:
70 return UniqueList()
71 # Serve from cache (fresh or stale)
72 artist_dicts = await self.mass.cache.get(
73 RECOMMENDATIONS_CACHE_KEY,
74 provider=self.provider.instance_id,
75 allow_expired_cache=True,
76 default=None,
77 )
78 if artist_dicts is None:
79 return UniqueList()
80 return UniqueList([Artist.from_dict(d) for d in artist_dicts])
81
82 def schedule_refresh(self) -> None:
83 """Scan the library and refresh the cached matches in the background (deduplicated)."""
84 self.mass.create_task(self._refresh(), task_id=self._refresh_task_id)
85
86 def cancel(self) -> None:
87 """Cancel any pending background refresh (called on provider unload)."""
88 self.mass.cancel_task(self._refresh_task_id)
89
90 # ------------------------------------------------------------------
91 # background refresh
92 # ------------------------------------------------------------------
93
94 async def _refresh(self) -> None:
95 """Scan the library once and cache matched artists as serialized dicts."""
96 try:
97 artists = await self._scan_matches()
98 except Exception as err:
99 self.logger.warning("Failed to compute MusicBrainz recommendations: %s", err)
100 return
101 # Expire at the next UTC midnight so the fresh path recomputes daily; the entry
102 # survives cleanup (allow_expired_cache) so it can be served as stale fallback.
103 await self.mass.cache.set(
104 RECOMMENDATIONS_CACHE_KEY,
105 [a.to_dict() for a in artists],
106 expiration=self._seconds_until_next_utc_midnight() + 60,
107 provider=self.provider.instance_id,
108 allow_expired_cache=True,
109 )
110
111 async def _scan_matches(self) -> list[Artist]:
112 """
113 Scan the library for artists whose birth/death/founding/disbanding date falls in the window.
114
115 Returns matched artists with their metadata populated (life_span and artist_entity_type)
116 so the frontend can compute event type and date offset independently.
117 """
118 days_before_after = self._days_window()
119 self.logger.info(
120 "MusicBrainz recommendations: scanning %d days before/after today",
121 days_before_after,
122 )
123 window = self._window_dates()
124
125 matched: list[Artist] = []
126 scanned = 0
127 async for artist in self.mass.music.artists.iter_library_items(order_by="name"):
128 mbid = artist.get_external_id(ExternalID.MB_ARTIST)
129 if not mbid:
130 continue
131 scanned += 1
132 last_refresh = artist.metadata.last_refresh if artist.metadata else None
133 if last_refresh is None:
134 self.mass.metadata.schedule_update_metadata(artist)
135 continue
136 life_span = artist.metadata.life_span if artist.metadata else None
137 if not life_span:
138 continue
139 entity_type = artist.metadata.artist_entity_type if artist.metadata else None
140 # Only process known artist types with date-based events (whitelist approach)
141 if entity_type not in (
142 ArtistEntityType.PERSON,
143 ArtistEntityType.GROUP,
144 ArtistEntityType.ORCHESTRA,
145 ArtistEntityType.CHOIR,
146 ):
147 continue
148 begin = life_span.begin
149 end = life_span.end
150 # Only full "YYYY-MM-DD" dates are usable; partial dates like "1990" are skipped
151 birth_in_window = begin and len(begin) >= 10 and begin[5:10] in window
152 death_in_window = life_span.ended and end and len(end) >= 10 and end[5:10] in window
153 if birth_in_window or death_in_window:
154 matched.append(artist)
155
156 self.logger.debug(
157 "Scanned %d library artist(s) with MB IDs, %d matched", scanned, len(matched)
158 )
159 return matched
160
161 # ------------------------------------------------------------------
162 # folder building
163 # ------------------------------------------------------------------
164
165 def _build_folder_metadata(self) -> RecommendationFolder:
166 """Build recommendation folder metadata without items."""
167 return RecommendationFolder(
168 item_id=TIMELINE_FOLDER_ID,
169 name="Artist Events",
170 provider=self.provider.instance_id,
171 translation_key="artist_timeline",
172 items=UniqueList(),
173 is_playable=False,
174 type=RecommendationFolderType.TIMELINE,
175 enabled_by_default=False,
176 )
177
178 # ------------------------------------------------------------------
179 # helpers
180 # ------------------------------------------------------------------
181
182 def _window_dates(self) -> set[str]:
183 """Return the set of MM-DD strings within the current scan window."""
184 today = datetime_utc().date()
185 days_before_after = self._days_window()
186 return {
187 f"{(today + timedelta(days=offset)).month:02d}-{(today + timedelta(days=offset)).day:02d}"
188 for offset in range(-days_before_after, days_before_after + 1)
189 }
190
191 def _days_window(self) -> int:
192 """Return the validated number of days to scan before/after today."""
193 days_config = self.provider.config.get_value("recommendation_days", 3)
194 try:
195 days_before_after = int(str(days_config))
196 except TypeError, ValueError:
197 days_before_after = 3
198 return max(1, min(15, days_before_after))
199
200 def _seconds_until_next_utc_midnight(self) -> int:
201 """Return the number of seconds until the next UTC midnight (at least 60)."""
202 now = datetime_utc()
203 next_midnight = datetime.combine(
204 now.date() + timedelta(days=1), time.min, tzinfo=now.tzinfo
205 )
206 return max(60, int((next_midnight - now).total_seconds()))
207