Phoenix
487b28e682
- Implemented a Python script for integrating Plex Media Server with Sonarr to manage TV show seasons based on viewing habits. - Added features for automated season management, exclusion list support, and detailed logging. - Created a requirements.txt file to include necessary packages: plexapi, requests, pytest, pytest-mock, and requests-mock. - Developed unit tests for the API verification function using pytest and requests_mock. - Included a standalone script to unmonitor excluded shows in Sonarr. - Added a test script for manual API verification with mock and real API tests.
208 lines
8.6 KiB
Python
208 lines
8.6 KiB
Python
from plexapi.server import PlexServer
|
||
import requests
|
||
|
||
PLEX_TOKEN = "gsyJxRUczrTPsyvopP6k"
|
||
PLEX_SERVER_URL = "http://192.168.50.111:32400"
|
||
SONARR_API_KEY = "2537de37fded4874ae83da9cf3c14f34"
|
||
SONARR_SERVER_URL = "http://192.168.50.111:8989"
|
||
|
||
plex = PlexServer(PLEX_SERVER_URL, PLEX_TOKEN)
|
||
tv_shows = plex.library.section('TV Shows')
|
||
|
||
# List of TV show titles to exclude
|
||
exclude_shows = ["Stargate SG-1", "Space Sheriff Gavan", "Spider-Man and His Amazing Friends","Super Sentai", "Superman & Lois", "UFO Robot Grendizer","Zorro",
|
||
"Saved by the Bell: The College Years","Saber Rider and the Star Sheriffs","Prison Break","Power Rangers","The Outer Limits (1995)","MacGyver","Knight Rider",
|
||
"Chuck","Breaking Bad","Amazing Stories (1985)","Airwolf","The Adventures of Superboy (1988)", "The Amazing Race"]
|
||
|
||
def unmonitor_all_excluded_shows():
|
||
for show_title in exclude_shows:
|
||
show = tv_shows.get(title=show_title)
|
||
print(f"Unmonitor Title: {show.title}")
|
||
tvdb_id = get_tvdb_id(show)
|
||
series_id = get_series_id_from_tvdb(tvdb_id)
|
||
seasons = show.seasons()
|
||
for season in seasons:
|
||
mark_season_unmonitored(series_id, season.index)
|
||
print(f"All seasons of '{show_title}' have been marked as unmonitored.")
|
||
|
||
|
||
# Function to check if the last 3 episodes of a season are watched
|
||
def last_3_episodes_watched(season):
|
||
episodes = sorted(season.episodes(), key=lambda ep: ep.index)
|
||
return all(ep.isWatched for ep in episodes[-3:])
|
||
|
||
# Function to check if the last 5 episodes of a show are watched, if not unmonitor the show
|
||
def check_last_5_episodes_and_unmonitor(show):
|
||
"""
|
||
Check if the last 5 episodes of a show are watched.
|
||
If not, mark the entire show as unmonitored in Sonarr.
|
||
"""
|
||
try:
|
||
# Get all episodes from all seasons
|
||
all_episodes = []
|
||
for season in show.seasons():
|
||
all_episodes.extend(season.episodes())
|
||
|
||
# Sort episodes by air date or episode index
|
||
all_episodes = sorted(all_episodes, key=lambda ep: (ep.seasonNumber, ep.index))
|
||
|
||
if len(all_episodes) < 5:
|
||
print(f" ⚠️ Show has less than 5 episodes total. Skipping check.")
|
||
return True
|
||
|
||
# Get the last 5 episodes
|
||
last_5_episodes = all_episodes[-5:]
|
||
watched_count = sum(1 for ep in last_5_episodes if ep.isWatched)
|
||
|
||
print(f" 📊 Last 5 episodes: {watched_count}/5 watched")
|
||
|
||
if watched_count >= 1:
|
||
print(f" ✅ At least 1 of last 5 episodes watched. Continuing normal processing...")
|
||
return True
|
||
elif watched_count == 0:
|
||
print(f" ❌ None of last 5 episodes watched. Marking show as unmonitored...")
|
||
tvdb_id = get_tvdb_id(show)
|
||
series_id = get_series_id_from_tvdb(tvdb_id)
|
||
mark_show_unmonitored(series_id)
|
||
return False
|
||
else:
|
||
print(f" ✅ Show remains monitored.")
|
||
return True
|
||
|
||
except Exception as e:
|
||
print(f" ❌ Error checking episodes: {e}")
|
||
return True
|
||
|
||
# Function to check if Sonarr API is alive and the token is correct
|
||
def verify_sonarr_api():
|
||
url = f"{SONARR_SERVER_URL}/api/v3/system/status?apikey={SONARR_API_KEY}"
|
||
try:
|
||
response = requests.get(url)
|
||
response.raise_for_status()
|
||
print("Sonarr API is alive and the token is correct.")
|
||
except requests.exceptions.RequestException as e:
|
||
print(f"Failed to verify Sonarr API: {e}")
|
||
raise
|
||
|
||
# Function to get the series ID from the TVDB ID
|
||
def get_series_id_from_tvdb(tvdb_id):
|
||
print(f"Fetching series ID for TVDB ID: {tvdb_id}")
|
||
url = f"{SONARR_SERVER_URL}/api/v3/series/lookup?term=tvdb:{tvdb_id}&apikey={SONARR_API_KEY}"
|
||
response = requests.get(url)
|
||
response.raise_for_status()
|
||
series = response.json()
|
||
if series:
|
||
print(f"Series found: {series[0]['title']} (ID: {series[0]['id']})")
|
||
return series[0]['id']
|
||
else:
|
||
raise ValueError(f"No series found for TVDB ID: {tvdb_id}")
|
||
|
||
def get_series_id_from_tvdb(tvdb_id):
|
||
url = f"{SONARR_SERVER_URL}/api/v3/series/lookup?term=tvdb:{tvdb_id}&apikey={SONARR_API_KEY}"
|
||
response = requests.get(url)
|
||
response.raise_for_status()
|
||
series = response.json()
|
||
if series:
|
||
return series[0]['id']
|
||
else:
|
||
raise ValueError(f"No series found for TVDB ID: {tvdb_id}")
|
||
|
||
# Function to mark a season as unmonitored in Sonarr v4
|
||
def mark_season_unmonitored_bak(series_id, season_number):
|
||
url = f"{SONARR_SERVER_URL}/api/v3/series/{series_id}/season/{season_number}?apikey={SONARR_API_KEY}"
|
||
response = requests.put(url, json={"monitored": False})
|
||
response.raise_for_status()
|
||
|
||
def mark_season_unmonitored(series_id, season_number):
|
||
url = f"{SONARR_SERVER_URL}/api/v3/series/{series_id}?apikey={SONARR_API_KEY}"
|
||
response = requests.get(url)
|
||
# print(f"GET Response Status Code: {response.status_code}")
|
||
# print(f"GET Response Content: {response.content}")
|
||
response.raise_for_status()
|
||
series = response.json()
|
||
|
||
for season in series['seasons']:
|
||
if season['seasonNumber'] == season_number:
|
||
if season['monitored'] == False:
|
||
# print(f"Season {season_number} is already unmonitored for series ID {series_id}. Skipping.")
|
||
return
|
||
season['monitored'] = False
|
||
break
|
||
|
||
response = requests.put(url, json=series)
|
||
# print(f"PUT Response Status Code: {response.status_code}")
|
||
# print(f"PUT Response Content: {response.content}")
|
||
response.raise_for_status()
|
||
# print(f"Season {season_number} marked as unmonitored for series ID {series_id}.")
|
||
|
||
def mark_show_unmonitored(series_id):
|
||
"""Mark an entire show as unmonitored in Sonarr"""
|
||
url = f"{SONARR_SERVER_URL}/api/v3/series/{series_id}?apikey={SONARR_API_KEY}"
|
||
response = requests.get(url)
|
||
response.raise_for_status()
|
||
series = response.json()
|
||
|
||
if series['monitored'] == False:
|
||
print(f" ℹ️ Show is already unmonitored in Sonarr. Skipping.")
|
||
return
|
||
|
||
# Mark the entire series as unmonitored
|
||
series['monitored'] = False
|
||
|
||
response = requests.put(url, json=series)
|
||
response.raise_for_status()
|
||
print(f" ✅ Show marked as unmonitored in Sonarr (Series ID: {series_id})")
|
||
|
||
|
||
def get_tvdb_id(show):
|
||
for guid in reversed(show.guids):
|
||
if guid.id.startswith("tvdb"):
|
||
return guid.id.split("://")[1]
|
||
raise ValueError(f"No series found for TVDB ID: {show.title}")
|
||
|
||
# Verify Sonarr API before proceeding
|
||
verify_sonarr_api()
|
||
|
||
# Call this function after processing the main shows
|
||
# unmonitor_all_excluded_shows()
|
||
|
||
# Iterate over all TV shows and apply the deletion rules
|
||
for show in tv_shows.all():
|
||
if show.title not in exclude_shows:
|
||
#print(f"TV Show: {show}")
|
||
|
||
print(f"\n📺 {show.title}")
|
||
print(f" 📅 Year: {show.year if show.year else 'Unknown'}")
|
||
print(f" ⭐ Rating: {show.rating if show.rating else 'N/A'}/10")
|
||
print(f" 🎬 Seasons: {show.childCount}")
|
||
print(" " + "─" * 40)
|
||
|
||
# Check if last 5 episodes are watched, if not unmonitor the show
|
||
if not check_last_5_episodes_and_unmonitor(show):
|
||
continue # Skip further processing if show was unmonitored
|
||
|
||
#print(f"Summary: {show.summary}")
|
||
#print(f"Studio: {show.studio}")
|
||
# print(f"Actors: {', '.join(actor.tag for actor in show.actors)}")
|
||
#print(f"Views: {show.viewCount}")
|
||
#print(f"Guid: {show.guid}")
|
||
#print("="*40)
|
||
|
||
seasons = sorted(show.seasons(), key=lambda s: s.index)
|
||
if len(seasons) > 1: # Ensure there is a previous season to delete
|
||
latest_season = seasons[-1]
|
||
if len(latest_season.episodes()) >= 3 and last_3_episodes_watched(latest_season):
|
||
# tvdb_id = show.guid.split('/')[-1]
|
||
tvdb_id = get_tvdb_id(show)
|
||
print(f"TVDB ID: {tvdb_id}")
|
||
series_id = get_series_id_from_tvdb(tvdb_id)
|
||
for season in seasons[:-1]: # Excluding the latest season
|
||
# Mark the season as unmonitored in Sonarr v4 before deleting
|
||
mark_season_unmonitored(series_id, season.index)
|
||
print(f" - Marking Season {season.index} as unmonitored and ready to delete")
|
||
# season.delete()
|
||
else:
|
||
for episode in latest_season.episodes():
|
||
if episode.isWatched:
|
||
print(f" - Watched Episode: {episode.title}")
|