From cc9521f7a402c0339d55911f3718967ec00c2666 Mon Sep 17 00:00:00 2001 From: Danilo Reyes Date: Fri, 14 Nov 2025 02:04:11 -0600 Subject: [PATCH] Implement album deletion feature after metadata migration - Added a new command-line argument `--delete` to allow users to delete single albums after successful metadata migration. - Integrated the `unmonitor_and_delete_album` function to handle the deletion process for albums that meet the migration criteria. - Enhanced the `migrate_plex_metadata` function to support exclusion of smart playlists during migration. - Updated logging to provide detailed feedback on the deletion process and migration results. --- src-cleanup/lidarr_delete.py | 49 ++++++++++++++++++++++ src-cleanup/main.py | 52 +++++++++++++++++++++++- src-cleanup/plex_metadata.py | 79 ++++++++++++++++++++++-------------- src-cleanup/pyproject.toml | 2 +- 4 files changed, 149 insertions(+), 33 deletions(-) create mode 100644 src-cleanup/lidarr_delete.py diff --git a/src-cleanup/lidarr_delete.py b/src-cleanup/lidarr_delete.py new file mode 100644 index 0000000..eab3678 --- /dev/null +++ b/src-cleanup/lidarr_delete.py @@ -0,0 +1,49 @@ +"""Lidarr album deletion functions""" + +import logging +from typing import Dict + +import requests + +logger = logging.getLogger(__name__) + + +def unmonitor_and_delete_album( + base_url: str, headers: Dict[str, str], album_id: int +) -> bool: + """Unmonitor and delete an album from Lidarr. Returns success status.""" + try: + # First unmonitor the album + logger.debug(f"Unmonitoring album {album_id}...") + album_resp = requests.get( + f"{base_url.rstrip('/')}/api/v1/album/{album_id}", + headers=headers, + timeout=30, + ) + album_resp.raise_for_status() + album_data = album_resp.json() + + album_data["monitored"] = False + + update_resp = requests.put( + f"{base_url.rstrip('/')}/api/v1/album/{album_id}", + headers=headers, + json=album_data, + timeout=30, + ) + update_resp.raise_for_status() + + # Then delete the album (deleteFiles=true to remove files, addImportListExclusion=false) + logger.debug(f"Deleting album {album_id}...") + delete_resp = requests.delete( + f"{base_url.rstrip('/')}/api/v1/album/{album_id}", + headers=headers, + params={"deleteFiles": "true", "addImportListExclusion": "false"}, + timeout=30, + ) + delete_resp.raise_for_status() + + return True + except Exception as e: + logger.error(f"Failed to delete album {album_id}: {e}") + return False diff --git a/src-cleanup/main.py b/src-cleanup/main.py index eead51b..fb64a63 100644 --- a/src-cleanup/main.py +++ b/src-cleanup/main.py @@ -15,7 +15,8 @@ from dotenv import load_dotenv from duplicate_finder import build_album_track_map, find_duplicate_singles from lidarr_client import fetch_all_artists, fetch_albums_for_artist -from plex_metadata import get_plex_server, migrate_plex_metadata +from lidarr_delete import unmonitor_and_delete_album +from plex_metadata import get_plex_server, get_smart_playlist_ids, migrate_plex_metadata load_dotenv() @@ -56,6 +57,11 @@ def main() -> None: action="store_true", help="Migrate metadata (ratings, play counts) from singles to album tracks. Only applies to perfect matches (confidence >= 95).", ) + parser.add_argument( + "--delete", + action="store_true", + help="Delete single albums after successful metadata migration. Only deletes perfect matches (confidence >= 95) with successful migration.", + ) args = parser.parse_args() logging.basicConfig( @@ -155,8 +161,11 @@ def main() -> None: ) return + smart_playlist_ids = get_smart_playlist_ids(plex_server) + logger.info("Migrating Plex metadata for perfect matches (confidence >= 95)...") migration_count = 0 + albums_to_delete = [] for dup in duplicates: for album_track in dup.get("verified_albums", []): @@ -170,7 +179,11 @@ def main() -> None: f"Migrating Plex metadata for '{dup['track_title']}' to album '{album_track['album_title']}'..." ) success, message = migrate_plex_metadata( - plex_server, single_file_path, album_file_path, docker_mount + plex_server, + single_file_path, + album_file_path, + docker_mount, + smart_playlist_ids, ) album_track["migration_message"] = message @@ -179,12 +192,47 @@ def main() -> None: if success: migration_count += 1 logger.info(f" ✓ {message}") + if args.delete and dup["single_album_id"] not in [ + a["album_id"] for a in albums_to_delete + ]: + albums_to_delete.append( + { + "album_id": dup["single_album_id"], + "album_title": dup["single_album_title"], + "artist_name": artist_map.get( + dup["artist_id"], "Unknown" + ), + } + ) else: logger.warning(f" ✗ {message}") logger.info(f"Completed Plex metadata migration for {migration_count} track(s)") logger.info("") + if args.delete and albums_to_delete: + logger.info( + f"Deleting {len(albums_to_delete)} single album(s) from Lidarr..." + ) + deleted_count = 0 + + for album_info in albums_to_delete: + album_id = album_info["album_id"] + album_title = album_info["album_title"] + artist_name = album_info["artist_name"] + + logger.info( + f"Deleting album: {artist_name} - {album_title} (albumId: {album_id})..." + ) + if unmonitor_and_delete_album(base_url, headers, album_id): + deleted_count += 1 + logger.info(f" ✓ Successfully deleted album {album_id}") + else: + logger.error(f" ✗ Failed to delete album {album_id}") + + logger.info(f"Deleted {deleted_count}/{len(albums_to_delete)} album(s)") + logger.info("") + verified_count = sum(1 for dup in duplicates if dup.get("verified_albums")) logger.info( f"Found {len(duplicates)} single track(s) that are duplicates of album tracks ({verified_count} verified by audio fingerprint):" diff --git a/src-cleanup/plex_metadata.py b/src-cleanup/plex_metadata.py index 49a922d..23ddbcf 100644 --- a/src-cleanup/plex_metadata.py +++ b/src-cleanup/plex_metadata.py @@ -29,55 +29,71 @@ def find_plex_track_by_path( try: mapped_path = map_docker_path(file_path, docker_mount) + logger.debug( + f"Searching for track: lidarr_path={file_path}, mapped_path={mapped_path}" + ) + music_sections = [ s for s in plex_server.library.sections() if s.type == "artist" ] + if not music_sections: + logger.warning("No music sections found in Plex") + return None - # Try searching by exact mapped path first - for section in music_sections: - results = section.search(filters={"track.file": mapped_path}) - if results: - logger.debug(f"Found track by mapped path: {mapped_path}") - return results[0] - - # Try original path (might be what Plex sees in Docker) - for section in music_sections: - results = section.search(filters={"track.file": file_path}) - if results: - logger.debug(f"Found track by original path: {file_path}") - return results[0] - - # Fallback: search by filename in all tracks - filename = os.path.basename(file_path) + # Strategy: Check track.locations (list of file paths) for section in music_sections: all_tracks = section.searchTracks() for track in all_tracks: - for media in track.media: - for part in media.parts: - if part.file and ( - part.file == mapped_path - or part.file == file_path - or part.file.endswith(filename) - ): - logger.debug(f"Found track by filename match: {part.file}") - return track + track_locations = getattr(track, "locations", []) + if mapped_path in track_locations or file_path in track_locations: + logger.debug( + f"Found track by locations match: {track.title} - {track_locations[0] if track_locations else 'unknown'}" + ) + return track logger.warning( - f"Could not find Plex track for path: {file_path} (mapped: {mapped_path})" + f"Could not find Plex track. Paths tried: {file_path}, {mapped_path}" ) return None except Exception as e: - logger.debug(f"Could not find Plex track for path {file_path}: {e}") + logger.error(f"Error finding Plex track for path {file_path}: {e}") + import traceback + + logger.debug(traceback.format_exc()) return None -def get_plex_playlists_for_track(plex_server, track) -> List: - """Get all playlists containing this track""" +def get_smart_playlist_ids(plex_server) -> set: + """Get set of smart playlist IDs to exclude from migration""" + try: + smart_playlists = [ + p.ratingKey + for p in plex_server.playlists() + if p.playlistType == "audio" and p.smart + ] + if smart_playlists: + logger.info( + f"Found {len(smart_playlists)} smart playlists (will exclude from migration)" + ) + return set(smart_playlists) + except Exception as e: + logger.debug(f"Could not get smart playlists: {e}") + return set() + + +def get_plex_playlists_for_track( + plex_server, track, exclude_smart_playlists: set = None +) -> List: + """Get all manual playlists containing this track (excludes smart playlists)""" + if exclude_smart_playlists is None: + exclude_smart_playlists = set() + try: return [ playlist for playlist in plex_server.playlists() if playlist.playlistType == "audio" + and playlist.ratingKey not in exclude_smart_playlists and any(item.ratingKey == track.ratingKey for item in playlist.items()) ] except Exception as e: @@ -216,6 +232,7 @@ def migrate_plex_metadata( single_file_path: str, album_file_path: str, docker_mount: Optional[str] = None, + exclude_smart_playlists: set = None, ) -> Tuple[bool, str]: """Migrate Plex metadata from single to album track. Returns (success, message)""" if not plex_server: @@ -231,7 +248,9 @@ def migrate_plex_metadata( single_rating = getattr(single_track, "userRating", None) single_plays = getattr(single_track, "viewCount", 0) or 0 - single_playlists = get_plex_playlists_for_track(plex_server, single_track) + single_playlists = get_plex_playlists_for_track( + plex_server, single_track, exclude_smart_playlists + ) logger.info( f" Single track metadata: rating={single_rating or 'none'}, plays={single_plays}, playlists={len(single_playlists)}" diff --git a/src-cleanup/pyproject.toml b/src-cleanup/pyproject.toml index c7b0e91..283ab09 100644 --- a/src-cleanup/pyproject.toml +++ b/src-cleanup/pyproject.toml @@ -3,7 +3,7 @@ requires = ["setuptools"] build-backend = "setuptools.build_meta" [tool.setuptools] -py-modules = ["main", "lidarr_client", "audio_verification", "track_verification", "plex_metadata", "duplicate_finder"] +py-modules = ["main", "lidarr_client", "audio_verification", "track_verification", "plex_metadata", "duplicate_finder", "lidarr_delete"] [project] name = "lidarr-cleanup-singles"