commit 6545d820954c2f4af28b88aabaf6fcdef17b6fe3 Author: unfaiyted Date: Mon May 6 21:23:06 2024 -0500 chore: initial diff --git a/.idea/.gitignore b/.idea/.gitignore new file mode 100644 index 0000000..13566b8 --- /dev/null +++ b/.idea/.gitignore @@ -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 diff --git a/.idea/misc.xml b/.idea/misc.xml new file mode 100644 index 0000000..bda831c --- /dev/null +++ b/.idea/misc.xml @@ -0,0 +1,9 @@ + + + + + + + + \ No newline at end of file diff --git a/.idea/modules.xml b/.idea/modules.xml new file mode 100644 index 0000000..41de656 --- /dev/null +++ b/.idea/modules.xml @@ -0,0 +1,8 @@ + + + + + + + + \ No newline at end of file diff --git a/emby-scripts.iml b/emby-scripts.iml new file mode 100644 index 0000000..70d05d5 --- /dev/null +++ b/emby-scripts.iml @@ -0,0 +1,9 @@ + + + + + + + + + \ No newline at end of file diff --git a/src/.cache b/src/.cache new file mode 100644 index 0000000..cc9ae45 --- /dev/null +++ b/src/.cache @@ -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"} \ No newline at end of file diff --git a/src/__init__.py b/src/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/__pycache__/__init__.cpython-312.pyc b/src/__pycache__/__init__.cpython-312.pyc new file mode 100644 index 0000000..f4bb91f Binary files /dev/null and b/src/__pycache__/__init__.cpython-312.pyc differ diff --git a/src/__pycache__/config.cpython-312.pyc b/src/__pycache__/config.cpython-312.pyc new file mode 100644 index 0000000..ef1ede1 Binary files /dev/null and b/src/__pycache__/config.cpython-312.pyc differ diff --git a/src/actions/.cache b/src/actions/.cache new file mode 100644 index 0000000..a936a3f --- /dev/null +++ b/src/actions/.cache @@ -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} \ No newline at end of file diff --git a/src/actions/__init__.py b/src/actions/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/actions/series_mix_playlist_generator.py b/src/actions/series_mix_playlist_generator.py new file mode 100644 index 0000000..db2a822 --- /dev/null +++ b/src/actions/series_mix_playlist_generator.py @@ -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.") \ No newline at end of file diff --git a/src/actions/sync_spotify_playlists.py b/src/actions/sync_spotify_playlists.py new file mode 100644 index 0000000..17f6a4c --- /dev/null +++ b/src/actions/sync_spotify_playlists.py @@ -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 diff --git a/src/actions/sync_user_playlists.py b/src/actions/sync_user_playlists.py new file mode 100644 index 0000000..50e29ea --- /dev/null +++ b/src/actions/sync_user_playlists.py @@ -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)}") \ No newline at end of file diff --git a/src/clients/__init__.py b/src/clients/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/clients/__pycache__/__init__.cpython-312.pyc b/src/clients/__pycache__/__init__.cpython-312.pyc new file mode 100644 index 0000000..f65a055 Binary files /dev/null and b/src/clients/__pycache__/__init__.cpython-312.pyc differ diff --git a/src/clients/__pycache__/emby_client.cpython-312.pyc b/src/clients/__pycache__/emby_client.cpython-312.pyc new file mode 100644 index 0000000..9a6e7e0 Binary files /dev/null and b/src/clients/__pycache__/emby_client.cpython-312.pyc differ diff --git a/src/clients/__pycache__/spotify_client.cpython-312.pyc b/src/clients/__pycache__/spotify_client.cpython-312.pyc new file mode 100644 index 0000000..ecae94b Binary files /dev/null and b/src/clients/__pycache__/spotify_client.cpython-312.pyc differ diff --git a/src/clients/emby_client.py b/src/clients/emby_client.py new file mode 100644 index 0000000..6409469 --- /dev/null +++ b/src/clients/emby_client.py @@ -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) diff --git a/src/clients/spotify_client.py b/src/clients/spotify_client.py new file mode 100644 index 0000000..cb9e085 --- /dev/null +++ b/src/clients/spotify_client.py @@ -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 \ No newline at end of file diff --git a/src/config.py b/src/config.py new file mode 100644 index 0000000..f38c2dd --- /dev/null +++ b/src/config.py @@ -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" diff --git a/src/health.py b/src/health.py new file mode 100644 index 0000000..66b9270 --- /dev/null +++ b/src/health.py @@ -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() \ No newline at end of file diff --git a/src/main.py b/src/main.py new file mode 100644 index 0000000..5805331 --- /dev/null +++ b/src/main.py @@ -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)}") \ No newline at end of file diff --git a/src/mesh-shows.py b/src/mesh-shows.py new file mode 100644 index 0000000..8eee2f2 --- /dev/null +++ b/src/mesh-shows.py @@ -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() \ No newline at end of file diff --git a/src/services/__init__.py b/src/services/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/services/__pycache__/__init__.cpython-312.pyc b/src/services/__pycache__/__init__.cpython-312.pyc new file mode 100644 index 0000000..deda8f6 Binary files /dev/null and b/src/services/__pycache__/__init__.cpython-312.pyc differ diff --git a/src/services/__pycache__/playlist_service.cpython-312.pyc b/src/services/__pycache__/playlist_service.cpython-312.pyc new file mode 100644 index 0000000..b6a196e Binary files /dev/null and b/src/services/__pycache__/playlist_service.cpython-312.pyc differ diff --git a/src/services/playlist_service.py b/src/services/playlist_service.py new file mode 100644 index 0000000..ed94bc5 --- /dev/null +++ b/src/services/playlist_service.py @@ -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) \ No newline at end of file diff --git a/src/tests/__init__.py b/src/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/tests/__pycache__/__init__.cpython-312.pyc b/src/tests/__pycache__/__init__.cpython-312.pyc new file mode 100644 index 0000000..a599400 Binary files /dev/null and b/src/tests/__pycache__/__init__.cpython-312.pyc differ diff --git a/src/tests/__pycache__/emby_integration_test.cpython-312.pyc b/src/tests/__pycache__/emby_integration_test.cpython-312.pyc new file mode 100644 index 0000000..f35b0fd Binary files /dev/null and b/src/tests/__pycache__/emby_integration_test.cpython-312.pyc differ diff --git a/src/tests/__pycache__/emby_test.cpython-312.pyc b/src/tests/__pycache__/emby_test.cpython-312.pyc new file mode 100644 index 0000000..bc49e3c Binary files /dev/null and b/src/tests/__pycache__/emby_test.cpython-312.pyc differ diff --git a/src/tests/emby_integration_test.py b/src/tests/emby_integration_test.py new file mode 100644 index 0000000..1ccc5ff --- /dev/null +++ b/src/tests/emby_integration_test.py @@ -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() diff --git a/src/tests/emby_test.py b/src/tests/emby_test.py new file mode 100644 index 0000000..b4f581c --- /dev/null +++ b/src/tests/emby_test.py @@ -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() \ No newline at end of file diff --git a/src/types/__init__.py b/src/types/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/utils/__init__.py b/src/utils/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/utils/__pycache__/__init__.cpython-312.pyc b/src/utils/__pycache__/__init__.cpython-312.pyc new file mode 100644 index 0000000..50b850c Binary files /dev/null and b/src/utils/__pycache__/__init__.cpython-312.pyc differ diff --git a/src/utils/__pycache__/string.cpython-312.pyc b/src/utils/__pycache__/string.cpython-312.pyc new file mode 100644 index 0000000..867cf7a Binary files /dev/null and b/src/utils/__pycache__/string.cpython-312.pyc differ diff --git a/src/utils/string.py b/src/utils/string.py new file mode 100644 index 0000000..735de9c --- /dev/null +++ b/src/utils/string.py @@ -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 +