discord-jellyseerr/utility_commands.py
2025-05-25 16:42:08 +02:00

409 lines
19 KiB
Python

import discord
from discord.ext import commands
import asyncio
import aiohttp
from typing import Optional, List, Dict, Union, Any
import logging
import config
from commands import safe_int_convert, get_api
logger = logging.getLogger('utility_commands')
class UtilityCommands(commands.Cog):
"""Commands for utility functions like help, status, and testing."""
def __init__(self, bot):
"""Initialize the UtilityCommands cog.
Args:
bot: The Discord bot instance
"""
self.bot = bot
self.embed_color = config.EMBED_COLOR
self.tmdb_base_url = "https://api.themoviedb.org/3"
@commands.command(name="help", aliases=["h"])
async def custom_help(self, ctx, command: str = None):
"""Show help information for all commands or a specific command.
Args:
ctx: The command context
command: Optional specific command to get help for
"""
if command:
cmd = self.bot.get_command(command)
if cmd:
embed = discord.Embed(
title=f"Help: {config.BOT_PREFIX}{cmd.name}",
description=cmd.help or "No description available",
color=self.embed_color
)
# Add aliases if any
if cmd.aliases:
embed.add_field(
name="Aliases",
value=", ".join([f"`{config.BOT_PREFIX}{alias}`" for alias in cmd.aliases]),
inline=False
)
# Add usage
embed.add_field(
name="Usage",
value=f"`{config.BOT_PREFIX}{cmd.name} {cmd.signature}`",
inline=False
)
await ctx.send(embed=embed)
return
else:
await ctx.send(f"Command `{command}` not found. Try `{config.BOT_PREFIX}help` to see all commands.")
return
embed = discord.Embed(
title="Jellyseerr Discord Bot Help",
description="Here are the commands you can use:",
color=self.embed_color
)
# Search & Discovery Commands
embed.add_field(
name="🔍 Search & Discovery",
value=(
f"`{config.BOT_PREFIX}search <query>` - Search for movies and TV shows\n"
f"`{config.BOT_PREFIX}info <id>` - Get detailed information about any media (auto-detects type)\n"
f"`{config.BOT_PREFIX}trending` - Show trending movies and TV shows"
),
inline=False
)
# Request Commands
embed.add_field(
name="📝 Requests",
value=(
f"`{config.BOT_PREFIX}request movie <id>` - Request a movie\n"
f"`{config.BOT_PREFIX}request tv <id> [seasons]` - Request a TV show (all or specific seasons)\n"
f"`{config.BOT_PREFIX}requests [status] [page]` - List media requests\n"
),
inline=False
)
# Other Commands
embed.add_field(
name="🔧 Other",
value=(
f"`{config.BOT_PREFIX}help` - Show this help message\n"
f"`{config.BOT_PREFIX}status` - Check Jellyseerr server status and authentication\n"
f"`{config.BOT_PREFIX}testmedia <type> <id>` - Test if media exists in TMDB and Jellyseerr\n"
),
inline=False
)
embed.set_footer(text=f"Bot Prefix: {config.BOT_PREFIX}")
await ctx.send(embed=embed)
@commands.command(name="testmedia")
async def test_media(self, ctx, media_type: str = None, media_id = None):
"""Test if a media ID exists directly on TMDB and check correct media type.
This command tests whether a given ID exists on TMDB and Jellyseerr,
and verifies the correct media type. Useful for troubleshooting.
Args:
ctx: The command context
media_type: Optional media type (movie or tv)
media_id: The TMDB ID to test
Example: !testmedia movie 83936
"""
if not media_type and not media_id:
await ctx.send("Please provide a media ID. Example: `!testmedia 83936` or `!testmedia movie 550`")
return
elif not media_id:
# If only one parameter is provided, assume it's the ID
media_id = media_type
media_type = None
await ctx.send(f"Testing both movie and TV show for ID {media_id}...")
if media_type and media_type.lower() not in ['movie', 'tv']:
await ctx.send("Invalid media type. Use 'movie' or 'tv' or just provide the ID to test both.")
return
media_id = safe_int_convert(media_id)
if not media_id or media_id <= 0:
await ctx.send(f"Please provide a valid media ID (a positive number). Example: `!testmedia 550`")
return
async with ctx.typing():
try:
# Try to get media info from Jellyseerr
api = get_api()
embed = discord.Embed(
title=f"Media Test: ID {media_id}",
description="Testing this ID against both movie and TV endpoints",
color=self.embed_color
)
# Test both movie and TV endpoints directly using HTTP requests
async with aiohttp.ClientSession() as session:
try:
# Test movie endpoint
async with session.get(f"{self.tmdb_base_url}/movie/{media_id}", timeout=aiohttp.ClientTimeout(total=15)) as movie_response:
movie_status = movie_response.status
# Test TV endpoint
async with session.get(f"{self.tmdb_base_url}/tv/{media_id}", timeout=aiohttp.ClientTimeout(total=15)) as tv_response:
tv_status = tv_response.status
# Check which type of media this ID belongs to
# 401 or 404 with specific message means it exists but auth failed
# Pure 404 means it doesn't exist
is_movie = movie_status == 401 or (movie_status == 404 and "authentication" in await movie_response.text().lower())
is_tv = tv_status == 401 or (tv_status == 404 and "authentication" in await tv_response.text().lower())
except asyncio.TimeoutError:
logger.error(f"TMDB request timed out for ID {media_id}", exc_info=True)
embed.add_field(
name="TMDB Status",
value="❌ Request to TMDB timed out. Please try again later.",
inline=False
)
return await interaction.followup.send(embed=embed)
except aiohttp.ClientConnectorError as e:
logger.error(f"TMDB connection error for ID {media_id}: {str(e)}", exc_info=True)
embed.add_field(
name="TMDB Status",
value="❌ Could not connect to TMDB. Please check your internet connection.",
inline=False
)
return await interaction.followup.send(embed=embed)
except Exception as e:
embed.add_field(
name="TMDB Status",
value=f"❌ Error checking TMDB: {str(e)}",
inline=False
)
return await interaction.followup.send(embed=embed)
if is_movie and not is_tv:
embed.add_field(
name="TMDB Status",
value="✅ This ID belongs to a **MOVIE**",
inline=False
)
correct_type = "movie"
elif is_tv and not is_movie:
embed.add_field(
name="TMDB Status",
value="✅ This ID belongs to a **TV SHOW**",
inline=False
)
correct_type = "tv"
elif is_movie and is_tv:
embed.add_field(
name="TMDB Status",
value="⚠️ Unusual: This ID seems to exist as both movie and TV show",
inline=False
)
correct_type = "both"
else:
embed.add_field(
name="TMDB Status",
value="❌ This ID was not found in TMDB database (neither movie nor TV show)",
inline=False
)
correct_type = None
# If media_type was specified, check if it matches
if media_type and correct_type and correct_type != "both" and media_type.lower() != correct_type:
embed.add_field(
name="Media Type Mismatch",
value=f"⚠️ You specified `{media_type}` but this ID is for a `{correct_type}`.\n"
f"Use `{config.BOT_PREFIX}{correct_type} {media_id}` instead.",
inline=False
)
# Test Jellyseerr access for the correct media type
try:
if correct_type == "movie" or (media_type and media_type.lower() == 'movie'):
media_details = await api.get_movie_details(media_id)
title = media_details.get('title', 'Unknown')
embed.add_field(
name="Jellyseerr Movie Status",
value=f"✅ Found in Jellyseerr: {title}",
inline=False
)
elif correct_type == "tv" or (media_type and media_type.lower() == 'tv'):
media_details = await api.get_tv_details(media_id)
title = media_details.get('name', 'Unknown')
embed.add_field(
name="Jellyseerr TV Status",
value=f"✅ Found in Jellyseerr: {title}",
inline=False
)
elif correct_type == "both":
# Try both endpoints
try:
movie_details = await api.get_movie_details(media_id)
movie_title = movie_details.get('title', 'Unknown')
embed.add_field(
name="Jellyseerr Movie Status",
value=f"✅ Found as movie in Jellyseerr: {movie_title}",
inline=False
)
except:
embed.add_field(
name="Jellyseerr Movie Status",
value="❌ Not found as movie in Jellyseerr",
inline=False
)
try:
tv_details = await api.get_tv_details(media_id)
tv_title = tv_details.get('name', 'Unknown')
embed.add_field(
name="Jellyseerr TV Status",
value=f"✅ Found as TV show in Jellyseerr: {tv_title}",
inline=False
)
except:
embed.add_field(
name="Jellyseerr TV Status",
value="❌ Not found as TV show in Jellyseerr",
inline=False
)
else: # No correct type identified
# Try both if we couldn't determine the type
tv_success = movie_success = False
try:
movie_details = await api.get_movie_details(media_id)
if movie_details.get('title'):
embed.add_field(
name="Jellyseerr Status (Unexpected)",
value=f"✅ Found as MOVIE in Jellyseerr: {movie_details.get('title')}",
inline=False
)
movie_success = True
except:
pass
try:
tv_details = await api.get_tv_details(media_id)
if tv_details.get('name'):
embed.add_field(
name="Jellyseerr Status (Unexpected)",
value=f"✅ Found as TV SHOW in Jellyseerr: {tv_details.get('name')}",
inline=False
)
tv_success = True
except:
pass
if not tv_success and not movie_success:
embed.add_field(
name="Jellyseerr Status",
value="❌ Not found in Jellyseerr (neither as movie nor TV show)",
inline=False
)
except Exception as e:
embed.add_field(
name="Jellyseerr Status",
value=f"❌ Error retrieving from Jellyseerr: {str(e)}",
inline=False
)
# Add recommendations
if not correct_type:
embed.add_field(
name="Recommendation",
value="This media ID doesn't exist in TMDB. Try searching for the correct ID with:\n"
f"`{config.BOT_PREFIX}search movie name` or `{config.BOT_PREFIX}search tv name`",
inline=False
)
elif "Error retrieving" in ''.join([f.value for f in embed.fields if hasattr(f, 'value')]):
embed.add_field(
name="Recommendation",
value="The media exists in TMDB but your Jellyseerr account cannot access it. This could be due to:\n"
"• Permission settings in your Jellyseerr account\n"
"• Content filtering in Jellyseerr\n"
"• The media may be excluded from your Jellyseerr instance",
inline=False
)
elif correct_type == "movie":
embed.add_field(
name="Command to Use",
value=f"Use `{config.BOT_PREFIX}movie {media_id}` to get details\n"
f"Use `{config.BOT_PREFIX}request movie {media_id}` to request it",
inline=False
)
elif correct_type == "tv":
embed.add_field(
name="Command to Use",
value=f"Use `{config.BOT_PREFIX}tv {media_id}` to get details\n"
f"Use `{config.BOT_PREFIX}request tv {media_id}` to request it",
inline=False
)
await ctx.send(embed=embed)
except Exception as e:
logger.error(f"Error in test_media command: {str(e)}", exc_info=True)
await ctx.send(f"Error testing media: {str(e)}")
@commands.command(name="status")
async def get_status(self, ctx):
"""Check Jellyseerr server status and authentication.
Provides information about the Jellyseerr server, including
version, update status, and authentication status.
Args:
ctx: The command context
"""
async with ctx.typing():
try:
api = get_api()
status_data = await api._request('GET', '/status')
embed = discord.Embed(
title="Jellyseerr Server Status",
color=self.embed_color
)
version = status_data.get('version', 'Unknown')
update_available = status_data.get('updateAvailable', False)
commits_behind = status_data.get('commitsBehind', 0)
embed.add_field(name="Version", value=version, inline=True)
embed.add_field(name="Update Available", value="Yes" if update_available else "No", inline=True)
if update_available:
embed.add_field(name="Commits Behind", value=str(commits_behind), inline=True)
embed.add_field(name="Server URL", value=config.JELLYSEERR_URL, inline=False)
embed.add_field(name="Bot Account", value=config.JELLYSEERR_EMAIL, inline=False)
# Check if authenticated
try:
api = get_api()
if api and api.last_login:
embed.add_field(name="Authentication", value="✅ Authenticated", inline=True)
else:
embed.add_field(name="Authentication", value="❌ Not authenticated", inline=True)
except:
embed.add_field(name="Authentication", value="❌ Not authenticated", inline=True)
await ctx.send(embed=embed)
except aiohttp.ClientConnectorError as e:
logger.error(f"Connection error while checking server status: {str(e)}", exc_info=True)
await ctx.send(f"Error connecting to Jellyseerr server at {config.JELLYSEERR_URL}. Please check if the server is running.")
except asyncio.TimeoutError:
logger.error("Status request timed out", exc_info=True)
await ctx.send(f"Request timed out. The Jellyseerr server at {config.JELLYSEERR_URL} might be slow or overloaded.")
except Exception as e:
logger.error(f"Error checking server status: {str(e)}", exc_info=True)
await ctx.send(f"Error checking server status: {str(e)}")