chore: update
This commit is contained in:
unfaiyted
parent
6545d82095
commit
62fc4201ca
|
|
@ -0,0 +1,5 @@
|
|||
.idea
|
||||
src/action/unmatched_songs.db
|
||||
/.env
|
||||
.env
|
||||
/src/actions/unmatched_songs.db
|
||||
|
|
@ -0,0 +1,15 @@
|
|||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<project version="4">
|
||||
<component name="DataSourceManagerImpl" format="xml" multifile-model="true">
|
||||
<data-source source="LOCAL" name="unmatched_songs" uuid="1e97849d-6a1a-4dd4-9931-8f37b238a790">
|
||||
<driver-ref>sqlite.xerial</driver-ref>
|
||||
<synchronize>true</synchronize>
|
||||
<jdbc-driver>org.sqlite.JDBC</jdbc-driver>
|
||||
<jdbc-url>jdbc:sqlite:C:\Users\Dane Miller\codebase\emby-scripts\src\actions\unmatched_songs.db</jdbc-url>
|
||||
<jdbc-additional-properties>
|
||||
<property name="com.intellij.clouds.kubernetes.db.enabled" value="false" />
|
||||
</jdbc-additional-properties>
|
||||
<working-dir>$ProjectFileDir$</working-dir>
|
||||
</data-source>
|
||||
</component>
|
||||
</project>
|
||||
|
|
@ -0,0 +1,6 @@
|
|||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<project version="4">
|
||||
<component name="SqlDialectMappings">
|
||||
<file url="file://$PROJECT_DIR$/src/actions/sync_spotify_liked.py" dialect="GenericSQL" />
|
||||
</component>
|
||||
</project>
|
||||
|
|
@ -0,0 +1,6 @@
|
|||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<project version="4">
|
||||
<component name="VcsDirectoryMappings">
|
||||
<mapping directory="$PROJECT_DIR$" vcs="Git" />
|
||||
</component>
|
||||
</project>
|
||||
|
|
@ -1,8 +1,14 @@
|
|||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<module type="PYTHON_MODULE" version="4">
|
||||
<component name="Go" enabled="true" />
|
||||
<component name="NewModuleRootManager" inherit-compiler-output="true">
|
||||
<exclude-output />
|
||||
<content url="file://$MODULE_DIR$" />
|
||||
<content url="file://$MODULE_DIR$">
|
||||
<excludeFolder url="file://$MODULE_DIR$/.idea/dataSources" />
|
||||
<excludeFolder url="file://$MODULE_DIR$/venv/Lib" />
|
||||
<excludeFolder url="file://$MODULE_DIR$/venv/Scripts" />
|
||||
<excludeFolder url="file://$MODULE_DIR$/venv/share" />
|
||||
</content>
|
||||
<orderEntry type="jdk" jdkName="Python 3.12 (emby-scripts)" jdkType="Python SDK" />
|
||||
<orderEntry type="sourceFolder" forTests="false" />
|
||||
</component>
|
||||
|
|
|
|||
Binary file not shown.
|
|
@ -1 +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}
|
||||
{"access_token": "BQAYOPW8HTsv1vW7-DIAmNRqqXqG7XqddSEB9217h1jxSBqIu5yjvh13FB8sE-aW5k0wtycRdTqiUPcj4fLGPXvE3iNgbO059EqDmOzPWrAdzXBIgwal3SUouI2j2z_8tYTcqFh0VzqgcVSnX5AsEyhQcdJA2QusfItEPTI1jkv69RuAXURYtloPoCPjhkfCoum4FFrvya5xDSN06sgB2ZPA9euT0_9_f8oQbXGnZFsBPA", "token_type": "Bearer", "expires_in": 3600, "scope": "playlist-read-private", "expires_at": 1715107319, "refresh_token": "AQBij3uDeV9G6Z71rKFATWpcHJLhggDPYabbTEuvebdSKEqRY_5VqS98wkx3u037d7nBlaTHIQpeSqkbbOmLdIut56TZ3Yiva56R4t6PV22LSY0wUdTYfwSjP_nWr_jP1wU"}
|
||||
|
|
@ -0,0 +1,82 @@
|
|||
import shlex
|
||||
import sqlite3
|
||||
import logging
|
||||
import subprocess
|
||||
import re
|
||||
|
||||
# Configure logging
|
||||
logging.basicConfig(
|
||||
level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s"
|
||||
)
|
||||
|
||||
# Set the path to the spotdl.exe file
|
||||
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"
|
||||
|
||||
# Connect to the SQLite database
|
||||
conn = sqlite3.connect('unmatched_songs.db')
|
||||
c = conn.cursor()
|
||||
|
||||
|
||||
def remove_special_characters(string):
|
||||
# Remove single quotes, double quotes, slashes, and other special characters
|
||||
cleaned_string = re.sub(r'[\'"\\/]', '', string)
|
||||
return cleaned_string
|
||||
|
||||
# Create the table to store downloaded songs if it doesn't exist
|
||||
c.execute('''CREATE TABLE IF NOT EXISTS downloaded_songs
|
||||
(track_name TEXT, artist_name TEXT, album_name TEXT)''')
|
||||
|
||||
# Remove duplicates from the unmatched songs table
|
||||
c.execute('''DELETE FROM unmatched_songs
|
||||
WHERE rowid NOT IN (
|
||||
SELECT MIN(rowid)
|
||||
FROM unmatched_songs
|
||||
GROUP BY track_name, artist_name, album_name
|
||||
)''')
|
||||
conn.commit()
|
||||
|
||||
# Retrieve the unmatched songs from the database
|
||||
c.execute('SELECT DISTINCT track_name, artist_name, album_name FROM unmatched_songs')
|
||||
unmatched_songs = c.fetchall()
|
||||
|
||||
# Get the total number of unmatched songs
|
||||
total_unmatched_songs = len(unmatched_songs)
|
||||
|
||||
# Process each unmatched song
|
||||
for index, song in enumerate(unmatched_songs, start=1):
|
||||
track_name, artist_name, album_name = song
|
||||
|
||||
# Check if the song has already been downloaded
|
||||
c.execute('SELECT COUNT(*) FROM downloaded_songs WHERE track_name = ? AND artist_name = ? AND album_name = ?',
|
||||
(track_name, artist_name, album_name))
|
||||
count = c.fetchone()[0]
|
||||
|
||||
if count > 0:
|
||||
logging.info(f"Skipping '{track_name}' by {artist_name} - already downloaded")
|
||||
continue
|
||||
|
||||
logging.info(f"Processing unmatched song: '{track_name}' by {artist_name}")
|
||||
|
||||
# Use spotdl to find and download the song
|
||||
try:
|
||||
command = f'"{spotdl_path}" download "{remove_special_characters(artist_name)} - {remove_special_characters(track_name)}" --output "{output_dir}"'
|
||||
logging.info(f"Running command: {command}")
|
||||
subprocess.run(command, shell=True, check=True)
|
||||
logging.info(f"Successfully downloaded '{track_name}' by {artist_name}")
|
||||
|
||||
# Insert the downloaded song into the database
|
||||
c.execute('INSERT INTO downloaded_songs VALUES (?, ?, ?)', (track_name, artist_name, album_name))
|
||||
conn.commit()
|
||||
except subprocess.CalledProcessError as e:
|
||||
logging.error(f"Error downloading '{track_name}' by {artist_name}")
|
||||
logging.error(f"Error message: {str(e)}")
|
||||
|
||||
# Calculate the percentage of songs processed
|
||||
percentage = (index / total_unmatched_songs) * 100
|
||||
logging.info(f"Progress: {index}/{total_unmatched_songs} ({percentage:.2f}%)")
|
||||
|
||||
# Close the database connection
|
||||
conn.close()
|
||||
|
|
@ -0,0 +1,173 @@
|
|||
import os
|
||||
import shutil
|
||||
import requests
|
||||
import json
|
||||
from rapidfuzz import fuzz, process
|
||||
|
||||
# Configuration
|
||||
SONARR_API_KEY = '760b6730464c443ab89bbcc96bc0b581'
|
||||
SONARR_URL = 'http://192.168.0.120:8989/api/v3'
|
||||
SOURCE_DIR = r"\\192.168.0.120\i\downloads\ORG_TV"
|
||||
DEST_BASE_PATH = r'\\192.168.0.120\raid\tv'
|
||||
MATCH_THRESHOLD = 80 # Confidence for fuzzy finding logic
|
||||
|
||||
|
||||
DRY_RUN = False
|
||||
|
||||
|
||||
def get_directories(path):
|
||||
"""Get a list of all direct children directories in the given path."""
|
||||
return [d for d in os.listdir(path) if os.path.isdir(os.path.join(path, d))]
|
||||
|
||||
|
||||
def get_all_shows():
|
||||
"""Get a list of all TV shows in Sonarr."""
|
||||
response = requests.get(f'{SONARR_URL}/series', headers={'X-Api-Key': SONARR_API_KEY})
|
||||
response.raise_for_status()
|
||||
return response.json()
|
||||
|
||||
|
||||
def search_sonarr(show_name):
|
||||
"""Search for a TV show in Sonarr by name."""
|
||||
response = requests.get(f'{SONARR_URL}/series/lookup', params={'term': show_name},
|
||||
headers={'X-Api-Key': SONARR_API_KEY})
|
||||
response.raise_for_status()
|
||||
return response.json()
|
||||
|
||||
|
||||
def get_show_path(show_id):
|
||||
"""Get the path of a TV show in Sonarr by its ID."""
|
||||
response = requests.get(f'{SONARR_URL}/series/{show_id}', headers={'X-Api-Key': SONARR_API_KEY})
|
||||
response.raise_for_status()
|
||||
return response.json()['path']
|
||||
|
||||
|
||||
def adjust_path_for_network(path):
|
||||
"""Adjust the Sonarr path to the network path."""
|
||||
network_path = path.replace('/raid/tv', DEST_BASE_PATH)
|
||||
return network_path.replace('/', '\\')
|
||||
|
||||
def merge_folders(src, dst):
|
||||
"""Merge source folder into destination folder."""
|
||||
if DRY_RUN:
|
||||
print(f'[DRY RUN] Would merge {src} into {dst}')
|
||||
return
|
||||
if not os.path.exists(dst):
|
||||
shutil.move(src, dst)
|
||||
print(f'Moved {src} to {dst}')
|
||||
else:
|
||||
for item in os.listdir(src):
|
||||
s = os.path.join(src, item)
|
||||
d = os.path.join(dst, item)
|
||||
if os.path.isdir(s):
|
||||
merge_folders(s, d)
|
||||
else:
|
||||
shutil.move(s, d)
|
||||
print(f'Moved {s} to {d}')
|
||||
os.rmdir(src)
|
||||
|
||||
def move_folders(source, destination):
|
||||
"""Move season folders from source to destination."""
|
||||
for item in os.listdir(source):
|
||||
s = os.path.join(source, item)
|
||||
d = os.path.join(destination, item)
|
||||
if os.path.isdir(s):
|
||||
merge_folders(s, d)
|
||||
|
||||
def rescan_series(show_id):
|
||||
"""Trigger a rescan of the series in Sonarr."""
|
||||
payload = {'name': 'RescanSeries', 'seriesId': show_id}
|
||||
response = requests.post(f'{SONARR_URL}/command', json=payload, headers={'X-Api-Key': SONARR_API_KEY})
|
||||
response.raise_for_status()
|
||||
return response.json()
|
||||
|
||||
def fuzzy_match(show_name, shows):
|
||||
"""Find the best fuzzy match for a show name from a list of shows."""
|
||||
choices = {show['title'].lower(): show for show in shows}
|
||||
best_match = process.extractOne(show_name.lower(), choices.keys(), scorer=fuzz.ratio)
|
||||
if best_match and best_match[1] >= MATCH_THRESHOLD:
|
||||
return choices[best_match[0]]
|
||||
return None
|
||||
|
||||
|
||||
def main():
|
||||
all_shows = get_all_shows()
|
||||
directories = get_directories(SOURCE_DIR)
|
||||
|
||||
for directory in directories:
|
||||
print(f'Processing directory: {directory}')
|
||||
|
||||
# shows = search_sonarr(directory)
|
||||
best_match = fuzzy_match(directory, all_shows)
|
||||
|
||||
if not best_match:
|
||||
# remove 4 digit year if exists.
|
||||
# redo fuzzy match w/out year
|
||||
# validate that the year of the tv show is within 2 years of the
|
||||
# year in sonarr.
|
||||
# assign match as best_match
|
||||
|
||||
# Remove 4 digit year if exists
|
||||
name_without_year = ' '.join([word for word in directory.split() if not word.isdigit() or len(word) != 4])
|
||||
best_match = fuzzy_match(name_without_year, all_shows)
|
||||
|
||||
if best_match:
|
||||
show_year = best_match.get('year', None)
|
||||
dir_year = next((word for word in directory.split() if word.isdigit() and len(word) == 4), None)
|
||||
|
||||
if dir_year and show_year:
|
||||
dir_year = int(dir_year)
|
||||
if abs(show_year - dir_year) <= 2:
|
||||
print(f'Using fuzzy match for show [{directory}] without year, matched with [{best_match["title"]}]')
|
||||
else:
|
||||
best_match = None
|
||||
else:
|
||||
best_match = None
|
||||
if best_match:
|
||||
show = best_match
|
||||
# print(json.dumps(show, indent=4))
|
||||
|
||||
if show.get("id"):
|
||||
print(f'Found show [{show["title"]}] with id [{show["id"]}]')
|
||||
|
||||
show_path = get_show_path(show['id'])
|
||||
adjusted_show_path = adjust_path_for_network(show_path)
|
||||
# print(f'Correct path for {show["title"]} is: {adjusted_show_path}')
|
||||
source_path = os.path.join(SOURCE_DIR, directory)
|
||||
# print(f'Source path: {source_path}')
|
||||
# print(f'Destination path: {adjusted_show_path}')
|
||||
|
||||
move_folders(source_path, adjusted_show_path)
|
||||
# Check if the parent directory is empty and remove it
|
||||
# if DRY_RUN:
|
||||
# print(f'[DRY RUN] Would delete {source_path} if empty')
|
||||
# else:
|
||||
# if not os.listdir(source_path):
|
||||
# os.rmdir(source_path)
|
||||
# print(f'Deleted empty directory: {source_path}')
|
||||
# else:
|
||||
# print(f'Directory not empty, did not delete: {source_path}')
|
||||
|
||||
if DRY_RUN:
|
||||
print(f'[DRY RUN] Would trigger rescan for series ID: {best_match["id"]}')
|
||||
else:
|
||||
rescan_response = rescan_series(best_match['id'])
|
||||
print(f'Triggered rescan for series ID: {best_match["id"]} with response: {rescan_response}')
|
||||
else:
|
||||
print(f'No matching show found for: {directory}')
|
||||
|
||||
#TODO: check if the directory is empty and delete it if it is.
|
||||
# Check if the directory is empty and delete it if it is
|
||||
if DRY_RUN:
|
||||
print(f'[DRY RUN] Would delete {os.path.join(SOURCE_DIR, directory)} if empty')
|
||||
else:
|
||||
directory_path = os.path.join(SOURCE_DIR, directory)
|
||||
if not os.listdir(directory_path):
|
||||
os.rmdir(directory_path)
|
||||
print(f'Deleted empty directory: {directory_path}')
|
||||
else:
|
||||
print(f'Directory not empty, did not delete: {directory_path}')
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
|
|
@ -0,0 +1,141 @@
|
|||
import os
|
||||
import re
|
||||
import shutil
|
||||
import logging
|
||||
|
||||
|
||||
def organize_episodes(root_folder, destination_root, dry_run=False):
|
||||
# Configure logging
|
||||
logging.basicConfig(level=logging.INFO, format="%(message)s")
|
||||
|
||||
# Define known movie file formats
|
||||
movie_extensions = [".avi", ".mp4", ".mkv", ".mov", ".wmv"]
|
||||
|
||||
for folder_name in os.listdir(root_folder):
|
||||
print('Looking in', folder_name)
|
||||
folder_path = os.path.join(root_folder, folder_name)
|
||||
if os.path.isdir(folder_path):
|
||||
match_full = re.match(r"(.+)\.S(\d+)E(\d+)\.(.+)", folder_name, flags=re.IGNORECASE | re.MULTILINE)
|
||||
match_season = re.match(r".*(?:S|Season)\s*(\d+).*", folder_name.replace("_","."), flags=re.IGNORECASE)
|
||||
|
||||
if match_full:
|
||||
try:
|
||||
series_name, season_number, episode_number, episode_name = match_full.groups()
|
||||
series_name = series_name.replace(".", " ").strip().title()
|
||||
episode_name = episode_name.replace(".", " ").strip()
|
||||
|
||||
# episode_name = folder_name.split(f".S{season_number}E{episode_number}.")[-1].split(".")[0].replace(".", " ").strip()
|
||||
except AttributeError:
|
||||
continue
|
||||
|
||||
season_folder = os.path.join(destination_root, series_name, f"Season {int(season_number)}")
|
||||
|
||||
# (
|
||||
# series_name,
|
||||
# season_number,
|
||||
# episode_number,
|
||||
# episode_name,
|
||||
# ) = match_full.groups()
|
||||
|
||||
# series_name = series_name.replace(
|
||||
# ".", " "
|
||||
# ).title() # Replace periods with spaces in the series name
|
||||
# season_folder = os.path.join(
|
||||
# destination_root, series_name, f"Season {int(season_number)}"
|
||||
# )
|
||||
|
||||
print('Found full match', series_name, season_number)
|
||||
|
||||
if not dry_run:
|
||||
os.makedirs(season_folder, exist_ok=True)
|
||||
|
||||
for file_name in os.listdir(folder_path):
|
||||
file_path = os.path.join(folder_path, file_name)
|
||||
if os.path.isfile(file_path):
|
||||
_, extension = os.path.splitext(file_name)
|
||||
if extension.lower() in movie_extensions:
|
||||
new_file_name = f"{series_name} S{season_number.zfill(2)}E{episode_number.zfill(2)} {episode_name.replace('.', ' ')}{extension}"
|
||||
new_file_path = os.path.join(season_folder, new_file_name)
|
||||
|
||||
if dry_run:
|
||||
logging.info(
|
||||
f"[DRY RUN] Would copy {file_name} to {new_file_path}"
|
||||
)
|
||||
else:
|
||||
print(f'[DRY RUN] Copying {file_name} to {new_file_path}')
|
||||
shutil.copy2(file_path, new_file_path)
|
||||
logging.info(f"Copied {file_name} to {new_file_path}")
|
||||
else:
|
||||
if dry_run:
|
||||
logging.info(
|
||||
f"[DRY RUN] Would delete {file_name} (unsupported format)"
|
||||
)
|
||||
else:
|
||||
os.remove(file_path)
|
||||
logging.info(
|
||||
f"Deleted {file_name} (unsupported format)"
|
||||
)
|
||||
|
||||
if dry_run:
|
||||
logging.info(f"[DRY RUN] Would remove empty folder: {folder_path}")
|
||||
else:
|
||||
shutil.rmtree(folder_path)
|
||||
logging.info(f"Removed empty folder: {folder_path}")
|
||||
elif match_season and not match_full:
|
||||
season_number = match_season.groups(1)[0]
|
||||
series_name = folder_name.split(f"S{season_number}")[0].replace(".", " ").strip().title()
|
||||
|
||||
print(f"Season {season_number} found match.")
|
||||
|
||||
season_folder = os.path.join(destination_root, series_name, f"Season {season_number}")
|
||||
|
||||
if not dry_run:
|
||||
os.makedirs(season_folder, exist_ok=True)
|
||||
|
||||
for file_name in os.listdir(folder_path):
|
||||
file_path = os.path.join(folder_path, file_name)
|
||||
if os.path.isfile(file_path):
|
||||
_, extension = os.path.splitext(file_name)
|
||||
if extension.lower() in movie_extensions:
|
||||
file_name = file_name.replace("_", ".")
|
||||
episode_match = re.match(r".*S(\d+)E(\d+)\.(.+)", file_name, flags=re.IGNORECASE)
|
||||
|
||||
if episode_match:
|
||||
print('Found episode match', episode_match)
|
||||
_, episode_number, episode_name = episode_match.groups()
|
||||
episode_name = episode_name.replace(".", " ").replace("_", " ").title()
|
||||
new_file_name = f"{series_name} S{season_number.zfill(2)}E{episode_number.zfill(2)} {episode_name}{extension}"
|
||||
new_file_path = os.path.join(season_folder, new_file_name)
|
||||
|
||||
if dry_run:
|
||||
logging.info(f"[DRY RUN] Would copy {file_name} to {new_file_path}")
|
||||
else:
|
||||
shutil.copy2(file_path, new_file_path)
|
||||
logging.info(f"Copied {file_name} to {new_file_path}")
|
||||
else:
|
||||
logging.warning(f"File {file_name} does not match episode pattern and was skipped.")
|
||||
else:
|
||||
if dry_run:
|
||||
logging.info(f"[DRY RUN] Would delete {file_name} (unsupported format)")
|
||||
else:
|
||||
os.remove(file_path)
|
||||
logging.info(f"Deleted {file_name} (unsupported format)")
|
||||
else:
|
||||
print("No match", match_full, match_season)
|
||||
|
||||
if dry_run:
|
||||
logging.info(f"[DRY RUN] Would remove empty folder: {folder_path}")
|
||||
else:
|
||||
if not os.listdir(folder_path):
|
||||
shutil.rmtree(folder_path)
|
||||
logging.info(f"Removed empty folder: {folder_path}")
|
||||
|
||||
|
||||
# Usage example
|
||||
root_folder = r"\\192.168.0.120\i\downloads\TV"
|
||||
destination_root = r"\\192.168.0.120\i\downloads\ORG_TV"
|
||||
|
||||
dry_run = False # Set False to perform actual file operations
|
||||
|
||||
print("Moving unorganized episodes to organized folder")
|
||||
organize_episodes(root_folder, destination_root, dry_run)
|
||||
|
|
@ -1,14 +1,64 @@
|
|||
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:
|
||||
season_number = int(match.group(1))
|
||||
episode_number = int(match.group(2))
|
||||
return season_number, episode_number
|
||||
else:
|
||||
raise ValueError(f"Invalid episode string format: {episode_string}")
|
||||
|
||||
# Create an instance of the Emby client
|
||||
|
||||
|
||||
# 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"]
|
||||
# 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 = {
|
||||
"The Office (US)": "S01E01", # Start from Season 2, Episode 3 of The Office ()
|
||||
"Bob's Burgers": "S01E01",
|
||||
"Parks and Recreations": "S01E01"
|
||||
}
|
||||
|
||||
# Toggle variable to control marking episodes as unwatched
|
||||
mark_as_unwatched = True
|
||||
|
||||
# Create a new playlist
|
||||
playlist_name = "Bob's Office Park"
|
||||
playlist_name = "Bob's Office Park - Mix"
|
||||
playlist_type = "shows"
|
||||
playlist = emby.create_playlist(playlist_name, playlist_type)
|
||||
|
||||
|
|
@ -31,7 +81,20 @@ for show_name in all_episodes:
|
|||
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}
|
||||
episode_counters = {}
|
||||
for show_name in tv_shows:
|
||||
start_episode_string = starting_episodes.get(show_name, "S01E01")
|
||||
start_season, start_episode = parse_episode_string(start_episode_string)
|
||||
|
||||
episode_counter = 0
|
||||
for episode in all_episodes[show_name]:
|
||||
if episode["ParentIndexNumber"] > start_season or (
|
||||
episode["ParentIndexNumber"] == start_season and episode["IndexNumber"] >= start_episode
|
||||
):
|
||||
break
|
||||
episode_counter += 1
|
||||
|
||||
episode_counters[show_name] = episode_counter
|
||||
|
||||
# Initialize a set to keep track of added episode IDs
|
||||
added_episode_ids = set()
|
||||
|
|
@ -46,12 +109,18 @@ for i in range(max_episodes):
|
|||
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}")
|
||||
|
||||
if mark_as_unwatched:
|
||||
emby.mark_as_unwatched(episode_id)
|
||||
print(f"Marked as unwatched: {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}")
|
||||
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.")
|
||||
print("Playlist creation completed.")
|
||||
|
|
|
|||
|
|
@ -0,0 +1,108 @@
|
|||
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"
|
||||
)
|
||||
|
||||
# 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"
|
||||
|
||||
# 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
|
||||
|
||||
# Regex pattern to extract year from date string
|
||||
year_pattern = r"\d{4}"
|
||||
|
||||
|
||||
def sanitize_filename(filename):
|
||||
# Replace problematic characters with underscores
|
||||
sanitized = re.sub(r'[<>:"/\\|?*]', "_", filename)
|
||||
return sanitized
|
||||
|
||||
|
||||
# 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}")
|
||||
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}")
|
||||
|
||||
# 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}")
|
||||
|
||||
# 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
|
||||
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)
|
||||
|
||||
try:
|
||||
shutil.copy2(file_path, output_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}")
|
||||
|
||||
# Check if a lyrics file exists and copy it to the new directory structure
|
||||
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}")
|
||||
|
|
@ -0,0 +1,141 @@
|
|||
import requests
|
||||
import logging
|
||||
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
|
||||
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)
|
||||
conn = sqlite3.connect('unmatched_songs.db')
|
||||
c = conn.cursor()
|
||||
|
||||
# Create the table to store unmatched songs if it doesn't exist
|
||||
c.execute('''CREATE TABLE IF NOT EXISTS unmatched_songs
|
||||
(playlist_name TEXT, track_name TEXT, artist_name TEXT, album_name TEXT)''')
|
||||
|
||||
# Get the user's playlists from Spotify
|
||||
playlists = spot.get_playlists()
|
||||
# Iterate over each Spotify playlist
|
||||
playlist_name = "Favorites"
|
||||
playlist_id = "0"
|
||||
playlist_owner = "Dane"
|
||||
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"]})"
|
||||
)
|
||||
# Get the playlist image from Spotify
|
||||
# playlist_image_data = spot.get_playlist_image(playlist_id)
|
||||
# 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}'")
|
||||
# except requests.exceptions.RequestException as e:
|
||||
# logging.warning(f"Error uploading playlist cover image: {str(e)}")
|
||||
# else:
|
||||
# logging.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)}")
|
||||
|
||||
# Get the tracks in the Spotify playlist
|
||||
tracks = spot.get_liked_songs()
|
||||
|
||||
logging.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:
|
||||
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"
|
||||
)
|
||||
unmatched_tracks.append((playlist_name, track_name, artist_name, album_name))
|
||||
|
||||
# Insert the unmatched songs into the database
|
||||
c.executemany('INSERT INTO unmatched_songs VALUES (?, ?, ?, ?)', unmatched_tracks)
|
||||
conn.commit()
|
||||
|
||||
# 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}%")
|
||||
|
||||
added_tracks = 0
|
||||
|
||||
# Close the database connection
|
||||
conn.close()
|
||||
|
|
@ -8,6 +8,7 @@ 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
|
||||
import sqlite3
|
||||
|
||||
# Configure logging
|
||||
logging.basicConfig(
|
||||
|
|
@ -19,6 +20,14 @@ spot = SpotifyClient(Config.SPOTIFY_CLIENT_ID, Config.SPOTIFY_CLIENT_SECRET, Con
|
|||
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)
|
||||
conn = sqlite3.connect('unmatched_songs.db')
|
||||
c = conn.cursor()
|
||||
|
||||
# Create the table to store unmatched songs if it doesn't exist
|
||||
c.execute('''CREATE TABLE IF NOT EXISTS unmatched_songs
|
||||
(playlist_name TEXT, track_name TEXT, artist_name TEXT, album_name TEXT)''')
|
||||
|
||||
playlists = spot.get_playlists()
|
||||
|
||||
# Iterate over each Spotify playlist
|
||||
|
|
@ -52,6 +61,22 @@ for playlist in playlists["items"]:
|
|||
logging.info(
|
||||
f"Created Emby playlist: {playlist_name} (ID: {emby_playlist["Id"]})"
|
||||
)
|
||||
# Get the playlist image from Spotify
|
||||
playlist_image_data = spot.get_playlist_image(playlist_id)
|
||||
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}'")
|
||||
except requests.exceptions.RequestException as e:
|
||||
logging.warning(f"Error uploading playlist cover image: {str(e)}")
|
||||
else:
|
||||
logging.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)}")
|
||||
|
|
@ -62,6 +87,7 @@ for playlist in playlists["items"]:
|
|||
logging.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:
|
||||
track_name = track["track"]["name"]
|
||||
artist_name = track["track"]["artists"][0]["name"]
|
||||
|
|
@ -109,5 +135,18 @@ for playlist in playlists["items"]:
|
|||
logging.warning(
|
||||
f"No match found for '{track_name}' by {artist_name} in Emby"
|
||||
)
|
||||
logging.info("Added tracks: " + str(added_tracks))
|
||||
unmatched_tracks.append((playlist_name, track_name, artist_name, album_name))
|
||||
|
||||
# Insert the unmatched songs into the database
|
||||
c.executemany('INSERT INTO unmatched_songs VALUES (?, ?, ?, ?)', unmatched_tracks)
|
||||
conn.commit()
|
||||
|
||||
# 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}%")
|
||||
|
||||
added_tracks = 0
|
||||
|
||||
# Close the database connection
|
||||
conn.close()
|
||||
|
|
@ -6,7 +6,7 @@ from src.config import Config
|
|||
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_usernames = ["Alyssa", "Dalton", "Laura", "Zuko", "Azula","Chris"]
|
||||
target_kids_usernames = ["Alyssa", "Azula", "Zuko"]
|
||||
|
||||
# Get all the playlists for the source user
|
||||
|
|
|
|||
Binary file not shown.
Binary file not shown.
|
|
@ -1,3 +1,4 @@
|
|||
import base64
|
||||
import logging
|
||||
import random
|
||||
|
||||
|
|
@ -7,41 +8,12 @@ import requests
|
|||
from requests.exceptions import Timeout
|
||||
import time
|
||||
from enum import Enum
|
||||
|
||||
from starlette.responses import guess_type
|
||||
|
||||
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):
|
||||
|
|
@ -419,16 +391,20 @@ class EmbyClient:
|
|||
# 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 upload_image(self, id, image_path, img_type='Primary'):
|
||||
mime_type = guess_type(image_path)[0]
|
||||
with open(image_path, 'rb') as f:
|
||||
image_data = f.read()
|
||||
return self.upload_image_data(id, image_data, img_type, mime_type)
|
||||
|
||||
def upload_image_data(self, id, image_data, img_type='Primary', mime_type='image/jpeg'):
|
||||
encoded_image_data = base64.b64encode(image_data)
|
||||
headers = self.headers
|
||||
headers['Content-Type'] = mime_type
|
||||
print('Uploading image')
|
||||
url = self._build_url(f'Items/{id}/Images/{img_type}')
|
||||
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'})
|
||||
|
|
@ -602,6 +578,15 @@ class EmbyClient:
|
|||
def get_unwatched_movies(self, limit=50):
|
||||
return self.get_movies(limit, is_played=False)
|
||||
|
||||
def mark_as_unwatched(self, item_id):
|
||||
url = self._build_url(f'Users/{self.user_id}/PlayedItems/{item_id}/')
|
||||
response = self._delete_request(url)
|
||||
|
||||
if response.status_code == 204:
|
||||
print(f'Item {item_id} has been marked as unwatched')
|
||||
else:
|
||||
print(f"Failed to mark item {item_id} as unwatched. Status code: {response.status_code}")
|
||||
|
||||
def get_watched_series(self, limit=50):
|
||||
return self.get_media(limit, "Series", is_played=True)
|
||||
|
||||
|
|
@ -693,20 +678,21 @@ class EmbyClient:
|
|||
# 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
|
||||
# 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)
|
||||
|
||||
# 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)
|
||||
|
|
|
|||
|
|
@ -1,5 +1,6 @@
|
|||
import logging
|
||||
|
||||
import requests
|
||||
import spotipy
|
||||
from fuzzywuzzy import fuzz
|
||||
from spotipy import SpotifyOAuth
|
||||
|
|
@ -72,4 +73,22 @@ class SpotifyClient:
|
|||
return self.sp.current_user_playlists()
|
||||
|
||||
def get_spotify(self):
|
||||
return self.sp
|
||||
return self.sp
|
||||
|
||||
def get_playlist_image(self, playlist_id):
|
||||
results = self.sp.playlist_cover_image(playlist_id)
|
||||
if results:
|
||||
image_url = results[0]['url']
|
||||
response = requests.get(image_url)
|
||||
if response.status_code == 200:
|
||||
image_data = response.content
|
||||
return image_data
|
||||
return None
|
||||
|
||||
def get_liked_songs(self):
|
||||
liked_songs = []
|
||||
results = self.sp.current_user_saved_tracks()
|
||||
while results:
|
||||
liked_songs.extend(results["items"])
|
||||
results = self.sp.next(results)
|
||||
return liked_songs
|
||||
|
|
|
|||
|
|
@ -1,35 +1,31 @@
|
|||
# config.py
|
||||
from dotenv import load_dotenv
|
||||
import os
|
||||
|
||||
load_dotenv()
|
||||
|
||||
class Config:
|
||||
|
||||
DRY_RUN = os.getenv('DRY_RUN')
|
||||
|
||||
# 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"
|
||||
SPOTIFY_CLIENT_ID = os.getenv('SPOTIFY_CLIENT_ID')
|
||||
SPOTIFY_CLIENT_SECRET = os.getenv('SPOTIFY_CLIENT_SECRET')
|
||||
SPOTIFY_REDIRECT_URI =os.getenv('SPOTIFY_REDIRECT_URI')
|
||||
SPOTIFY_SCOPE = os.getenv('SPOTIFY_SCOPE')
|
||||
|
||||
# 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"
|
||||
EMBY_URL: str = os.getenv('EMBY_URL')
|
||||
EMBY_API_KEY = os.getenv('EMBY_API_KEY')
|
||||
EMBY_USER_ID = os.getenv('EMBY_USER_ID')
|
||||
EMBY_USERNAME = os.getenv('EMBY_USERNAME')
|
||||
EMBY_PASSWORD = os.getenv('EMBY_PASSWORD')
|
||||
EMBY_CLIENT = os.getenv('EMBY_CLIENT')
|
||||
EMBY_DEVICE = os.getenv('EMBY_DEVICE')
|
||||
EMBY_DEVICE_ID = os.getenv('EMBY_DEVICE_ID')
|
||||
EMBY_VERSION = os.getenv('EMBY_VERSION')
|
||||
|
||||
SONARR_API_KEY = os.getenv('SONARR_API_KEY')
|
||||
SONARR_URL = os.getenv('SONARR_URL')
|
||||
|
||||
|
||||
# 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"
|
||||
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.
|
||||
|
|
|
|||
|
|
@ -1,12 +1,12 @@
|
|||
from src.clients.emby_client import Emby
|
||||
from src.clients.emby_client import EmbyClient
|
||||
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)
|
||||
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","Brittany","Adam"]
|
||||
target_usernames = ["Alyssa", "Dalton", "Laura", "Zuko", "Azula","Brittany","Adam","Chris"]
|
||||
|
||||
# Get all the playlists for the source user
|
||||
playlists = source_emby.get_playlists()
|
||||
|
|
@ -19,7 +19,7 @@ for playlist in playlists:
|
|||
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, '')
|
||||
target_emby = EmbyClient(Config.EMBY_URL, target_username, '')
|
||||
|
||||
# Create an instance of the PlaylistService
|
||||
playlist_service = PlaylistService(source_emby, target_emby)
|
||||
|
|
|
|||
Binary file not shown.
|
|
@ -1,5 +1,9 @@
|
|||
import unittest
|
||||
from src.clients.emby_client import Emby
|
||||
from unittest.mock import patch
|
||||
from io import BytesIO
|
||||
import requests
|
||||
|
||||
from src.clients.emby_client import EmbyClient
|
||||
from src.config import Config
|
||||
|
||||
|
||||
|
|
@ -10,7 +14,20 @@ class TestEmbyIntegration(unittest.TestCase):
|
|||
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)
|
||||
self.emby = EmbyClient(self.server_url, self.username, self.password)
|
||||
|
||||
self.test_playlist_name = "Test Playlist"
|
||||
self.test_playlist = self.emby.create_playlist(self.test_playlist_name, "Audio")
|
||||
self.test_playlist_id = self.test_playlist["Id"]
|
||||
|
||||
def tearDown(self):
|
||||
# Clean up by deleting the test playlist
|
||||
self.emby.delete_playlist(self.test_playlist_id)
|
||||
# Delete all playlists that have the name Test Playlist
|
||||
playlists = self.emby.get_playlists()
|
||||
for playlist in playlists:
|
||||
if playlist["Name"] == "Test Playlist":
|
||||
self.emby.delete_playlist(playlist["Id"])
|
||||
|
||||
def test_get_collections(self):
|
||||
collections = self.emby.get_collections()
|
||||
|
|
@ -168,5 +185,14 @@ class TestEmbyIntegration(unittest.TestCase):
|
|||
# self.assertGreater(len(search_results), 0)
|
||||
# self.assertEqual(len(search_results), 0)
|
||||
|
||||
def test_upload_playlist_cover_image(self):
|
||||
image_data = BytesIO(b'test image data')
|
||||
res = self.emby.upload_image_data(self.test_playlist_id, image_data.getvalue(), 'Primary')
|
||||
print(res, res.status_code)
|
||||
# Verify that the playlist cover image was uploaded successfully
|
||||
playlist_details = self.emby.get_list(self.test_playlist_id)
|
||||
print(playlist_details)
|
||||
self.assertTrue(playlist_details["ImageTags"].get("Primary"))
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
|
|
|
|||
|
|
@ -0,0 +1,33 @@
|
|||
from enum import Enum
|
||||
|
||||
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"
|
||||
Loading…
Reference in New Issue