-
Notifications
You must be signed in to change notification settings - Fork 1
/
cached.py
143 lines (109 loc) · 4.08 KB
/
cached.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
"""Classes for cached playlists."""
import json
from typing import List
from categories import CATEGORIES
class CachedPlaylist:
id: str
name: str
track_ids: List[str]
def __init__(self, playlist_id, name):
self.id = playlist_id
self.name = name
self.track_ids = []
def __len__(self):
return len(self.track_ids)
@classmethod
def from_playlist_id(cls, playlist_id, spotify, expected_name=None):
playlist = spotify.playlist(playlist_id)
if expected_name and expected_name != playlist.name:
raise RuntimeError("Expected playlist name {expected_name}, but "
"actual name is {playlist.name}")
return cls.from_tekore_playlist(playlist, spotify)
@classmethod
def from_tekore_playlist(cls, playlist, spotify):
obj = cls(playlist.id, playlist.name)
if hasattr(playlist.tracks, "items"):
items = spotify.all_items(playlist.tracks)
else:
items = spotify.all_items(spotify.playlist_items(playlist.id))
obj.track_ids = [item.track.id for item in items if item.track.id is not None]
return obj
@classmethod
def from_cached_dict(cls, data):
obj = cls(data['id'], data['name'])
obj.track_ids = data['track_ids']
return obj
def contains_track(self, track):
return self.contains_track_id(track.id)
def contains_track_id(self, track_id):
return track_id in self.track_ids
def add_track_id(self, track_id):
self.track_ids.append(track_id)
def serialize(self):
return {
'id': self.id,
'name': self.name,
'track_ids': self.track_ids,
}
class CachedPlaylistGroup:
playlists: List[CachedPlaylist]
def __init__(self):
self.playlists = []
def __iter__(self):
return iter(self.playlists)
def add_from_filename(self, filename):
fp = open(filename)
self.add_from_file(fp)
fp.close()
def add_from_file(self, fp):
objs = json.load(fp)
self.playlists.extend(CachedPlaylist.from_cached_dict(obj) for obj in objs)
@classmethod
def from_filename(cls, filename):
group = cls()
group.add_from_filename(filename)
return group
@classmethod
def from_filenames(cls, filenames):
group = cls()
for filename in filenames:
group.add_from_filename(filename)
return group
@classmethod
def from_file(cls, fp):
group = cls()
group.add_from_file(fp)
return group
def playlists_containing_track(self, track_id):
return [playlist for playlist in self.playlists if track_id in playlist.track_ids]
def playlists_containing_track_str(self, track_id, sep=", ", remove_prefix="WCS "):
playlists = self.playlists_containing_track(track_id)
names = [playlist.name for playlist in playlists]
if remove_prefix:
nchars = len(remove_prefix)
names = [name[nchars:] if name.startswith(remove_prefix) else name for name in names]
return sep.join(names)
def add_playlist(self, playlist):
self.playlists.append(playlist)
def add_playlists(self, playlists):
self.playlists.extend(playlists)
def remove_playlist(self, playlist_id):
self.playlists = [p for p in self.playlists if p.id != playlist_id]
def contains_playlist_id(self, playlist_id):
"""Checks only the ID, not the entire object."""
return playlist_id in [p.id for p in self.playlists]
def playlist_by_name(self, name, allow_prefix="WCS "):
for playlist in self.playlists:
if playlist.name == name:
return playlist
if playlist.name == allow_prefix + name:
return playlist
else:
return None
def serialize(self):
return [obj.serialize() for obj in self.playlists]
def all_cached_playlists():
group = CachedPlaylistGroup()
for filename in CATEGORIES.keys():
group.add_from_filename(filename)
return group