Skip to content

Commit

Permalink
refactor/ocp_modernize (#7)
Browse files Browse the repository at this point in the history 
* refactor/ocp_modernize

return the new OCP MediaEntry/Playlist objects instead of needing to know a json structure

* fix/allow_ocp_objects

MediaEntry/Playlist/PluginStream objects should be allowed in ocp decorators

* utils 0.0.38 compat
  • Loading branch information
@JarbasAl
JarbasAl authored Jun 21, 2024
1 parent d96ec62 commit e49fb5f
Show file tree
Hide file tree
Showing 3 changed files with 93 additions and 112 deletions.
4 changes: 2 additions & 2 deletions requirements.txt
View file
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
ovos-utils >= 0.1.0a7
ovos-utils >= 0.0.38
ovos-bus-client>=0.0.9a2
ovos-workshop>=0.0.16a3
ovos-workshop>=0.0.16a39
ovos-backend-client>=0.1.1a5
spotipy
199 changes: 90 additions & 109 deletions skill_spotify/__init__.py
View file
Original file line number Diff line number Diff line change
@@ -1,9 +1,8 @@
from os.path import join, dirname

from typing import Iterable, Union
from ovos_utils import classproperty
from ovos_utils.ocp import MediaType, PlaybackType
from ovos_workshop.backwards_compat import MediaType, PlaybackType, MediaEntry, Playlist
from ovos_utils.process_utils import RuntimeRequirements

from ovos_workshop.decorators.ocp import ocp_search
from ovos_workshop.skills.common_play import OVOSCommonPlaybackSkill
from skill_spotify.spotify import SpotifyClient
Expand All @@ -20,127 +19,109 @@ def runtime_requirements(self):
return RuntimeRequirements(internet_before_load=True,
requires_internet=True)

def search_artists(self, query):
def search_artists(self, query) -> Iterable[Playlist]:
score, data = self.spotify.query_artist(query)

for artist in data["data"]["artists"]["items"]:

uri = artist["uri"]
playlist = []

playlist = Playlist(
title=artist["name"] + " (Featured Tracks)",
image=artist["images"][-1]["url"] if artist["images"] else "",
match_confidence=score,
media_type=MediaType.MUSIC,
playback=PlaybackType.AUDIO,
skill_id=self.skill_id,
skill_icon=self.skill_icon
)
for t in self.spotify.tracks_from_artist(uri):
playlist.append({
"title": t["name"],
"duration": t["duration_ms"] / 1000,
"artist": artist["name"],
"match_confidence": score,
"media_type": MediaType.MUSIC,
"uri": t["uri"],
"playback": PlaybackType.AUDIO,
"skill_icon": self.skill_icon,
"skill_id": self.skill_id,
"image": artist["images"][-1]["url"] if artist["images"] else "",
"bg_image": artist["images"][0]["url"] if artist["images"] else ""
})

entry = {
"match_confidence": score,
"media_type": MediaType.MUSIC,
"playlist": playlist[:25],
"playback": PlaybackType.AUDIO,
"skill_icon": self.skill_icon,
"skill_id": self.skill_id,
"image": artist["images"][-1]["url"] if artist["images"] else "",
"bg_image": artist["images"][0]["url"] if artist["images"] else "",
"title": artist["name"] + " (Featured Tracks)"
}
yield entry
playlist.append(MediaEntry(media_type=MediaType.MUSIC,
uri=t["uri"],
title=t["name"],
playback=PlaybackType.AUDIO,
image=artist["images"][-1]["url"] if artist["images"] else "",
skill_id=self.skill_id,
artist=artist["name"],
match_confidence=min(100, score),
length=t["duration_ms"] / 1000,
skill_icon=self.skill_icon))
if len(playlist) > 25:
break
yield playlist

def search_albums(self, query):
def search_albums(self, query) -> Iterable[Playlist]:
score, data = self.spotify.query_album(query)

for album in data["data"]["albums"]["items"]:
uri = album["uri"]
playlist = []
playlist = Playlist(
title=album["name"] + " (Full Album)",
image=album["images"][-1]["url"] if album["images"] else "",
match_confidence=score,
media_type=MediaType.MUSIC,
playback=PlaybackType.AUDIO,
skill_id=self.skill_id,
skill_icon=self.skill_icon
)
for t in self.spotify.tracks_from_album(uri):
artist = t["artists"][0]
playlist.append({
"title": t["name"],
"duration": t["duration_ms"] / 1000,
"artist": artist["name"],
"match_confidence": score,
"media_type": MediaType.MUSIC,
"uri": t["uri"],
"playback": PlaybackType.AUDIO,
"skill_icon": self.skill_icon,
"skill_id": self.skill_id,
"image": album["images"][-1]["url"] if album["images"] else "",
"bg_image": album["images"][0]["url"] if album["images"] else ""
})
entry = {
"match_confidence": score,
"media_type": MediaType.MUSIC,
"playlist": playlist[:25],
"playback": PlaybackType.AUDIO,
"skill_icon": self.skill_icon,
"skill_id": self.skill_id,
"image": album["images"][-1]["url"] if album["images"] else "",
"bg_image": album["images"][0]["url"] if album["images"] else "",
"title": album["name"] + " (Full Album)"
}
yield entry
playlist.append(MediaEntry(media_type=MediaType.MUSIC,
uri=t["uri"],
title=t["name"],
playback=PlaybackType.AUDIO,
image=album["images"][-1]["url"] if album["images"] else "",
skill_id=self.skill_id,
artist=artist["name"],
match_confidence=min(100, score),
length=t["duration_ms"] / 1000,
skill_icon=self.skill_icon))
if len(playlist) > 25:
break
yield playlist

def search_tracks(self, query):
def search_tracks(self, query) -> Iterable[MediaEntry]:
score, data = self.spotify.query_song(query)

for track in data["data"]["tracks"]["items"]:
album = track["album"]
entry = {
"title": track["name"],
"duration": track["duration_ms"] / 1000,
"match_confidence": score,
"media_type": MediaType.MUSIC,
"uri": track["uri"],
"playback": PlaybackType.AUDIO,
"skill_icon": self.skill_icon,
"skill_id": self.skill_id,
"image": album["images"][-1]["url"] if album["images"] else "",
"bg_image": album["images"][0]["url"] if album["images"] else ""
}
yield entry
yield MediaEntry(media_type=MediaType.MUSIC,
uri=track["uri"],
title=track["name"],
playback=PlaybackType.AUDIO,
image=album["images"][-1]["url"] if album["images"] else "",
skill_id=self.skill_id,
# artist=ch.artist,
match_confidence=min(100, score),
length=track["duration_ms"] / 1000,
skill_icon=self.skill_icon)

def search_playlists(self, query):
def search_playlists(self, query) -> Iterable[Playlist]:
data, score = self.spotify.get_best_user_playlist(query)
uri = data["uri"]
playlist = []
playlist = Playlist(
title=data["name"],
image=data["images"][-1]["url"] if data["images"] else "",
match_confidence=score,
media_type=MediaType.MUSIC,
playback=PlaybackType.AUDIO,
skill_id=self.skill_id,
skill_icon=self.skill_icon
)
for t in self.spotify.tracks_from_playlist(uri)["items"]:
t = t["track"]
artist = t["artists"][0]
playlist.append({
"title": t["name"],
"duration": t["duration_ms"] / 1000,
"artist": artist["name"],
"match_confidence": score,
"media_type": MediaType.MUSIC,
"uri": t["uri"],
"playback": PlaybackType.AUDIO,
"skill_icon": self.skill_icon,
"skill_id": self.skill_id,
"image": data["images"][-1]["url"] if data["images"] else "",
"bg_image": data["images"][0]["url"] if data["images"] else ""
})
entry = {
"match_confidence": score,
"media_type": MediaType.MUSIC,
"playlist": playlist[:25],
"playback": PlaybackType.AUDIO,
"skill_icon": self.skill_icon,
"skill_id": self.skill_id,
"image": data["images"][-1]["url"] if data["images"] else "",
"bg_image": data["images"][0]["url"] if data["images"] else "",
"title": data["name"]
}
yield entry
playlist.append(MediaEntry(media_type=MediaType.MUSIC,
uri=t["uri"],
title=t["name"],
playback=PlaybackType.AUDIO,
image=data["images"][-1]["url"] if data["images"] else "",
skill_id=self.skill_id,
artist=artist["name"],
match_confidence=min(100, score),
length=t["duration_ms"] / 1000,
skill_icon=self.skill_icon))
if len(playlist) > 25:
break
yield playlist

# multiple decorators are used to execute the search in parallel
@ocp_search()
Expand All @@ -152,8 +133,8 @@ def search_spotify_artist(self, phrase, media_type=MediaType.GENERIC):
base_score += 15
phrase = self.remove_voc(phrase, "spotify")
for res in self.search_artists(phrase):
res["match_confidence"] += base_score
res["match_confidence"] = min(100, res["match_confidence"])
res.match_confidence += base_score
res.match_confidence = min(100, res.match_confidence)
yield res

@ocp_search()
Expand All @@ -165,8 +146,8 @@ def search_spotify_album(self, phrase, media_type=MediaType.GENERIC):
base_score += 15
phrase = self.remove_voc(phrase, "spotify")
for res in self.search_albums(phrase):
res["match_confidence"] += base_score
res["match_confidence"] = min(100, res["match_confidence"])
res.match_confidence += base_score
res.match_confidence = min(100, res.match_confidence)
yield res

@ocp_search()
Expand All @@ -178,8 +159,8 @@ def search_spotify_tracks(self, phrase, media_type=MediaType.GENERIC):
base_score += 15
phrase = self.remove_voc(phrase, "spotify")
for res in self.search_tracks(phrase):
res["match_confidence"] += base_score
res["match_confidence"] = min(100, res["match_confidence"])
res.match_confidence += base_score
res.match_confidence = min(100, res.match_confidence)
yield res

@ocp_search()
Expand All @@ -191,8 +172,8 @@ def search_spotify_playlists(self, phrase, media_type=MediaType.GENERIC):
base_score += 15
phrase = self.remove_voc(phrase, "spotify")
for res in self.search_playlists(phrase):
res["match_confidence"] += base_score
res["match_confidence"] = min(100, res["match_confidence"])
res.match_confidence += base_score
res.match_confidence = min(100, res.match_confidence)
yield res


Expand Down
2 changes: 1 addition & 1 deletion skill_spotify/spotify.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@
import requests
import spotipy
from ovos_utils.log import LOG
from ovos_utils.ocp import PlaybackType, MediaType
from ovos_workshop.backwards_compat import PlaybackType, MediaType
from ovos_utils.parse import match_one, fuzzy_match
from requests.exceptions import HTTPError
from spotipy.oauth2 import SpotifyAuthBase
Expand Down

0 comments on commit e49fb5f

Please sign in to comment.