import datetime
import json
from argparse import ArgumentParser, Namespace, _SubParsersAction
from pprint import pformat
from typing import Any, List, Optional, Union, cast, no_type_check
from pycliarr.api import base_api, base_media, exceptions, radarr, sonarr
from pycliarr.cli.utils import size_to_str
[docs]class CliCommand:
"""Base command, all command should extend this class."""
name = ""
description = ""
def __init__(self) -> None:
pass
[docs] def run(self, cli: Any, args: Namespace) -> None:
pass
[docs]class CliApiCommand:
"""Definition of an API client.
Allows instantiating the relevant communication client, and execute a subcommmand from its name.
"""
def __init__(self, name: str, cli_class: Any, commands: List[CliCommand]) -> None:
self.name = name
self.cli_class = cli_class
self.cmd_list = {cmd.name: cmd for cmd in commands}
def _new_client(self, host: str, api_key: str, username: Optional[str], password: Optional[str]) -> Any:
cli = self.cli_class(host, api_key, username=username, password=password)
return cli
[docs] def add_commands_args(self, cmd_subparser: _SubParsersAction) -> None:
for cmd in self.cmd_list:
self.cmd_list[cmd].configure_args(cmd_subparser)
[docs] def run_command(self, cmd_name: str, args: Namespace) -> None:
cli = self._new_client(args.host, args.api_key, username=args.user, password=args.password)
self.cmd_list[cmd_name].run(cli, args)
##############################################
########## media specific commands ##########
##############################################
[docs]def select_profile(cli: base_media.BaseCliMediaApi) -> int:
res = cli.get_quality_profiles()
for profile in res:
qualities = []
for qual in profile["items"]:
if qual["allowed"]:
# Quality items
if "quality" in qual:
qualities.append(qual["quality"]["name"])
# Quality groups
elif "name" in qual:
qualities.append(qual["name"])
print(f"[{profile['id']}]: {profile['name']} ({qualities})")
profile_id = input(f"Profile id to use (1-{len(res)}):")
if profile_id.isdigit():
return int(profile_id)
else:
raise Exception("Invalid profile selection: {profile_id}")
[docs]def select_language_profile(cli: base_media.BaseCliMediaApi) -> int:
res = cli.get_language_profiles()
for profile in res:
print(f"[{profile['id']}]: {profile['name']}")
profile_id = input(f"Profile id to use (1-{len(res)}):")
if profile_id.isdigit():
return int(profile_id)
else:
raise Exception("Invalid profile selection: {profile_id}")
[docs]def select_item(
terms: str, choices: List[Union[radarr.RadarrMovieItem, sonarr.SonarrSerieItem]]
) -> Union[radarr.RadarrMovieItem, sonarr.SonarrSerieItem]:
if not choices or (isinstance(choices, list) and len(choices) == 0):
raise Exception(f"No match found for terms {terms}")
elif issubclass(choices.__class__, base_api.BaseCliApiItem):
# Only one result is returned
return choices # type: ignore
for item in choices:
print(f"[{choices.index(item)+1}]: {item.title} ({item.year})")
item_id = input(f"Select the item to add (1-{len(choices)}):")
if item_id.isdigit() and int(item_id) <= len(choices):
return choices[int(item_id) - 1]
else:
raise Exception("Invalid selection: {}")
[docs]def print_root_folder(cli: base_media.BaseCliMediaApi, raw=bool) -> None:
res = cli.get_root_folder()
if raw:
print(res)
else:
print("Id Free Path")
for root_folder in res:
print(f"{root_folder['id']:<3} {size_to_str(root_folder.get('freeSpace')):<10} {root_folder['path']}")
[docs]def select_root_folder(cli: base_media.BaseCliMediaApi) -> int:
print_root_folder(cli)
root_folder_id = input("Root folder to use (Id):")
if root_folder_id.isdigit():
return int(root_folder_id)
else:
raise Exception("Invalid root folder selection, must be the folder id: {root_folder_id}")
[docs]def root_folder_id_from_arg(cli: base_media.BaseCliMediaApi, root_arg: str) -> int:
if root_arg:
if root_arg == "auto":
# Interactive selection
return select_root_folder(cli)
elif root_arg.isdigit():
# Id specified directly
return int(root_arg)
else:
# Match the path with registered roots to find id
res = cli.get_root_folder()
for root_path in res:
if root_path["path"] == root_arg:
return int(root_path["id"])
raise exceptions.CliArrError(f"No root folder with path '{root_arg}'")
return 0
[docs]class CliGetProfilesCommand(CliCommand):
name = "profiles"
description = "Get list of quality profiles"
[docs] def run(self, cli: base_media.BaseCliMediaApi, args: Namespace) -> None:
super().run(cli, args)
res = cli.get_quality_profiles()
print("Available quality profiles:\n")
for profile in res:
qualities = ",".join([qual["quality"]["name"] for qual in profile["items"] if qual["allowed"]])
print(f"{profile['id']}: {profile['name']} ({qualities})")
[docs]class CliSystemStatusCommand(CliCommand):
name = "system-status"
description = "Get system status"
[docs] def run(self, cli: base_media.BaseCliMediaApi, args: Namespace) -> None:
super().run(cli, args)
res = cli.get_system_status()
print(f"{pformat(res)}\n")
[docs]class CliGetDiskSpaceCommand(CliCommand):
name = "disk-space"
description = "Get disk space"
[docs] def run(self, cli: base_media.BaseCliMediaApi, args: Namespace) -> None:
super().run(cli, args)
res = cli.get_disk_space()
print(f"{pformat(res)}\n")
[docs]class CliGetQueueCommand(CliCommand):
name = "queue"
description = "Get current downloading queue"
[docs] def run(self, cli: base_media.BaseCliMediaApi, args: Namespace) -> None:
super().run(cli, args)
res = cli.get_queue(args.page, args.sort_key, args.page_size, args.sort_dir, not args.exclude_unknown)
print(f"{pformat(res)}\n")
[docs]class CliGetCalendarCommand(CliCommand):
name = "calendar"
description = "Get events from calendar"
[docs] def run(self, cli: base_media.BaseCliMediaApi, args: Namespace) -> None:
super().run(cli, args)
start_date = datetime.datetime.strptime(args.start, "%Y-%m-%d") if args.start else None
end_date = datetime.datetime.strptime(args.end, "%Y-%m-%d") if args.end else None
res = cli.get_calendar(start_date=start_date, end_date=end_date)
print(f"{pformat(res)}\n")
[docs]class CliDeleteQueueCommand(CliCommand):
name = "delete-queue"
description = "Get list of quality profiles"
[docs] def run(self, cli: base_media.BaseCliMediaApi, args: Namespace) -> None:
super().run(cli, args)
res = cli.delete_queue(args.id)
print(f"{pformat(res)}\n")
[docs]class CliWantedCommand(CliCommand):
name = "wanted"
description = "List wanted/missing"
[docs] def run(self, cli: base_media.BaseCliMediaApi, args: Namespace) -> None:
super().run(cli, args)
res = cli.get_wanted(page=args.page, sort_key=args.sort_key, page_size=args.page_size, sort_dir=args.sort_dir)
print(f"{pformat(res)}\n")
[docs]class CliStatusCommand(CliCommand):
name = "status"
description = "Get status of 1 or all currently running commands"
[docs] def run(self, cli: base_media.BaseCliMediaApi, args: Namespace) -> None:
super().run(cli, args)
res = cli.get_command(args.id)
print(f"{pformat(res)}\n")
[docs]class CliGetBlocklistCommand(CliCommand):
name = "blocklist"
description = "Get blocklisted items"
[docs] def run(self, cli: base_media.BaseCliMediaApi, args: Namespace) -> None:
super().run(cli, args)
res = cli.get_blocklist(
page=args.page, sort_key=args.sort_key, page_size=args.page_size, sort_dir=args.sort_dir
)
print(f"{pformat(res)}\n")
[docs]class CliDeleteBlocklistCommand(CliCommand):
name = "delete-blocklist"
description = "Get list of quality profiles"
[docs] def run(self, cli: base_media.BaseCliMediaApi, args: Namespace) -> None:
super().run(cli, args)
res = cli.delete_blocklist(args.id)
print(f"{pformat(res)}\n")
[docs]class CliGetNotificationCommand(CliCommand):
name = "notification"
description = "Get notification(s)"
[docs] def run(self, cli: base_media.BaseCliMediaApi, args: Namespace) -> None:
super().run(cli, args)
res = cli.get_notification(args.id)
print(f"{pformat(res)}\n")
[docs]class CliDeleteNotificationCommand(CliCommand):
name = "delete-notification"
description = "Delete the specified notification or all"
[docs] def run(self, cli: base_media.BaseCliMediaApi, args: Namespace) -> None:
super().run(cli, args)
res = cli.delete_notification(args.id)
print(f"{pformat(res)}\n")
[docs]class CliPutNotificationCommand(CliCommand):
name = "put-notification"
description = "Create the specified notification"
[docs] def run(self, cli: base_media.BaseCliMediaApi, args: Namespace) -> None:
super().run(cli, args)
# Use json data from argument by default, but load json file if specified
if args.file:
with open(args.file, "r") as f:
notification_data = json.load(f)
else:
notification_data = json.loads(args.json)
res = cli.put_notification(args.id, notification_data)
print(f"{pformat(res)}\n")
[docs]class CliGetTagCommand(CliCommand):
name = "tag"
description = "Get tag(s)"
[docs] def run(self, cli: base_media.BaseCliMediaApi, args: Namespace) -> None:
super().run(cli, args)
res = cli.get_tag(args.id)
print(f"{pformat(res)}\n")
[docs]class CliGetTagDetailCommand(CliCommand):
name = "tag-detail"
description = "Get tag(s) details"
[docs] def run(self, cli: base_media.BaseCliMediaApi, args: Namespace) -> None:
super().run(cli, args)
res = cli.get_tag_detail(args.id)
print(f"{pformat(res)}\n")
[docs]class CliDeleteTagCommand(CliCommand):
name = "delete-tag"
description = "Delete the specified tag"
[docs] def run(self, cli: base_media.BaseCliMediaApi, args: Namespace) -> None:
super().run(cli, args)
res = cli.delete_tag(args.id)
print(f"{pformat(res)}\n")
[docs]class CliEditTagCommand(CliCommand):
name = "edit-tag"
description = "Edit the specified tag"
[docs] def run(self, cli: base_media.BaseCliMediaApi, args: Namespace) -> None:
super().run(cli, args)
res = cli.edit_tag(args.id, args.label)
print(f"{pformat(res)}\n")
[docs]class CliGetTagItemsCommand(CliCommand):
name = "tag-items"
description = "List items with specifed tag"
[docs] @no_type_check
def run(self, cli: base_media.BaseCliMediaApi, args: Namespace) -> None:
super().run(cli, args)
res = None
if args.label:
tags = cli.get_tag_detail()
for tag in tags:
if tag["label"] == args.label:
res = tag
else:
res = cli.get_tag_detail(args.id)
if res:
print(f"Items with tag \"{res['label']}\" ({res['id']}):")
if "seriesIds" in res:
for tag_item in res["seriesIds"]:
item = cli.get_serie(tag_item)
print(f" {item.title} ({item.year})")
elif "movieIds" in res:
for tag_item in res["movieIds"]:
item = cli.get_movie(tag_item)
print(f" {item.title} ({item.year})")
else:
print("no such tag")
[docs]class CliCreateTagCommand(CliCommand):
name = "create-tag"
description = "Create the specified tag"
[docs] def run(self, cli: base_media.BaseCliMediaApi, args: Namespace) -> None:
super().run(cli, args)
res = cli.create_tag(args.label)
print(f"{pformat(res)}\n")
[docs]class CliGetExclusionCommand(CliCommand):
name = "exclusion"
description = "Get exclusion(s)"
[docs] def run(self, cli: base_media.BaseCliMediaApi, args: Namespace) -> None:
super().run(cli, args)
res = cli.get_exclusion(args.id)
print(f"{pformat(res)}\n")
[docs]class CliDeleteExclusionCommand(CliCommand):
name = "delete-exclusion"
description = "Delete the specified exclusion"
[docs] def run(self, cli: base_media.BaseCliMediaApi, args: Namespace) -> None:
super().run(cli, args)
res = cli.delete_exclusion(args.id)
print(f"{pformat(res)}\n")
[docs]class CliRootFoldersCommand(CliCommand):
name = "root-folders"
description = "Get root folder list"
[docs] def run(self, cli: base_media.BaseCliMediaApi, args: Namespace) -> None:
super().run(cli, args)
print_root_folder(cli, raw=args.json)
##############################################
########## radarr specific commands ##########
##############################################
[docs]class CliGetMovieCommand(CliCommand):
name = "get"
description = "Get info on a of movie"
[docs] def run(self, cli: radarr.RadarrCli, args: Namespace) -> None:
super().run(cli, args)
res = cli.get_movie(args.mid)
if args.json:
if isinstance(res, list):
json_objs = [item.to_json() for item in res]
print(f"[{','.join(json_objs)}]")
else:
print(f"{res.to_json()}")
else:
print(res)
[docs]class CliDeleteMovieCommand(CliCommand):
name = "delete"
description = "Delete a movie"
[docs] def run(self, cli: radarr.RadarrCli, args: Namespace) -> None:
super().run(cli, args)
res = cli.delete_movie(args.mid, delete_files=args.delfiles, add_exclusion=args.exclude)
print(res)
[docs]class CliGetRefreshMovieCommand(CliCommand):
name = "refresh"
description = "Refresh movies"
[docs] def run(self, cli: radarr.RadarrCli, args: Namespace) -> None:
super().run(cli, args)
res = cli.refresh_movie(args.mid)
print(res)
[docs]class CliGetRescanMovieCommand(CliCommand):
name = "rescan"
description = "Rescan movies"
[docs] def run(self, cli: radarr.RadarrCli, args: Namespace) -> None:
super().run(cli, args)
res = cli.rescan_movie(args.mid)
print(res)
[docs]class CliAddMovieCommand(CliCommand):
name = "add"
description = "Add a movie from the imdb/tmdb id, or look for keywords"
[docs] def run(self, cli: radarr.RadarrCli, args: Namespace) -> None:
super().run(cli, args)
# If keywords were specified, look for results and prompt for choice
movie_info = None
if args.terms:
choices = cli.lookup_movie(term=args.terms)
movie_info = select_item(args.terms, choices) # type: ignore
# If no quality profile specified, list them qnd prompt for choice
if not args.quality:
args.quality = select_profile(cli)
root_id = root_folder_id_from_arg(cli, args.root_folder)
res = cli.add_movie(
quality=args.quality,
tmdb_id=args.tmdb,
imdb_id=args.imdb,
movie_info=movie_info, # type: ignore
path=args.path,
root_id=root_id,
)
print(f"{json.dumps(res)}")
[docs]class CliEditMovieCommand(CliCommand):
name = "edit"
description = "Push an updated item to the movie library"
[docs] def run(self, cli: radarr.RadarrCli, args: Namespace) -> None:
super().run(cli, args)
json_data = args.json
if not json_data and args.file:
with open(args.file, "r") as f:
json_data = f.read()
info = radarr.RadarrMovieItem.from_json(json_data)
res = cli.edit_movie(info)
print(f"{json.dumps(res)}")
[docs]class CliCreateRadarrExclusionCommand(CliCommand):
name = "create-exclusion"
description = "Create the specified exclusion"
[docs] def run(self, cli: radarr.RadarrCli, args: Namespace) -> None:
super().run(cli, args)
res = cli.create_exclusion(args.title, args.id, args.year)
print(f"{pformat(res)}\n")
[docs]class CliSearchMissingMovies(CliCommand):
name = "search-missing"
description = "Search missing movies"
[docs] def run(self, cli: radarr.RadarrCli, args: Namespace) -> None:
super().run(cli, args)
res = cli.missing_movies_search()
print(f"{json.dumps(res)}")
##############################################
########## sonarr specific commands ##########
##############################################
[docs]class CliGetSerieCommand(CliCommand):
name = "get"
description = "Get info on a of serie"
[docs] def run(self, cli: sonarr.SonarrCli, args: Namespace) -> None:
super().run(cli, args)
res = cli.get_serie(args.sid)
if args.json:
if isinstance(res, list):
json_objs = [item.to_json() for item in res]
print(f"[{','.join(json_objs)}]")
else:
print(f"{res.to_json()}")
else:
print(res)
[docs]class CliDeleteSerieCommand(CliCommand):
name = "delete"
description = "Delete a serie"
[docs] def run(self, cli: sonarr.SonarrCli, args: Namespace) -> None:
super().run(cli, args)
res = cli.delete_serie(args.sid, delete_files=args.delfiles, add_exclusion=args.exclude)
print(f"Result:\n{res}")
[docs]class CliGetRefreshSerieCommand(CliCommand):
name = "refresh"
description = "Refresh series"
[docs] def run(self, cli: sonarr.SonarrCli, args: Namespace) -> None:
super().run(cli, args)
res = cli.refresh_serie(args.sid)
print(f"Result:\n{json.dumps(res)}\n")
[docs]class CliGetRescanSerieCommand(CliCommand):
name = "rescan"
description = "Rescan series"
[docs] def run(self, cli: sonarr.SonarrCli, args: Namespace) -> None:
super().run(cli, args)
res = cli.rescan_serie(args.sid)
print(f"Result: {json.dumps(res)}\n")
[docs]class CliAddSerieCommand(CliCommand):
name = "add"
description = "Add a serie from the tvdb id, or look for keywords"
[docs] def run(self, cli: sonarr.SonarrCli, args: Namespace) -> None:
super().run(cli, args)
# If keywords were specified, look for results and prompt for choice
serie_info = None
if args.terms:
choices = cli.lookup_serie(term=args.terms)
serie_info = select_item(args.terms, choices) # type: ignore
# If no quality profile specified, list them qnd prompt for choice
if not args.quality:
args.quality = select_profile(cli)
if not args.language:
args.language = select_language_profile(cli)
root_id = root_folder_id_from_arg(cli, args.root_folder)
# Get the optional season list
seasons_str = args.seasons.replace(" ", "").split(",") if args.seasons else []
try:
seasons = [int(season_num) for season_num in seasons_str]
except ValueError as e:
raise Exception(f"Error, invalid season list: {args.seasons} ({e})")
res = cli.add_serie(
quality=args.quality,
tvdb_id=args.tvdb,
serie_info=serie_info, # type: ignore
monitored_seasons=seasons,
season_folder=args.season_folders,
path=args.path,
root_id=root_id,
language=args.language,
)
print(f"Result:\n{res}")
[docs]class CliEpisodeCommand(CliCommand):
name = "get-episode"
description = "Get info on an episode"
[docs] def run(self, cli: sonarr.SonarrCli, args: Namespace) -> None:
super().run(cli, args)
res = cli.get_episode(serie_id=args.sid, episode_id=args.epid)
print(f"Episode info:\n{res}")
[docs]class CliGetEpisodeFileCommand(CliCommand):
name = "get-episode-file"
description = "Get info on an episode file"
[docs] def run(self, cli: sonarr.SonarrCli, args: Namespace) -> None:
super().run(cli, args)
res = cli.get_episode_file(serie_id=args.sid, episode_id=args.epid)
print(f"Episode file:\n{res}")
[docs]class CliDeleteEpisodeFileCommand(CliCommand):
name = "delete-episode-file"
description = "Get info on a of serie"
[docs] def run(self, cli: sonarr.SonarrCli, args: Namespace) -> None:
super().run(cli, args)
res = cli.delete_episode_file(args.sid)
print(f"Res:\n{res}")
[docs]class CliCreateSonarrExclusionCommand(CliCommand):
name = "create-exclusion"
description = "Create the specified exclusion"
[docs] def run(self, cli: sonarr.SonarrCli, args: Namespace) -> None:
super().run(cli, args)
res = cli.create_exclusion(args.title, args.id)
print(f"{pformat(res)}\n")
[docs]class CliSearchMissingEpisodes(CliCommand):
name = "search-missing"
description = "Search missing episods"
[docs] def run(self, cli: sonarr.SonarrCli, args: Namespace) -> None:
super().run(cli, args)
res = cli.missing_episodes_search()
print(f"Result: {json.dumps(res)}\n")
#####################################################
########## Clients and commands definition ##########
#####################################################
CLI_LIST: List[CliApiCommand] = [
CliApiCommand(
"sonarr",
sonarr.SonarrCli,
[
CliGetSerieCommand(),
CliDeleteSerieCommand(),
CliAddSerieCommand(),
CliGetRefreshSerieCommand(),
CliGetRescanSerieCommand(),
CliEpisodeCommand(),
CliGetEpisodeFileCommand(),
CliDeleteEpisodeFileCommand(),
CliGetProfilesCommand(),
CliSystemStatusCommand(),
CliGetDiskSpaceCommand(),
CliGetQueueCommand(),
CliGetCalendarCommand(),
CliDeleteQueueCommand(),
CliWantedCommand(),
CliStatusCommand(),
CliGetBlocklistCommand(),
CliDeleteBlocklistCommand(),
CliGetNotificationCommand(),
CliDeleteNotificationCommand(),
CliPutNotificationCommand(),
CliGetTagCommand(),
CliGetTagDetailCommand(),
CliDeleteTagCommand(),
CliEditTagCommand(),
CliCreateTagCommand(),
CliGetTagItemsCommand(),
CliGetExclusionCommand(),
CliDeleteExclusionCommand(),
CliCreateSonarrExclusionCommand(),
CliSearchMissingEpisodes(),
CliRootFoldersCommand(),
],
),
CliApiCommand(
"radarr",
radarr.RadarrCli,
[
CliGetMovieCommand(),
CliDeleteMovieCommand(),
CliAddMovieCommand(),
CliEditMovieCommand(),
CliGetRefreshMovieCommand(),
CliGetRescanMovieCommand(),
CliGetProfilesCommand(),
CliSystemStatusCommand(),
CliGetDiskSpaceCommand(),
CliGetQueueCommand(),
CliGetCalendarCommand(),
CliDeleteQueueCommand(),
CliWantedCommand(),
CliStatusCommand(),
CliGetBlocklistCommand(),
CliDeleteBlocklistCommand(),
CliGetNotificationCommand(),
CliDeleteNotificationCommand(),
CliPutNotificationCommand(),
CliGetTagCommand(),
CliGetTagDetailCommand(),
CliDeleteTagCommand(),
CliEditTagCommand(),
CliCreateTagCommand(),
CliGetTagItemsCommand(),
CliGetExclusionCommand(),
CliDeleteExclusionCommand(),
CliCreateRadarrExclusionCommand(),
CliSearchMissingMovies(),
CliRootFoldersCommand(),
],
),
]