-
Notifications
You must be signed in to change notification settings - Fork 0
/
utilities.py
165 lines (138 loc) · 5.56 KB
/
utilities.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
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
""" StreamSort utility functions
Copyright (c) 2020 IdmFoundInHim, except where otherwise credited
"""
from collections.abc import Iterator, Mapping
from urllib import parse as urlparse
from more_itertools import flatten
import requests
from spotipy import Spotify, SpotifyPKCE
from .constants import (MOBNAMES, MOB_URI_PREFIX, MOB_URL_PREFIX,
SPID_VALID_CHARS)
from .musictypes import Mob
def get_header(oauth: str) -> dict:
""" Returns header with given oauth for the Spotify API """
return {
"Accept": "application/json",
"Content-Type": "application/json",
"Authorization": f"Bearer {oauth}",
}
def results_generator(auth: SpotifyPKCE, page_zero: Mapping) -> Iterator[Mob]:
""" Cycles through multi-page responses from Spotify """
try:
yield from page_zero['items']
if not page_zero['next']:
# Included seperately from loop condition because it avoids
# unnecessary token access (and undesired indentation)
return
except KeyError as err:
raise ValueError from err
page = page_zero
header = get_header(auth.get_access_token())
while nexturl := page['next']:
try:
page_r = requests.get(nexturl, headers=header)
page_r.raise_for_status()
page = page_r.json()
yield from page['items']
except requests.exceptions.HTTPError:
if 'offset=1000' in nexturl:
return
header = get_header(auth.get_access_token(check_cache=False))
page = {'next': nexturl}
except KeyError:
key = next((n + 's' for n in MOBNAMES if n + 's' in page), 'items')
page = page[key]
yield from page['items']
def as_uri(uri: str):
""" Returns the URI in standard format only if present """
if uri.startswith(MOB_URL_PREFIX):
try:
url = urlparse.urlparse(uri)
except ValueError:
return ''
uri = MOB_URI_PREFIX + ':'.join(url.path.split('/')[-2:])
uri_parts = uri.strip().split(':')
if (len(uri_parts) == 3
and uri_parts[0] == 'spotify'
and uri_parts[1] in MOBNAMES
and all(c in SPID_VALID_CHARS for c in uri_parts[2])):
return uri
return ''
def _track_in_mob(auth: SpotifyPKCE, track: Mob, mob: Mob) -> bool:
if mob['type'] == 'artist':
return mob['uri'] in (a['uri'] for a in track['artists'])
if mob['type'] == 'album':
return track['uri'] in (t['uri'] for t
in results_generator(auth,
mob['tracks']))
if mob['type'] == 'playlist':
return (track['uri']
in (t['track']['uri'] for t
in results_generator(auth, mob['tracks'])))
return False
def _album_in_mob(auth: SpotifyPKCE, album: Mob, mob: Mob) -> bool:
if mob['type'] == 'artist':
return mob['uri'] in (a['uri'] for a in album['artists'])
if mob['type'] == 'playlist':
return (album['uri'] in
(t['track']['album']['uri'] for t
in results_generator(auth, mob['tracks'])))
return False
def _artist_in_mob(auth: SpotifyPKCE, artist: Mob, mob: Mob) -> bool:
if mob['type'] == 'track':
return artist['uri'] in (a['uri'] for a in mob['artists'])
if mob['type'] == 'album':
return (artist['uri'] in
(a['uri'] for a in
flatten(t['artists'] for t
in results_generator(auth, mob['tracks']))))
if mob['type'] == 'playlist':
return (artist['uri'] in
(a['uri'] for a in
flatten(t['track']['artists'] for t
in results_generator(auth, mob['tracks']))))
return False
def _playlist_in_playlist(auth: SpotifyPKCE, sub: Mob, lst: Mob) -> bool:
if lst['type'] != 'playlist':
return False
lst_gen = (o['track'] for o in results_generator(auth, lst['tracks']))
sub_gen = (o['track'] for o in results_generator(auth, sub['tracks']))
sub_gen_first = next(sub_gen, {'uri': None})['uri']
for track in lst_gen:
if track['uri'] == sub_gen_first:
break
for track in sub_gen:
if next(lst_gen, {'uri': None})['uri'] != track['uri']:
return False
return True
_MOB_SPECIFIC_TESTS = {
'track': _track_in_mob,
'album': _album_in_mob,
'artist': _artist_in_mob,
'playlist': _playlist_in_playlist
}
def mob_in_mob(api: Spotify, obj: Mob, lst: Mob) -> bool:
""" Check if a mob is found in another mob
All items contain themselves in addition to anything else.
Albums are considered to contain their tracks + any
artists credited on those tracks. Artists are considered to contain
any tracks they are credited on + any albums or playlists
containing their tracks. Playlists contain their tracks + any
albums and artists represented in those tracks.
TODO doctests here
Episodes and non-mobs match nothing, but throw no error.
TODO that especially needs doctesting
"""
if obj['uri'] == lst['uri']:
return True
if test := _MOB_SPECIFIC_TESTS.get(obj['type']):
return test(api.auth_manager, obj, lst)
return False
def iter_mob(auth: SpotifyPKCE, mob: Mob) -> Iterator[str]:
if tracks := mob.get('tracks', mob.get('episodes')):
mob_tracks = results_generator(auth, tracks)
else:
mob_tracks = [mob]
if mob['type'] == 'playlist':
mob_tracks = (t['track'] for t in mob_tracks)
yield from (t['uri'] for t in mob_tracks)