Skip to content

Instantly share code, notes, and snippets.

@thegrunge36
Last active October 11, 2025 20:45
Show Gist options
  • Select an option

  • Save thegrunge36/25acd61586bbb17ac545320e346c38df to your computer and use it in GitHub Desktop.

Select an option

Save thegrunge36/25acd61586bbb17ac545320e346c38df to your computer and use it in GitHub Desktop.
torrent
#!/usr/bin/env python3
"""
🎬 ULTIMATE RADARR/SONARR CATEGORIES MANAGER 🎬
Version corrigée avec gestion d'erreurs améliorée
"""
import requests
import json
import time
import logging
from datetime import datetime
from pathlib import Path
from typing import Dict, List
import colorama
from colorama import Fore, Style
# Initialisation des couleurs
colorama.init(autoreset=True)
class UltimateCategoriesManager:
"""🎯 Gestionnaire ultime des catégories pour Radarr/Sonarr"""
def __init__(self):
self.setup_logging()
self.stats = {
'updated': 0,
'skipped': 0,
'errors': 0,
'total_time': 0
}
# ⚙️ CONFIGURATION - Modifiez ici vos paramètres
self.config = {
'radarr': {
'name': 'Radarr',
'url': 'http://localhost:7878',
'api_key': 'votre clé api',
'categories': self._get_radarr_categories()
},
'sonarr': {
'name': 'Sonarr',
'url': 'http://localhost:8989',
'api_key': 'votre clé api',
'categories': self._get_sonarr_categories()
}
}
def setup_logging(self):
"""🔧 Configuration des logs avec couleurs"""
# Logger principal
self.logger = logging.getLogger('CategoriesManager')
self.logger.setLevel(logging.INFO)
# Handler console avec couleurs
console_handler = logging.StreamHandler()
console_handler.setLevel(logging.INFO)
# Format simple et coloré
class ColoredFormatter(logging.Formatter):
COLORS = {
'INFO': Fore.GREEN,
'WARNING': Fore.YELLOW,
'ERROR': Fore.RED,
'DEBUG': Fore.BLUE,
}
def format(self, record):
log_color = self.COLORS.get(record.levelname, '')
record.levelname = f"{log_color}{record.levelname}{Style.RESET_ALL}"
return super().format(record)
formatter = ColoredFormatter('%(asctime)s | %(levelname)s | %(message)s', datefmt='%H:%M:%S')
console_handler.setFormatter(formatter)
self.logger.addHandler(console_handler)
def test_connection(self, app_name: str, url: str, api_key: str) -> bool:
"""🔍 Test de connexion rapide"""
try:
headers = {"X-Api-Key": api_key}
response = requests.get(f"{url}/api/v3/system/status", headers=headers, timeout=10)
response.raise_for_status()
version = response.json().get('version', 'Unknown')
self.logger.info(f"🟢 {app_name} connecté (v{version})")
return True
except Exception as e:
self.logger.error(f"🔴 {app_name} connexion échouée: {e}")
return False
def validate_indexer_data(self, indexer: Dict) -> bool:
"""✅ Validation des données de l'indexeur"""
required_fields = ['id', 'name', 'fields']
for field in required_fields:
if field not in indexer:
self.logger.error(f"❌ Champ manquant '{field}' dans l'indexeur")
return False
if not isinstance(indexer['fields'], list):
self.logger.error(f"❌ Le champ 'fields' doit être une liste")
return False
return True
def create_backup(self, app_name: str, indexer: Dict):
"""💾 Création d'une sauvegarde avant modification"""
try:
backup_dir = Path("backups")
backup_dir.mkdir(exist_ok=True)
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
backup_file = backup_dir / f"{app_name.lower()}_{indexer['name']}_{timestamp}.json"
with open(backup_file, 'w', encoding='utf-8') as f:
json.dump(indexer, f, indent=2, ensure_ascii=False)
self.logger.debug(f"💾 Sauvegarde: {backup_file}")
except Exception as e:
self.logger.warning(f"⚠️ Erreur sauvegarde: {e}")
def update_categories(self, app_name: str, url: str, api_key: str, categories_map: Dict[str, List[int]]) -> bool:
"""🔧 Mise à jour des catégories pour une app"""
headers = {
"X-Api-Key": api_key,
"Content-Type": "application/json"
}
try:
# Récupération des indexeurs
response = requests.get(f"{url}/api/v3/indexer", headers=headers, timeout=15)
response.raise_for_status()
indexers = response.json()
self.logger.info(f"📡 {len(indexers)} indexeurs trouvés dans {app_name}")
# Traitement de chaque indexeur
for indexer in indexers:
try:
indexer_name = indexer.get("name", "Unknown")
indexer_id = indexer.get("id")
# Validation des données
if not self.validate_indexer_data(indexer):
self.logger.error(f"❌ {indexer_name}: données invalides")
continue
if indexer_name not in categories_map:
self.logger.debug(f"⏭️ {indexer_name}: pas dans la configuration")
continue
categories = categories_map[indexer_name]
# Recherche du champ categories
categories_field = None
for field in indexer.get("fields", []):
if field.get("name") == "categories":
categories_field = field
break
if not categories_field:
self.logger.warning(f"⚠️ {indexer_name}: pas de champ 'categories'")
continue
# Vérification si mise à jour nécessaire
current_cats = set(categories_field.get("value", []))
new_cats = set(categories)
if current_cats == new_cats:
self.logger.info(f"➡️ {indexer_name}: déjà à jour")
self.stats['skipped'] += 1
continue
# Sauvegarde avant modification
self.create_backup(app_name, indexer)
# Préparation des données pour la mise à jour
update_data = indexer.copy()
# Mise à jour du champ categories
for field in update_data.get("fields", []):
if field.get("name") == "categories":
field["value"] = categories
break
# Nettoyage des champs qui peuvent poser problème
fields_to_remove = ['supportsRss', 'supportsSearch', 'protocol']
for field in fields_to_remove:
update_data.pop(field, None)
# Log des données envoyées (pour debug)
self.logger.debug(f"🔍 Données envoyées pour {indexer_name}: {json.dumps(update_data, indent=2)}")
# Mise à jour via API
update_response = requests.put(
f"{url}/api/v3/indexer/{indexer_id}",
headers=headers,
json=update_data,
timeout=15
)
# Gestion détaillée des erreurs
if update_response.status_code == 400:
error_detail = "Détails non disponibles"
try:
error_data = update_response.json()
error_detail = json.dumps(error_data, indent=2)
except:
error_detail = update_response.text
self.logger.error(f"❌ {indexer_name}: Erreur 400 - {error_detail}")
self.stats['errors'] += 1
continue
update_response.raise_for_status()
self.logger.info(f"✅ {indexer_name}: {len(categories)} catégories mises à jour")
self.stats['updated'] += 1
# Petit délai pour éviter de surcharger l'API
time.sleep(0.5)
except Exception as e:
self.logger.error(f"❌ Erreur avec {indexer_name}: {e}")
self.stats['errors'] += 1
continue
return True
except Exception as e:
self.logger.error(f"❌ Erreur générale {app_name}: {e}")
self.stats['errors'] += 1
return False
def run(self):
"""🚀 Exécution principale"""
start_time = time.time()
# Banner
print(f"""
{Fore.CYAN}{'='*70}
{Fore.YELLOW}🎬 ULTIMATE CATEGORIES MANAGER 🎬
{Fore.GREEN}Gestion automatisée des catégories Radarr/Sonarr
{Fore.CYAN}{'='*70}{Style.RESET_ALL}
""")
results = []
# Traitement de chaque application
for app_key, config in self.config.items():
self.logger.info(f"\n🚀 Traitement de {config['name']}")
# Test de connexion
if not self.test_connection(config['name'], config['url'], config['api_key']):
results.append(False)
continue
# Mise à jour des catégories
result = self.update_categories(
config['name'],
config['url'],
config['api_key'],
config['categories']
)
results.append(result)
# Statistiques finales
self.stats['total_time'] = time.time() - start_time
self.print_final_stats()
return all(results)
def print_final_stats(self):
"""📊 Statistiques finales"""
print(f"""
{Fore.CYAN}{'='*50}
{Fore.YELLOW}📊 RÉSULTATS
{Fore.GREEN}✅ Mis à jour: {self.stats['updated']}
{Fore.BLUE}➡️ Déjà à jour: {self.stats['skipped']}
{Fore.RED}❌ Erreurs: {self.stats['errors']}
{Fore.MAGENTA}⏱️ Temps: {self.stats['total_time']:.1f}s
{Fore.CYAN}{'='*50}{Style.RESET_ALL}
""")
def _get_radarr_categories(self) -> Dict[str, List[int]]:
"""🎬 Catégories Radarr"""
return {
"Abnormal (Prowlarr)": [2000, 100002, 100003, 100004],
"CrazySpirits (Prowlarr)": [2000, 2020, 2030, 2040, 2045, 2050, 2070, 2080, 100050, 100051, 100052, 100053, 100056, 100057, 100058, 100059, 100060, 100061, 100063, 100065, 100067, 100068, 100069, 100070, 100071, 100072, 100073, 100098, 100099, 100122, 100123, 100126, 100129, 100131, 100136, 100137, 100138, 100139, 1001340, 100143],
"Generation-Free (API) (Prowlarr)": [2000, 100001],
"HD-Forever (Prowlarr)": [2000, 100001, 100002, 100003, 100004, 100007],
"HD-Space (Prowlarr)": [2000, 2020, 2040, 2045, 2050, 100015, 100018, 100019, 100024, 100025, 100027, 100028, 100040, 100041, 100046, 100047, 100048],
"LeSaloon (Prowlarr)": [2000, 2030, 2040, 2045, 2050, 2070, 2080, 8000, 100013, 100015, 100018, 100019, 100020, 100026, 100027,100028, 100029, 100030, 100032, 100033, 100034, 100035, 100037,100038, 100039, 100040, 100041, 100042, 100071, 100072, 100074, 100075, 100077, 100078, 100079, 100080, 100082, 100084],
"SceneTime (Prowlarr)": [2000, 2020, 2030, 2040, 2045, 100016, 100047, 100057, 100059, 100064],
"Sharewood (API) (Prowlarr)": [2000, 2020, 8000, 8010, 100009, 100011, 100013, 100016, 100025, 100049, 100052],
"Team CT Game (Prowlarr)": [2000, 2040, 2045, 2050, 2070, 2080, 8000, 100414, 100415, 100416, 100417, 100418, 100419, 100420, 100421, 100422, 100423, 100424, 100425, 100426, 100427, 100428, 100429, 100430, 100431, 100432, 100433, 100434],
"The Old School (API) (Prowlarr)": [2000, 100001, 100006, 100010],
"TorrentLeech (Prowlarr)": [2000, 2010, 2020, 2030, 2040, 2045, 2050, 2070, 2080, 8000, 100011, 100012, 100013, 100014, 100015, 100029, 100036, 100037, 100043, 100047],
"Xthor (API) (Prowlarr)": [2000, 2020, 2030, 2040, 2045, 2050, 2070, 2080, 100001, 100002, 100004, 100005, 100006, 100007, 100008, 100009, 100012, 100020, 100031, 100033, 100094, 100095, 100100, 100107, 100118, 100119, 100122, 100125, 100126, 100127],
"YggAPI (Prowlarr)": [2000, 2020, 102145, 102178, 102179, 102180, 102181, 102182, 102183, 102184, 102185, 102186, 102187],
}
def _get_sonarr_categories(self) -> Dict[str, List[int]]:
"""📺 Catégories Sonarr"""
return {
"Abnormal (Prowlarr)": [5000, 5070, 5080, 100001, 100009],
"CrazySpirits (Prowlarr)": [5000, 5030, 5040, 5045, 5050, 5060, 5070, 5080, 100074, 100075, 100077, 100084, 100085, 100088, 100092, 100093, 100094, 100095, 100096, 100097, 100128, 100130, 100133],
"Generation-Free (API) (Prowlarr)": [5000, 100002],
"HD-Forever (Prowlarr)": [5000, 5070, 100005, 100006],
"HD-Space (Prowlarr)": [5000, 5040, 5045, 5070, 5080, 100021, 100022, 100045],
"LeSaloon (Prowlarr)": [5000, 5030, 5040, 5050, 5060, 5070, 5080, 100017, 100021, 100022, 100023, 100043, 100044, 100045, 100046, 100047, 100048, 100049, 100050, 100051, 100052, 100083, 100085, 100086, 100087, 100088],
"SceneTime (Prowlarr)": [5000, 5030, 5040, 5045, 5070, 100001, 100002, 100009, 100043, 100077],
"Sharewood (API) (Prowlarr)": [5000, 5050, 5060, 5070, 5080, 100010, 100012, 100014],
"Team CT Game (Prowlarr)": [5000, 5030, 5040, 5060, 5070, 5080, 100440, 100441, 100442, 100443, 100444, 100445, 100446, 100447, 100448, 100449, 100451, 100452, 100453],
"The Old School (API) (Prowlarr)": [5000, 5060, 100002, 100007, 100008, 100009],
"TorrentLeech (Prowlarr)": [5000, 5020, 5030, 5040, 5070, 100026, 100027, 100032, 100034, 100035, 100044, 100093, 100094, 100095, 100096, 100097, 100128, 100130, 100133],
"Xthor (API) (Prowlarr)": [5000, 5030, 5040, 5050, 5060, 5070, 5080, 100013, 100014, 100015, 100016, 100017, 100030, 100032, 100034, 100098, 100101, 100104, 100109, 100110, 100123],
"YggAPI (Prowlarr)": [5000, 5050, 5060, 5070, 5080],
}
def main():
"""🎯 Fonction principale avec mode debug"""
import argparse
parser = argparse.ArgumentParser(description='🎬 Ultimate Categories Manager')
parser.add_argument('--debug', '-d', action='store_true', help='Mode debug avec logs détaillés')
args = parser.parse_args()
try:
manager = UltimateCategoriesManager()
if args.debug:
manager.logger.setLevel(logging.DEBUG)
manager.logger.info("🔍 Mode DEBUG activé")
success = manager.run()
if success:
print(f"{Fore.GREEN}🎉 Traitement terminé avec succès !{Style.RESET_ALL}")
else:
print(f"{Fore.RED}❌ Des erreurs sont survenues{Style.RESET_ALL}")
except KeyboardInterrupt:
print(f"\n{Fore.YELLOW}⚠️ Interruption par l'utilisateur{Style.RESET_ALL}")
except Exception as e:
print(f"{Fore.RED}❌ Erreur: {e}{Style.RESET_ALL}")
if __name__ == "__main__":
main()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment