chore: initial
This commit is contained in:
unfaiyted
commit
6545d82095
|
|
@ -0,0 +1,8 @@
|
|||
# Default ignored files
|
||||
/shelf/
|
||||
/workspace.xml
|
||||
# Editor-based HTTP Client requests
|
||||
/httpRequests/
|
||||
# Datasource local storage ignored files
|
||||
/dataSources/
|
||||
/dataSources.local.xml
|
||||
|
|
@ -0,0 +1,9 @@
|
|||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<project version="4">
|
||||
<component name="Black">
|
||||
<option name="sdkName" value="Python 3.12 (emby-scripts)" />
|
||||
</component>
|
||||
<component name="ProjectRootManager">
|
||||
<output url="file://$PROJECT_DIR$/out" />
|
||||
</component>
|
||||
</project>
|
||||
|
|
@ -0,0 +1,8 @@
|
|||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<project version="4">
|
||||
<component name="ProjectModuleManager">
|
||||
<modules>
|
||||
<module fileurl="file://$PROJECT_DIR$/emby-scripts.iml" filepath="$PROJECT_DIR$/emby-scripts.iml" />
|
||||
</modules>
|
||||
</component>
|
||||
</project>
|
||||
|
|
@ -0,0 +1,9 @@
|
|||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<module type="PYTHON_MODULE" version="4">
|
||||
<component name="NewModuleRootManager" inherit-compiler-output="true">
|
||||
<exclude-output />
|
||||
<content url="file://$MODULE_DIR$" />
|
||||
<orderEntry type="jdk" jdkName="Python 3.12 (emby-scripts)" jdkType="Python SDK" />
|
||||
<orderEntry type="sourceFolder" forTests="false" />
|
||||
</component>
|
||||
</module>
|
||||
|
|
@ -0,0 +1 @@
|
|||
{"access_token": "BQCdJ8TcebCvCeo2VM75nwko_tshza2WqpKlhWK_EZufuP4NXOWEtaks1sUfnHlnJt258CbM2Hsx78XUNbu84_4u-uKEHgj51fPRJzAVg91O6MeSd-3ZpIUHbXLaNjnUVBS7ltPClDRHyCuaFaRW8BKlA_F-Nq-BrbPD4vgJnqboI3prb3BH4GTl1tfkInodNAfIog5oeReyHF8vy85KtrSugMZeSptlUInUd4Z-IZLqXg", "token_type": "Bearer", "expires_in": 3600, "scope": "playlist-read-private", "expires_at": 1715037521, "refresh_token": "AQAA4rKcyTFqaVctXWaxmE3d0LqLJu3ms91BNXrlJeYfy_1fYmCyQ3VyX6rO6K0O4iGyRVNWT8lcJB6MGRuC5rLXG85ZFB2XQtpcab6ECyKrIQNcMNU7AJj_eiZgfTwNtFk"}
|
||||
Binary file not shown.
Binary file not shown.
|
|
@ -0,0 +1 @@
|
|||
{"access_token": "BQCWbHB-kY3mUtocqK43wX8Ljl4wluFQnOIA9zCF3JlvKW0jtZHd65RJN_06wX5pRZdbrjeZfBQcE0--0o89f_lvI2tKiDjynhD4h44GruTSa4a8QFE9JkrdA27MdVJ5IlbPKoBQFpNOrYan6xmeTjme1YCARuF_tkfcy8E-9RcRKDAvH5eThu6Q41H-Rr4", "token_type": "Bearer", "expires_in": 3600, "refresh_token": "AQBij3uDeV9G6Z71rKFATWpcHJLhggDPYabbTEuvebdSKEqRY_5VqS98wkx3u037d7nBlaTHIQpeSqkbbOmLdIut56TZ3Yiva56R4t6PV22LSY0wUdTYfwSjP_nWr_jP1wU", "scope": "playlist-read-private", "expires_at": 1715044235}
|
||||
|
|
@ -0,0 +1,57 @@
|
|||
from src.clients.emby_client import EmbyClient
|
||||
from src.config import Config
|
||||
|
||||
# Create an instance of the Emby client
|
||||
emby = EmbyClient(Config.EMBY_URL, Config.EMBY_USERNAME, Config.EMBY_PASSWORD)
|
||||
|
||||
# Define an array of TV show names
|
||||
tv_shows = ["The Office (US)", "Parks and Recreation", "Bob's Burgers"]
|
||||
|
||||
# Create a new playlist
|
||||
playlist_name = "Bob's Office Park"
|
||||
playlist_type = "shows"
|
||||
playlist = emby.create_playlist(playlist_name, playlist_type)
|
||||
|
||||
# Get all the episodes for all of the TV shows in the array
|
||||
all_episodes = {}
|
||||
for show_name in tv_shows:
|
||||
show = emby.search(show_name, "Series")[0] # Assuming the first search result is the desired show
|
||||
seasons = emby.get_seasons(show["Id"])
|
||||
|
||||
all_episodes[show_name] = []
|
||||
for season in seasons:
|
||||
episodes = emby.get_episodes(show["Id"], season["Id"])
|
||||
all_episodes[show_name].extend(episodes)
|
||||
|
||||
# Sort the episodes of each show by season and episode number
|
||||
for show_name in all_episodes:
|
||||
all_episodes[show_name].sort(key=lambda x: (x["ParentIndexNumber"], x["IndexNumber"]))
|
||||
|
||||
# Find the maximum number of episodes among all shows
|
||||
max_episodes = max(len(episodes) for episodes in all_episodes.values())
|
||||
|
||||
# Initialize episode counters for each show
|
||||
episode_counters = {show_name: 0 for show_name in tv_shows}
|
||||
|
||||
# Initialize a set to keep track of added episode IDs
|
||||
added_episode_ids = set()
|
||||
|
||||
# Loop over the episodes and add them sequentially
|
||||
for i in range(max_episodes):
|
||||
for show_name in tv_shows:
|
||||
if episode_counters[show_name] < len(all_episodes[show_name]):
|
||||
episode = all_episodes[show_name][episode_counters[show_name]]
|
||||
episode_id = episode["Id"]
|
||||
|
||||
if episode_id not in added_episode_ids:
|
||||
emby.add_item_to_playlist(playlist["Id"], episode_id)
|
||||
print(f"Added episode: {show_name} - S{episode['ParentIndexNumber']:02}E{episode['IndexNumber']:02}")
|
||||
added_episode_ids.add(episode_id)
|
||||
else:
|
||||
print(f"Skipped duplicate episode: {show_name} - S{episode['ParentIndexNumber']:02}E{episode['IndexNumber']:02}")
|
||||
|
||||
episode_counters[show_name] += 1
|
||||
else:
|
||||
print(f"No more episodes for {show_name}")
|
||||
|
||||
print("Playlist creation completed.")
|
||||
|
|
@ -0,0 +1,113 @@
|
|||
import spotipy
|
||||
from spotipy.oauth2 import SpotifyOAuth
|
||||
import requests
|
||||
import logging
|
||||
from fuzzywuzzy import fuzz
|
||||
import re
|
||||
from src.config import Config
|
||||
from src.utils.string import StringUtils
|
||||
from src.clients.emby_client import EmbyClient
|
||||
from src.clients.spotify_client import SpotifyClient
|
||||
|
||||
# Configure logging
|
||||
logging.basicConfig(
|
||||
level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s"
|
||||
)
|
||||
|
||||
# Get the user's playlists from Spotify
|
||||
spot = SpotifyClient(Config.SPOTIFY_CLIENT_ID, Config.SPOTIFY_CLIENT_SECRET, Config.SPOTIFY_REDIRECT_URI,
|
||||
Config.SPOTIFY_SCOPE)
|
||||
emby = EmbyClient(Config.EMBY_URL, Config.EMBY_USERNAME, Config.EMBY_PASSWORD)
|
||||
|
||||
playlists = spot.get_playlists()
|
||||
|
||||
# Iterate over each Spotify playlist
|
||||
for playlist in playlists["items"]:
|
||||
playlist_name = playlist["name"]
|
||||
playlist_id = playlist["id"]
|
||||
playlist_owner = playlist["owner"]["display_name"]
|
||||
logging.info(f"Processing Spotify playlist: {playlist_name} ({playlist_owner})")
|
||||
|
||||
emby_playlist_name = f"{playlist_name} ({playlist_owner})"
|
||||
|
||||
emby_playlist_search_results = emby.search(emby_playlist_name, 'Playlist')
|
||||
|
||||
if emby_playlist_search_results:
|
||||
# Delete the existing playlist
|
||||
existing_playlist_id = emby_playlist_search_results[0]["Id"]
|
||||
|
||||
for existing_playlist in emby_playlist_search_results:
|
||||
if (
|
||||
existing_playlist["Name"] == emby_playlist_name
|
||||
and existing_playlist["Type"] == "Playlist"
|
||||
):
|
||||
emby.delete_playlist(existing_playlist['Id'])
|
||||
logging.info(
|
||||
f"Deleted existing Emby playlist: {emby_playlist_name} (ID: {existing_playlist_id})"
|
||||
)
|
||||
|
||||
# Create a new playlist in Emby
|
||||
try:
|
||||
emby_playlist = emby.create_playlist(emby_playlist_name, 'Audio')
|
||||
logging.info(
|
||||
f"Created Emby playlist: {playlist_name} (ID: {emby_playlist["Id"]})"
|
||||
)
|
||||
except (requests.exceptions.RequestException, KeyError) as e:
|
||||
logging.error(f"Error creating Emby playlist: {playlist_name}")
|
||||
logging.error(f"Error message: {str(e)}")
|
||||
|
||||
# Get the tracks in the Spotify playlist
|
||||
tracks = spot.get_playlist_tracks(playlist_id)
|
||||
|
||||
logging.info(f"Processing {len(tracks)} tracks in Spotify playlist")
|
||||
# Iterate over each track in the Spotify playlist
|
||||
added_tracks = 0
|
||||
for track in tracks:
|
||||
track_name = track["track"]["name"]
|
||||
artist_name = track["track"]["artists"][0]["name"]
|
||||
album_name = track["track"]["album"]["name"]
|
||||
clean_track_name = StringUtils.remove_special_characters(track_name)
|
||||
clean_artist_name = StringUtils.remove_special_characters(artist_name)
|
||||
|
||||
emby_search_results = emby.search_for_track(track_name, artist_name)
|
||||
emby_search_results_cleaned = emby.search_for_track(
|
||||
clean_track_name, clean_artist_name
|
||||
)
|
||||
|
||||
if emby_search_results:
|
||||
found_match = False
|
||||
for result in emby_search_results:
|
||||
if found_match:
|
||||
break
|
||||
|
||||
emby_item_id = result["Id"]
|
||||
|
||||
logging.debug(f'Matching {track["track"]["name"]} with {result["Name"]}')
|
||||
if spot.match_song(track["track"], result):
|
||||
logging.debug(f"Matched track: {track_name}")
|
||||
try:
|
||||
emby.add_item_to_playlist(emby_playlist['Id'], emby_item_id)
|
||||
logging.info(
|
||||
f"Added '{track_name}' by {artist_name} to Emby playlist"
|
||||
)
|
||||
found_match = True
|
||||
added_tracks += 1
|
||||
except requests.exceptions.RequestException as e:
|
||||
logging.warning(
|
||||
f"Error adding track to Emby playlist: {track_name}"
|
||||
)
|
||||
logging.warning(f"Error message: {str(e)}")
|
||||
else:
|
||||
logging.warning(
|
||||
f"No match found for '{track_name}' by {artist_name} in Emby, failed match_song"
|
||||
)
|
||||
logging.warning(f"SPOTIFY: {track_name} by {artist_name}")
|
||||
logging.warning(
|
||||
f"EMBY: {result['Name']} by {result.get('Artists', [])}"
|
||||
)
|
||||
else:
|
||||
logging.warning(
|
||||
f"No match found for '{track_name}' by {artist_name} in Emby"
|
||||
)
|
||||
logging.info("Added tracks: " + str(added_tracks))
|
||||
added_tracks = 0
|
||||
|
|
@ -0,0 +1,64 @@
|
|||
from src.clients.emby_client import EmbyClient
|
||||
from src.services.playlist_service import PlaylistService
|
||||
from src.config import Config
|
||||
|
||||
# Create an instance of the Emby client for the source user
|
||||
source_emby = EmbyClient(Config.EMBY_URL, Config.EMBY_USERNAME, Config.EMBY_PASSWORD)
|
||||
|
||||
# Create an array of target usernames
|
||||
target_usernames = ["Alyssa", "Dalton", "Laura", "Zuko", "Azula"]
|
||||
target_kids_usernames = ["Alyssa", "Azula", "Zuko"]
|
||||
|
||||
# Get all the playlists for the source user
|
||||
playlists = source_emby.get_playlists()
|
||||
kids_playlists = source_emby.get_tagged_playlist('Kids')
|
||||
|
||||
# Iterate over each playlist
|
||||
for playlist in playlists:
|
||||
# Iterate over the target usernames
|
||||
for target_username in target_usernames:
|
||||
# Check if the target username is in the playlist name
|
||||
if target_username in playlist["Name"]:
|
||||
try:
|
||||
# Create an instance of the Emby client for the target user
|
||||
target_emby = EmbyClient(Config.EMBY_URL, target_username, '')
|
||||
# Create an instance of the PlaylistService
|
||||
playlist_service = PlaylistService(source_emby, target_emby)
|
||||
|
||||
# Check if the playlist already exists for the target user
|
||||
existing_playlist = next((p for p in target_emby.get_playlists() if p["Name"] == playlist["Name"]), None)
|
||||
|
||||
if existing_playlist:
|
||||
# Delete the existing playlist
|
||||
target_emby.delete_playlist(existing_playlist["Id"])
|
||||
print(f"Existing playlist '{playlist['Name']}' deleted for user '{target_username}'.")
|
||||
|
||||
# Copy the playlist to the target user
|
||||
new_playlist = playlist_service.copy_playlist_by_usernames(Config.EMBY_USERNAME, target_username, playlist["Id"])
|
||||
print(f"Playlist '{playlist['Name']}' copied successfully to user '{target_username}'. New playlist ID: {new_playlist['Id']}")
|
||||
except ValueError as e:
|
||||
print(f"Error copying playlist '{playlist['Name']}' to user '{target_username}': {str(e)}")
|
||||
|
||||
# Check if the playlist has the tag 'Kids'
|
||||
for playlist in kids_playlists:
|
||||
# Iterate over the target kids usernames
|
||||
for target_username in target_kids_usernames:
|
||||
try:
|
||||
# Create an instance of the Emby client for the target user
|
||||
target_emby = EmbyClient(Config.EMBY_URL, target_username, '')
|
||||
# Create an instance of the PlaylistService
|
||||
playlist_service = PlaylistService(source_emby, target_emby)
|
||||
|
||||
# Check if the playlist already exists for the target user
|
||||
existing_playlist = next((p for p in target_emby.get_playlists() if p["Name"] == playlist["Name"]), None)
|
||||
|
||||
if existing_playlist:
|
||||
# Delete the existing playlist
|
||||
target_emby.delete_playlist(existing_playlist["Id"])
|
||||
print(f"Existing playlist '{playlist['Name']}' deleted for user '{target_username}'.")
|
||||
|
||||
# Copy the playlist to the target user
|
||||
new_playlist = playlist_service.copy_playlist_by_usernames(Config.EMBY_USERNAME, target_username, playlist["Id"])
|
||||
print(f"Kids playlist '{playlist['Name']}' copied successfully to user '{target_username}'. New playlist ID: {new_playlist['Id']}")
|
||||
except ValueError as e:
|
||||
print(f"Error copying kids playlist '{playlist['Name']}' to user '{target_username}': {str(e)}")
|
||||
Binary file not shown.
Binary file not shown.
Binary file not shown.
|
|
@ -0,0 +1,712 @@
|
|||
import logging
|
||||
import random
|
||||
|
||||
from PIL import Image
|
||||
from io import BytesIO
|
||||
import requests
|
||||
from requests.exceptions import Timeout
|
||||
import time
|
||||
from enum import Enum
|
||||
from src.config import Config
|
||||
|
||||
|
||||
class EmbyLibraryItemType(Enum):
|
||||
AUDIO = "Audio"
|
||||
VIDEO = "Video"
|
||||
FOLDER = "Folder"
|
||||
EPISODE = "Episode"
|
||||
MOVIE = "Movie"
|
||||
TRAILER = "Trailer"
|
||||
ADULT_VIDEO = "AdultVideo"
|
||||
MUSIC_VIDEO = "MusicVideo"
|
||||
BOX_SET = "BoxSet"
|
||||
MUSIC_ALBUM = "MusicAlbum"
|
||||
MUSIC_ARTIST = "MusicArtist"
|
||||
SEASON = "Season"
|
||||
SERIES = "Series"
|
||||
GAME = "Game"
|
||||
GAME_SYSTEM = "GameSystem"
|
||||
BOOK = "Book"
|
||||
|
||||
|
||||
class EmbyImageType(Enum):
|
||||
PRIMARY = "Primary"
|
||||
ART = "Art"
|
||||
BACKDROP = "Backdrop"
|
||||
BANNER = "Banner"
|
||||
LOGO = "Logo"
|
||||
THUMB = "Thumb"
|
||||
DISC = "Disc"
|
||||
BOX = "Box"
|
||||
SCREENSHOT = "Screenshot"
|
||||
MENU = "Menu"
|
||||
CHAPTER = "Chapter"
|
||||
|
||||
|
||||
class EmbyClient:
|
||||
def __init__(self, server_url, username, password):
|
||||
self.server_url = server_url
|
||||
self.username = username
|
||||
self.password = password
|
||||
|
||||
emby_auth_url = f"{server_url}/Users/AuthenticateByName"
|
||||
emby_auth_data = {"username": username, "pw": password}
|
||||
emby_auth_headers = {
|
||||
"Authorization": f'Emby UserId="{username}", Client="{Config.EMBY_CLIENT}", Device="{Config.EMBY_DEVICE}", DeviceId="{Config.EMBY_DEVICE_ID}", Version="{Config.EMBY_VERSION}"',
|
||||
"Content-Type": "application/json",
|
||||
}
|
||||
|
||||
emby_auth_response = requests.post(
|
||||
emby_auth_url, json=emby_auth_data, headers=emby_auth_headers
|
||||
)
|
||||
|
||||
self.user = emby_auth_response.json()["User"]
|
||||
|
||||
self.headers = {
|
||||
"Authorization": f'Emby UserId="{username}", Client="{Config.EMBY_CLIENT}", Device="{Config.EMBY_DEVICE}", DeviceId="{Config.EMBY_DEVICE_ID}", Version="{Config.EMBY_VERSION}"',
|
||||
"X-Emby-Token": emby_auth_response.json()["AccessToken"],
|
||||
"Content-Type": "application/json",
|
||||
}
|
||||
|
||||
# self.user = self.get_user_by_username(username)
|
||||
self.user_id = self.user['Id']
|
||||
|
||||
def _build_url(self, path, params=None):
|
||||
url = f'{self.server_url}/emby/{path}'
|
||||
if params:
|
||||
url += '?' + '&'.join(f'{key}={value}' for key, value in params.items())
|
||||
return url
|
||||
|
||||
def _get_request_with_retry(self, url, retries=6, delay=1, stream=False):
|
||||
for attempt in range(retries):
|
||||
try:
|
||||
response = requests.get(url, headers=self.headers, timeout=61, stream=stream)
|
||||
response.raise_for_status() # Raise an exception for non-1xx status codes
|
||||
if stream is True:
|
||||
return response
|
||||
else:
|
||||
return response.json()
|
||||
except (Timeout, requests.exceptions.RequestException, requests.exceptions.ReadTimeout) as e:
|
||||
print(f"Request failed: {e}")
|
||||
if attempt < retries - 2:
|
||||
print(f"Retrying in {delay} seconds...")
|
||||
time.sleep(delay)
|
||||
raise Exception(f"Failed to make the request after {retries} attempts.")
|
||||
|
||||
def _post_request_with_retry(self, url, data=None, files=None, retries=6, delay=1):
|
||||
for attempt in range(retries):
|
||||
try:
|
||||
response = requests.post(url, data=data, files=files, headers=self.headers, timeout=61)
|
||||
response.raise_for_status() # Raise an exception for non-1xx status codes
|
||||
return response
|
||||
except (Timeout, requests.exceptions.RequestException, requests.exceptions.ReadTimeout) as e:
|
||||
print(f"Request failed: {e}")
|
||||
if attempt < retries - 2:
|
||||
print(f"Retrying in {delay} seconds...")
|
||||
time.sleep(delay)
|
||||
raise Exception(f"Failed to make the request after {retries} attempts.")
|
||||
|
||||
# Modify your existing methods to use the new _get_request_with_retry and _post_request_with_retry methods
|
||||
|
||||
def _delete_request_with_retry(self, url, retries=6, delay=1):
|
||||
for attempt in range(retries):
|
||||
try:
|
||||
response = requests.delete(url, headers=self.headers, timeout=61)
|
||||
response.raise_for_status() # Raise an exception for non-1xx status codes
|
||||
return response
|
||||
except (Timeout, requests.exceptions.RequestException, requests.exceptions.ReadTimeout) as e:
|
||||
print(f"Request failed: {e}")
|
||||
if attempt < retries - 2:
|
||||
print(f"Retrying in {delay} seconds...")
|
||||
time.sleep(delay)
|
||||
raise Exception(f"Failed to make the request after {retries} attempts.")
|
||||
|
||||
def _get_request(self, url, stream=False, retries=6, delay=1):
|
||||
return self._get_request_with_retry(url, stream=stream, retries=retries, delay=delay)
|
||||
|
||||
def _post_request(self, url, data=None, files=None):
|
||||
return self._post_request_with_retry(url, data=data, files=files)
|
||||
|
||||
def _delete_request(self, url, retries=6, delay=1):
|
||||
return self._delete_request_with_retry(url, retries=retries, delay=delay)
|
||||
|
||||
def create_collection(self, name, type, sort_name=None, poster=None):
|
||||
|
||||
# Get the first items id of the correct type (so the collection is sorted right)
|
||||
initial_item_id = self.get_items_by_type(type, 2)[0]['Id']
|
||||
|
||||
url = self._build_url('Collections', {'Name': name, 'Ids': initial_item_id, 'userId': self.user_id})
|
||||
response = self._post_request(url)
|
||||
collection = response.json()
|
||||
|
||||
# TODO: Add sort name if other than None
|
||||
|
||||
print(f"Created collection: {collection['Name']} ({collection['Id']})")
|
||||
|
||||
if sort_name:
|
||||
self.update_item_sort_name(collection['Id'], sort_name)
|
||||
|
||||
# Remove the initial item from the collection,
|
||||
# since we don't want the item and I had errors trying
|
||||
# to create a collection without an initial item.
|
||||
|
||||
try:
|
||||
self.delete_item_from_collection(collection['Id'], initial_item_id)
|
||||
except:
|
||||
print(f"Failed to remove initial item #{initial_item_id} from collection")
|
||||
|
||||
return collection
|
||||
|
||||
def create_playlist(self, name, type, user_id=None, sort_name=None, poster=None):
|
||||
|
||||
if user_id is None:
|
||||
user_id = self.user_id
|
||||
# Get the first items id of the correct type (so the collection is sorted right)
|
||||
# initial_item_id = self.get_items_by_type(type, 2)[0]['Id']
|
||||
|
||||
url = self._build_url('Playlists', {'Name': name, 'userId': user_id})
|
||||
response = self._post_request(url)
|
||||
playlist = response.json()
|
||||
|
||||
print(f"Created playlist: {playlist['Name']} ({playlist['Id']})")
|
||||
|
||||
return playlist
|
||||
|
||||
def update_item_sort_name(self, item_id, sort_name):
|
||||
emby_watchlist_metadata = self.get_item_metadata(item_id)
|
||||
|
||||
emby_watchlist_metadata['ForcedSortName'] = sort_name
|
||||
emby_watchlist_metadata['SortName'] = sort_name
|
||||
emby_watchlist_metadata['LockedFields'] = ['SortName']
|
||||
|
||||
self.update_item_metadata(emby_watchlist_metadata)
|
||||
|
||||
def get_collections(self):
|
||||
url = self._build_url(f'users/{self.user_id}/items',
|
||||
{'Fields': 'ChildCount,RecursiveItemCount',
|
||||
'Recursive': 'true',
|
||||
'SortBy': 'SortName',
|
||||
'SortOrder': 'Ascending',
|
||||
'IncludeItemTypes': 'boxset'})
|
||||
response = self._get_request(url)
|
||||
return response.get('Items', [])
|
||||
|
||||
def get_playlists(self):
|
||||
url = self._build_url(f'users/{self.user_id}/items',
|
||||
{'Fields': 'ChildCount,RecursiveItemCount,Taglines',
|
||||
'Recursive': 'true',
|
||||
'SortBy': 'SortName',
|
||||
'SortOrder': 'Ascending',
|
||||
'IncludeItemTypes': 'playlist'})
|
||||
response = self._get_request(url)
|
||||
return response.get('Items', [])
|
||||
|
||||
def get_tagged_playlist(self, tag):
|
||||
url = self._build_url(f'users/{self.user_id}/items',
|
||||
{'Fields': 'ChildCount,RecursiveItemCount,Taglines',
|
||||
'Recursive': 'true',
|
||||
'SortBy': 'SortName',
|
||||
'SortOrder': 'Ascending',
|
||||
'Tags': tag,
|
||||
'IncludeItemTypes': 'playlist'})
|
||||
response = self._get_request(url)
|
||||
return response.get('Items', [])
|
||||
|
||||
|
||||
def get_collection_by_name(self, name, item_type=None):
|
||||
collections = self.get_collections()
|
||||
|
||||
if item_type is not None:
|
||||
return next((item for item in collections if item.get('Type') == name and item.get('Name') == name), None)
|
||||
return next((item for item in collections if item.get('Name') == name), None)
|
||||
|
||||
def get_collection(self, collection_id):
|
||||
return self.get_list(collection_id)
|
||||
|
||||
def get_list(self, list_id):
|
||||
url = self._build_url(f'users/{self.user_id}/items/{list_id}')
|
||||
response = self._get_request(url)
|
||||
return response
|
||||
|
||||
def get_collection_items(self, collection_id):
|
||||
return self.get_list_items(collection_id)
|
||||
|
||||
def get_items_from_parent(self, parent_id, image_type_limit=2,
|
||||
fields='BasicSyncInfo,CanDelete,Container,PrimaryImageAspectRatio,ProductionYear,ExternalUrls,Status,EndDate,ProviderIds',
|
||||
enable_total_record_count=True,
|
||||
limit: int = 51, offset: int = 0):
|
||||
params = {
|
||||
'ParentId': parent_id,
|
||||
'ImageTypeLimit': image_type_limit,
|
||||
'Fields': fields,
|
||||
'IncludeItemTypes': 'Movie,Series,Season,Episode',
|
||||
'Recursive': 'true',
|
||||
'EnableTotalRecordCount': enable_total_record_count,
|
||||
'StartIndex': offset,
|
||||
'Limit': limit
|
||||
}
|
||||
|
||||
url = self._build_url(f'Users/{self.user_id}/Items', params=params)
|
||||
response = self._get_request(url)
|
||||
items = response.get('Items', [])
|
||||
total_count = response.get('TotalRecordCount', 1)
|
||||
# print(items, total_count)
|
||||
return items, total_count
|
||||
|
||||
def get_libraries(self):
|
||||
url = self._build_url(f'Users/{self.user_id}/views')
|
||||
response = self._get_request(url)
|
||||
return response.get('Items', [])
|
||||
|
||||
def get_library(self, library_id):
|
||||
libraries = self.get_libraries()
|
||||
library = next((item for item in libraries if item.get('Id') == library_id), None)
|
||||
return library
|
||||
|
||||
def get_items_from_library(self, library_name):
|
||||
libraries = self.get_libraries()
|
||||
library = next((item for item in libraries if item.get('Name') == library_name), None)
|
||||
if library:
|
||||
return self.get_items_from_parent(library['Id'])
|
||||
return None, 1
|
||||
|
||||
def get_list_items(self, list_id):
|
||||
url = self._build_url(f'Users/{self.user_id}/Items', {
|
||||
'ParentId': list_id,
|
||||
})
|
||||
response = self._get_request(url)
|
||||
items = response.get('Items', [])
|
||||
total_count = response.get('TotalRecordCount', 0)
|
||||
print(f'Found {total_count} items in playlist')
|
||||
return items, total_count
|
||||
|
||||
def get_seasons(self, series_id):
|
||||
print(f"Getting seasons for series {series_id}")
|
||||
url = self._build_url(f'Shows/{series_id}/Seasons')
|
||||
response = self._get_request(url)
|
||||
return response.get('Items', [])
|
||||
|
||||
def get_episodes(self, series_id, season_id):
|
||||
print(f"Getting episodes for series {series_id} season {season_id}")
|
||||
url = self._build_url(f'Shows/{series_id}/Episodes', {'SeasonId': season_id})
|
||||
response = self._get_request(url)
|
||||
return response.get('Items', [])
|
||||
|
||||
def does_collection_exist(self, collection_name):
|
||||
collections = self.get_collections()
|
||||
for collection in collections:
|
||||
if collection.get('Name') == collection_name:
|
||||
return True
|
||||
return False
|
||||
|
||||
def get_collection_poster(self, collection_id):
|
||||
url = self._build_url(f'Items/{collection_id}/Images/Primary')
|
||||
response = requests.get(url, headers=self.headers)
|
||||
return response
|
||||
|
||||
def add_item_to_collection(self, collection_id, item_id):
|
||||
url = self._build_url(f'Collections/{collection_id}/Items', {'Ids': item_id})
|
||||
response = self._post_request(url)
|
||||
return response
|
||||
|
||||
def add_item_to_playlist(self, playlist_id, item_id):
|
||||
url = self._build_url(f'Playlists/{playlist_id}/Items', {'Ids': item_id})
|
||||
response = self._post_request(url)
|
||||
return response
|
||||
|
||||
def delete_item_from_collection(self, collection_id, item_id):
|
||||
url = self._build_url(f'Collections/{collection_id}/Items', {'Ids': item_id})
|
||||
response = self._delete_request(url)
|
||||
return response
|
||||
|
||||
def delete_item_from_playlist(self, playlist_id, item_id):
|
||||
url = self._build_url(f'Playlists/{playlist_id}/Items', {'EntryIds': item_id})
|
||||
response = self._delete_request(url)
|
||||
return response
|
||||
|
||||
def get_item_image(self, item_id):
|
||||
url = self._build_url(f'Items/{item_id}/Images/Primary')
|
||||
print(url)
|
||||
response = self._get_request(url, stream=True, retries=1)
|
||||
|
||||
if response.status_code == 200:
|
||||
# Assuming _get_request is using the requests library.
|
||||
# Use BytesIO to convert the response content into a file-like object so it can be opened by PIL
|
||||
img = Image.open(BytesIO(response.content)).convert('RGBA')
|
||||
return img
|
||||
else:
|
||||
raise Exception(f"Failed to fetch the image for collection {item_id}. Status code: {response.status_code}")
|
||||
|
||||
def delete_collection(self, collection_id):
|
||||
return self.delete_item(collection_id)
|
||||
|
||||
def delete_playlist(self, playlist_id):
|
||||
return self.delete_item(playlist_id)
|
||||
|
||||
def delete_item(self, item_id):
|
||||
url = self._build_url(f'Items/{item_id}')
|
||||
response = self._delete_request(url)
|
||||
return response
|
||||
|
||||
def delete_all_collections(self):
|
||||
collections = self.get_collections()
|
||||
for collection in collections:
|
||||
# Skip this ALWAYS
|
||||
# TODO: implement some sort of "skip" list
|
||||
if (collection.get('Name') == 'Watchlist'):
|
||||
continue
|
||||
|
||||
print(f"Deleting collection {collection.get('Name')} ({collection.get('Id')})")
|
||||
self.delete_collection(collection.get('Id'))
|
||||
return
|
||||
|
||||
def delete_collection_by_name(self, collection_name):
|
||||
collection = self.get_collection_by_name(collection_name)
|
||||
if collection:
|
||||
self.delete_item(collection.get('Id'))
|
||||
return
|
||||
|
||||
def add_search_results_to_collection(self, collection_id, results):
|
||||
for item in results.get('Items', []):
|
||||
item_id = item.get('Id')
|
||||
print(f"Found {item.get('Name')} with id {item_id}")
|
||||
self.add_item_to_collection(collection_id, item_id)
|
||||
print(f"Added {item.get('Name')} to {collection_id}")
|
||||
|
||||
def delete_search_results_from_collection(self, collection_id, results):
|
||||
for item in results.get('Items', []):
|
||||
item_id = item.get('Id')
|
||||
print(f"Found {item.get('Name')} with id {item_id}")
|
||||
self.delete_item_from_collection(collection_id, item_id)
|
||||
print(f"Removed {item.get('Name')} from {collection_id}")
|
||||
|
||||
def get_items_by_type(self, item_types='Series', limit=50):
|
||||
url = self._build_url(f'Users/{self.user_id}/Items',
|
||||
{'SortBy': 'SortName',
|
||||
'SortOrder': 'Ascending',
|
||||
'IncludeItemTypes': item_types,
|
||||
'Recursive': 'true',
|
||||
'Fields': 'BasicSyncInfo,CanDelete,Container,PrimaryImageAspectRatio,Prefix',
|
||||
'StartIndex': '0',
|
||||
'EnableImageTypes': 'Primary,Backdrop,Thumb',
|
||||
'ImageTypeLimit': '1',
|
||||
'Limit': limit})
|
||||
response = self._get_request_with_retry(url)
|
||||
return response.get('Items', [])
|
||||
|
||||
def get_items_in_collection(self, collection_id):
|
||||
url = self._build_url(f'users/{self.user_id}/items', {'Parentid': collection_id})
|
||||
response = self._get_request(url)
|
||||
items = response.get('Items', [])
|
||||
return items, len(items)
|
||||
|
||||
def get_all_trailers(self):
|
||||
"""
|
||||
Retrieves all trailers from the Emby collection.
|
||||
"""
|
||||
item_type = 'Trailer'
|
||||
url = self._build_url(f'Users/{self.user_id}/Items',
|
||||
{
|
||||
'IncludeItemTypes': item_type,
|
||||
'Recursive': 'true',
|
||||
'Fields': 'Title,Year,Type,Description'
|
||||
})
|
||||
response = self._get_request(url)
|
||||
return response.get('Items', [])
|
||||
|
||||
# def get_all_trailers(self):
|
||||
# url = self._build_url(f'Users/{self.user_id}/Items', {'IncludeItemTypes': 'Trailer'})
|
||||
# response = self._get_request(url)
|
||||
# return response.get('Items', [])
|
||||
|
||||
# def upload_image(self, id, image_path, imgType='Primary'):
|
||||
# mime_type = guess_type(image_path)[0]
|
||||
# with open(image_path, 'rb') as f:
|
||||
# image_data = f.read()
|
||||
# encoded_image_data = base64.b64encode(image_data)
|
||||
# headers = {'Content-Type': mime_type}
|
||||
# print('Uploading collection image: ', image_path, mime_type)
|
||||
# url = self._build_url(f'Items/{id}/Images/{imgType}')
|
||||
# response = requests.post(url, data=encoded_image_data, headers=headers)
|
||||
# return response
|
||||
|
||||
def get_item_metadata(self, item_id):
|
||||
url = self._build_url(f'Users/{self.user_id}/Items/{item_id}', {'Fields': 'ChannelMappingInfo'})
|
||||
response = self._get_request(url)
|
||||
return response
|
||||
|
||||
def update_item_metadata(self, metadata):
|
||||
url = self._build_url(f'Items/{metadata["Id"]}')
|
||||
response = requests.post(url, json=metadata)
|
||||
return response.text
|
||||
|
||||
def get_user_by_username(self, username):
|
||||
users = self.get_users()
|
||||
print(users)
|
||||
return next((user for user in users if user.get('Name') == username), None)
|
||||
|
||||
def set_favorite(self, item_id):
|
||||
url = self._build_url(f'Users/{self.user_id}/FavoriteItems/{item_id}')
|
||||
response = self._post_request(url)
|
||||
return response
|
||||
|
||||
def get_users(self):
|
||||
# https://emby.faiyts.media/emby/users/public?X-Emby-Client=Emby%20Web&X-Emby-Device-Name=Google%20Chrome%20Linux&X-Emby-Device-Id=ea453a6f-4ba4-4901-a3c5-dd875239c834&X-Emby-Client-Version=4.7.13.0&X-Emby-Language=en-us
|
||||
# url = self._build_url(f'Users')
|
||||
headers = {
|
||||
"X-Emby-Token": Config.EMBY_API_KEY,
|
||||
"X-Emby-Client": Config.EMBY_CLIENT,
|
||||
"X-Emby-Device-Name": Config.EMBY_DEVICE,
|
||||
"X-Emby-Device-Id": Config.EMBY_DEVICE_ID,
|
||||
"X-Emby-Client-Version": Config.EMBY_VERSION
|
||||
}
|
||||
|
||||
response = requests.get(f"{Config.EMBY_URL}/emby/Users", headers=headers)
|
||||
return response.json()
|
||||
|
||||
def search(self, query, item_type):
|
||||
url = self._build_url(f'Users/{self.user_id}/Items',
|
||||
{'SortBy': 'SortName',
|
||||
'SortOrder': 'Ascending',
|
||||
'IncludeItemTypes': item_type,
|
||||
'Fields': 'BasicSyncInfo,CanDelete,Container,PrimaryImageAspectRatio,ProductionYear,Status,EndDate',
|
||||
'StartIndex': '0',
|
||||
'EnableImageTypes': 'Primary,Backdrop,Thumb',
|
||||
'ImageTypeLimit': '1',
|
||||
'Recursive': 'true',
|
||||
'SearchTerm': query,
|
||||
'Limit': '50',
|
||||
'IncludeSearchTypes': 'false'})
|
||||
response = self._get_request(url)
|
||||
return response.get('Items', [])
|
||||
|
||||
|
||||
def playlist_search(self, query, item_type):
|
||||
url = self._build_url(f'Users/{self.user_id}/Items',{
|
||||
'Fields': 'BasicSyncInfo,CanDelete,PrimaryImageAspectRatio,ProductionYear,Status,EndDate',
|
||||
'StartIndex': '0',
|
||||
'SortBy': 'SortName',
|
||||
'SortOrder': 'Ascending',
|
||||
'EnableImageTypes': 'Primary,Backdrop,Thumb',
|
||||
'ImageTypeLimit': '1',
|
||||
'Recursive': 'true',
|
||||
'SearchTerm': query,
|
||||
'GroupProgramsBySeries': 'true',
|
||||
'Limit': '50',
|
||||
})
|
||||
|
||||
# @staticmethod
|
||||
# def build_query_parameters(filter_data: EmbyFilters) -> Dict[str, Any]:
|
||||
# filter_dict = filter_data.dict(exclude_none=True)
|
||||
#
|
||||
# # Remap keys as necessary
|
||||
# remapped_keys = {
|
||||
# 'search': 'SearchTerm',
|
||||
# 'limit': 'Limit',
|
||||
# 'listId': 'ParentId',
|
||||
# 'offset': 'StartIndex',
|
||||
# # ... add other remapped keys here ...
|
||||
# }
|
||||
#
|
||||
# for key, new_key in remapped_keys.items():
|
||||
# if key in filter_dict:
|
||||
# filter_dict[new_key] = filter_dict.pop(key)
|
||||
#
|
||||
# # Special handling for Filters
|
||||
# filters_list = []
|
||||
#
|
||||
# if 'isPlayed' in filter_dict:
|
||||
# filters_list.append('IsUnplayed' if not filter_dict['isPlayed'] else 'IsPlayed')
|
||||
# del filter_dict['isPlayed']
|
||||
#
|
||||
# if 'isFavorite' in filter_dict and filter_dict['isFavorite']:
|
||||
# filters_list.append('IsFavorite')
|
||||
# del filter_dict['isFavorite']
|
||||
#
|
||||
# if filters_list:
|
||||
# filter_dict['Filters'] = ','.join(filters_list)
|
||||
#
|
||||
# # Capitalizing the first letter of each key in the dictionary
|
||||
# filter_dict = {key.capitalize(): value for key, value in filter_dict.items()}
|
||||
#
|
||||
# return filter_dict
|
||||
#
|
||||
# def filter_search(self, filters: EmbyFilters) -> Dict[str, Any]:
|
||||
# # Building the request URL
|
||||
# url = self._build_url(f'Users/{self.user_id}/Items', self.build_query_parameters(filters))
|
||||
#
|
||||
# response = self._get_request(url)
|
||||
# return response.get('Items', [])
|
||||
|
||||
def search_for_track(self, track_name, artist_name):
|
||||
emby_search_results = None
|
||||
url = self._build_url(
|
||||
f"Items?SearchTerm={track_name}&Artists={artist_name}&Recursive=true&IncludeItemTypes=Audio&ExcludeItemTypes=Podcast&Limit=10")
|
||||
try:
|
||||
emby_search_response = self._get_request(url)
|
||||
# print(emby_search_response)
|
||||
# emby_search_response.raise_for_status()
|
||||
emby_search_results = emby_search_response["Items"]
|
||||
# print(emby_search_results)
|
||||
except (requests.exceptions.RequestException, KeyError) as e:
|
||||
logging.warning(f"Error searching for track in Emby: {track_name}")
|
||||
logging.warning(f"Error message: {str(e)}")
|
||||
|
||||
return emby_search_results
|
||||
|
||||
def get_sessions(self):
|
||||
url = self._build_url(f'Sessions')
|
||||
response = self._get_request(url)
|
||||
return response
|
||||
|
||||
def play_item(self, session_id, item_id):
|
||||
url = self._build_url(f'Sessions/{session_id}/Playing', {'ItemIds': item_id, 'PlayCommand': 'PlayNow'})
|
||||
response = self._post_request(url)
|
||||
return response
|
||||
|
||||
def send_message(self, session_id, message):
|
||||
url = self._build_url(f'Sessions/{session_id}/Message', {'Text': message})
|
||||
response = self._post_request(url)
|
||||
return response
|
||||
|
||||
def get_movies(self, limit=50, is_played=None, is_favorite=None):
|
||||
return self.get_media(limit, "Movie", is_played, is_favorite)
|
||||
|
||||
def get_media(self, limit=50, item_types="Movie", genre=None, is_played=None, is_favorite=None,
|
||||
external_id=None, name=None, year=None):
|
||||
|
||||
params = {'Recursive': 'true', 'IncludeItemTypes': item_types, "Limit": limit}
|
||||
|
||||
if is_played is not None:
|
||||
params['IsPlayed'] = str(is_played)
|
||||
if is_favorite is not None:
|
||||
params['IsFavorite'] = str(is_favorite)
|
||||
if genre is not None:
|
||||
params['Genres'] = genre
|
||||
if external_id is not None:
|
||||
params['AnyProviderIdEquals'] = external_id
|
||||
if name is not None:
|
||||
params['Name'] = name
|
||||
if year is not None:
|
||||
params['ProductionYear'] = str(year)
|
||||
|
||||
url = self._build_url(f'Users/{self.user_id}/Items', params)
|
||||
response = self._get_request(url)
|
||||
items = response.get('Items', [])
|
||||
random.shuffle(items)
|
||||
return items[:limit]
|
||||
|
||||
def get_liked_movies(self, limit=50):
|
||||
return self.get_movies(limit, is_favorite=True)
|
||||
|
||||
def get_unwatched_movies(self, limit=50):
|
||||
return self.get_movies(limit, is_played=False)
|
||||
|
||||
def get_watched_series(self, limit=50):
|
||||
return self.get_media(limit, "Series", is_played=True)
|
||||
|
||||
def get_movies_by_genre(self, limit=50, genre="Action"):
|
||||
return self.get_media(limit, "Movie", genre=genre)
|
||||
|
||||
# @staticmethod
|
||||
# def create_poster(path, text, root_path, icon_path=f'/resources/icons/tv.png'):
|
||||
# width, height = 400, 600
|
||||
# start, end = (233, 0, 4), (88, 76, 76)
|
||||
# angle = -160
|
||||
# font_path = f'{root_path}/resources/fonts/OpenSans-SemiBold.ttf' # path to your .ttf font file
|
||||
#
|
||||
# image_creator = PosterImageCreator(width, height, "cyan-teal", angle, font_path)
|
||||
# img = image_creator.create_gradient().add_icon_with_text(icon_path, text)
|
||||
#
|
||||
# img.save(path, quality=95)
|
||||
# return img
|
||||
|
||||
####
|
||||
# MediaList based Methods
|
||||
####
|
||||
|
||||
# def search_for_external_ids(self, media_item: MediaItem) -> Optional[dict]:
|
||||
# item = None
|
||||
#
|
||||
# def search_id(external_id: str) -> Optional[dict]:
|
||||
# try:
|
||||
# search_results = self.get_media(external_id=external_id)
|
||||
# if search_results and search_results[0]['Type'] != 'Trailer':
|
||||
# return search_results[0]
|
||||
# except Exception as e:
|
||||
# print(f"Failed searching for {external_id} due to {e}")
|
||||
# return None
|
||||
#
|
||||
# try:
|
||||
# imdb_result = search_id(f"imdb.{media_item.providers.imdbId}")
|
||||
# tvdb_result = search_id(f"tvdb.{media_item.providers.tvdbId}")
|
||||
# except Exception as e:
|
||||
# print(f"Failed searching for {media_item} due to {e}")
|
||||
# return None
|
||||
#
|
||||
# if imdb_result:
|
||||
# item = imdb_result
|
||||
# elif tvdb_result:
|
||||
# item = tvdb_result
|
||||
# return item
|
||||
#
|
||||
# def create_collection_from_list(self, media_list: MediaList):
|
||||
# collection = self.create_collection(media_list.name, media_list.type, media_list.sortName)
|
||||
# # search emby for the items
|
||||
# # add the first result to the collection
|
||||
#
|
||||
# print('-------------', media_list.items)
|
||||
# for item in media_list.items:
|
||||
# print('-------------', item)
|
||||
# media_item = self.search_for_external_ids(item)
|
||||
# if media_item:
|
||||
# self.add_item_to_collection(collection['Id'], media_item['Id'])
|
||||
# return collection
|
||||
|
||||
def delete_collection_items(self, collection_id):
|
||||
items = self.get_collection_items(collection_id)
|
||||
for item in items:
|
||||
self.delete_item_from_collection(collection_id, item['Id'])
|
||||
return
|
||||
|
||||
# def update_collection_from_list(self, media_list: MediaList):
|
||||
# collection = self.get_list(media_list.sourceListId)
|
||||
# if collection:
|
||||
# self.delete_collection_items(collection['Id'])
|
||||
# return self.create_collection_from_list(media_list)
|
||||
#
|
||||
# def create_playlist_from_list(self, media_list: MediaList):
|
||||
# playlist = self.create_playlist(media_list.name, media_list.type, media_list.sortName)
|
||||
# # search emby for the items
|
||||
# # add the first result to the collection
|
||||
#
|
||||
# for item in media_list.items:
|
||||
# print('-------------', item)
|
||||
# emby_media_item = self.search_for_external_ids(item)
|
||||
# if emby_media_item:
|
||||
# self.add_item_to_playlist(playlist['Id'], emby_media_item['Id'])
|
||||
# return playlist
|
||||
#
|
||||
# def update_playlist_from_list(self, media_list: MediaList):
|
||||
# playlist = self.get_list(media_list.sourceListId)
|
||||
# if playlist:
|
||||
# self.delete_collection_items(playlist['Id'])
|
||||
# return self.create_playlist_from_list(media_list)
|
||||
|
||||
def upload_image_from_url(self, sourceListId, poster, root_path):
|
||||
if poster is None:
|
||||
print('no poster provided')
|
||||
return None
|
||||
|
||||
# if the media_list.poster is a url, download the image and upload it to the provider
|
||||
|
||||
print('downloading image from url')
|
||||
response = requests.get(poster, stream=True)
|
||||
if response.status_code == 200:
|
||||
# Assuming _get_request is using the requests library.
|
||||
# Use BytesIO to convert the response content into a file-like object so it can be opened by PIL
|
||||
img = Image.open(BytesIO(response.content)).convert('RGBA')
|
||||
|
||||
poster_location = f'{root_path}/poster.png'
|
||||
img.save(poster_location, quality=95)
|
||||
self.upload_image(sourceListId, poster_location)
|
||||
|
|
@ -0,0 +1,75 @@
|
|||
import logging
|
||||
|
||||
import spotipy
|
||||
from fuzzywuzzy import fuzz
|
||||
from spotipy import SpotifyOAuth
|
||||
from src.utils.string import StringUtils
|
||||
|
||||
|
||||
class SpotifyClient:
|
||||
def __init__(self, client_id, client_secret, redirect_uri, scope):
|
||||
# Authenticate with Spotify API
|
||||
self.sp = spotipy.Spotify(
|
||||
auth_manager=SpotifyOAuth(
|
||||
client_id=client_id,
|
||||
client_secret=client_secret,
|
||||
redirect_uri=redirect_uri,
|
||||
scope=scope,
|
||||
)
|
||||
)
|
||||
|
||||
def match_artists(self, spotify_artists, emby_artists):
|
||||
"""
|
||||
Match Spotify artists with Emby artists using fuzzy string matching.
|
||||
Returns the maximum artist similarity score found.
|
||||
"""
|
||||
max_artist_score = 0
|
||||
for spotify_artist in spotify_artists:
|
||||
spotify_artist_name = StringUtils.clean_string(spotify_artist["name"])
|
||||
for emby_artist in emby_artists:
|
||||
emby_artist_name = StringUtils.clean_string(emby_artist)
|
||||
artist_score = fuzz.ratio(spotify_artist_name, emby_artist_name)
|
||||
max_artist_score = max(max_artist_score, artist_score)
|
||||
return max_artist_score
|
||||
|
||||
def match_song(self, spotify_song, emby_song):
|
||||
"""
|
||||
Match a Spotify song with an Emby song using fuzzy string matching.
|
||||
"""
|
||||
try:
|
||||
spotify_title = StringUtils.clean_string(spotify_song["name"])
|
||||
spotify_artists = spotify_song["artists"]
|
||||
|
||||
emby_title = StringUtils.clean_string(emby_song["Name"])
|
||||
emby_artists = emby_song["Artists"]
|
||||
|
||||
title_ratio = fuzz.ratio(spotify_title, emby_title)
|
||||
artist_ratio = self.match_artists(spotify_artists, emby_artists)
|
||||
|
||||
# Adjust the thresholds as needed
|
||||
title_threshold = 90
|
||||
artist_threshold = 80
|
||||
combined_threshold = 170 # Sum of title and artist thresholds
|
||||
|
||||
if title_ratio >= title_threshold and artist_ratio >= artist_threshold:
|
||||
combined_score = title_ratio + artist_ratio
|
||||
if combined_score >= combined_threshold:
|
||||
return True
|
||||
return False
|
||||
except Exception as e:
|
||||
logging.error(f"Error matching song: {str(e)}")
|
||||
return False
|
||||
|
||||
def get_playlist_tracks(self, playlist_id):
|
||||
results = self.sp.playlist_tracks(playlist_id)
|
||||
tracks = results["items"]
|
||||
while results["next"]:
|
||||
results = self.sp.next(results)
|
||||
tracks.extend(results["items"])
|
||||
return tracks
|
||||
|
||||
def get_playlists(self):
|
||||
return self.sp.current_user_playlists()
|
||||
|
||||
def get_spotify(self):
|
||||
return self.sp
|
||||
|
|
@ -0,0 +1,35 @@
|
|||
# config.py
|
||||
|
||||
class Config:
|
||||
# Spotify API credentials
|
||||
SPOTIFY_CLIENT_ID = "402b29e6ea6e4296b351d9d95c65b6d9"
|
||||
SPOTIFY_CLIENT_SECRET = "5ec3c861ca7141d9b01f75e1eca71bdc"
|
||||
SPOTIFY_REDIRECT_URI = "https://my.home-assistant.io/redirect/oauth"
|
||||
SPOTIFY_SCOPE = "playlist-read-private"
|
||||
|
||||
# Emby API credentials
|
||||
EMBY_URL: str = "http://192.168.0.120:8096"
|
||||
EMBY_API_KEY = "8dc2564d98064333bae3e6eae4a5f8b4"
|
||||
EMBY_USER_ID = "174eb36806924abdad60d6e18ca6da61"
|
||||
EMBY_USERNAME = 'Faiyt'
|
||||
EMBY_PASSWORD = 'dragon'
|
||||
EMBY_CLIENT = "SpotEmbySync"
|
||||
EMBY_DEVICE = "PC"
|
||||
EMBY_DEVICE_ID = "your_emby_device_id"
|
||||
EMBY_VERSION = "1.0.0"
|
||||
|
||||
|
||||
|
||||
# spotify_client_id = "402b29e6ea6e4296b351d9d95c65b6d9"
|
||||
# spotify_client_secret = "5ec3c861ca7141d9b01f75e1eca71bdc"
|
||||
# spotify_redirect_uri = "https://my.home-assistant.io/redirect/oauth"
|
||||
# spotify_scope = "playlist-read-private"
|
||||
#
|
||||
# # Emby API credentials
|
||||
# emby_url = "https://emby.faiyts.media"
|
||||
# emby_api_key = "8dc2564d98064333bae3e6eae4a5f8b4"
|
||||
# emby_user_id = "174eb36806924abdad60d6e18ca6da61"
|
||||
# emby_client = "SpotEmbySync"
|
||||
# emby_device = "PC"
|
||||
# emby_device_id = "your_emby_device_id"
|
||||
# emby_version = "1.0.0"
|
||||
|
|
@ -0,0 +1,74 @@
|
|||
# health.py
|
||||
from config import Config
|
||||
import spotipy
|
||||
from spotipy.oauth2 import SpotifyOAuth
|
||||
import requests
|
||||
|
||||
def test_spotify_connection():
|
||||
try:
|
||||
spotify = spotipy.Spotify(auth_manager=SpotifyOAuth(
|
||||
client_id=Config.SPOTIFY_CLIENT_ID,
|
||||
client_secret=Config.SPOTIFY_CLIENT_SECRET,
|
||||
redirect_uri=Config.SPOTIFY_REDIRECT_URI,
|
||||
scope=Config.SPOTIFY_SCOPE
|
||||
))
|
||||
user = spotify.current_user()
|
||||
print("Spotify connection successful.")
|
||||
print(f"Logged in as: {user['display_name']} ({user['id']})")
|
||||
except Exception as e:
|
||||
print(f"Spotify connection failed: {str(e)}")
|
||||
|
||||
def test_emby_connection():
|
||||
try:
|
||||
headers = {
|
||||
"X-Emby-Token": Config.EMBY_API_KEY,
|
||||
"X-Emby-Client": Config.EMBY_CLIENT,
|
||||
"X-Emby-Device-Name": Config.EMBY_DEVICE,
|
||||
"X-Emby-Device-Id": Config.EMBY_DEVICE_ID,
|
||||
"X-Emby-Client-Version": Config.EMBY_VERSION
|
||||
}
|
||||
response = requests.get(f"{Config.EMBY_URL}/emby/System/Info", headers=headers)
|
||||
|
||||
if response.status_code == 200:
|
||||
print("Emby connection successful.")
|
||||
data = response.json()
|
||||
print(f"Emby server version: {data['Version']}")
|
||||
else:
|
||||
print(f"Emby connection failed. Status code: {response.status_code}")
|
||||
except Exception as e:
|
||||
print(f"Emby connection failed: {str(e)}")
|
||||
|
||||
|
||||
def get_emby_profiles():
|
||||
try:
|
||||
headers = {
|
||||
"X-Emby-Token": Config.EMBY_API_KEY,
|
||||
"X-Emby-Client": Config.EMBY_CLIENT,
|
||||
"X-Emby-Device-Name": Config.EMBY_DEVICE,
|
||||
"X-Emby-Device-Id": Config.EMBY_DEVICE_ID,
|
||||
"X-Emby-Client-Version": Config.EMBY_VERSION
|
||||
}
|
||||
response = requests.get(f"{Config.EMBY_URL}/emby/Users", headers=headers)
|
||||
|
||||
if response.status_code == 200:
|
||||
profiles_data = response.json()
|
||||
print("Valid profiles:")
|
||||
for profile in profiles_data:
|
||||
print(f"- ID: {profile['Id']}, Name: {profile['Name']}")
|
||||
else:
|
||||
print(f"Failed to retrieve profiles. Status code: {response.status_code}")
|
||||
except Exception as e:
|
||||
print(f"Failed to retrieve profiles: {str(e)}")
|
||||
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
print("Testing Spotify connection...")
|
||||
test_spotify_connection()
|
||||
print()
|
||||
|
||||
print("Testing Emby connection...")
|
||||
test_emby_connection()
|
||||
|
||||
print("Retrieving Emby profiles...")
|
||||
get_emby_profiles()
|
||||
|
|
@ -0,0 +1,31 @@
|
|||
from src.clients.emby_client import Emby
|
||||
from services.playlist_service import PlaylistService
|
||||
from src.config import Config
|
||||
|
||||
# Create an instance of the Emby client for the source user
|
||||
source_emby = Emby(Config.EMBY_URL, Config.EMBY_USERNAME, Config.EMBY_PASSWORD)
|
||||
|
||||
# Create an array of target usernames
|
||||
target_usernames = ["Alyssa", "Dalton", "Laura", "Zuko", "Azula","Brittany","Adam"]
|
||||
|
||||
# Get all the playlists for the source user
|
||||
playlists = source_emby.get_playlists()
|
||||
|
||||
# Iterate over each playlist
|
||||
for playlist in playlists:
|
||||
# Iterate over the target usernames
|
||||
for target_username in target_usernames:
|
||||
# Check if the target username is in the playlist name
|
||||
if target_username in playlist["Name"]:
|
||||
try:
|
||||
# Create an instance of the Emby client for the target user
|
||||
target_emby = Emby(Config.EMBY_URL, target_username, '')
|
||||
|
||||
# Create an instance of the PlaylistService
|
||||
playlist_service = PlaylistService(source_emby, target_emby)
|
||||
|
||||
# Copy the playlist to the target user
|
||||
new_playlist = playlist_service.copy_playlist_by_usernames(Config.EMBY_USERNAME, target_username, playlist["Id"])
|
||||
print(f"Playlist '{playlist['Name']}' copied successfully to user '{target_username}'. New playlist ID: {new_playlist['Id']}")
|
||||
except ValueError as e:
|
||||
print(f"Error copying playlist '{playlist['Name']}' to user '{target_username}': {str(e)}")
|
||||
|
|
@ -0,0 +1,49 @@
|
|||
from src.clients.emby_client import Emby, EmbyLibraryItemType
|
||||
|
||||
# Array of TV shows
|
||||
shows = [
|
||||
{"id": "show1-id", "name": "First Show"},
|
||||
{"id": "show2-id", "name": "Second Show"},
|
||||
{"id": "show3-id", "name": "Third Show"}
|
||||
]
|
||||
|
||||
def mesh_shows(emby_client):
|
||||
all_episodes = []
|
||||
for show in shows:
|
||||
seasons = emby_client.get_seasons(show["id"])
|
||||
for season in seasons:
|
||||
episodes = emby_client.get_episodes(show["id"], season["Id"])
|
||||
all_episodes.append(episodes)
|
||||
|
||||
max_episodes = max(len(episodes) for episodes in all_episodes)
|
||||
|
||||
playlist = emby_client.create_playlist("Meshed Shows", EmbyLibraryItemType.VIDEO.value)
|
||||
playlist_id = playlist["Id"]
|
||||
|
||||
for i in range(max_episodes):
|
||||
for j, episodes in enumerate(all_episodes):
|
||||
if i < len(episodes):
|
||||
episode = episodes[i]
|
||||
emby_client.add_item_to_playlist(playlist_id, episode["Id"])
|
||||
print(f"Adding {shows[j]['name']} S{episode['ParentIndexNumber']:02}E{episode['IndexNumber']:02} to the playlist")
|
||||
|
||||
# Balance the shows based on the number of episodes
|
||||
remaining_episodes = len(episodes) - (i + 1)
|
||||
other_remaining_episodes = [len(eps) - (i + 1) for eps in all_episodes[:j] + all_episodes[j+1:]]
|
||||
if remaining_episodes < min(other_remaining_episodes):
|
||||
# Add additional episodes of the current show
|
||||
extra_episodes = min(other_remaining_episodes) - remaining_episodes
|
||||
for k in range(extra_episodes):
|
||||
if i + k + 1 < len(episodes):
|
||||
extra_episode = episodes[i + k + 1]
|
||||
emby_client.add_item_to_playlist(playlist_id, extra_episode["Id"])
|
||||
print(f"Adding {shows[j]['name']} S{extra_episode['ParentIndexNumber']:02}E{extra_episode['IndexNumber']:02} to the playlist")
|
||||
|
||||
print("Playlist created successfully!")
|
||||
|
||||
def main():
|
||||
emby_client = Emby(EMBY_URL, EMBY_USERNAME, EMBY_API_KEY)
|
||||
mesh_shows(emby_client)
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Binary file not shown.
Binary file not shown.
|
|
@ -0,0 +1,38 @@
|
|||
from src.clients.emby_client import EmbyClient
|
||||
|
||||
class PlaylistService:
|
||||
def __init__(self, source_emby: EmbyClient, target_emby: EmbyClient):
|
||||
self.emby = source_emby
|
||||
self.target_emby = target_emby
|
||||
|
||||
def copy_playlist(self, source_user_id: int , target_user_id: int, playlist_id: str):
|
||||
# Get the playlist details
|
||||
playlist = self.emby.get_list(playlist_id)
|
||||
|
||||
print(f"Copying playlist {playlist_id} to {target_user_id}")
|
||||
# print(playlist)
|
||||
|
||||
# Create a new playlist for the target user
|
||||
new_playlist = self.target_emby.create_playlist(playlist["Name"], playlist["Type"], user_id=target_user_id)
|
||||
|
||||
# Get the items from the source playlist
|
||||
playlist_items, _ = self.emby.get_list_items(playlist_id)
|
||||
|
||||
# Add the items to the new playlist
|
||||
for item in playlist_items:
|
||||
self.target_emby.add_item_to_playlist(new_playlist["Id"], item["Id"])
|
||||
|
||||
return new_playlist
|
||||
|
||||
def copy_playlist_by_usernames(self, source_username: str, target_username: str, playlist_id: str):
|
||||
|
||||
source_user = self.emby.get_user_by_username(source_username)
|
||||
target_user = self.target_emby.get_user_by_username(target_username)
|
||||
|
||||
if source_user is None:
|
||||
raise ValueError(f"Source user '{source_username}' not found.")
|
||||
if target_user is None:
|
||||
raise ValueError(f"Target user '{target_username}' not found.")
|
||||
|
||||
# Call the copy_playlist method with the user IDs
|
||||
return self.copy_playlist(source_user['Id'], target_user["Id"], playlist_id)
|
||||
Binary file not shown.
Binary file not shown.
Binary file not shown.
|
|
@ -0,0 +1,172 @@
|
|||
import unittest
|
||||
from src.clients.emby_client import Emby
|
||||
from src.config import Config
|
||||
|
||||
|
||||
class TestEmbyIntegration(unittest.TestCase):
|
||||
|
||||
def setUp(self):
|
||||
self.server_url = Config.EMBY_URL
|
||||
self.username = Config.EMBY_USERNAME
|
||||
self.api_key = Config.EMBY_API_KEY
|
||||
self.password = Config.EMBY_PASSWORD
|
||||
self.emby = Emby(self.server_url, self.username, self.password)
|
||||
|
||||
def test_get_collections(self):
|
||||
collections = self.emby.get_collections()
|
||||
self.assertIsInstance(collections, list)
|
||||
# Assert that the collections have the expected properties
|
||||
for collection in collections:
|
||||
self.assertIn("Name", collection)
|
||||
self.assertIn("Id", collection)
|
||||
|
||||
def test_add_and_remove_items_from_collection(self):
|
||||
# Get the ID of the first item in the movies library
|
||||
movies_library = next((library for library in self.emby.get_libraries() if library["Name"] == "Movies"), None)
|
||||
if movies_library is None:
|
||||
self.skipTest("Movies library not found")
|
||||
|
||||
movies, _ = self.emby.get_items_from_library(movies_library["Name"])
|
||||
if not movies:
|
||||
self.skipTest("No movies found in the library")
|
||||
|
||||
item_id = movies[0]["Id"]
|
||||
|
||||
# Create a new collection
|
||||
collection_name = "Test Collection"
|
||||
collection_type = "movies"
|
||||
collection = self.emby.create_collection(collection_name, collection_type)
|
||||
self.assertEqual(collection["Name"], collection_name)
|
||||
self.assertIsNotNone(collection["Id"])
|
||||
|
||||
# Add the item to the collection
|
||||
self.emby.add_item_to_collection(collection["Id"], item_id)
|
||||
|
||||
# Verify that the item is added to the collection
|
||||
collection_items, _ = self.emby.get_collection_items(collection["Id"])
|
||||
self.assertIn(item_id, [item["Id"] for item in collection_items])
|
||||
|
||||
# Delete the item from the collection
|
||||
self.emby.delete_item_from_collection(collection["Id"], item_id)
|
||||
|
||||
# Verify that the item is deleted from the collection
|
||||
collection_items, _ = self.emby.get_collection_items(collection["Id"])
|
||||
self.assertNotIn(item_id, [item["Id"] for item in collection_items])
|
||||
|
||||
def test_create_collection(self):
|
||||
# Create a new collection
|
||||
collection_name = "Test Collection"
|
||||
collection_type = "Movie"
|
||||
collection = self.emby.create_collection(collection_name, collection_type)
|
||||
self.assertEqual(collection["Name"], collection_name)
|
||||
self.assertIsNotNone(collection["Id"])
|
||||
|
||||
# Delete the collection
|
||||
self.emby.delete_collection(collection["Id"])
|
||||
|
||||
# Verify that the collection is deleted
|
||||
collections = self.emby.get_collections()
|
||||
self.assertNotIn(collection["Id"], [c["Id"] for c in collections])
|
||||
|
||||
# Add more integration tests for other methods
|
||||
|
||||
def test_create_and_delete_playlist(self):
|
||||
# Get the ID of the first item in the movies library
|
||||
movies_library = next((library for library in self.emby.get_libraries() if library["Name"] == "Movies"), None)
|
||||
if movies_library is None:
|
||||
self.skipTest("Movies library not found")
|
||||
|
||||
movies, _ = self.emby.get_items_from_library(movies_library["Name"])
|
||||
if not movies:
|
||||
self.skipTest("No movies found in the library")
|
||||
|
||||
item_id = movies[0]["Id"]
|
||||
|
||||
# Create a new playlist
|
||||
playlist_name = "Test Playlist"
|
||||
playlist_type = "movies"
|
||||
playlist = self.emby.create_playlist(playlist_name, playlist_type)
|
||||
self.assertEqual(playlist["Name"], playlist_name)
|
||||
self.assertIsNotNone(playlist["Id"])
|
||||
|
||||
# Add the item to the playlist
|
||||
self.emby.add_item_to_playlist(playlist["Id"], item_id)
|
||||
|
||||
# Verify that the item is added to the playlist
|
||||
playlist_items, _ = self.emby.get_list_items(playlist["Id"])
|
||||
self.assertIn(item_id, [item["Id"] for item in playlist_items])
|
||||
|
||||
# Delete the item from the playlist
|
||||
self.emby.delete_item_from_playlist(playlist["Id"], item_id)
|
||||
|
||||
# Verify that the item is deleted from the playlist
|
||||
playlist_items, _ = self.emby.get_list_items(playlist["Id"])
|
||||
self.assertNotIn([item_id], [item["Id"] for item in playlist_items])
|
||||
|
||||
# Delete the playlist
|
||||
self.emby.delete_playlist(playlist["Id"])
|
||||
|
||||
# Verify that the playlist is deleted
|
||||
playlists = self.emby.get_playlists()
|
||||
self.assertNotIn(playlist["Id"], [p["Id"] for p in playlists])
|
||||
|
||||
def test_search_for_track_existing(self):
|
||||
# Test searching for a track that exists in the Emby library
|
||||
track_name = "Away from the Sun"
|
||||
artist_name = "3 Doors Down"
|
||||
search_results = self.emby.search_for_track(track_name, artist_name)
|
||||
print(search_results)
|
||||
self.assertIsNotNone(search_results)
|
||||
self.assertGreater(len(search_results), 0)
|
||||
print(len(search_results))
|
||||
has_match = False
|
||||
for track in search_results:
|
||||
if track["Name"].lower() == track_name.lower():
|
||||
for artist in track["Artists"]:
|
||||
if artist.lower() == artist_name.lower():
|
||||
has_match = True
|
||||
self.assertTrue(has_match)
|
||||
|
||||
def test_search_for_track_nonexistent(self):
|
||||
# Test searching for a track that doesn't exist in the Emby library
|
||||
track_name = "Nonexistent Track"
|
||||
artist_name = "Nonexistent Artist"
|
||||
search_results = self.emby.search_for_track(track_name, artist_name)
|
||||
self.assertIsNotNone(search_results)
|
||||
self.assertEqual(len(search_results), 0)
|
||||
|
||||
# def test_search_for_track_partial_match(self):
|
||||
# # Test searching for a track with partial track name and artist name
|
||||
# track_name = "away from"
|
||||
# artist_name = "3 doors"
|
||||
# search_results = self.emby.search_for_track(track_name, artist_name)
|
||||
# self.assertIsNotNone(search_results)
|
||||
# self.assertGreater(len(search_results), 0)
|
||||
# for track in search_results:
|
||||
# self.assertIn(track_name.lower(), track["Name"].lower())
|
||||
# self.assertIn(artist_name.lower(), [artist["Name"].lower() for artist in track["Artists"]])
|
||||
|
||||
def test_search_for_track_empty_query(self):
|
||||
# Test searching for a track with empty track name and artist name
|
||||
track_name = ""
|
||||
artist_name = ""
|
||||
search_results = self.emby.search_for_track(track_name, artist_name)
|
||||
self.assertIsNotNone(search_results)
|
||||
# Depending on your Emby server, empty search query may return all tracks or no tracks
|
||||
# Adjust the assertion based on your expected behavior
|
||||
# self.assertGreater(len(search_results), 0)
|
||||
# self.assertEqual(len(search_results), 0)
|
||||
|
||||
def test_search_for_track_special_characters(self):
|
||||
# Test searching for a track with special characters in track name and artist name
|
||||
track_name = "Track with !@#$%^&*()"
|
||||
artist_name = "Artist with !@#$%^&*()"
|
||||
search_results = self.emby.search_for_track(track_name, artist_name)
|
||||
self.assertIsNotNone(search_results)
|
||||
# Depending on your Emby server, special characters may affect the search results
|
||||
# Adjust the assertions based on your expected behavior
|
||||
# self.assertGreater(len(search_results), 0)
|
||||
# self.assertEqual(len(search_results), 0)
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
|
|
@ -0,0 +1,55 @@
|
|||
import unittest
|
||||
from unittest.mock import patch, MagicMock
|
||||
from src.clients.emby_client import Emby
|
||||
from src.config import Config
|
||||
|
||||
class TestEmby(unittest.TestCase):
|
||||
|
||||
def setUp(self):
|
||||
self.server_url = Config.EMBY_URL
|
||||
self.username = Config.EMBY_USERNAME
|
||||
self.api_key = Config.EMBY_API_KEY
|
||||
self.emby = Emby(self.server_url, self.username, self.api_key)
|
||||
self.emby.user = {"Id": Config.EMBY_USER_ID}
|
||||
self.emby.user_id = Config.EMBY_USER_ID
|
||||
|
||||
@patch('src.clients.emby.Emby._get_request')
|
||||
def test_get_collections(self, mock_get_request):
|
||||
mock_response = {"Items": [{"Name": "Collection1"}, {"Name": "Collection2"}]}
|
||||
mock_get_request.return_value = mock_response
|
||||
collections = self.emby.get_collections()
|
||||
self.assertEqual(len(collections), 2)
|
||||
self.assertEqual(collections[0]["Name"], "Collection1")
|
||||
self.assertEqual(collections[1]["Name"], "Collection2")
|
||||
|
||||
@patch('src.clients.emby.Emby._get_request')
|
||||
def test_get_collection_by_name(self, mock_get_request):
|
||||
mock_response = {"Items": [{"Name": "Collection1", "Type": "boxset"}, {"Name": "Collection2", "Type": "boxset"}]}
|
||||
mock_get_request.return_value = mock_response
|
||||
collection = self.emby.get_collection_by_name("Collection1")
|
||||
self.assertIsNotNone(collection)
|
||||
self.assertEqual(collection["Name"], "Collection1")
|
||||
|
||||
@patch('src.clients.emby.Emby._post_request')
|
||||
def test_create_collection(self, mock_post_request):
|
||||
mock_response = MagicMock()
|
||||
mock_response.json.return_value = {"Name": "NewCollection", "Id": "newcollectionid"}
|
||||
mock_post_request.return_value = mock_response
|
||||
|
||||
with patch('src.clients.emby.Emby.get_items_by_type') as mock_get_items_by_type:
|
||||
mock_get_items_by_type.return_value = [{"Id": "itemid"}]
|
||||
with patch('src.clients.emby.Emby.delete_item_from_collection') as mock_delete_item_from_collection:
|
||||
collection = self.emby.create_collection("NewCollection", "boxset")
|
||||
self.assertEqual(collection["Name"], "NewCollection")
|
||||
self.assertEqual(collection["Id"], "newcollectionid")
|
||||
mock_delete_item_from_collection.assert_called_once_with("newcollectionid", "itemid")
|
||||
|
||||
@patch('src.clients.emby.Emby._post_request')
|
||||
def test_delete_collection(self, mock_post_request):
|
||||
self.emby.delete_collection("collectionid")
|
||||
mock_post_request.assert_called_once_with(f"{self.server_url}/emby/Items/collectionid/Delete?api_key={self.api_key}&X-Emby-Token={self.api_key}")
|
||||
|
||||
# Add more test methods for other functions in the Emby class
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
Binary file not shown.
Binary file not shown.
|
|
@ -0,0 +1,16 @@
|
|||
import re
|
||||
|
||||
|
||||
class StringUtils:
|
||||
def clean_string(s):
|
||||
"""
|
||||
Helper function to remove non-alphanumeric characters and convert to lowercase.
|
||||
"""
|
||||
return re.sub(r"[^a-zA-Z0-9\s]", "", s).lower()
|
||||
|
||||
def remove_special_characters(text):
|
||||
# Remove special characters, brackets, parentheses, and their contents
|
||||
cleaned_text = re.sub(r"[\[\]\(\)]*", "", text)
|
||||
cleaned_text = re.sub(r"[^a-zA-Z0-9\s]", "", cleaned_text)
|
||||
return cleaned_text
|
||||
|
||||
Loading…
Reference in New Issue