SickGear/lib/pytvmaze/tvmaze.py

1537 lines
53 KiB
Python

import re
from datetime import datetime
import requests
from urllib.parse import quote
from urllib3.util.retry import Retry
from requests.adapters import HTTPAdapter
from . import endpoints
from .exceptions import *
# noinspection PyUnreachableCode
if False:
from typing import Any, AnyStr, Dict, List, Optional, Union
class Show(object):
def __init__(self, data):
self.status = data.get('status') # type: Optional[AnyStr]
self.rating = data.get('rating') # type: Optional[int]
self.genres = data.get('genres') # type: Optional[List[AnyStr]]
self.weight = data.get('weight') # type: Optional[int]
self.updated = data.get('updated')
self.name = data.get('name') # type: Optional[AnyStr]
self.language = data.get('language') # type: Optional[AnyStr]
self.schedule = data.get('schedule')
self.url = data.get('url') # type: Optional[AnyStr]
self.official_site = data.get('officialSite') # type: Optional[AnyStr]
self.image = data.get('image')
self.externals = data.get('externals')
self.premiered = data.get('premiered') # type: Optional[AnyStr]
self.summary = _remove_tags(data.get('summary', '')) # type: Optional[AnyStr]
self.links = data.get('_links')
if data.get('webChannel'):
self.web_channel = WebChannel(data.get('webChannel')) # type: Optional[WebChannel]
else:
self.web_channel = None
self.runtime = data.get('runtime') # type: Optional[int]
self.average_runtime = data.get('averageRuntime')
self.type = data.get('type') # type: Optional[AnyStr]
self.id = data.get('id') # type: int
self.maze_id = self.id # type: int
if data.get('network'):
self.network = Network(data.get('network')) # type: Optional[Network]
else:
self.network = None
self.__episodes = None # type: Optional[List[Episode]]
self._specials_loaded = False # type: bool
self._seasons = None # type: Optional[List[Season]]
self._cast = None # type: Optional[List[Cast]]
self._crew = None # type: Optional[List[Crew]]
self._images = None # type: Optional[List[Image]]
self._akas = None # type: Optional[List[AnyStr]]
self.__nextepisode = None # type: Optional[Episode]
self.__previousepisode = None # type: Optional[Episode]
self.populate(data)
def __repr__(self):
if self.premiered:
year = str(self.premiered[:4])
else:
year = None
if self.web_channel:
platform = ',show_web_channel='
network = self.web_channel.name
elif self.network:
platform = ',network='
network = self.network.name
else:
platform = ''
network = ''
return f'<Show(maze_id={self.maze_id}, name={self.name}, year={year}{platform}{network})>'
def __str__(self):
n = ''
if self.network:
n = ' (%s)' % self.network.name
return f'{self.name}{n} [{self.premiered}]'
def __iter__(self):
return iter(self.seasons.values())
# Python 3 bool evaluation
def __bool__(self):
return bool(self.id)
# Python 2 bool evaluation
def __nonzero__(self):
return bool(self.id)
def __len__(self):
return len(self.seasons)
def __getitem__(self, item):
try:
return self.seasons[item]
except KeyError:
raise SeasonNotFound(f'Season {item} does not exist for show {self.name}.')
@property
def akas(self):
if None is self._akas:
self._akas = show_akas(self.maze_id, raise_error=False)
return self._akas
@property
def crew(self):
if None is self._crew:
self._crew = get_show_crew(self.maze_id, raise_error=False)
return self._crew
@property
def images(self):
if None is self._images:
self._images = get_show_images(self.maze_id, raise_error=False)
return self._images
@property
def cast(self):
if None is self._cast:
self._cast = show_cast(self.maze_id, raise_error=False)
return self._cast
@property
def next_episode(self):
if self.__nextepisode is None and 'nextepisode' in self.links and 'href' in self.links['nextepisode']:
episode_id = self.links['nextepisode']['href'].rsplit('/', 1)[1]
if episode_id.isdigit():
self.__nextepisode = episode_by_id(episode_id, show=self, raise_error=False)
return self.__nextepisode
@property
def previous_episode(self):
if self.__previousepisode is None and 'previousepisode' in self.links\
and 'href' in self.links['previousepisode']:
episode_id = self.links['previousepisode']['href'].rsplit('/', 1)[1]
if episode_id.isdigit():
self.__previousepisode = episode_by_id(episode_id, show=self, raise_error=False)
return self.__previousepisode
def _set_season_numbers(self):
seasons = show_seasons(self.maze_id, show=self, raise_error=False)
self._seasons = dict()
for episode in self.__episodes:
season_num = int(episode.season_number)
if season_num not in self._seasons:
self._seasons[season_num] = seasons[season_num]
self._seasons[season_num].show = self
@property
def seasons(self):
# type: (...) -> Dict
if None is self._seasons:
# noinspection PyStatementEffect
self.episodes
return self._seasons
@property
def episodes(self):
if None is self.__episodes or not self._specials_loaded:
self.__episodes = episode_list(self.maze_id, specials=True, raise_error=False, show=self)
self._specials_loaded = True
self._set_season_numbers()
return self.__episodes
def populate(self, data):
embedded = data.get('_embedded')
if embedded:
episodes = embedded.get('episodeswithspecials') or embedded.get('episodes')
if episodes:
self.__episodes = []
for episode in episodes:
self.__episodes.append(Episode(episode, self))
self._set_season_numbers()
if 'episodeswithspecials' in embedded:
self._specials_loaded = True
if embedded.get('cast'):
self._cast = Cast(embedded.get('cast'))
class Season(object):
def __init__(self, data, show=None, season_number=None):
self.show = show
if None is not season_number and None is not show and show.episodes:
self.__episodes = [ep for ep in show.episodes if season_number == ep.season_number]
else:
self.__episodes = None
self.id = data.get('id')
self.url = data.get('url')
self.season_number = data.get('number')
self.name = data.get('name')
self.episode_order = data.get('episodeOrder')
self.premiere_date = data.get('premiereDate')
self.end_date = data.get('endDate')
if data.get('network'):
self.network = Network(data.get('network'))
else:
self.network = None
if data.get('webChannel'):
self.web_channel = WebChannel(data.get('webChannel'))
else:
self.web_channel = None
self.image = data.get('image')
self.summary = data.get('summary')
self.links = data.get('_links')
def __repr__(self):
return f'<Season(id={self.id}, season_number={self.season_number})>'
def _get_showname(self):
sn = ''
if self.show:
sn = '%s - ' % self.show.name
return f'{sn}Season {self.season_number:0>2} ({len(self.episodes)}/{self.episode_order})'
def __str__(self):
return self._get_showname()
def __iter__(self):
return iter(self.episodes)
def __len__(self):
return len(self.episodes)
def __getitem__(self, item):
try:
return self.episodes[item]
except KeyError:
raise EpisodeNotFound(f'Episode {item} does not exist for season {self.season_number} of show {self.show}.')
# Python 3 bool evaluation
def __bool__(self):
return bool(self.id)
# Python 2 bool evaluation
def __nonzero__(self):
return bool(self.id)
@property
def episodes(self):
# type: (...) -> List[Episode]
if None is self.__episodes or not self.show._specials_loaded:
self.__episodes = [ep for ep in episode_list(self.show.maze_id, specials=True, raise_error=False,
show=self.show) if ep.season_number == self.season_number]
self.show._specials_loaded = True
return self.__episodes
class Episode(object):
def __init__(self, data, show=None):
self.title = data.get('name')
self.airdate = data.get('airdate')
self.url = data.get('url')
self.season_number = data.get('season')
self.episode_number = data.get('number')
self.image = data.get('image')
self.airstamp = data.get('airstamp')
self.airtime = data.get('airtime')
self.runtime = data.get('runtime')
self.summary = _remove_tags(data.get('summary'))
self.maze_id = data.get('id')
self.type = data.get('type') # type: AnyStr # ("regular", "significant_special", "insignificant_special")
self.special = self.is_special()
# Reference to show for when using get_schedule()
self.show = show
if not show:
if data.get('show'):
self.show = Show(data.get('show'))
# Reference to show for when using get_full_schedule()
if not self.show and data.get('_embedded'):
if data['_embedded'].get('show'):
self.show = Show(data['_embedded']['show'])
if not self.show and data.get('_links') and 'show' in data['_links']:
self.show = Show({})
self.show.id = int(re.search(r'/(\d+)$', data['_links']['show']['href']).group(1))
def __repr__(self):
if self.special:
epnum = 'Special'
else:
epnum = self.episode_number
return f'<Episode(season={self.season_number:0>2}, episode_number={epnum:0>2})>'
def _gen_ep_name(self):
season = f'S{self.season_number:0>2}'
if self.special:
episode = ' Special'
else:
episode = f'E{self.episode_number:0>2}'
sn = ''
if self.show:
sn = f'{self.show.name} - ' % self.show.name
return f'{sn}{season}{episode} {self.title}'
def __str__(self):
return self._gen_ep_name()
def is_special(self):
if self.type in ('significant_special', 'insignificant_special'):
return True
elif 'regular' == self.type:
return False
# fallback for previous api without type field
if self.episode_number:
return False
return True
class Person(object):
def __init__(self, data):
if data.get('person'):
data = data['person']
self.links = data.get('_links')
self.id = data.get('id')
self.image = data.get('image')
self.name = data.get('name')
self.score = data.get('score')
self.url = data.get('url')
self.birthday = data.get('birthday')
self.death_day = data.get('deathday')
self.gender = data.get('gender')
self.country = data.get('country')
self.character = None
self._castcredits = None
self._guestcastcredits = None
self._crewcredits = None
# self.populate(data)
# def populate(self, data):
# if data.get('_embedded'):
# if data['_embedded'].get('castcredits'):
# self._castcredits = [CastCredit(credit)
# for credit in data['_embedded']['castcredits']]
# elif data['_embedded'].get('crewcredits'):
# self.crewcredits = [CrewCredit(credit)
# for credit in data['_embedded']['crewcredits']]
@property
def castcredits(self):
if None is self._castcredits:
self._castcredits = person_cast_credits(self.id, embed='show,character', raise_error=False)
return self._castcredits
@property
def guestcastcredits(self):
if None is self._guestcastcredits:
self._guestcastcredits = person_guestcast_credits(self.id, embed='episode,character', raise_error=False)
return self._guestcastcredits
@property
def crewcredits(self):
if None is self._crewcredits:
self._crewcredits = person_crew_credits(self.id, embed='show', raise_error=False)
return self._crewcredits
def __repr__(self):
return f'<Person(name={self.name}, maze_id={self.id})>'
def _gen_lifetime(self):
life = ''
if self.birthday:
life = f'{self.birthday}'
if self.death_day:
if 0 < len(life):
life += ' - '
life = f'{self.death_day}'
if 0 < len(life):
life = f' ({life})'
return life
def __str__(self):
return f'{self.name}{self._gen_lifetime()}'
class Character(object):
def __init__(self, data, base_data=None):
base_data = {} if None is base_data else base_data
self.id = data.get('id')
self.url = data.get('url')
self.name = data.get('name')
self.image = data.get('image')
self.links = data.get('_links')
self.plays_self = base_data.get('self')
self.voice = base_data.get('voice')
self.person = None # type: Optional[Person]
def __repr__(self):
return f'<Character(name={self.name}, maze_id={self.id})>'
def _get_person(self):
p = ''
if self.person:
p = f' ({self.person.name})'
return p
def __str__(self):
return f'{self.name}{self._get_person()}'
class Cast(object):
def __init__(self, data):
self.people = []
self.characters = [] # type: List[Character]
self.populate(data)
def populate(self, data):
for cast_member in data:
self.people.append(Person(cast_member['person']))
self.characters.append(Character(cast_member['character'], cast_member))
self.people[-1].character = self.characters[-1] # add reference to character
self.characters[-1].person = self.people[-1] # add reference to cast member
def __repr__(self):
return self.__str__()
def __str__(self):
return f'{self.characters} ({self.people})'
class CastCredit(object):
def __init__(self, data):
self.links = data.get('_links')
self.character = None # type: Optional[Character]
self.show = None # type: Optional[Show]
self.episode = None # type: Optional[Episode]
self.populate(data)
def populate(self, data):
if data.get('_embedded'):
if data['_embedded'].get('character'):
self.character = Character(data['_embedded']['character'])
if data['_embedded'].get('show'):
self.show = Show(data['_embedded']['show'])
if data['_embedded'].get('episode'):
self.episode = Episode(data['_embedded']['episode'])
def __repr__(self):
return self.__str__()
def __str__(self):
return f'{self.character} ({self.show})'
class CrewCredit(object):
def __init__(self, data):
self.links = data.get('_links')
self.type = data.get('type')
self.show = None
self.populate(data)
def populate(self, data):
if data.get('_embedded'):
if data['_embedded'].get('show'):
self.show = Show(data['_embedded']['show'])
def __repr__(self):
return self.__str__()
def __str__(self):
return f'{self.type} ({self.show})'
class Crew(object):
def __init__(self, data):
self.person = Person(data.get('person'))
self.type = data.get('type')
def __str__(self):
return f'{self.type}: {self.person}'
def __repr__(self):
return f'<Crew(name={self.person.name}, maze_id={self.person.id}, type={self.type})>'
class Image(object):
def __init__(self, data):
self.main = data.get('main') # type: bool
self.type = data.get('type') # type: AnyStr
self.id = data.get('id') # type: int
self.resolutions = data.get('resolutions') # type: Dict[AnyStr, AnyStr]
def _get_type_name(self):
return ('unknown', self.type)[isinstance(self.type, str)]
def _get_main_str(self):
return ('Other', 'Main')[True is self.main]
def __str__(self):
return f'{self._get_type_name()}: {self._get_main_str()}'
def __repr__(self):
return f'<Image(main={self.main}, maze_id={self.id}, type={self._get_type_name()})>'
class Updates(object):
def __init__(self, data):
self.updates = dict()
self.populate(data)
def populate(self, data):
for maze_id, time in data.items():
self.updates[int(maze_id)] = Update(maze_id, time)
def __getitem__(self, item):
try:
return self.updates[item]
except KeyError:
logger.error(f'No update found for Maze id {item}.')
raise UpdateNotFound(f'No update found for Maze id {item}.')
def __iter__(self):
return iter(self.updates.values())
class Update(object):
def __init__(self, maze_id, time):
self.maze_id = int(maze_id)
self.seconds_since_epoch = time
self.timestamp = datetime.fromtimestamp(time)
def __repr__(self):
return f'<Update(maze_id={self.maze_id}, time={self.seconds_since_epoch})>'
class AKA(object):
def __init__(self, data):
self.name = data.get('name')
self.country = data.get('country')
def __repr__(self):
return f'<AKA(name={self.name}, country={self.country})>'
def _get_country(self):
c = ''
if self.country:
c = f' ({self.country.get("name")})'
return c
def __str__(self):
return f'{self.name}{self.name}'
class NetworkBase(object):
def __init__(self, data, *args, **kwargs):
self.name = data.get('name')
self.maze_id = data.get('id')
if data.get('country'):
self.country = data['country'].get('name')
self.timezone = data['country'].get('timezone')
self.code = data['country'].get('code')
else:
self.country = None
self.timezone = None
self.code = None
def _get_country(self):
c = ''
if self.country:
c = f' ({self.country})'
return c
def __str__(self):
return f'{self.name}{self._get_country()}'
class Network(NetworkBase):
def __init__(self, data):
super(Network, self).__init__(data)
def __repr__(self):
return f'<Network(name={self.name}, country={self.country})>'
class WebChannel(NetworkBase):
def __init__(self, data):
super(WebChannel, self).__init__(data)
def __repr__(self):
return f'<WebChannel(name={self.name}, country={self.country})>'
class FollowedShow(object):
def __init__(self, data):
self.maze_id = data.get('show_id')
self.show = None
if data.get('_embedded'):
self.show = Show(data['_embedded'].get('show'))
def __repr__(self):
return f'<FollowedShow(maze_id={self.maze_id})>'
class FollowedPerson(object):
def __init__(self, data):
self.person_id = data.get('person_id')
self.person = None
if data.get('_embedded'):
self.person = Person(data['_embedded'].get('person'))
def __repr__(self):
return f'<FollowedPerson(person_id={self.person_id})>'
class FollowedNetwork(object):
def __init__(self, data):
self.network_id = data.get('network_id')
self.network = None
if data.get('_embedded'):
self.network = Network(data['_embedded'].get('network'))
def __repr__(self):
return f'<FollowedNetwork(network_id={self.network_id})>'
class FollowedWebChannel(object):
def __init__(self, data):
self.web_channel_id = data.get('webchannel_id')
self.web_channel = None
if data.get('_embedded'):
self.web_channel = WebChannel(data['_embedded'].get('webchannel'))
def __repr__(self):
return f'<FollowedWebChannel(web_channel_id={self.web_channel_id})>'
class MarkedEpisode(object):
def __init__(self, data):
self.episode_id = data.get('episode_id')
self.marked_at = data.get('marked_at')
type_ = data.get('type')
types = {0: 'watched', 1: 'acquired', 2: 'skipped'}
self.type = types[type_]
def __repr__(self):
return f'<MarkedEpisode(episode_id={self.episode_id}, marked_at={self.marked_at}, type={self.type})>'
class VotedShow(object):
def __init__(self, data):
self.maze_id = data.get('show_id')
self.voted_at = data.get('voted_at')
self.vote = data.get('vote')
if data.get('_embedded'):
self.show = Show(data['_embedded'].get('show'))
def __repr__(self):
return f'<VotedShow(maze_id={self.maze_id}, voted_at={self.voted_at}, vote={self.vote})>'
class VotedEpisode(object):
def __init__(self, data):
self.episode_id = data.get('episode_id')
self.voted_at = data.get('voted_at')
self.vote = data.get('vote')
def __repr__(self):
return f'<VotedEpisode(episode_id={self.episode_id}, voted_at={self.voted_at}, vote={self.vote})>'
def _url_quote(show):
return quote(show.encode('UTF-8'))
def _remove_tags(text):
if not text:
return None
return re.sub(r'<.*?>', '', text)
# noinspection PyUnusedLocal
def _record_hook(r, *args, **kwargs):
r.hook_called = True
if 301 == r.status_code and isinstance(r.headers.get('Location'), str) \
and r.headers.get('Location').startswith('http://api.tvmaze'):
r.headers['Location'] = r.headers['Location'].replace('http://', 'https://')
return r
def _embed_url(base_url, embed, possible_embeds, glue):
if isinstance(embed, str):
embed_words = [em.strip() for em in embed.split(',')]
elif None is embed:
embed_words = []
else:
embed_words = None
if None is embed_words or any(ew not in possible_embeds for ew in embed_words):
raise InvalidEmbedValue('Value for embed must be %s' % possible_embeds)
if embed_words:
if 1 == len(embed_words):
return '%s%sembed=%s' % (base_url, glue, embed_words[0])
else:
eu = ''
for word in embed_words:
eu += '%sembed[]=%s' % (glue, word)
glue = '&'
return '%s%s' % (base_url, eu)
return base_url
class TVmaze(object):
"""This is the main class of the module enabling interaction with both free and Premium
TVmaze features.
Attributes:
username (str): Username for https://www.tvmaze.com
api_key (str): TVmaze api key. Find your key at https://www.tvmaze.com/dashboard
"""
def __init__(self, username=None, api_key=None):
self.username = username
self.api_key = api_key
# Query TVmaze free endpoints
@staticmethod
def endpoint_standard_get(url):
# type: (str) -> Any
s = requests.Session()
retries = Retry(total=5,
backoff_factor=0.1,
status_forcelist=[429])
s.mount('http://', HTTPAdapter(max_retries=retries))
s.mount('https://', HTTPAdapter(max_retries=retries))
try:
r = s.get(url, hooks={'response': _record_hook})
except requests.exceptions.ConnectionError as e:
logger.error(repr(e))
raise ConnectionError(repr(e))
s.close()
if r.status_code in [404, 422]:
return None
if r.status_code == 400:
raise BadRequest(f'Bad Request for url {url}')
results = r.json()
return results
# Query TVmaze Premium endpoints
def _endpoint_premium_get(self, url):
# type: (str) -> Any
s = requests.Session()
retries = Retry(total=5,
backoff_factor=0.1,
status_forcelist=[429])
s.mount('http://', HTTPAdapter(max_retries=retries))
s.mount('https://', HTTPAdapter(max_retries=retries))
try:
r = s.get(url, auth=(self.username, self.api_key), hooks={'response': _record_hook})
except requests.exceptions.ConnectionError as e:
logger.error(repr(e))
raise ConnectionError(repr(e))
s.close()
if r.status_code in [404, 422]:
return None
if r.status_code == 400:
raise BadRequest(f'Bad Request for url {url}')
results = r.json()
return results
def _endpoint_premium_delete(self, url):
# type: (str) -> Any
s = requests.Session()
retries = Retry(total=5,
backoff_factor=0.1,
status_forcelist=[429])
s.mount('http://', HTTPAdapter(max_retries=retries))
s.mount('https://', HTTPAdapter(max_retries=retries))
try:
r = s.delete(url, auth=(self.username, self.api_key), hooks={'response': _record_hook})
except requests.exceptions.ConnectionError as e:
logger.error(repr(e))
raise ConnectionError(repr(e))
s.close()
if r.status_code == 400:
logger.error(f'Bad Request for url {url}')
raise BadRequest(f'Bad Request for url {url}')
if r.status_code == 200:
return True
if r.status_code == 404:
return None
def _endpoint_premium_put(self, url, payload=None):
# type: (str, Any) -> Any
s = requests.Session()
retries = Retry(total=5,
backoff_factor=0.1,
status_forcelist=[429])
s.mount('http://', HTTPAdapter(max_retries=retries))
s.mount('https://', HTTPAdapter(max_retries=retries))
try:
r = s.put(url, data=payload, auth=(self.username, self.api_key), hooks={'response': _record_hook})
except requests.exceptions.ConnectionError as e:
logger.error(repr(e))
raise ConnectionError(repr(e))
s.close()
if r.status_code == 400:
raise BadRequest(f'Bad Request for url {url}')
if r.status_code == 200:
return True
if r.status_code in [404, 422]:
return None
# Get Show object
def get_show(self, maze_id=None, tvdb_id=None, tvrage_id=None, imdb_id=None, show_name=None,
show_year=None, show_network=None, show_language=None, show_country=None,
show_web_channel=None, embed=None):
# type: (...) -> Show
"""
Get Show object directly via id or indirectly via name + optional qualifiers
If only a show_name is given, the show with the highest score using the
tvmaze algorithm will be returned.
If you provide extra qualifiers such as network or language they will be
used for a more specific match, if one exists.
Args:
maze_id: Show maze_id
tvdb_id: Show tvdb_id
tvrage_id: Show tvrage_id
imdb_id: Show tvrage_id
show_name: Show name to be searched
show_year: Show premiere year
show_network: Show TV Network (like ABC, NBC, etc.)
show_web_channel: Show Web Channel (like Netflix, Amazon, etc.)
show_language: Show language
show_country: Show country
embed: embed parameter to include additional data. Values: 'episodes', 'cast', 'episodeswithspecials'
"""
errors = []
if not (maze_id or tvdb_id or tvrage_id or imdb_id or show_name):
raise MissingParameters(
'Either maze_id, tvdb_id, tvrage_id, imdb_id or show_name are required to get show, none provided,')
if maze_id:
try:
return show_main_info(maze_id, embed=embed)
except IDNotFound as e:
errors.append(e.value)
if tvdb_id:
try:
return show_main_info(lookup_tvdb(tvdb_id).id, embed=embed)
except IDNotFound as e:
errors.append(e.value)
if tvrage_id:
try:
return show_main_info(lookup_tvrage(tvrage_id).id, embed=embed)
except IDNotFound as e:
errors.append(e.value)
if imdb_id:
try:
return show_main_info(lookup_imdb(imdb_id).id, embed=embed)
except IDNotFound as e:
errors.append(e.value)
if show_name:
try:
show = self._get_show_by_search(show_name, show_year, show_network, show_language,
show_country, show_web_channel, embed=embed)
return show
except ShowNotFound as e:
errors.append(e.value)
raise ShowNotFound(' ,'.join(errors))
@staticmethod
def _get_show_with_qualifiers(show_name, qualifiers):
# type: (str, List) -> Show
shows = get_show_list(show_name)
best_match = -1 # Initialize match value score
show_match = None
for show in shows:
if show.premiered:
premiered = show.premiered[:-6].lower()
else:
premiered = None
if show.network and show.network.name:
network = show.network.name.lower()
else:
network = None
if show.web_channel and show.web_channel.name:
web_channel = show.web_channel.name.lower()
else:
web_channel = None
if show.network and show.network.code:
country = show.network.code.lower()
else:
if show.web_channel and show.web_channel.code:
country = show.web_channel.code.lower()
else:
country = None
if show.language:
language = show.language.lower()
else:
language = None
attributes = [premiered, country, network, language, web_channel]
show_score = len(set(qualifiers) & set(attributes))
if show_score > best_match:
best_match = show_score
show_match = show
return show_match
# Search with user-defined qualifiers, used by get_show() method
def _get_show_by_search(self, show_name, show_year, show_network, show_language, show_country,
show_web_channel, embed):
# type: (str, int, str, str, str, str, str) -> Show
if show_year:
show_year = str(show_year)
qualifiers = list(filter(None, [show_year, show_network, show_language, show_country, show_web_channel]))
if qualifiers:
qualifiers = [q.lower() for q in qualifiers if q]
show = self._get_show_with_qualifiers(show_name, qualifiers)
else:
return show_single_search(show=show_name, embed=embed)
if embed:
return show_main_info(maze_id=show.id, embed=embed)
else:
return show
# TVmaze Premium Endpoints
# NOT DONE OR TESTED
def get_followed_shows(self, embed=None):
# type: (str) -> List[FollowedShow]
if embed not in [None, 'show']:
raise InvalidEmbedValue('Value for embed must be "show" or None')
url = endpoints.followed_shows.format('/')
if embed == 'show':
url = endpoints.followed_shows.format('?embed=show')
q = self._endpoint_premium_get(url)
if q:
return [FollowedShow(show) for show in q]
else:
raise NoFollowedShows('You have not followed any shows yet')
def get_followed_show(self, maze_id):
# type: (int) -> FollowedShow
url = endpoints.followed_shows.format(f'/{maze_id}')
q = self._endpoint_premium_get(url)
if q:
return FollowedShow(q)
else:
raise ShowNotFollowed(f'Show with ID {maze_id} is not followed')
def follow_show(self, maze_id):
# type: (int) -> None
url = endpoints.followed_shows.format(f'/{maze_id}')
q = self._endpoint_premium_put(url)
if not q:
raise ShowNotFound(f'Show with ID {maze_id} does not exist')
def unfollow_show(self, maze_id):
# type: (int) -> None
url = endpoints.followed_shows.format(f'/{maze_id}')
q = self._endpoint_premium_delete(url)
if not q:
raise ShowNotFollowed(f'Show with ID {maze_id} was not followed')
def get_followed_people(self, embed=None):
# type: (str) -> List[FollowedPerson]
if embed not in [None, 'person']:
raise InvalidEmbedValue('Value for embed must be "person" or None')
url = endpoints.followed_people.format('/')
if embed == 'person':
url = endpoints.followed_people.format('?embed=person')
q = self._endpoint_premium_get(url)
if q:
return [FollowedPerson(person) for person in q]
else:
raise NoFollowedPeople('You have not followed any people yet')
def get_followed_person(self, person_id):
# type: (int) -> FollowedPerson
url = endpoints.followed_people.format(f'/{person_id}')
q = self._endpoint_premium_get(url)
if q:
return FollowedPerson(q)
else:
raise PersonNotFound(f'Person with ID {person_id} is not followed')
def follow_person(self, person_id):
# type: (int) -> None
url = endpoints.followed_people.format(f'/{person_id}')
q = self._endpoint_premium_put(url)
if not q:
raise PersonNotFound(f'Person with ID {person_id} does not exist')
def unfollow_person(self, person_id):
# type: (int) -> None
url = endpoints.followed_people.format(f'/{person_id}')
q = self._endpoint_premium_delete(url)
if not q:
raise PersonNotFollowed(f'Person with ID {person_id} was not followed')
def get_followed_networks(self, embed=None):
# type: (str) -> List[FollowedNetwork]
if embed not in [None, 'network']:
raise InvalidEmbedValue('Value for embed must be "network" or None')
url = endpoints.followed_networks.format('/')
if embed == 'network':
url = endpoints.followed_networks.format('?embed=network')
q = self._endpoint_premium_get(url)
if q:
return [FollowedNetwork(network) for network in q]
else:
raise NoFollowedNetworks('You have not followed any networks yet')
def get_followed_network(self, network_id):
# type: (int) -> FollowedNetwork
url = endpoints.followed_networks.format(f'/{network_id}')
q = self._endpoint_premium_get(url)
if q:
return FollowedNetwork(q)
else:
raise NetworkNotFound(f'Network with ID {network_id} is not followed')
def follow_network(self, network_id):
# type: (int) -> None
url = endpoints.followed_networks.format(f'/{network_id}')
q = self._endpoint_premium_put(url)
if not q:
raise NetworkNotFound(f'Network with ID {network_id} does not exist')
def unfollow_network(self, network_id):
# type: (int) -> None
url = endpoints.followed_networks.format(f'/{network_id}')
q = self._endpoint_premium_delete(url)
if not q:
raise NetworkNotFollowed(f'Network with ID {network_id} was not followed')
def get_followed_web_channels(self, embed=None):
# type: (str) -> List[FollowedWebChannel]
if embed not in [None, 'webchannel']:
raise InvalidEmbedValue('Value for embed must be "webchannel" or None')
url = endpoints.followed_web_channels.format('/')
if embed == 'webchannel':
url = endpoints.followed_web_channels.format('?embed=webchannel')
q = self._endpoint_premium_get(url)
if q:
return [FollowedWebChannel(webchannel) for webchannel in q]
else:
raise NoFollowedWebChannels('You have not followed any Web Channels yet')
def get_followed_web_channel(self, webchannel_id):
# type: (int) -> FollowedWebChannel
url = endpoints.followed_web_channels.format(f'/{webchannel_id}')
q = self._endpoint_premium_get(url)
if q:
return FollowedWebChannel(q)
else:
raise NetworkNotFound('Web Channel with ID {} is not followed'.format(webchannel_id))
def follow_web_channel(self, webchannel_id):
# type: (int) -> None
url = endpoints.followed_web_channels.format(f'/{webchannel_id}')
q = self._endpoint_premium_put(url)
if not q:
raise WebChannelNotFound(f'Web Channel with ID {webchannel_id} does not exist')
def unfollow_web_channel(self, webchannel_id):
# type: (int) -> None
url = endpoints.followed_web_channels.format(f'/{webchannel_id}')
q = self._endpoint_premium_delete(url)
if not q:
raise WebChannelNotFollowed(f'Web Channel with ID {webchannel_id} was not followed')
def get_marked_episodes(self, maze_id=None):
# type: (int) -> List[MarkedEpisode]
if not maze_id:
url = endpoints.marked_episodes.format('/')
else:
show_id = '?show_id={}'.format(maze_id)
url = endpoints.marked_episodes.format(show_id)
q = self._endpoint_premium_get(url)
if q:
return [MarkedEpisode(episode) for episode in q]
else:
raise NoMarkedEpisodes('You have not marked any episodes yet')
def get_marked_episode(self, episode_id):
# type: (int) -> MarkedEpisode
url = endpoints.marked_episodes.format(f'/{episode_id}')
q = self._endpoint_premium_get(url)
if q:
return MarkedEpisode(q)
else:
raise EpisodeNotMarked(f'Episode with ID {episode_id} is not marked')
def mark_episode(self, episode_id, mark_type):
# type: (int, str) -> None
types = {'watched': 0, 'acquired': 1, 'skipped': 2}
try:
status = types[mark_type]
except IndexError:
raise InvalidMarkedEpisodeType('Episode must be marked as "watched", "acquired", or "skipped"')
payload = {'type': str(status)}
url = endpoints.marked_episodes.format(f'/{episode_id}')
q = self._endpoint_premium_put(url, payload=payload)
if not q:
raise EpisodeNotFound(f'Episode with ID {episode_id} does not exist')
def unmark_episode(self, episode_id):
# type: (int) -> None
url = endpoints.marked_episodes.format(f'/{episode_id}')
q = self._endpoint_premium_delete(url)
if not q:
raise EpisodeNotMarked(f'Episode with ID {episode_id} was not marked')
def get_voted_shows(self, embed=None):
# type: (str) -> List[VotedShow]
if embed not in [None, 'show']:
raise InvalidEmbedValue('Value for embed must be "show" or None')
url = endpoints.voted_shows.format('/')
if embed == 'show':
url = endpoints.voted_shows.format('?embed=show')
q = self._endpoint_premium_get(url)
if q:
return [VotedShow(show) for show in q]
else:
raise NoVotedShows('You have not voted for any shows yet')
def get_voted_show(self, maze_id):
# type: (int) -> VotedShow
url = endpoints.voted_shows.format(f'/{maze_id}')
q = self._endpoint_premium_get(url)
if q:
return VotedShow(q)
else:
raise ShowNotVotedFor(f'Show with ID {maze_id} not voted for')
def remove_show_vote(self, maze_id):
# type: (int) -> None
url = endpoints.voted_shows.format(f'/{maze_id}')
q = self._endpoint_premium_delete(url)
if not q:
raise ShowNotVotedFor(f'Show with ID {maze_id} was not voted for')
def vote_show(self, maze_id, vote):
# type: (int, int) -> None
if not 1 <= vote <= 10:
raise InvalidVoteValue('Vote must be an integer between 1 and 10')
payload = {'vote': int(vote)}
url = endpoints.voted_shows.format(f'/{maze_id}')
q = self._endpoint_premium_put(url, payload=payload)
if not q:
raise ShowNotFound(f'Show with ID {maze_id} does not exist')
def get_voted_episodes(self):
# type: (...) -> List[VotedEpisode]
url = endpoints.voted_episodes.format('/')
q = self._endpoint_premium_get(url)
if q:
return [VotedEpisode(episode) for episode in q]
else:
raise NoVotedEpisodes('You have not voted for any episodes yet')
def get_voted_episode(self, episode_id):
# type: (int) -> VotedEpisode
url = endpoints.voted_episodes.format(f'/{episode_id}')
q = self._endpoint_premium_get(url)
if q:
return VotedEpisode(q)
else:
raise EpisodeNotVotedFor(f'Episode with ID {episode_id} not voted for')
def remove_episode_vote(self, episode_id):
# type: (int) -> None
url = endpoints.voted_episodes.format(f'/{episode_id}')
q = self._endpoint_premium_delete(url)
if not q:
raise EpisodeNotVotedFor(f'Episode with ID {episode_id} was not voted for')
def vote_episode(self, episode_id, vote):
# type: (int, int) -> None
if not 1 <= vote <= 10:
raise InvalidVoteValue('Vote must be an integer between 1 and 10')
payload = {'vote': int(vote)}
url = endpoints.voted_episodes.format(f'/{episode_id}')
q = self._endpoint_premium_put(url, payload=payload)
if not q:
raise EpisodeNotFound(f'Episode with ID {episode_id} does not exist')
# Return list of Show objects
def get_show_list(show_name):
# type: (str) -> List[Show]
"""
Return list of Show objects from the TVmaze "Show Search" endpoint
List will be ordered by tvmaze score and should mimic the results you see
by doing a show search on the website.
:param show_name: Name of show
:return: List of Show(s)
"""
shows = show_search(show_name)
return shows
# Get list of Person objects
def get_people(name):
# type: (str) -> List[Person]
"""
Return list of Person objects from the TVmaze "People Search" endpoint
:param name: Name of person
:return: List of Person(s)
"""
people = people_search(name)
if people:
return people
def show_search(show):
# type: (str) -> List[Show]
_show = _url_quote(show)
url = endpoints.show_search.format(_show)
q = TVmaze.endpoint_standard_get(url)
if q:
shows = []
for result in q:
show = Show(result['show'])
show.score = result['score']
shows.append(show)
return shows
else:
raise ShowNotFound(f'Show {show} not found')
def show_single_search(show, embed=None):
# type: (str, str) -> Show
_show = _url_quote(show)
url = _embed_url(endpoints.show_single_search.format(_show), embed,
[None, 'episodes', 'cast', 'previousepisode', 'nextepisode'], '&')
q = TVmaze.endpoint_standard_get(url)
if q:
return Show(q)
else:
raise ShowNotFound(f'show name "{show}" not found')
def lookup_tvrage(tvrage_id):
# type: (Union[int, str]) -> Show
url = endpoints.lookup_tvrage.format(tvrage_id)
q = TVmaze.endpoint_standard_get(url)
if q:
return Show(q)
else:
raise IDNotFound(f'TVRage id {tvrage_id} not found')
def lookup_tvdb(tvdb_id):
# type: (Union[int, str]) -> Show
url = endpoints.lookup_tvdb.format(tvdb_id)
q = TVmaze.endpoint_standard_get(url)
if q:
return Show(q)
else:
raise IDNotFound(f'TVDB ID {tvdb_id} not found')
def lookup_imdb(imdb_id):
# type: (str) -> Show
url = endpoints.lookup_imdb.format(imdb_id)
q = TVmaze.endpoint_standard_get(url)
if q:
return Show(q)
else:
raise IDNotFound(f'IMDB ID {imdb_id} not found')
def get_schedule(country='US', date=str(datetime.today().date())):
# type: (str, str) -> List[Episode]
url = endpoints.get_schedule.format(country, date)
q = TVmaze.endpoint_standard_get(url)
if q:
return [Episode(episode) for episode in q]
else:
raise ScheduleNotFound(f'Schedule for country {country} at date {date} not found')
# ALL known future episodes, several MB large, cached for 24 hours
def get_full_schedule():
# type: (...) -> List[Episode]
url = endpoints.get_full_schedule
q = TVmaze.endpoint_standard_get(url)
if q:
return [Episode(episode) for episode in q]
else:
raise GeneralError('Something went wrong, www.tvmaze.com may be down')
def show_main_info(maze_id, embed=None):
# type: (int, str) -> Show
url = _embed_url(endpoints.show_main_info.format(maze_id), embed,
[None, 'episodes', 'cast', 'previousepisode', 'nextepisode', 'episodeswithspecials'], '?')
q = TVmaze.endpoint_standard_get(url)
if q:
return Show(q)
else:
raise IDNotFound(f'Maze id {maze_id} not found')
def episode_list(maze_id, specials=None, raise_error=True, show=None):
# type: (int, bool, bool, Show) -> List[Episode]
if specials:
url = f'{endpoints.episode_list.format(maze_id)}?specials=1'
else:
url = endpoints.episode_list.format(maze_id)
q = TVmaze.endpoint_standard_get(url)
if type(q) == list:
return [Episode(episode, show) for episode in q]
elif raise_error:
raise IDNotFound(f'Maze id {maze_id} not found'.format)
return []
def episode_by_number(maze_id, season_number, episode_number):
# type: (int, int, int) -> Episode
url = endpoints.episode_by_number.format(maze_id,
season_number,
episode_number)
q = TVmaze.endpoint_standard_get(url)
if q:
return Episode(q)
else:
raise EpisodeNotFound(f'Couldn\'t find season {season_number} episode {episode_number} for TVmaze ID {maze_id}')
def episodes_by_date(maze_id, airdate):
# type: (int, str) -> List[Episode]
try:
datetime.strptime(airdate, '%Y-%m-%d')
except ValueError:
raise IllegalAirDate('Airdate must be string formatted as \"YYYY-MM-DD\"')
url = endpoints.episodes_by_date.format(maze_id, airdate)
q = TVmaze.endpoint_standard_get(url)
if q:
return [Episode(episode) for episode in q]
else:
raise NoEpisodesForAirdate(f'Couldn\'t find an episode airing {airdate} for TVmaze ID {maze_id}')
def show_cast(maze_id, raise_error=True):
# type: (int, bool) -> Cast
url = endpoints.show_cast.format(maze_id)
q = TVmaze.endpoint_standard_get(url)
if q:
return Cast(q)
elif raise_error:
raise CastNotFound(f'Couldn\'nt find show cast for TVmaze ID {maze_id}')
return Cast({})
def show_index(page=1):
# type: (int) -> List[Show]
url = endpoints.show_index.format(page)
q = TVmaze.endpoint_standard_get(url)
if q:
return [Show(show) for show in q]
else:
raise ShowIndexError('Error getting show index, www.tvmaze.com may be down')
def people_search(person):
# type: (AnyStr) -> List[Person]
person = _url_quote(person)
url = endpoints.people_search.format(person)
q = TVmaze.endpoint_standard_get(url)
if q:
return [Person(person) for person in q]
else:
raise PersonNotFound(f'Couldn\'t find person {person}')
def person_main_info(person_id, embed=None):
# type: (int, AnyStr) -> Person
url = _embed_url(endpoints.person_main_info.format(person_id), embed,
[None, 'castcredits', 'crewcredits'], '?')
q = TVmaze.endpoint_standard_get(url)
if q:
return Person(q)
else:
raise PersonNotFound(f'Couldn\'t find person {person_id}')
def person_cast_credits(person_id, embed=None, raise_error=True):
# type: (int, str, bool) -> List[CastCredit]
url = _embed_url(endpoints.person_cast_credits.format(person_id), embed,
[None, 'show', 'character'], '?')
q = TVmaze.endpoint_standard_get(url)
if q:
return [CastCredit(credit) for credit in q]
elif raise_error:
raise CreditsNotFound(f'Couldn\'t find cast credits for person ID {person_id}')
return []
def person_guestcast_credits(person_id, embed=None, raise_error=True):
# type: (int, str, bool) -> List[CastCredit]
url = _embed_url(endpoints.person_guestcast_credits.format(person_id), embed,
[None, 'episode', 'character'], '?')
q = TVmaze.endpoint_standard_get(url)
if q:
return [CastCredit(credit) for credit in q]
elif raise_error:
raise CreditsNotFound(f'Couldn\'t find cast credits for person ID {person_id}')
return []
def person_crew_credits(person_id, embed=None, raise_error=True):
# type: (int, str, bool) -> List[CrewCredit]
url = _embed_url(endpoints.person_crew_credits.format(person_id), embed,
[None, 'show'], '?')
q = TVmaze.endpoint_standard_get(url)
if q:
return [CrewCredit(credit) for credit in q]
elif raise_error:
raise CreditsNotFound(f'Couldn\'t find crew credits for person ID {person_id}')
return []
def get_show_crew(maze_id, raise_error=True):
# type: (int, bool) -> List[Crew]
url = endpoints.show_crew.format(maze_id)
q = TVmaze.endpoint_standard_get(url)
if q:
return [Crew(crew) for crew in q]
elif raise_error:
raise CrewNotFound(f'Couldn\'t find crew for TVmaze ID {maze_id}')
return []
def get_show_images(maze_id, raise_error=True):
# type: (int, bool) -> List[Image]
url = endpoints.show_images.format(maze_id)
q = TVmaze.endpoint_standard_get(url)
if q:
return [Image(img) for img in q]
elif raise_error:
raise ShowImagesNotFound(f'Couldn\'t find images for TVmaze ID {maze_id}')
return []
def show_updates(since=None):
# type: (AnyStr) -> Updates
"""
returns all or in given timeframe changed shows
:param since: None, "day", "week", "month"
"""
if since not in ('day', 'week', 'month', None):
raise InvalidTimeFrame('Only supported are: None, "day", "week", "month"')
url = '%s%s' % (endpoints.show_updates, ('', '?since=%s' % since)[None is not since])
q = TVmaze.endpoint_standard_get(url)
if q:
return Updates(q)
else:
raise ShowIndexError('Error getting show updates, www.tvmaze.com may be down')
def show_akas(maze_id, raise_error=True):
# type: (int, bool) -> List[AKA]
url = endpoints.show_akas.format(maze_id)
q = TVmaze.endpoint_standard_get(url)
if q:
return [AKA(aka) for aka in q]
elif raise_error:
raise AKASNotFound(f'Couldn\'t find AKA\'s for TVmaze ID {maze_id}')
return []
def show_seasons(maze_id, raise_error=True, show=None):
# type: (int, bool, Show) -> Dict[int, Season]
url = endpoints.show_seasons.format(maze_id)
q = TVmaze.endpoint_standard_get(url)
if q:
season_dict = dict()
for season in q:
season_dict[season['number']] = Season(season, show=show, season_number=season['number'])
return season_dict
elif raise_error:
raise SeasonNotFound(f'Couldn\'t find Season\'s for TVmaze ID {maze_id}')
return {}
def season_by_id(season_id, embed=None):
# type: (int, str) -> Season
url = _embed_url(endpoints.season_by_id.format(season_id), embed,
[None, 'episodes'], '?')
q = TVmaze.endpoint_standard_get(url)
if q:
return Season(q)
else:
raise SeasonNotFound(f'Couldn\'t find Season with ID {season_id}')
def episode_by_id(episode_id, show=None, raise_error=True, embed=None):
# type: (int, Show, bool, str) -> Episode
url = _embed_url(endpoints.episode_by_id.format(episode_id), embed,
[None, 'show', 'guestcast', 'guestcrew'], '?')
q = TVmaze.endpoint_standard_get(url)
if q:
return Episode(q, show=show)
elif raise_error:
raise EpisodeNotFound(f'Couldn\'t find Episode with ID {episode_id}')
def episode_guestcast_credits(episode_id, raise_error=True):
# type: (int, bool) -> List[CastCredit]
url = endpoints.episode_guestcast.format(episode_id)
q = TVmaze.endpoint_standard_get(url)
if q:
return [CastCredit(credit) for credit in q]
elif raise_error:
raise CreditsNotFound(f'Couldn\'t find cast credits for episode ID {episode_id}')
return []
def episode_crew_credits(episode_id, raise_error=True):
# type: (int, bool) -> List[CrewCredit]
url = endpoints.episode_guestcrew.format(episode_id)
q = TVmaze.endpoint_standard_get(url)
if q:
return [CrewCredit(credit) for credit in q]
elif raise_error:
raise CreditsNotFound(f'Couldn\'t find crew credits for episode ID {episode_id}')
return []