feature: add navidome
This commit is contained in:
unfaiyted
parent
c0cb7ecbee
commit
1d57f05e5f
|
|
@ -1 +1 @@
|
|||
{"access_token": "BQCOqPiTZyLWYouW3TxjCpFO3m15BcmPJXXYXhLZ-k36BaTCh7IVkoFiHh2uXXydEIMabbHuk9mZsImyw-lfZ1DS-_NNCEdR1z1Hy0xzBlM_OJ3ItOHMIW-xodY3czcUJWcQUvcLduUqpGiWEod_lbxN9JaYx35JA9zFozGfwba82UjrMtDsRqYQg0PMhqhqb-VCSBClg8ue0RGR8D3IdmketZRLq7Se46YvWAOBG_td-YI", "token_type": "Bearer", "expires_in": 3600, "scope": "playlist-read-private", "expires_at": 1722629879, "refresh_token": "AQAA4rKcyTFqaVctXWaxmE3d0LqLJu3ms91BNXrlJeYfy_1fYmCyQ3VyX6rO6K0O4iGyRVNWT8lcJB6MGRuC5rLXG85ZFB2XQtpcab6ECyKrIQNcMNU7AJj_eiZgfTwNtFk"}
|
||||
{"access_token": "BQAF13mnRW0H1bV2OpTwaODCsS29vIOvgMgYi0Yx6mzmzX8wakeLiD_Cdc33BBNyx_qId5EpdjyWqOLNsQ22Gc24iqZCue9IwYmekDyD4mTRqwsz9UrW8kpxLxk5m6zm1mYKTr6TOLw41qLbEezQ0qmxnBEWmJF_MVv1OAOJHO-9uYRYSU-Q43KI3OuQoSZfhXddRXewxeSQ7Dxz353bdAxQ0rNR7_h17f_XOvJtB4d25To", "token_type": "Bearer", "expires_in": 3600, "scope": "playlist-read-private", "expires_at": 1722697785, "refresh_token": "AQAA4rKcyTFqaVctXWaxmE3d0LqLJu3ms91BNXrlJeYfy_1fYmCyQ3VyX6rO6K0O4iGyRVNWT8lcJB6MGRuC5rLXG85ZFB2XQtpcab6ECyKrIQNcMNU7AJj_eiZgfTwNtFk"}
|
||||
Binary file not shown.
Binary file not shown.
|
|
@ -1 +1 @@
|
|||
{"access_token": "BQDLUIwaf9wl9YwDlaD-kcAg1imJM99GzbVxIRkdasZ-i1cc0UlxCjkO9-m_dRMPVLjHedkX2mcffSj2XTJDFnrv92DrImZZmOD4AZJMo0FQsXWFyQHSYWdKqgxJg0xIx9a0Xz6-_sFQoPidjYxCS6o34WBOv7agj5V0ARJif90AUxvE9j5O813JRZjLFioCgmicjgcCXylMEm80OMXYCwfEWM3ZUgNVmZ3uqRLlUcHIw5c", "token_type": "Bearer", "expires_in": 3600, "scope": "playlist-read-private", "expires_at": 1722622536, "refresh_token": "AQBij3uDeV9G6Z71rKFATWpcHJLhggDPYabbTEuvebdSKEqRY_5VqS98wkx3u037d7nBlaTHIQpeSqkbbOmLdIut56TZ3Yiva56R4t6PV22LSY0wUdTYfwSjP_nWr_jP1wU"}
|
||||
{"access_token": "BQAxlmvegGDayLdhwE18y2aZbW0saezyrJvq7xl1W_1mylT-MMgVrhqNkt1duZILVuoJ49EcmwCHNAYOx02q8B-DM-XDq4-HNL0UDU_Y1QvKyyUstSiqN8ogbcp3Xz8y9gDXbv9-5fF7t4US-DquOy58Ldq58BcTTo7pM15hQwI1c8EjJv-WNPGoXh9sikPnqk3df1DAcOxc-CGZKx3AYZEeAwYyGMhLqCD5Bo9jmuC8nBs", "token_type": "Bearer", "expires_in": 3600, "scope": "playlist-read-private", "expires_at": 1722809560, "refresh_token": "AQBij3uDeV9G6Z71rKFATWpcHJLhggDPYabbTEuvebdSKEqRY_5VqS98wkx3u037d7nBlaTHIQpeSqkbbOmLdIut56TZ3Yiva56R4t6PV22LSY0wUdTYfwSjP_nWr_jP1wU"}
|
||||
Binary file not shown.
|
|
@ -0,0 +1,80 @@
|
|||
# codebase/emby-scripts/src/actions/copy_org_songs_to_server.py
|
||||
|
||||
import os
|
||||
import re
|
||||
import shutil
|
||||
from src.utils.logger import logger
|
||||
import sys
|
||||
import filecmp
|
||||
|
||||
# Set the source and destination directories
|
||||
SOURCE_DIR = r"C:\Music\org_spotdl_downloads"
|
||||
DEST_DIR = r"\\192.168.0.101\download\music"
|
||||
|
||||
def sanitize_filename(filename):
|
||||
# Replace problematic characters with underscores
|
||||
sanitized = re.sub(r'[<>:"/\\|?*]', "_", filename)
|
||||
# Remove trailing dots and spaces
|
||||
sanitized = sanitized.rstrip('. ')
|
||||
# Ensure the filename isn't empty after sanitization
|
||||
return sanitized or "Unknown"
|
||||
|
||||
def copy_and_delete_files(source, destination, dry_run=False):
|
||||
for root, dirs, files in os.walk(source):
|
||||
# Create relative path
|
||||
rel_path = os.path.relpath(root, source)
|
||||
dest_path = os.path.join(destination, rel_path)
|
||||
|
||||
# Create destination directory if it doesn't exist
|
||||
if not dry_run:
|
||||
os.makedirs(dest_path, exist_ok=True)
|
||||
|
||||
for file in files:
|
||||
src_file = os.path.join(root, file)
|
||||
dest_file = os.path.join(dest_path, sanitize_filename(file))
|
||||
|
||||
if dry_run:
|
||||
logger.info(f"Would copy {src_file} to {dest_file} and then delete {src_file}")
|
||||
else:
|
||||
try:
|
||||
# Copy the file
|
||||
shutil.copy2(src_file, dest_file)
|
||||
logger.info(f"Copied {src_file} to {dest_file}")
|
||||
|
||||
# Verify the copy was successful
|
||||
if filecmp.cmp(src_file, dest_file):
|
||||
# Delete the original file
|
||||
os.remove(src_file)
|
||||
logger.info(f"Deleted original file: {src_file}")
|
||||
else:
|
||||
logger.error(f"Copy verification failed for {src_file}. File not deleted.")
|
||||
except Exception as e:
|
||||
logger.error(f"Error processing {src_file}: {e}")
|
||||
|
||||
def main():
|
||||
# Check if --dry-run flag is provided
|
||||
dry_run = "--dry-run" in sys.argv
|
||||
|
||||
if dry_run:
|
||||
logger.info("Running in dry-run mode. No files will be copied or deleted.")
|
||||
|
||||
# Ensure the destination directory exists
|
||||
if not os.path.exists(DEST_DIR):
|
||||
logger.error(f"Destination directory does not exist: {DEST_DIR}")
|
||||
return
|
||||
|
||||
logger.info(f"Starting to copy files from {SOURCE_DIR} to {DEST_DIR}")
|
||||
copy_and_delete_files(SOURCE_DIR, DEST_DIR, dry_run)
|
||||
logger.info("File copying and deletion process completed.")
|
||||
|
||||
# Clean up empty directories in the source
|
||||
if not dry_run:
|
||||
for root, dirs, files in os.walk(SOURCE_DIR, topdown=False):
|
||||
for dir in dirs:
|
||||
dir_path = os.path.join(root, dir)
|
||||
if not os.listdir(dir_path):
|
||||
os.rmdir(dir_path)
|
||||
logger.info(f"Removed empty directory: {dir_path}")
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
|
|
@ -1,13 +1,11 @@
|
|||
# codebase/emby-scripts/src/actions/delete_duplicate_playlists.py
|
||||
|
||||
import logging
|
||||
from src.config import Config
|
||||
from src.clients.emby_client import EmbyClient
|
||||
from src.utils.logger import logger
|
||||
|
||||
# Configure logging
|
||||
logging.basicConfig(
|
||||
level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s"
|
||||
)
|
||||
|
||||
|
||||
def delete_duplicate_playlists(emby):
|
||||
# Initialize EmbyClient
|
||||
|
|
@ -30,16 +28,17 @@ def delete_duplicate_playlists(emby):
|
|||
if playlist_item_count > existing_item_count:
|
||||
# Delete the existing playlist with fewer items
|
||||
emby.delete_playlist(existing_playlist['Id'])
|
||||
logging.info(f"Deleted duplicate playlist: {existing_playlist['Name']} (ID: {existing_playlist['Id']}) with {existing_item_count} items")
|
||||
logger.info(f"Deleted duplicate playlist: {existing_playlist['Name']} (ID: {existing_playlist['Id']}) with {existing_item_count} items")
|
||||
playlists_by_name[playlist_name] = playlist
|
||||
else:
|
||||
# Delete the current playlist with fewer items
|
||||
emby.delete_playlist(playlist_id)
|
||||
logging.info(f"Deleted duplicate playlist: {playlist_name} (ID: {playlist_id}) with {playlist_item_count} items")
|
||||
logger.info(f"Deleted duplicate playlist: {playlist_name} (ID: {playlist_id}) with {playlist_item_count} items")
|
||||
else:
|
||||
playlists_by_name[playlist_name] = playlist
|
||||
|
||||
logging.info("Finished deleting duplicate playlists")
|
||||
logger.info("Finished deleting duplicate playlists")
|
||||
|
||||
if __name__ == "__main__":
|
||||
delete_duplicate_playlists()
|
||||
emby = EmbyClient(Config.EMBY_URL, Config.EMBY_USERNAME, Config.EMBY_PASSWORD)
|
||||
delete_duplicate_playlists(emby)
|
||||
|
|
@ -24,7 +24,7 @@ logging.basicConfig(
|
|||
spotdl_path = "C:\\Users\\Dane Miller\\Downloads\\spotdl-4.2.5-win32.exe"
|
||||
|
||||
# Set the output directory where the songs will be saved
|
||||
output_dir = "C:\\Users\\Dane Miller\\Music\\spotdl_downloads"
|
||||
output_dir = "C:\\Music\\spotdl_downloads"
|
||||
|
||||
def remove_special_characters(string, skip_non_ascii=False):
|
||||
if skip_non_ascii and not all(ord(char) < 128 for char in string):
|
||||
|
|
|
|||
|
|
@ -2,7 +2,6 @@ from src.clients.emby_client import EmbyClient
|
|||
from src.config import Config
|
||||
import re
|
||||
|
||||
|
||||
def parse_episode_string(episode_string):
|
||||
match = re.match(r"S(\d+)E(\d+)", episode_string)
|
||||
if match:
|
||||
|
|
@ -12,40 +11,8 @@ def parse_episode_string(episode_string):
|
|||
else:
|
||||
raise ValueError(f"Invalid episode string format: {episode_string}")
|
||||
|
||||
# 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 = ["Superstore", "Fresh Off the Boat", "What If...?", "Friends", "Avatar: The Last Airbender"]
|
||||
#
|
||||
# starting_episodes = {
|
||||
# "Superstore": "S03E09", # Start from Season 2, Episode 3 of The Office ()
|
||||
# "Fresh Off the Boat": "S03E08",
|
||||
# "What If...?": "S01E05",
|
||||
# "Friends": "S01E01",
|
||||
# "Avatar: The Last Airbender": "S01E19"
|
||||
# }
|
||||
# tv_shows = ["The Following", "Ozark", "Better Call Saul",
|
||||
# "The Leftovers", "Halo"]
|
||||
#
|
||||
#
|
||||
# starting_episodes = {
|
||||
# "The Following": "S01E09", # Start from Season 2, Episode 3 of The Office ()
|
||||
# "Ozark": "S01E02",
|
||||
# "Better Call Saul": "S05E03",
|
||||
# "The Leftovers": "S01E01",
|
||||
# "Halo": "S01E01"
|
||||
# }
|
||||
#
|
||||
# Superstore S03E09
|
||||
# "The Following": "S01E09"
|
||||
# "Ozark": "S01E02",
|
||||
# "Fresh Off the Boat": "S03E08",
|
||||
# "Better Call Saul": "S05E03",
|
||||
# "What If...?" "S01E05"
|
||||
|
||||
tv_shows = ["The Office (US)", "Bob's Burgers", "Parks and Recreation"]
|
||||
|
||||
starting_episodes = {
|
||||
|
|
|
|||
|
|
@ -1,108 +1,148 @@
|
|||
import os
|
||||
import shutil
|
||||
import logging
|
||||
import tinytag
|
||||
import re
|
||||
import sys
|
||||
|
||||
# Set up logging
|
||||
logging.basicConfig(
|
||||
level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s"
|
||||
)
|
||||
import filecmp
|
||||
import src.utils.logger as logger
|
||||
|
||||
# Set the input and output directories
|
||||
input_dir = r"C:\Users\Dane Miller\Music\spotdl_downloads"
|
||||
output_dir = r"C:\Users\Dane Miller\Music\org_spotdl_downloads"
|
||||
input_dir = r"C:\Music\spotdl_downloads"
|
||||
output_dir = r"C:\Music\org_spotdl_downloads"
|
||||
|
||||
# Set the dry run option based on command-line argument
|
||||
if len(sys.argv) > 1 and sys.argv[1] == "--dry-run":
|
||||
dry_run = True
|
||||
else:
|
||||
dry_run = False
|
||||
# Set the dry run and delete source options based on command-line arguments
|
||||
dry_run = "--dry-run" in sys.argv
|
||||
keep_source = "--dont-delete-source" in sys.argv
|
||||
|
||||
# Regex pattern to extract year from date string
|
||||
year_pattern = r"\d{4}"
|
||||
|
||||
import re
|
||||
|
||||
def sanitize_filename(filename):
|
||||
|
||||
def shorten_album_name(album_name, max_length=50):
|
||||
# Remove common words and abbreviate
|
||||
album_name = re.sub(r'\bOriginal\s+|Television\s+|Soundtrack\s+|\bOST\b', '', album_name, flags=re.IGNORECASE)
|
||||
album_name = album_name.replace('Season', 'S').replace('Special', 'Sp')
|
||||
|
||||
# Remove text in square brackets
|
||||
album_name = re.sub(r'\[.*?\]', '', album_name).strip()
|
||||
|
||||
# Truncate if still too long
|
||||
if len(album_name) > max_length:
|
||||
album_name = album_name[:max_length - 3] + '...'
|
||||
|
||||
return album_name.strip()
|
||||
|
||||
|
||||
def sanitize_filename(filename, max_length=100):
|
||||
# Replace problematic characters with underscores
|
||||
sanitized = re.sub(r'[<>:"/\\|?*]', "_", filename)
|
||||
return sanitized
|
||||
# Remove trailing dots and spaces
|
||||
sanitized = sanitized.rstrip('. ')
|
||||
|
||||
# Truncate if too long
|
||||
if len(sanitized) > max_length:
|
||||
base, ext = os.path.splitext(sanitized)
|
||||
sanitized = base[:max_length - len(ext) - 3] + '...' + ext
|
||||
|
||||
return sanitized or "Unknown"
|
||||
# def sanitize_filename(filename):
|
||||
# # Replace problematic characters with underscores
|
||||
# sanitized = re.sub(r'[<>:"/\\|?*]', "_", filename)
|
||||
# # Remove trailing dots and spaces
|
||||
# sanitized = sanitized.rstrip('. ')
|
||||
# return sanitized or "Unknown"
|
||||
|
||||
def process_file(file_path, output_file_path):
|
||||
if os.path.exists(output_file_path):
|
||||
if filecmp.cmp(file_path, output_file_path):
|
||||
logger.info(f"File already exists and is identical: {output_file_path}")
|
||||
return True
|
||||
else:
|
||||
logger.warning(f"File exists but is different: {output_file_path}")
|
||||
return False
|
||||
|
||||
if not dry_run:
|
||||
try:
|
||||
shutil.copy2(file_path, output_file_path)
|
||||
logger.info(f"Copied {file_path} to {output_file_path}")
|
||||
return True
|
||||
except Exception as e:
|
||||
logger.error(f"Error copying {file_path} to {output_file_path}: {e}")
|
||||
return False
|
||||
else:
|
||||
logger.info(f"Dry run: Would have copied {file_path} to {output_file_path}")
|
||||
return True
|
||||
|
||||
# Loop through all files in the input directory
|
||||
for root, dirs, files in os.walk(input_dir):
|
||||
for file in files:
|
||||
if file.endswith(".mp3"):
|
||||
# Construct the full file path
|
||||
file_path = os.path.join(root, file)
|
||||
|
||||
try:
|
||||
# Read the metadata from the MP3 file
|
||||
tag = tinytag.TinyTag.get(file_path)
|
||||
artists = [
|
||||
artist.strip()
|
||||
for artist in (tag.artist or "Unknown Artist").split("/")
|
||||
]
|
||||
logging.info(f"Artists: {artists}")
|
||||
artists = [artist.strip() for artist in (tag.artist or "Unknown Artist").split("/")]
|
||||
album = tag.album or "Unknown Album"
|
||||
year_str = str(tag.year) if tag.year else ""
|
||||
year = re.search(year_pattern, year_str).group() if year_str else ""
|
||||
track_number = str(tag.track).zfill(2) if tag.track else "00"
|
||||
title = tag.title or os.path.splitext(file)[0]
|
||||
|
||||
except Exception as e:
|
||||
# Handle errors reading metadata
|
||||
artists = ["Unknown Artist"]
|
||||
album = "Unknown Album"
|
||||
year = ""
|
||||
track_number = "00"
|
||||
title = os.path.splitext(file)[0]
|
||||
logging.warning(f"Error reading metadata from {file_path}: {e}")
|
||||
logger.warning(f"Error reading metadata from {file_path}: {e}")
|
||||
|
||||
# Create the output folder structure if it doesn't exist
|
||||
album_name = f"{album} ({year})" if year else album
|
||||
album_name = sanitize_filename(album_name) # Sanitize the album name
|
||||
output_artist_dir = os.path.join(
|
||||
output_dir, sanitize_filename(artists[0])
|
||||
) # Use only the first artist and sanitize the name
|
||||
output_album_dir = os.path.join(
|
||||
output_artist_dir, album_name.replace("_", " ")
|
||||
)
|
||||
if not os.path.exists(output_album_dir):
|
||||
os.makedirs(output_album_dir)
|
||||
logging.info(f"Created directory: {output_album_dir}")
|
||||
album_name = shorten_album_name(album_name)
|
||||
album_name = sanitize_filename(album_name)
|
||||
output_artist_dir = os.path.join(output_dir, sanitize_filename(artists[0], max_length=30))
|
||||
output_album_dir = os.path.join(output_artist_dir, album_name.replace("_", " "))
|
||||
|
||||
# Rename the file with the new format
|
||||
new_file_name = f"{album_name.replace('_', ' ')} - {track_number} - {sanitize_filename(title)}.mp3" # Remove artist names from file name and sanitize the title
|
||||
if not os.path.exists(output_album_dir) and not dry_run:
|
||||
try:
|
||||
os.makedirs(output_album_dir)
|
||||
logger.info(f"Created directory: {output_album_dir}")
|
||||
except Exception as e:
|
||||
logger.error(f"Error creating directory: {output_album_dir}: {e}")
|
||||
continue
|
||||
|
||||
new_file_name = f"{track_number} - {sanitize_filename(title, max_length=50)}.mp3"
|
||||
output_file_path = os.path.join(output_album_dir, new_file_name)
|
||||
|
||||
logging.info(f"Source file path: {file_path}")
|
||||
logging.info(f"Destination file path: {output_file_path}")
|
||||
|
||||
if os.path.exists(file_path):
|
||||
logging.info(f'File path exists.')
|
||||
if dry_run:
|
||||
logging.info(f"Dry run: Would have moved {file_path} to {output_file_path}")
|
||||
else:
|
||||
logging.info(f"Moving {file_path} to {output_file_path}")
|
||||
# shutil.move(file_path, output_file_path)
|
||||
|
||||
if process_file(file_path, output_file_path):
|
||||
if not keep_source and not dry_run:
|
||||
try:
|
||||
shutil.copy2(file_path, output_file_path)
|
||||
os.remove(file_path)
|
||||
logger.info(f"Deleted source file: {file_path}")
|
||||
except Exception as e:
|
||||
logging.error(f"Error moving {file_path} to {output_file_path}: {e}")
|
||||
logging.info(f"Moved {file_path} to {output_file_path}")
|
||||
else:
|
||||
logging.warning(f"Source file does not exist: {file_path}")
|
||||
logger.error(f"Error deleting source file {file_path}: {e}")
|
||||
|
||||
# Check if a lyrics file exists and copy it to the new directory structure
|
||||
# Process lyrics file
|
||||
lyrics_file = os.path.splitext(file)[0] + ".lrc"
|
||||
lyrics_file_path = os.path.join(root, lyrics_file)
|
||||
if os.path.exists(lyrics_file_path):
|
||||
output_lyrics_file_path = os.path.join(output_album_dir, os.path.splitext(new_file_name)[0] + ".lrc")
|
||||
if dry_run:
|
||||
logging.info(f"Dry run: Would have copied {lyrics_file_path} to {output_lyrics_file_path}")
|
||||
else:
|
||||
shutil.copy2(lyrics_file_path, output_lyrics_file_path)
|
||||
logging.info(f"Copied {lyrics_file_path} to {output_lyrics_file_path}")
|
||||
if process_file(lyrics_file_path, output_lyrics_file_path):
|
||||
if not keep_source and not dry_run:
|
||||
try:
|
||||
os.remove(lyrics_file_path)
|
||||
logger.info(f"Deleted source lyrics file: {lyrics_file_path}")
|
||||
except Exception as e:
|
||||
logger.error(f"Error deleting source lyrics file {lyrics_file_path}: {e}")
|
||||
|
||||
# Clean up empty directories in the source if deleting source files
|
||||
if not keep_source and not dry_run:
|
||||
for root, dirs, files in os.walk(input_dir, topdown=False):
|
||||
for dir in dirs:
|
||||
dir_path = os.path.join(root, dir)
|
||||
if not os.listdir(dir_path):
|
||||
try:
|
||||
os.rmdir(dir_path)
|
||||
logger.info(f"Removed empty directory: {dir_path}")
|
||||
except Exception as e:
|
||||
logger.error(f"Error removing empty directory {dir_path}: {e}")
|
||||
|
|
|
|||
|
|
@ -1,7 +1,6 @@
|
|||
import spotipy
|
||||
from spotipy.oauth2 import SpotifyOAuth
|
||||
import requests
|
||||
import logging
|
||||
import json
|
||||
from fuzzywuzzy import fuzz
|
||||
import re
|
||||
|
|
@ -9,21 +8,10 @@ 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, clean_title
|
||||
from src.utils.logger import logger
|
||||
|
||||
import sqlite3
|
||||
|
||||
# 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)
|
||||
|
||||
# Connect to the SQLite database (it will be created if it doesn't exist)
|
||||
|
||||
|
||||
def try_match_and_add(track_name, artist_name, emby_playlist, emby, spot):
|
||||
|
||||
emby_search_results = emby.search_for_track(track_name, artist_name)
|
||||
|
|
@ -31,15 +19,15 @@ def try_match_and_add(track_name, artist_name, emby_playlist, emby, spot):
|
|||
if emby_search_results:
|
||||
for result in emby_search_results:
|
||||
emby_item_id = result["Id"]
|
||||
logging.debug(f'Matching {track_name} with {result["Name"]}')
|
||||
logger.debug(f'Matching {track_name} with {result["Name"]}')
|
||||
if spot.match_song({"name": track_name, "artists": [{"name": artist_name}]}, result):
|
||||
try:
|
||||
emby.add_item_to_playlist(emby_playlist['Id'], emby_item_id)
|
||||
# logging.info(f"Added '{track_name}' by {artist_name} to Emby playlist")
|
||||
return True
|
||||
except requests.exceptions.RequestException as e:
|
||||
logging.warning(f"Error adding track to Emby playlist: {track_name}")
|
||||
logging.warning(f"Error message: {str(e)}")
|
||||
logger.warning(f"Error adding track to Emby playlist: {track_name}")
|
||||
logger.warning(f"Error message: {str(e)}")
|
||||
return False
|
||||
|
||||
def sync_spotify_playlists(spot, emby):
|
||||
|
|
@ -52,16 +40,29 @@ def sync_spotify_playlists(spot, emby):
|
|||
|
||||
playlists = spot.get_playlists()
|
||||
featured_playlists = spot.get_featured_playlists()
|
||||
# print(json.dumps(featured_playlists, indent=4))
|
||||
# made_for_you = spot.get_made_for_you()
|
||||
|
||||
# list of categories to get playlists from
|
||||
categories = ["Made for You", "Pop", "Country", "Summer", "Discover", "In the Car"]
|
||||
|
||||
all_playlists = playlists["items"] + featured_playlists["items"]
|
||||
# get all the playlists from the categories
|
||||
for category in categories:
|
||||
logger.info(f"Getting playlists from category: {category}")
|
||||
all_playlists += spot.get_category_playlists_by_name(category)["items"]
|
||||
|
||||
# print(json.dumps(featured_playlists, indent=4))
|
||||
|
||||
# print(made_for_you)
|
||||
|
||||
# all_playlists = playlists["items"] + featured_playlists["items"] #+ made_for_you["items"]
|
||||
|
||||
# Iterate over each Spotify playlist
|
||||
for playlist in all_playlists:
|
||||
playlist_name = playlist["name"]
|
||||
playlist_id = playlist["id"]
|
||||
playlist_owner = playlist["owner"]["display_name"]
|
||||
logging.info(f"Processing Spotify playlist: {playlist_name} ({playlist_owner})")
|
||||
logger.info(f"Processing Spotify playlist: {playlist_name} ({playlist_owner})")
|
||||
|
||||
emby_playlist_name = f"{playlist_name} ({playlist_owner})"
|
||||
|
||||
|
|
@ -77,14 +78,14 @@ def sync_spotify_playlists(spot, emby):
|
|||
and existing_playlist["Type"] == "Playlist"
|
||||
):
|
||||
emby.delete_playlist(existing_playlist['Id'])
|
||||
logging.info(
|
||||
logger.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(
|
||||
logger.info(
|
||||
f"Created Emby playlist: {playlist_name} (ID: {emby_playlist["Id"]})"
|
||||
)
|
||||
# Get the playlist image from Spotify
|
||||
|
|
@ -92,33 +93,42 @@ def sync_spotify_playlists(spot, emby):
|
|||
if playlist_image_data:
|
||||
try:
|
||||
emby.upload_image_data(emby_playlist['Id'], playlist_image_data)
|
||||
logging.info(f"Uploaded playlist cover image for '{playlist_name}'")
|
||||
logger.info(f"Uploaded playlist cover image for '{playlist_name}'")
|
||||
except requests.exceptions.RequestException as e:
|
||||
logging.warning(f"Error uploading playlist cover image: {str(e)}")
|
||||
logger.warning(f"Error uploading playlist cover image: {str(e)}")
|
||||
else:
|
||||
logging.warning(f"No playlist cover image found for '{playlist_name}'")
|
||||
|
||||
|
||||
|
||||
|
||||
logger.warning(f"No playlist cover image found for '{playlist_name}'")
|
||||
|
||||
|
||||
except (requests.exceptions.RequestException, KeyError) as e:
|
||||
logging.error(f"Error creating Emby playlist: {playlist_name}")
|
||||
logging.error(f"Error message: {str(e)}")
|
||||
logger.error(f"Error creating Emby playlist: {playlist_name}")
|
||||
logger.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")
|
||||
logger.info(f"Processing {len(tracks)} tracks in Spotify playlist")
|
||||
# Iterate over each track in the Spotify playlist
|
||||
added_tracks = 0
|
||||
unmatched_tracks = []
|
||||
for track in tracks:
|
||||
|
||||
# skil if track is not available
|
||||
if track["track"] is None:
|
||||
logger.warning(f"Track is not available: {track}")
|
||||
continue
|
||||
|
||||
if track["track"]["name"] is None:
|
||||
logger.warning(f"Track name is None: {track}")
|
||||
continue
|
||||
|
||||
track_name = track["track"]["name"]
|
||||
artist_name = track["track"]["artists"][0]["name"]
|
||||
album_name = track["track"]["album"]["name"]
|
||||
|
||||
if album_name is None:
|
||||
album_name = "Unknown Album"
|
||||
|
||||
if try_match_and_add(track_name, artist_name, emby_playlist, emby, spot):
|
||||
added_tracks += 1
|
||||
continue
|
||||
|
|
@ -130,9 +140,9 @@ def sync_spotify_playlists(spot, emby):
|
|||
added_tracks += 1
|
||||
continue
|
||||
|
||||
logging.warning(f"No match found for '{track_name}' by {artist_name} in Emby")
|
||||
logging.warning(f"Cleaned track name: {clean_track_name}")
|
||||
logging.warning(f"Cleaned artist name: {clean_artist_name}")
|
||||
logger.warning(f"No match found for '{track_name}' by {artist_name} in Emby / clean: {clean_track_name}")
|
||||
# logging.warning(f"Cleaned track name: {clean_track_name}")
|
||||
# logging.warning(f"Cleaned artist name: {clean_artist_name}")
|
||||
|
||||
unmatched_tracks.append((playlist_name, track_name, artist_name, album_name))
|
||||
|
||||
|
|
@ -143,7 +153,7 @@ def sync_spotify_playlists(spot, emby):
|
|||
# Calculate the match percentage for the current playlist
|
||||
if len(tracks) > 0:
|
||||
match_percentage = (added_tracks / len(tracks)) * 100
|
||||
logging.info(f"Match percentage for playlist '{playlist_name}': {match_percentage:.2f}%")
|
||||
logger.info(f"Match percentage for playlist '{playlist_name}': {match_percentage:.2f}%")
|
||||
|
||||
added_tracks = 0
|
||||
|
||||
|
|
@ -151,8 +161,6 @@ def sync_spotify_playlists(spot, emby):
|
|||
conn.close()
|
||||
|
||||
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
spot = SpotifyClient(Config.SPOTIFY_CLIENT_ID, Config.SPOTIFY_CLIENT_SECRET, Config.SPOTIFY_REDIRECT_URI,
|
||||
Config.SPOTIFY_SCOPE)
|
||||
|
|
@ -0,0 +1,71 @@
|
|||
# codebase/emby-scripts/src/actions/sync_spotify_to_navidrome_playlists.py
|
||||
|
||||
import requests
|
||||
from src.config import Config
|
||||
from src.clients.spotify_client import SpotifyClient
|
||||
from src.clients.navidrome_client import NavidromeClient
|
||||
from src.utils.logger import logger
|
||||
|
||||
def try_match_and_add(track_name, artist_name, navidrome_playlist, spotify, navidrome):
|
||||
spotify_track = {"name": track_name, "artist": artist_name}
|
||||
|
||||
try:
|
||||
navidrome_track = navidrome.search_track(track_name, artist_name)
|
||||
if navidrome_track:
|
||||
if navidrome.match_song(spotify_track, navidrome_track):
|
||||
navidrome.add_track_to_playlist(navidrome_playlist['id'], navidrome_track['id'])
|
||||
logger.info(f"Added '{track_name}' by {artist_name} to Navidrome playlist")
|
||||
return True
|
||||
else:
|
||||
logger.warning(f"Found track but it didn't match closely enough: '{track_name}' by {artist_name}")
|
||||
else:
|
||||
logger.warning(f"No match found for '{track_name}' by {artist_name} in Navidrome")
|
||||
except requests.exceptions.RequestException as e:
|
||||
logger.warning(f"Error adding track to Navidrome playlist: {track_name}")
|
||||
logger.warning(f"Error message: {str(e)}")
|
||||
|
||||
return False
|
||||
|
||||
|
||||
def sync_navidrome_playlists(spotify, navidrome):
|
||||
spotify_playlists = spotify.get_playlists()
|
||||
|
||||
for spotify_playlist in spotify_playlists['items']:
|
||||
playlist_name = spotify_playlist["name"]
|
||||
playlist_id = spotify_playlist["id"]
|
||||
logger.info(f"Processing Spotify playlist: {playlist_name}")
|
||||
|
||||
# Check if playlist exists in Navidrome, create if not
|
||||
navidrome_playlist = navidrome.get_playlist_by_name(playlist_name)
|
||||
if not navidrome_playlist:
|
||||
navidrome_playlist = navidrome.create_playlist(playlist_name)
|
||||
logger.info(f"Created Navidrome playlist: {playlist_name}")
|
||||
else:
|
||||
# Clear existing tracks in Navidrome playlist
|
||||
navidrome.clear_playlist(navidrome_playlist['id'])
|
||||
logger.info(f"Cleared existing tracks in Navidrome playlist: {playlist_name}")
|
||||
|
||||
# Get tracks from Spotify playlist
|
||||
spotify_tracks = spotify.get_playlist_tracks(playlist_id)
|
||||
|
||||
added_tracks = 0
|
||||
for item in spotify_tracks:
|
||||
track = item['track']
|
||||
track_name = track['name']
|
||||
artist_name = track['artists'][0]['name'] # Assuming the first artist
|
||||
|
||||
if try_match_and_add(track_name, artist_name, navidrome_playlist, spotify, navidrome):
|
||||
added_tracks += 1
|
||||
|
||||
# Calculate the match percentage for the current playlist
|
||||
if len(spotify_tracks) > 0:
|
||||
match_percentage = (added_tracks / len(spotify_tracks)) * 100
|
||||
logger.info(f"Match percentage for playlist '{playlist_name}': {match_percentage:.2f}%")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
spotify = SpotifyClient(Config.SPOTIFY_CLIENT_ID, Config.SPOTIFY_CLIENT_SECRET, Config.SPOTIFY_REDIRECT_URI,
|
||||
Config.SPOTIFY_SCOPE)
|
||||
navidrome = NavidromeClient(Config.NAVIDROME_URL, Config.NAVIDROME_USERNAME, Config.NAVIDROME_PASSWORD)
|
||||
|
||||
sync_navidrome_playlists(spotify, navidrome)
|
||||
File diff suppressed because it is too large
Load Diff
Binary file not shown.
Binary file not shown.
|
|
@ -0,0 +1,91 @@
|
|||
# codebase/emby-scripts/src/clients/navidrome_client.py
|
||||
|
||||
import requests
|
||||
import hashlib
|
||||
import time
|
||||
import random
|
||||
import string
|
||||
from urllib.parse import urlencode
|
||||
import logging
|
||||
from fuzzywuzzy import fuzz
|
||||
from src.utils.string import StringUtils
|
||||
|
||||
|
||||
class NavidromeClient:
|
||||
def __init__(self, base_url, username, password):
|
||||
self.base_url = base_url
|
||||
self.username = username
|
||||
self.password = password
|
||||
self.salt = self.generate_salt()
|
||||
self.logger = logging.getLogger(__name__)
|
||||
|
||||
def generate_salt(self, length=6):
|
||||
return ''.join(random.choices(string.ascii_lowercase + string.digits, k=length))
|
||||
|
||||
def get_auth_params(self):
|
||||
t = int(time.time() * 1000)
|
||||
token = hashlib.md5(f"{self.password}{self.salt}".encode()).hexdigest()
|
||||
return {
|
||||
'u': self.username,
|
||||
't': token,
|
||||
's': self.salt,
|
||||
'v': '1.16.1',
|
||||
'c': 'myapp',
|
||||
'f': 'json'
|
||||
}
|
||||
|
||||
def make_request(self, endpoint, params=None):
|
||||
url = f"{self.base_url}/rest/{endpoint}"
|
||||
auth_params = self.get_auth_params()
|
||||
if params:
|
||||
auth_params.update(params)
|
||||
response = requests.get(url, params=auth_params)
|
||||
response.raise_for_status()
|
||||
return response.json()
|
||||
|
||||
def get_playlists(self):
|
||||
response = self.make_request('getPlaylists')
|
||||
return response.get('subsonic-response', {}).get('playlists', {}).get('playlist', [])
|
||||
|
||||
def get_playlist_by_name(self, name):
|
||||
playlists = self.get_playlists()
|
||||
return next((pl for pl in playlists if pl['name'] == name), None)
|
||||
|
||||
def create_playlist(self, name):
|
||||
response = self.make_request('createPlaylist', {'name': name})
|
||||
return response.get('subsonic-response', {}).get('playlist')
|
||||
|
||||
def clear_playlist(self, playlist_id):
|
||||
self.make_request('updatePlaylist', {'playlistId': playlist_id, 'songIdToRemove': ''})
|
||||
|
||||
def search_track(self, title, artist):
|
||||
response = self.make_request('search3', {'query': f"{title} {artist}"})
|
||||
songs = response.get('subsonic-response', {}).get('searchResult3', {}).get('song', [])
|
||||
|
||||
best_match = None
|
||||
best_score = 0
|
||||
for song in songs:
|
||||
title_score = fuzz.ratio(StringUtils.clean_string(title), StringUtils.clean_string(song['title']))
|
||||
artist_score = fuzz.ratio(StringUtils.clean_string(artist), StringUtils.clean_string(song['artist']))
|
||||
avg_score = (title_score + artist_score) / 2
|
||||
if avg_score > best_score:
|
||||
best_score = avg_score
|
||||
best_match = song
|
||||
|
||||
return best_match if best_score > 80 else None
|
||||
|
||||
def add_track_to_playlist(self, playlist_id, track_id):
|
||||
self.make_request('updatePlaylist', {'playlistId': playlist_id, 'songIdToAdd': track_id})
|
||||
|
||||
def match_song(self, spotify_track, navidrome_track):
|
||||
spotify_title = StringUtils.clean_string(spotify_track.get("name", "").lower())
|
||||
spotify_artist = StringUtils.clean_string(spotify_track.get("artist", "").lower())
|
||||
|
||||
navidrome_title = StringUtils.clean_string(navidrome_track.get("title", "").lower())
|
||||
navidrome_artist = StringUtils.clean_string(navidrome_track.get("artist", "").lower())
|
||||
|
||||
title_ratio = fuzz.ratio(spotify_title, navidrome_title)
|
||||
artist_ratio = fuzz.ratio(spotify_artist, navidrome_artist)
|
||||
|
||||
# Adjust these thresholds as needed
|
||||
return title_ratio > 80 and artist_ratio > 80
|
||||
|
|
@ -1,10 +1,9 @@
|
|||
import logging
|
||||
import re
|
||||
import requests
|
||||
import spotipy
|
||||
from fuzzywuzzy import fuzz
|
||||
from spotipy import SpotifyOAuth
|
||||
|
||||
from src.utils.logger import logger
|
||||
from src.utils.string import StringUtils
|
||||
|
||||
|
||||
|
|
@ -85,7 +84,7 @@ class SpotifyClient:
|
|||
return False
|
||||
except Exception as e:
|
||||
|
||||
logging.error(f"Error matching song: {str(e)}")
|
||||
logger.error(f"Error matching song: {str(e)}")
|
||||
return False
|
||||
|
||||
def get_playlist_tracks(self, playlist_id):
|
||||
|
|
@ -125,6 +124,44 @@ class SpotifyClient:
|
|||
return self.sp.featured_playlists()["playlists"]
|
||||
pass
|
||||
|
||||
|
||||
|
||||
def get_categorys(self):
|
||||
return self.sp.categories(country="US")["categories"]
|
||||
|
||||
def get_category_by_name(self, name):
|
||||
for category in self.sp.categories(country="US")["categories"]["items"]:
|
||||
logger.info(f"Checking category: {category['name']} == {name}")
|
||||
if category["name"].lower() == name.lower():
|
||||
logger.info(f"Found category: {category['name']}")
|
||||
return category
|
||||
return None
|
||||
|
||||
def get_category_playlists_by_name(self, name):
|
||||
category = self.get_category_by_name(name)
|
||||
if category:
|
||||
logger.info(f"Getting playlists from category: {category['name']}")
|
||||
return self.sp.category_playlists(category["id"])["playlists"]
|
||||
return None
|
||||
|
||||
|
||||
def get_made_for_you(self):
|
||||
# get the category id for made for you
|
||||
playlists = self.get_category_playlists_by_name("Made for You")
|
||||
logger.info(f"Made for you playlists: {playlists}")
|
||||
|
||||
# pretty print all the playlists
|
||||
# for playlist in playlists['items']:
|
||||
# logger.debug(f"Playlist name: {playlist['name']}")
|
||||
# logger.debug(f"Playlist description: {playlist['description']}")
|
||||
# logger.debug(f"Playlist owner: {playlist['owner']['display_name']}")
|
||||
# logger.debug(f"Playlist tracks: {playlist['tracks']['total']}")
|
||||
# logger.debug(f"Playlist images: {playlist['images']}")
|
||||
# logger.debug(f"Playlist collaborative: {playlist['collaborative']}")
|
||||
# get the tracks in the playlists
|
||||
return playlists
|
||||
|
||||
|
||||
def remove_parentheses(text):
|
||||
if text:
|
||||
return re.sub(r'\([^)]*\)', '', text).strip()
|
||||
|
|
@ -135,7 +172,7 @@ def clean_title(title):
|
|||
# Convert to lowercase
|
||||
title = title.lower()
|
||||
|
||||
logging.info(f"Current title: {title}")
|
||||
# logging.info(f"Current title: {title}")
|
||||
|
||||
# Remove text after featuring, feat, ft, etc.
|
||||
patterns = [
|
||||
|
|
@ -190,5 +227,5 @@ def clean_title(title):
|
|||
# Remove any extra whitespace
|
||||
title = ' '.join(title.split())
|
||||
|
||||
logging.info(f"Cleaned Title: {title}")
|
||||
# logging.info(f"Cleaned Title: {title}")
|
||||
return title.strip()
|
||||
|
|
|
|||
|
|
@ -29,3 +29,7 @@ class Config:
|
|||
SONARR_URL = os.getenv('SONARR_URL')
|
||||
|
||||
LOCAL_SONARR_ROOT_PATH = os.getenv('LOCAL_SONARR_ROOT_PATH') # The local path that gets the user to the same path that sonarr uses for root folder.
|
||||
|
||||
NAVIDROME_URL = os.getenv('NAVIDROME_URL')
|
||||
NAVIDROME_USERNAME = os.getenv('NAVIDROME_USERNAME')
|
||||
NAVIDROME_PASSWORD = os.getenv('NAVIDROME_PASSWORD')
|
||||
|
|
|
|||
|
|
@ -5,7 +5,7 @@ from src.config import Config
|
|||
from src.clients.emby_client import EmbyClient
|
||||
from src.clients.spotify_client import SpotifyClient
|
||||
from src.services.playlist_service import PlaylistService
|
||||
from src.actions.sync_spotify_playlists import sync_spotify_playlists
|
||||
from src.actions.sync_spotify_to_emby_playlists import sync_spotify_playlists
|
||||
from src.actions.find_unmatched_songs import find_unmatched_songs
|
||||
from src.actions.delete_duplicate_playlists import delete_duplicate_playlists
|
||||
from src.actions.copy_emby_playlist_to_users import copy_playlists_to_users
|
||||
|
|
|
|||
Binary file not shown.
Binary file not shown.
Binary file not shown.
|
|
@ -0,0 +1,45 @@
|
|||
# codebase/emby-scripts/src/utils/logger.py
|
||||
|
||||
import logging
|
||||
from colorama import Fore, Style, init
|
||||
|
||||
init(autoreset=True)
|
||||
|
||||
class ColoredFormatter(logging.Formatter):
|
||||
COLORS = {
|
||||
'DEBUG': Fore.BLUE,
|
||||
'WARNING': Fore.YELLOW,
|
||||
'INFO': Fore.WHITE,
|
||||
'ERROR': Fore.RED,
|
||||
'CRITICAL': Fore.RED + Style.BRIGHT
|
||||
}
|
||||
|
||||
def format(self, record):
|
||||
log_color = self.COLORS.get(record.levelname, Fore.WHITE)
|
||||
log_message = super().format(record)
|
||||
return f"{log_color}{log_message}{Style.RESET_ALL}"
|
||||
|
||||
def setup_logger(name='root', level=logging.INFO):
|
||||
logger = logging.getLogger(name)
|
||||
logger.setLevel(level)
|
||||
|
||||
# Remove all handlers associated with the logger object
|
||||
for handler in logger.handlers[:]:
|
||||
logger.removeHandler(handler)
|
||||
|
||||
# Create console handler and set formatter
|
||||
ch = logging.StreamHandler()
|
||||
formatter = ColoredFormatter("%(asctime)s - %(levelname)s - %(message)s")
|
||||
ch.setFormatter(formatter)
|
||||
logger.addHandler(ch)
|
||||
|
||||
return logger
|
||||
|
||||
# Create a default logger instance
|
||||
logger = setup_logger()
|
||||
|
||||
debug = logger.debug
|
||||
info = logger.info
|
||||
warning = logger.warning
|
||||
error = logger.error
|
||||
critical = logger.critical
|
||||
|
|
@ -8,9 +8,15 @@ class StringUtils:
|
|||
"""
|
||||
return re.sub(r"[^a-zA-Z0-9\s]", "", s).lower()
|
||||
|
||||
@staticmethod
|
||||
def remove_special_characters(text):
|
||||
# Remove special characters, brackets, parentheses, and their contents
|
||||
if text is None:
|
||||
return ""
|
||||
if not isinstance(text, str):
|
||||
text = str(text)
|
||||
|
||||
cleaned_text = re.sub(r"[\[\]\(\)]*", "", text)
|
||||
cleaned_text = re.sub(r"[^a-zA-Z0-9\s]", "", cleaned_text)
|
||||
return cleaned_text
|
||||
return cleaned_text.strip()
|
||||
|
||||
|
|
|
|||
Loading…
Reference in New Issue