music-assistant-server
6.7 KB•PY
api_response_converter_mapping.py
6.7 KB • 203 lines • python
1"""API type to converter function mappings."""
2
3from __future__ import annotations
4
5from collections.abc import Callable
6from dataclasses import dataclass
7from typing import TYPE_CHECKING, cast
8
9from mashumaro import DataClassDictMixin
10from niconico.objects.nvapi import (
11 FollowingMylistsData,
12 HistoryData,
13 LikeHistoryData,
14 ListSearchData,
15 OwnVideosData,
16 RecommendData,
17 RelationshipUsersData,
18 SeriesData,
19 UserVideosData,
20 VideoSearchData,
21)
22from niconico.objects.user import NicoUser, UserMylistItem, UserSeriesItem
23from niconico.objects.video import EssentialVideo, Mylist
24from niconico.objects.video.watch import WatchData
25from pydantic import BaseModel
26
27from music_assistant.providers.nicovideo.converters.stream import (
28 StreamConversionData,
29)
30from tests.providers.nicovideo.fixture_data.shared_types import StreamFixtureData
31
32if TYPE_CHECKING:
33 from music_assistant.providers.nicovideo.converters.manager import NicovideoConverterManager
34
35
36# Type definitions for converter results
37type SnapshotableItem = DataClassDictMixin
38type ConvertedResult = SnapshotableItem | list[SnapshotableItem] | None
39
40
41@dataclass(frozen=True)
42class APIResponseConverterMapping[T: BaseModel]:
43 """Maps API type to converter function."""
44
45 source_type: type[T]
46 convert_func: Callable[[T, NicovideoConverterManager], ConvertedResult]
47
48
49# API type to converter function mappings
50API_RESPONSE_CONVERTER_MAPPINGS = (
51 # Track Types
52 APIResponseConverterMapping(
53 source_type=EssentialVideo,
54 convert_func=lambda data, cm: cm.track.convert_by_essential_video(data),
55 ),
56 APIResponseConverterMapping(
57 source_type=WatchData,
58 convert_func=lambda data, cm: cm.track.convert_by_watch_data(data),
59 ),
60 APIResponseConverterMapping(
61 source_type=UserVideosData,
62 convert_func=lambda data, cm: [
63 track
64 for item in data.items
65 if (track := cm.track.convert_by_essential_video(item.essential)) is not None
66 ],
67 ),
68 APIResponseConverterMapping(
69 source_type=OwnVideosData,
70 convert_func=lambda data, cm: [
71 track
72 for item in data.items
73 if (track := cm.track.convert_by_essential_video(item.essential)) is not None
74 ],
75 ),
76 # Playlist Types
77 APIResponseConverterMapping(
78 source_type=Mylist,
79 convert_func=lambda data, cm: cm.playlist.convert_with_tracks_by_mylist(data),
80 ),
81 APIResponseConverterMapping(
82 source_type=UserMylistItem,
83 convert_func=lambda data, cm: cm.playlist.convert_by_mylist(data),
84 ),
85 APIResponseConverterMapping(
86 source_type=FollowingMylistsData,
87 convert_func=lambda data, cm: [
88 cm.playlist.convert_following_by_mylist(item) for item in data.mylists
89 ],
90 ),
91 # Album Types
92 APIResponseConverterMapping(
93 source_type=SeriesData,
94 convert_func=lambda data, cm: cm.album.convert_series_to_album_with_tracks(data),
95 ),
96 APIResponseConverterMapping(
97 source_type=UserSeriesItem,
98 convert_func=lambda data, cm: cm.album.convert_by_series(data),
99 ),
100 # Artist Types
101 APIResponseConverterMapping(
102 source_type=RelationshipUsersData,
103 convert_func=lambda data, cm: [
104 cm.artist.convert_by_owner_or_user(item) for item in data.items
105 ],
106 ),
107 APIResponseConverterMapping(
108 source_type=NicoUser,
109 convert_func=lambda data, cm: cm.artist.convert_by_owner_or_user(data),
110 ),
111 # Search Types
112 APIResponseConverterMapping(
113 source_type=VideoSearchData,
114 convert_func=lambda data, cm: [
115 track
116 for item in data.items
117 if (track := cm.track.convert_by_essential_video(item)) is not None
118 ],
119 ),
120 APIResponseConverterMapping(
121 source_type=ListSearchData,
122 convert_func=lambda data, cm: [
123 cm.playlist.convert_by_mylist(item)
124 if item.type_ == "mylist"
125 else cm.album.convert_by_series(item)
126 for item in data.items
127 ],
128 ),
129 # History Types
130 APIResponseConverterMapping(
131 source_type=HistoryData,
132 convert_func=lambda data, cm: [
133 track
134 for item in data.items
135 if (track := cm.track.convert_by_essential_video(item.video)) is not None
136 ],
137 ),
138 APIResponseConverterMapping(
139 source_type=LikeHistoryData,
140 convert_func=lambda data, cm: [
141 track
142 for item in data.items
143 if (track := cm.track.convert_by_essential_video(item.video)) is not None
144 ],
145 ),
146 # Recommendation Types
147 APIResponseConverterMapping(
148 source_type=RecommendData,
149 convert_func=lambda data, cm: [
150 track
151 for item in data.items
152 if isinstance(item.content, EssentialVideo)
153 and (track := cm.track.convert_by_essential_video(item.content)) is not None
154 ],
155 ),
156 # Stream Types
157 APIResponseConverterMapping(
158 source_type=StreamConversionData,
159 convert_func=lambda data, cm: cm.stream.convert_from_conversion_data(data),
160 ),
161 APIResponseConverterMapping(
162 source_type=StreamFixtureData,
163 convert_func=lambda data, cm: cm.stream.convert_from_conversion_data(
164 StreamConversionData(
165 watch_data=data.watch_data,
166 selected_audio=data.selected_audio,
167 hls_url="https://example.com/stub.m3u8",
168 domand_bid="stub_bid",
169 hls_playlist_text=(
170 "#EXTM3U\n"
171 "#EXT-X-VERSION:6\n"
172 "#EXT-X-TARGETDURATION:6\n"
173 "#EXT-X-MEDIA-SEQUENCE:1\n"
174 "#EXT-X-PLAYLIST-TYPE:VOD\n"
175 '#EXT-X-MAP:URI="https://example.com/init.mp4"\n'
176 '#EXT-X-KEY:METHOD=AES-128,URI="https://example.com/key"\n'
177 "#EXTINF:6.0,\n"
178 "segment1.m4s\n"
179 "#EXTINF:6.0,\n"
180 "segment2.m4s\n"
181 "#EXT-X-ENDLIST\n"
182 ),
183 )
184 ),
185 ),
186)
187
188
189class APIResponseConverterMappingRegistry:
190 """Maps API response types to converter functions."""
191
192 def __init__(self) -> None:
193 """Initialize the registry."""
194 self._registry: dict[type, APIResponseConverterMapping[BaseModel]] = {}
195 for mapping in API_RESPONSE_CONVERTER_MAPPINGS:
196 self._registry[mapping.source_type] = cast(
197 "APIResponseConverterMapping[BaseModel]", mapping
198 )
199
200 def get_by_type(self, source_type: type) -> APIResponseConverterMapping[BaseModel] | None:
201 """Get mapping by type with O(1) lookup."""
202 return self._registry.get(source_type)
203