king/ularsawa
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add Plex Sonarr integration script and testing framework

Browse Source 
- 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.
This commit is contained in:
Phoenix
2025-08-20 23:59:05 +08:00
parent 5413d621d4
commit 487b28e682
13 changed files with 874 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files
plexsonarr.py
+78 -8
View File
@@ -1,7 +1,7 @@
from plexapi.server import PlexServer
import requests
PLEX_TOKEN = "uZn42JMVkQpyb_duFsvT"
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"
@@ -12,7 +12,7 @@ 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)"]
"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:
@@ -31,6 +31,48 @@ 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}"
@@ -81,13 +123,35 @@ def mark_season_unmonitored(series_id, season_number):
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}.")
# 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):
@@ -107,13 +171,19 @@ for show in tv_shows.all():
if show.title not in exclude_shows:
#print(f"TV Show: {show}")
print(f"Title: {show.title}")
#print(f"Year: {show.year}")
#print(f"Rating: {show.rating}")
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"Seasons: {show.childCount}")
# print(f"Actors: {', '.join(actor.tag for actor in show.actors)}")
#print(f"Views: {show.viewCount}")
#print(f"Guid: {show.guid}")
#print("="*40)