#!/usr/bin/env python3 """ Script de récapitulatif hebdomadaire Plex pour Discord Récupère les nouveautés de Jellyseerr, Radarr et Sonarr et les envoie sur Discord Usage: python3 plex-weekly-recap.py --movies # Envoie uniquement les films python3 plex-weekly-recap.py --series # Envoie uniquement les séries python3 plex-weekly-recap.py --requests # Envoie uniquement les demandes python3 plex-weekly-recap.py --all # Envoie tout (par défaut) """ import json import sys import os import argparse from datetime import datetime, timedelta from typing import List, Dict, Optional import requests from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry # Configuration du chemin du fichier de config CONFIG_FILE = os.path.join(os.path.dirname(__file__), 'plex-recap-config.json') LOG_FILE = os.path.join(os.path.dirname(__file__), 'plex-recap.log') def log(message: str, level: str = "INFO"): """Écrit un message dans le fichier de log""" timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S") log_message = f"[{timestamp}] [{level}] {message}\n" try: with open(LOG_FILE, 'a', encoding='utf-8') as f: f.write(log_message) except Exception as e: print(f"Erreur lors de l'écriture du log: {e}") if level == "ERROR": print(log_message.strip()) def load_config() -> Dict: """Charge la configuration depuis le fichier JSON""" try: with open(CONFIG_FILE, 'r', encoding='utf-8') as f: config = json.load(f) return config except FileNotFoundError: log(f"Fichier de configuration non trouvé: {CONFIG_FILE}", "ERROR") sys.exit(1) except json.JSONDecodeError as e: log(f"Erreur de parsing JSON dans la configuration: {e}", "ERROR") sys.exit(1) def create_session() -> requests.Session: """Crée une session requests avec retry automatique""" session = requests.Session() retry_strategy = Retry( total=3, backoff_factor=1, status_forcelist=[429, 500, 502, 503, 504], ) adapter = HTTPAdapter(max_retries=retry_strategy) session.mount("http://", adapter) session.mount("https://", adapter) return session def get_jellyseerr_requests(config: Dict, days: int = 7) -> List[Dict]: """Récupère les demandes approuvées de Jellyseerr des derniers jours""" jellyseerr_config = config.get('jellyseerr', {}) url = jellyseerr_config.get('url', 'http://localhost:5055') api_key = jellyseerr_config.get('api_key', '') if not api_key: log("Clé API Jellyseerr non configurée", "ERROR") return [] # Calcul de la date limite cutoff_date = datetime.now() - timedelta(days=days) try: session = create_session() headers = {'X-Api-Key': api_key} # Récupération de toutes les demandes disponibles # On récupère toutes les demandes et on filtre ensuite response = session.get( f"{url}/api/v1/request?take=100", headers=headers, timeout=10 ) response.raise_for_status() requests_data = response.json() results = requests_data.get('results', []) # Filtrage par date (demandes approuvées/disponibles dans les derniers jours) filtered_requests = [] for req in results: # Vérifier que la demande est disponible status = req.get('status', 0) # Status 2 = Available, Status 3 = Partially Available if status in [2, 3]: # Vérifier la date de modification ou d'approbation modified = req.get('modified', '') if modified: try: # Gérer différents formats de date modified_str = modified.replace('Z', '+00:00') if '.' in modified_str and '+' in modified_str: # Format ISO avec timezone modified_date = datetime.fromisoformat(modified_str) else: # Format ISO simple modified_date = datetime.fromisoformat(modified_str.split('+')[0]) if modified_date.replace(tzinfo=None) >= cutoff_date: filtered_requests.append(req) except (ValueError, AttributeError) as e: log(f"Erreur de parsing de date pour demande {req.get('id', 'unknown')}: {e}", "ERROR") continue log(f"Récupéré {len(filtered_requests)} demandes Jellyseerr des {days} derniers jours") return filtered_requests except requests.exceptions.RequestException as e: log(f"Erreur lors de la récupération des demandes Jellyseerr: {e}", "ERROR") return [] def get_radarr_movies(config: Dict, days: int = 7) -> List[Dict]: """Récupère les nouveaux films de Radarr des derniers jours""" radarr_config = config.get('radarr', {}) hostname = radarr_config.get('hostname', 'localhost') port = radarr_config.get('port', 7878) api_key = radarr_config.get('api_key', '') use_ssl = radarr_config.get('use_ssl', False) if not api_key: log("Clé API Radarr non configurée", "ERROR") return [] protocol = 'https' if use_ssl else 'http' url = f"{protocol}://{hostname}:{port}" # Calcul de la date limite cutoff_date = datetime.now() - timedelta(days=days) try: session = create_session() params = {'apikey': api_key} response = session.get( f"{url}/api/v3/movie", params=params, timeout=10 ) response.raise_for_status() movies = response.json() # Filtrage par date d'ajout new_movies = [] for movie in movies: added = movie.get('added', '') if added: try: added_date = datetime.fromisoformat(added.replace('Z', '+00:00')) if added_date.replace(tzinfo=None) >= cutoff_date: # Vérifier que le film est disponible if movie.get('hasFile', False): new_movies.append(movie) except (ValueError, AttributeError): pass log(f"Récupéré {len(new_movies)} nouveaux films Radarr des {days} derniers jours") return new_movies except requests.exceptions.RequestException as e: log(f"Erreur lors de la récupération des films Radarr: {e}", "ERROR") return [] def get_sonarr_series(config: Dict, days: int = 7) -> List[Dict]: """Récupère les séries avec de nouveaux épisodes de Sonarr des derniers jours""" sonarr_config = config.get('sonarr', {}) hostname = sonarr_config.get('hostname', 'localhost') port = sonarr_config.get('port', 8989) api_key = sonarr_config.get('api_key', '') use_ssl = sonarr_config.get('use_ssl', False) if not api_key: log("Clé API Sonarr non configurée", "ERROR") return [] protocol = 'https' if use_ssl else 'http' url = f"{protocol}://{hostname}:{port}" # Calcul de la date limite cutoff_date = datetime.now() - timedelta(days=days) log(f"Recherche d'épisodes ajoutés après le {cutoff_date.strftime('%Y-%m-%d %H:%M:%S')}") try: session = create_session() params = {'apikey': api_key} # D'abord récupérer toutes les séries pour avoir les infos response_series = session.get( f"{url}/api/v3/series", params=params, timeout=10 ) response_series.raise_for_status() all_series = response_series.json() series_map = {s.get('id'): s for s in all_series} log(f"Récupéré {len(series_map)} séries depuis Sonarr") # Utiliser l'historique pour trouver les épisodes récemment téléchargés # L'historique contient les événements de téléchargement avec les dates cutoff_timestamp = int(cutoff_date.timestamp()) history_params = { 'apikey': api_key, 'page': 1, 'pageSize': 1000, 'sortKey': 'date', 'sortDirection': 'descending' } response_history = session.get( f"{url}/api/v3/history", params=history_params, timeout=10 ) response_history.raise_for_status() history_data = response_history.json() history_records = history_data.get('records', []) log(f"Récupéré {len(history_records)} enregistrements d'historique depuis Sonarr") # Filtrer les épisodes téléchargés récemment new_episodes_by_series = {} event_types_found = set() for record in history_records: event_type = record.get('eventType', '') event_types_found.add(event_type) # Vérifier que c'est un événement de téléchargement d'épisode # Les types peuvent varier selon la version de Sonarr if event_type in ['downloadFolderImported', 'grabbed', 'episodeFileDeleted', 'episodeFileRenamed']: date_str = record.get('date', '') if date_str: try: # Gérer différents formats de date date_str_clean = date_str.replace('Z', '+00:00') if '.' in date_str_clean and '+' in date_str_clean: event_date = datetime.fromisoformat(date_str_clean) else: event_date = datetime.fromisoformat(date_str_clean.split('+')[0]) if event_date.replace(tzinfo=None) >= cutoff_date: # Récupérer les infos de l'épisode depuis l'historique episode_info = record.get('episode', {}) if not episode_info: # Parfois l'info est directement dans le record episode_info = record series_id = episode_info.get('seriesId', 0) if series_id and series_id in series_map: series_info = series_map[series_id] if series_id not in new_episodes_by_series: # Récupérer la langue originale original_language = series_info.get('originalLanguage', {}) language_name = original_language.get('name', '') if isinstance(original_language, dict) else '' new_episodes_by_series[series_id] = { 'seriesId': series_id, 'title': series_info.get('title', 'Titre inconnu'), 'year': series_info.get('year', ''), 'originalLanguage': language_name, 'episodeCount': 0, 'episodes': [] } new_episodes_by_series[series_id]['episodeCount'] += 1 except (ValueError, AttributeError) as e: log(f"Erreur de parsing de date pour historique: {date_str} - {e}", "ERROR") pass log(f"Types d'événements trouvés dans l'historique: {', '.join(sorted(event_types_found))}") log(f"Épisodes trouvés dans l'historique: {len(new_episodes_by_series)} séries") # Si l'historique ne donne pas assez d'infos, parcourir les séries et leurs épisodes if len(new_episodes_by_series) == 0: log("Aucun épisode trouvé dans l'historique, parcours des séries...") series_checked = 0 episodes_found_in_series = 0 for series in all_series: series_id = series.get('id', 0) if not series_id: continue series_checked += 1 # Récupérer les épisodes de cette série episode_params = {'apikey': api_key, 'seriesId': series_id} try: response_episodes = session.get( f"{url}/api/v3/episode", params=episode_params, timeout=10 ) response_episodes.raise_for_status() episodes = response_episodes.json() # Filtrer les épisodes récemment ajoutés for episode in episodes: if episode.get('hasFile', False): episode_file = episode.get('episodeFile', {}) if episode_file: date_added = episode_file.get('dateAdded', '') if date_added: try: date_str_clean = date_added.replace('Z', '+00:00') if '.' in date_str_clean and '+' in date_str_clean: added_date = datetime.fromisoformat(date_str_clean) else: added_date = datetime.fromisoformat(date_str_clean.split('+')[0]) if added_date.replace(tzinfo=None) >= cutoff_date: episodes_found_in_series += 1 if series_id not in new_episodes_by_series: # Récupérer la langue originale original_language = series.get('originalLanguage', {}) language_name = original_language.get('name', '') if isinstance(original_language, dict) else '' new_episodes_by_series[series_id] = { 'seriesId': series_id, 'title': series.get('title', 'Titre inconnu'), 'year': series.get('year', ''), 'originalLanguage': language_name, 'episodeCount': 0, 'episodes': [] } new_episodes_by_series[series_id]['episodeCount'] += 1 except (ValueError, AttributeError) as e: log(f"Erreur de parsing date épisode: {date_added} - {e}", "ERROR") pass except requests.exceptions.RequestException as e: log(f"Erreur lors de la récupération des épisodes pour série {series_id}: {e}", "ERROR") continue log(f"Parcours terminé: {series_checked} séries vérifiées, {episodes_found_in_series} épisodes trouvés") # Convertir en liste new_series = list(new_episodes_by_series.values()) log(f"Récupéré {len(new_series)} séries avec nouveaux épisodes Sonarr des {days} derniers jours ({sum(s['episodeCount'] for s in new_series)} épisodes au total)") return new_series except requests.exceptions.RequestException as e: log(f"Erreur lors de la récupération des séries Sonarr: {e}", "ERROR") if hasattr(e, 'response') and e.response is not None: try: log(f"Réponse API: {e.response.text[:200]}", "ERROR") except: pass return [] def format_movie_title(movie: Dict) -> str: """Formate le titre d'un film avec année et qualité""" title = movie.get('title', 'Titre inconnu') year = movie.get('year', '') quality = '' # Récupération de la qualité depuis le fichier movie_file = movie.get('movieFile', {}) if movie_file: quality_profile = movie_file.get('quality', {}) if quality_profile: quality_name = quality_profile.get('quality', {}).get('name', '') if quality_name: quality = f" ({quality_name})" if year: return f"**{title}** ({year}){quality}" return f"**{title}**{quality}" def format_series_title(series: Dict) -> str: """Formate le titre d'une série avec année et nombre d'épisodes""" title = series.get('title', 'Titre inconnu') year = series.get('year', '') episode_count = series.get('episodeCount', 0) if year: return f"**{title}** ({year}) - {episode_count} épisode(s)" return f"**{title}** - {episode_count} épisode(s)" def format_request_title(request: Dict) -> str: """Formate le titre d'une demande Jellyseerr""" media = request.get('media', {}) media_type = request.get('type', '') if media_type == 'movie': title = media.get('title', 'Titre inconnu') year = media.get('releaseDate', '') if year: year = year[:4] # Extraire l'année return f"**{title}** ({year})" return f"**{title}**" else: # TV show title = media.get('name', 'Titre inconnu') season = request.get('seasons', []) if season: season_num = season[0].get('seasonNumber', '') return f"**{title}** - Saison {season_num}" return f"**{title}**" def split_text_field(text: str, max_length: int = 1024) -> List[str]: """Divise un texte en plusieurs parties si trop long pour Discord""" if len(text) <= max_length: return [text] parts = [] lines = text.split('\n') current_part = "" for line in lines: # Si la ligne seule dépasse la limite, on la tronque if len(line) > max_length: if current_part: parts.append(current_part) current_part = "" # Tronquer la ligne parts.append(line[:max_length-3] + "...") # Si ajouter la ligne dépasse la limite, sauvegarder la partie actuelle elif len(current_part) + len(line) + 1 > max_length: if current_part: parts.append(current_part) current_part = line else: # Ajouter la ligne à la partie actuelle if current_part: current_part += "\n" + line else: current_part = line if current_part: parts.append(current_part) return parts def create_movies_embed(config: Dict, movies: List[Dict], days: int = 7) -> List[Dict]: """Crée l'embed Discord pour les films uniquement""" embed_color = config.get('discord', {}).get('embed_color', 0x00ff00) # Calcul des dates de la semaine now = datetime.now() week_start = now - timedelta(days=days) week_start_str = week_start.strftime("%d/%m/%Y") week_end_str = now.strftime("%d/%m/%Y") embeds = [] if movies: movies_text = "\n".join([format_movie_title(m) for m in movies]) movies_parts = split_text_field(movies_text, 1024) for i, part in enumerate(movies_parts): embed = { "title": "🎬 Nouveaux Films Plex" if i == 0 else "🎬 Nouveaux Films Plex (suite)", "description": f"Films ajoutés du {week_start_str} au {week_end_str}" if i == 0 else "", "color": embed_color, "timestamp": now.isoformat(), "fields": [{ "name": f"🎬 Nouveaux Films ({len(movies)})" if i == 0 else f"🎬 Suite ({len(movies)} au total)", "value": part, "inline": False }], "footer": { "text": "Récapitulatif automatique Plex" } } embeds.append(embed) else: # Aucun film embed = { "title": "🎬 Nouveaux Films Plex", "description": f"Films ajoutés du {week_start_str} au {week_end_str}", "color": embed_color, "timestamp": now.isoformat(), "fields": [{ "name": "ℹ️ Aucun nouveau film", "value": "Aucun nouveau film cette semaine.", "inline": False }], "footer": { "text": "Récapitulatif automatique Plex" } } embeds.append(embed) return embeds def create_series_embed(config: Dict, series: List[Dict], days: int = 7) -> List[Dict]: """Crée l'embed Discord pour les séries uniquement, séparées en séries normales et manga""" embed_color = config.get('discord', {}).get('embed_color', 0x00ff00) # Calcul des dates de la semaine now = datetime.now() week_start = now - timedelta(days=days) week_start_str = week_start.strftime("%d/%m/%Y") week_end_str = now.strftime("%d/%m/%Y") embeds = [] if series: # Séparer les séries normales et les manga (Japanese/Chinese) normal_series = [] manga_series = [] for s in series: original_language = s.get('originalLanguage', '').lower() if original_language in ['japanese', 'chinese']: manga_series.append(s) else: normal_series.append(s) # Créer l'embed principal main_embed = { "title": "📺 Nouveautés Séries Plex", "description": f"Séries ajoutées du {week_start_str} au {week_end_str}", "color": embed_color, "timestamp": now.isoformat(), "fields": [], "footer": { "text": "Récapitulatif automatique Plex" } } # Section Séries normales if normal_series: normal_text = "\n".join([format_series_title(s) for s in normal_series]) normal_parts = split_text_field(normal_text, 1024) for i, part in enumerate(normal_parts): if i == 0: main_embed["fields"].append({ "name": f"📺 Nouveautés Séries ({len(normal_series)})", "value": part, "inline": False }) else: # Créer un embed supplémentaire pour la suite if len(embeds) == 0: embeds.append({ "color": embed_color, "fields": [] }) embeds[0]["fields"].append({ "name": f"📺 Nouveautés Séries (suite)", "value": part, "inline": False }) # Section Manga if manga_series: manga_text = "\n".join([format_series_title(s) for s in manga_series]) manga_parts = split_text_field(manga_text, 1024) for i, part in enumerate(manga_parts): if i == 0: main_embed["fields"].append({ "name": f"🏮 Nouveautés Manga ({len(manga_series)})", "value": part, "inline": False }) else: # Utiliser le dernier embed ou en créer un nouveau if len(embeds) == 0: embeds.append({ "color": embed_color, "fields": [] }) embeds[0]["fields"].append({ "name": f"🏮 Nouveautés Manga (suite)", "value": part, "inline": False }) # Si aucune série dans aucune catégorie (ne devrait pas arriver) if not normal_series and not manga_series: main_embed["fields"].append({ "name": "ℹ️ Aucune Nouveautés série", "value": "Aucune Nouveautés série cette semaine.", "inline": False }) embeds.insert(0, main_embed) else: # Aucune série embed = { "title": "📺 Nouveautés Séries Plex", "description": f"Séries ajoutées du {week_start_str} au {week_end_str}", "color": embed_color, "timestamp": now.isoformat(), "fields": [{ "name": "ℹ️ Aucune Nouveautés série", "value": "Aucune Nouveautés série cette semaine.", "inline": False }], "footer": { "text": "Récapitulatif automatique Plex" } } embeds.append(embed) return embeds def create_requests_embed(config: Dict, requests: List[Dict], days: int = 7) -> List[Dict]: """Crée l'embed Discord pour les demandes approuvées uniquement""" embed_color = config.get('discord', {}).get('embed_color', 0x00ff00) # Calcul des dates de la semaine now = datetime.now() week_start = now - timedelta(days=days) week_start_str = week_start.strftime("%d/%m/%Y") week_end_str = now.strftime("%d/%m/%Y") embeds = [] if requests: requests_text = "\n".join([format_request_title(r) for r in requests]) requests_parts = split_text_field(requests_text, 1024) for i, part in enumerate(requests_parts): embed = { "title": "✅ Demandes approuvées Jellyseerr" if i == 0 else "✅ Demandes approuvées Jellyseerr (suite)", "description": f"Demandes approuvées du {week_start_str} au {week_end_str}" if i == 0 else "", "color": embed_color, "timestamp": now.isoformat(), "fields": [{ "name": f"✅ Demandes approuvées ({len(requests)})" if i == 0 else f"✅ Suite ({len(requests)} au total)", "value": part, "inline": False }], "footer": { "text": "Récapitulatif automatique Plex" } } embeds.append(embed) else: # Aucune demande embed = { "title": "✅ Demandes approuvées Jellyseerr", "description": f"Demandes approuvées du {week_start_str} au {week_end_str}", "color": embed_color, "timestamp": now.isoformat(), "fields": [{ "name": "ℹ️ Aucune demande approuvée", "value": "Aucune demande approuvée cette semaine.", "inline": False }], "footer": { "text": "Récapitulatif automatique Plex" } } embeds.append(embed) return embeds def send_discord_webhook(config: Dict, embeds: List[Dict]) -> bool: """Envoie le message Discord via webhook (peut envoyer plusieurs embeds)""" discord_config = config.get('discord', {}) webhook_url = discord_config.get('webhook_url', '') if not webhook_url: log("URL du webhook Discord non configurée", "ERROR") return False # Discord limite à 10 embeds par message max_embeds_per_message = 10 try: session = create_session() # Diviser les embeds en groupes de 10 for i in range(0, len(embeds), max_embeds_per_message): embeds_batch = embeds[i:i + max_embeds_per_message] payload = { "embeds": embeds_batch } response = session.post( webhook_url, json=payload, timeout=10 ) response.raise_for_status() log(f"Message Discord envoyé avec succès ({len(embeds_batch)} embed(s))") return True except requests.exceptions.RequestException as e: log(f"Erreur lors de l'envoi du message Discord: {e}", "ERROR") if hasattr(e, 'response') and e.response is not None: try: error_detail = e.response.json() log(f"Détails de l'erreur Discord: {error_detail}", "ERROR") except: log(f"Réponse brute: {e.response.text[:200]}", "ERROR") return False def send_movies(config: Dict, days: int = 7) -> bool: """Envoie uniquement les films sur Discord""" log("Récupération des films...") movies = get_radarr_movies(config, days) embeds = create_movies_embed(config, movies, days) return send_discord_webhook(config, embeds) def send_series(config: Dict, days: int = 7) -> bool: """Envoie uniquement les séries sur Discord""" log("Récupération des séries...") series = get_sonarr_series(config, days) embeds = create_series_embed(config, series, days) return send_discord_webhook(config, embeds) def send_requests(config: Dict, days: int = 7) -> bool: """Envoie uniquement les demandes approuvées sur Discord""" log("Récupération des demandes...") requests = get_jellyseerr_requests(config, days) embeds = create_requests_embed(config, requests, days) return send_discord_webhook(config, embeds) def send_all(config: Dict, days: int = 7) -> bool: """Envoie tous les récapitulatifs (films, séries, demandes) sur Discord""" log("Récupération de toutes les données...") movies = get_radarr_movies(config, days) series = get_sonarr_series(config, days) requests = get_jellyseerr_requests(config, days) # Envoyer chaque type séparément success_movies = send_movies(config, days) success_series = send_series(config, days) success_requests = send_requests(config, days) return success_movies and success_series and success_requests def main(): """Fonction principale""" parser = argparse.ArgumentParser( description='Récapitulatif hebdomadaire Plex pour Discord', formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" Exemples: python3 plex-weekly-recap.py --movies # Envoie uniquement les films python3 plex-weekly-recap.py --series # Envoie uniquement les séries python3 plex-weekly-recap.py --requests # Envoie uniquement les demandes python3 plex-weekly-recap.py --all # Envoie tout (par défaut) """ ) parser.add_argument('--movies', action='store_true', help='Envoyer uniquement les films') parser.add_argument('--series', action='store_true', help='Envoyer uniquement les séries') parser.add_argument('--requests', action='store_true', help='Envoyer uniquement les demandes approuvées') parser.add_argument('--all', action='store_true', help='Envoyer tout (films, séries, demandes)') parser.add_argument('--days', type=int, default=None, help='Nombre de jours à considérer (défaut: depuis config)') args = parser.parse_args() log("Démarrage du récapitulatif hebdomadaire Plex") # Chargement de la configuration config = load_config() # Nombre de jours à considérer days = args.days if args.days is not None else config.get('days', 7) success = False # Déterminer ce qu'on doit envoyer if args.movies: log("Mode: Films uniquement") success = send_movies(config, days) elif args.series: log("Mode: Séries uniquement") success = send_series(config, days) elif args.requests: log("Mode: Demandes uniquement") success = send_requests(config, days) else: # Par défaut, envoyer tout log("Mode: Tout (films, séries, demandes)") success = send_all(config, days) if success: log("Récapitulatif envoyé avec succès") sys.exit(0) else: log("Échec de l'envoi du récapitulatif", "ERROR") sys.exit(1) if __name__ == "__main__": main()