516 lines
22 KiB
Python
516 lines
22 KiB
Python
# -*- coding: UTF-8 -*-
|
|
#
|
|
# Copyright (C) 2020, Team Kodi
|
|
#
|
|
# This program is free software: you can redistribute it and/or modify
|
|
# it under the terms of the GNU General Public License as published by
|
|
# the Free Software Foundation, either version 3 of the License, or
|
|
# (at your option) any later version.
|
|
#
|
|
# This program is distributed in the hope that it will be useful,
|
|
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
# GNU General Public License for more details.
|
|
#
|
|
# You should have received a copy of the GNU General Public License
|
|
# along with this program. If not, see <https://www.gnu.org/licenses/>.
|
|
# pylint: disable=missing-docstring
|
|
|
|
"""Functions to interact with TMDb API"""
|
|
|
|
from __future__ import absolute_import, unicode_literals
|
|
|
|
import unicodedata
|
|
from math import floor
|
|
from pprint import pformat
|
|
from . import cache, data_utils, api_utils, settings, imdbratings, traktratings
|
|
from .utils import logger
|
|
try:
|
|
from typing import Text, Optional, Union, List, Dict, Any # pylint: disable=unused-import
|
|
InfoType = Dict[Text, Any] # pylint: disable=invalid-name
|
|
except ImportError:
|
|
pass
|
|
|
|
HEADERS = (
|
|
('User-Agent', 'Kodi TV Show scraper by Team Kodi; contact pkscout@kodi.tv'),
|
|
('Accept', 'application/json'),
|
|
)
|
|
|
|
TMDB_PARAMS = {'api_key': settings.TMDB_CLOWNCAR}
|
|
BASE_URL = 'https://api.themoviedb.org/3/{}'
|
|
EPISODE_GROUP_URL = BASE_URL.format('tv/episode_group/{}')
|
|
SEARCH_URL = BASE_URL.format('search/tv')
|
|
FIND_URL = BASE_URL.format('find/{}')
|
|
SHOW_URL = BASE_URL.format('tv/{}')
|
|
SEASON_URL = BASE_URL.format('tv/{}/season/{}')
|
|
EPISODE_URL = BASE_URL.format('tv/{}/season/{}/episode/{}')
|
|
FANARTTV_URL = 'https://webservice.fanart.tv/v3/tv/{}'
|
|
FANARTTV_PARAMS = {'api_key': settings.FANARTTV_CLOWNCAR}
|
|
|
|
|
|
def _get_params():
|
|
source_settings = settings.getSourceSettings()
|
|
params = TMDB_PARAMS.copy()
|
|
params['language'] = source_settings["LANG_DETAILS"]
|
|
return params
|
|
|
|
|
|
def search_show(title, year=None):
|
|
# type: (Text, Text) -> List
|
|
"""
|
|
Search for a single TV show
|
|
|
|
:param title: TV show title to search
|
|
: param year: the year to search (optional)
|
|
:return: a list with found TV shows
|
|
"""
|
|
source_settings = settings.getSourceSettings()
|
|
api_utils.set_headers(dict(HEADERS))
|
|
params = _get_params()
|
|
results = []
|
|
ext_media_id = data_utils.parse_media_id(title)
|
|
if ext_media_id:
|
|
logger.debug('using %s of %s to find show' %
|
|
(ext_media_id['type'], ext_media_id['title']))
|
|
if ext_media_id['type'] == 'tmdb_id':
|
|
search_url = SHOW_URL.format(ext_media_id['title'])
|
|
else:
|
|
search_url = FIND_URL.format(ext_media_id['title'])
|
|
params['external_source'] = ext_media_id['type']
|
|
else:
|
|
logger.debug('using title of %s to find show' % title)
|
|
search_url = SEARCH_URL
|
|
params['query'] = unicodedata.normalize('NFKC', title)
|
|
if year:
|
|
params['first_air_date_year'] = str(year)
|
|
resp = api_utils.load_info(
|
|
search_url, params=params, verboselog=source_settings["VERBOSELOG"])
|
|
if resp is not None:
|
|
if ext_media_id:
|
|
if ext_media_id['type'] == 'tmdb_id':
|
|
if resp.get('success') == 'false':
|
|
results = []
|
|
else:
|
|
results = [resp]
|
|
else:
|
|
results = resp.get('tv_results', [])
|
|
else:
|
|
results = resp.get('results', [])
|
|
api_utils.set_headers({})
|
|
return results
|
|
|
|
|
|
def find_by_id(unique_ids):
|
|
"""
|
|
Find a show by external IDs
|
|
:param unique_ids: dict of external IDs
|
|
:return: a list with found TV shows
|
|
"""
|
|
supported_ids = ['imdb', 'facebook', 'instagram',
|
|
'tvdb', 'tiktok', 'twitter', 'wikidata', 'youtube']
|
|
source_settings = settings.getSourceSettings()
|
|
api_utils.set_headers(dict(HEADERS))
|
|
params = _get_params()
|
|
for key, value in unique_ids.items():
|
|
if key in supported_ids:
|
|
params['external_source'] = key + '_id'
|
|
search_url = FIND_URL.format(value)
|
|
resp = api_utils.load_info(
|
|
search_url, params=params, verboselog=source_settings["VERBOSELOG"])
|
|
if resp is not None:
|
|
return resp.get('tv_results', [])
|
|
api_utils.set_headers()
|
|
return []
|
|
|
|
|
|
def load_episode_list(show_info, season_map, ep_grouping):
|
|
# type: (InfoType, Dict, Text) -> Optional[InfoType]
|
|
"""get the IMDB ratings details"""
|
|
"""Load episode list from themoviedb.org API"""
|
|
source_settings = settings.getSourceSettings()
|
|
api_utils.set_headers(dict(HEADERS))
|
|
episode_list = []
|
|
if ep_grouping is not None:
|
|
logger.debug(
|
|
'Getting episodes with episode grouping of ' + ep_grouping)
|
|
episode_group_url = EPISODE_GROUP_URL.format(ep_grouping)
|
|
custom_order = api_utils.load_info(
|
|
episode_group_url, params=TMDB_PARAMS, verboselog=source_settings["VERBOSELOG"])
|
|
if custom_order is not None:
|
|
show_info['seasons'] = []
|
|
for custom_season in custom_order.get('groups', []):
|
|
season_episodes = []
|
|
try:
|
|
current_season = season_map.get(
|
|
str(custom_season['episodes'][0]['season_number']), {}).copy()
|
|
except IndexError:
|
|
continue
|
|
current_season['name'] = custom_season['name']
|
|
current_season['season_number'] = custom_season['order']
|
|
for episode in custom_season['episodes']:
|
|
episode['org_seasonnum'] = episode['season_number']
|
|
episode['org_epnum'] = episode['episode_number']
|
|
episode['season_number'] = custom_season['order']
|
|
episode['episode_number'] = episode['order'] + 1
|
|
season_episodes.append(episode)
|
|
episode_list.append(episode)
|
|
current_season['episodes'] = season_episodes
|
|
show_info['seasons'].append(current_season)
|
|
else:
|
|
logger.debug('Getting episodes from standard season list')
|
|
show_info['seasons'] = []
|
|
for key, value in season_map.items():
|
|
show_info['seasons'].append(value)
|
|
for season in show_info.get('seasons', []):
|
|
for episode in season.get('episodes', []):
|
|
episode['org_seasonnum'] = episode['season_number']
|
|
episode['org_epnum'] = episode['episode_number']
|
|
episode_list.append(episode)
|
|
show_info['episodes'] = episode_list
|
|
api_utils.set_headers({})
|
|
return show_info
|
|
|
|
|
|
def load_show_info(show_id, ep_grouping=None, named_seasons=None):
|
|
# type: (Text, Text, Dict) -> Optional[InfoType]
|
|
"""
|
|
Get full info for a single show
|
|
|
|
:param show_id: themoviedb.org show ID
|
|
:param ep_grouping: the episode group from TMDb
|
|
:param named_seasons: the named seasons from the NFO file
|
|
:return: show info or None
|
|
"""
|
|
|
|
api_utils.set_headers(dict(HEADERS))
|
|
source_settings = settings.getSourceSettings()
|
|
if named_seasons == None:
|
|
named_seasons = []
|
|
show_info = cache.load_show_info_from_cache(show_id)
|
|
if show_info is None:
|
|
logger.debug('no cache file found, loading from scratch')
|
|
show_url = SHOW_URL.format(show_id)
|
|
params = _get_params()
|
|
params['append_to_response'] = 'credits,content_ratings,external_ids,images,videos,keywords'
|
|
params['include_image_language'] = '%s,en,null' % source_settings["LANG_IMAGES"][0:2]
|
|
params['include_video_language'] = '%s,en,null' % source_settings["LANG_IMAGES"][0:2]
|
|
show_info = api_utils.load_info(
|
|
show_url, params=params, verboselog=source_settings["VERBOSELOG"])
|
|
if show_info is None:
|
|
return None
|
|
if show_info['overview'] == '' and source_settings["LANG_DETAILS"] != 'en-US':
|
|
params['language'] = 'en-US'
|
|
del params['append_to_response']
|
|
show_info_backup = api_utils.load_info(
|
|
show_url, params=params, verboselog=source_settings["VERBOSELOG"])
|
|
if show_info_backup is not None:
|
|
show_info['overview'] = show_info_backup.get('overview', '')
|
|
params['language'] = source_settings["LANG_DETAILS"]
|
|
season_map = {}
|
|
params['append_to_response'] = 'credits,images'
|
|
for season in show_info.get('seasons', []):
|
|
season_url = SEASON_URL.format(
|
|
show_id, season.get('season_number', 0))
|
|
season_info = api_utils.load_info(
|
|
season_url, params=params, default={}, verboselog=source_settings["VERBOSELOG"])
|
|
if (season_info.get('overview', '') == '' or season_info.get('name', '').lower().startswith('season')) and source_settings["LANG_DETAILS"] != 'en-US':
|
|
params['language'] = 'en-US'
|
|
season_info_backup = api_utils.load_info(
|
|
season_url, params=params, default={}, verboselog=source_settings["VERBOSELOG"])
|
|
params['language'] = source_settings["LANG_DETAILS"]
|
|
if season_info.get('overview', '') == '':
|
|
season_info['overview'] = season_info_backup.get(
|
|
'overview', '')
|
|
if season_info.get('name', '').lower().startswith('season'):
|
|
season_info['name'] = season_info_backup.get('name', '')
|
|
# this is part of a work around for xbmcgui.ListItem.addSeasons() not respecting NFO file information
|
|
for named_season in named_seasons:
|
|
if str(named_season[0]) == str(season.get('season_number')):
|
|
logger.debug('adding season name of %s from named seasons in NFO for season %s' % (
|
|
named_season[1], season['season_number']))
|
|
season_info['name'] = named_season[1]
|
|
break
|
|
# end work around
|
|
season_info['images'] = _sort_image_types(
|
|
season_info.get('images', {}))
|
|
season_map[str(season.get('season_number', 0))] = season_info
|
|
show_info = load_episode_list(show_info, season_map, ep_grouping)
|
|
show_info['ratings'] = load_ratings(show_info)
|
|
show_info = load_fanarttv_art(show_info)
|
|
show_info['images'] = _sort_image_types(show_info.get('images', {}))
|
|
show_info = trim_artwork(show_info)
|
|
cast_check = []
|
|
cast = []
|
|
for season in reversed(show_info.get('seasons', [])):
|
|
for cast_member in season.get('credits', {}).get('cast', []):
|
|
if cast_member.get('name', '') not in cast_check:
|
|
cast.append(cast_member)
|
|
cast_check.append(cast_member.get('name', ''))
|
|
show_info['credits']['cast'] = cast
|
|
logger.debug('saving show info to the cache')
|
|
if source_settings["VERBOSELOG"]:
|
|
logger.debug(format(pformat(show_info)))
|
|
cache.cache_show_info(show_info)
|
|
else:
|
|
logger.debug('using cached show info')
|
|
api_utils.set_headers({})
|
|
return show_info
|
|
|
|
|
|
def load_episode_info(show_id, episode_id):
|
|
# type: (Text, Text) -> Optional[InfoType]
|
|
"""
|
|
Load episode info
|
|
|
|
:param show_id:
|
|
:param episode_id:
|
|
:return: episode info or None
|
|
"""
|
|
source_settings = settings.getSourceSettings()
|
|
api_utils.set_headers(dict(HEADERS))
|
|
show_info = load_show_info(show_id)
|
|
if show_info is not None:
|
|
try:
|
|
episode_info = show_info['episodes'][int(episode_id)]
|
|
except KeyError:
|
|
return None
|
|
# this ensures we are using the season/ep from the episode grouping if provided
|
|
ep_url = EPISODE_URL.format(
|
|
show_info['id'], episode_info['org_seasonnum'], episode_info['org_epnum'])
|
|
params = _get_params()
|
|
params['append_to_response'] = 'credits,external_ids,images'
|
|
params['include_image_language'] = '%s,en,null' % source_settings["LANG_IMAGES"][0:2]
|
|
ep_return = api_utils.load_info(
|
|
ep_url, params=params, verboselog=source_settings["VERBOSELOG"])
|
|
if ep_return is None:
|
|
return None
|
|
bad_return_name = False
|
|
bad_return_overview = False
|
|
check_name = ep_return.get('name')
|
|
if check_name == None:
|
|
bad_return_name = True
|
|
ep_return['name'] = 'Episode ' + \
|
|
str(episode_info['episode_number'])
|
|
elif check_name.lower().startswith('episode') or check_name == '':
|
|
bad_return_name = True
|
|
if ep_return.get('overview', '') == '':
|
|
bad_return_overview = True
|
|
if (bad_return_overview or bad_return_name) and source_settings["LANG_DETAILS"] != 'en-US':
|
|
params['language'] = 'en-US'
|
|
del params['append_to_response']
|
|
ep_return_backup = api_utils.load_info(
|
|
ep_url, params=params, verboselog=source_settings["VERBOSELOG"])
|
|
if ep_return_backup is not None:
|
|
if bad_return_overview:
|
|
ep_return['overview'] = ep_return_backup.get(
|
|
'overview', '')
|
|
if bad_return_name:
|
|
ep_return['name'] = ep_return_backup.get(
|
|
'name', 'Episode ' + str(episode_info['episode_number']))
|
|
ep_return['images'] = _sort_image_types(ep_return.get('images', {}))
|
|
ep_return['season_number'] = episode_info['season_number']
|
|
ep_return['episode_number'] = episode_info['episode_number']
|
|
ep_return['org_seasonnum'] = episode_info['org_seasonnum']
|
|
ep_return['org_epnum'] = episode_info['org_epnum']
|
|
ep_return['ratings'] = load_ratings(
|
|
ep_return, show_imdb_id=show_info.get('external_ids', {}).get('imdb_id'))
|
|
for season in show_info.get('seasons', []):
|
|
if season.get('season_number') == episode_info['season_number']:
|
|
ep_return['season_cast'] = season.get(
|
|
'credits', {}).get('cast', [])
|
|
break
|
|
show_info['episodes'][int(episode_id)] = ep_return
|
|
cache.cache_show_info(show_info)
|
|
api_utils.set_headers({})
|
|
return ep_return
|
|
api_utils.set_headers({})
|
|
return None
|
|
|
|
|
|
def load_ratings(the_info, show_imdb_id=''):
|
|
# type: (InfoType, Text) -> Dict
|
|
"""
|
|
Load the ratings for the show/episode
|
|
|
|
:param the_info: show or episode info
|
|
:param show_imdb_id: show IMDB
|
|
:return: ratings or empty dict
|
|
"""
|
|
source_settings = settings.getSourceSettings()
|
|
ratings = {}
|
|
imdb_id = the_info.get('external_ids', {}).get('imdb_id')
|
|
for rating_type in source_settings["RATING_TYPES"]:
|
|
logger.debug('setting rating using %s' % rating_type)
|
|
if rating_type == 'tmdb':
|
|
ratings['tmdb'] = {'votes': the_info['vote_count'],
|
|
'rating': the_info['vote_average']}
|
|
elif rating_type == 'imdb' and imdb_id:
|
|
imdb_rating = imdbratings.get_details(imdb_id).get('ratings')
|
|
if imdb_rating:
|
|
ratings.update(imdb_rating)
|
|
elif rating_type == 'trakt':
|
|
if show_imdb_id:
|
|
season = the_info['org_seasonnum']
|
|
episode = the_info['org_epnum']
|
|
resp = traktratings.get_details(
|
|
show_imdb_id, season=season, episode=episode)
|
|
else:
|
|
resp = traktratings.get_details(imdb_id)
|
|
trakt_rating = resp.get('ratings')
|
|
if trakt_rating:
|
|
ratings.update(trakt_rating)
|
|
logger.debug('returning ratings of\n{}'.format(pformat(ratings)))
|
|
return ratings
|
|
|
|
|
|
def load_fanarttv_art(show_info):
|
|
# type: (InfoType) -> Optional[InfoType]
|
|
"""
|
|
Add fanart.tv images for a show
|
|
|
|
:param show_info: the current show info
|
|
:return: show info
|
|
"""
|
|
source_settings = settings.getSourceSettings()
|
|
api_utils.set_headers(dict(HEADERS))
|
|
if source_settings["FANARTTV_CLIENTKEY"]:
|
|
FANARTTV_PARAMS['client_key'] = source_settings["FANARTTV_CLIENTKEY"]
|
|
|
|
tvdb_id = show_info.get('external_ids', {}).get('tvdb_id')
|
|
if tvdb_id and source_settings["FANARTTV_ENABLE"]:
|
|
fanarttv_url = FANARTTV_URL.format(tvdb_id)
|
|
artwork = api_utils.load_info(
|
|
fanarttv_url, params=FANARTTV_PARAMS, verboselog=source_settings["VERBOSELOG"])
|
|
if artwork is None:
|
|
return show_info
|
|
for fanarttv_type, tmdb_type in settings.FANARTTV_MAPPING.items():
|
|
if not show_info['images'].get(tmdb_type) and not tmdb_type.startswith('season'):
|
|
show_info['images'][tmdb_type] = []
|
|
for item in artwork.get(fanarttv_type, []):
|
|
lang = item.get('lang')
|
|
if lang == '' or lang == '00':
|
|
lang = None
|
|
filepath = ''
|
|
if lang is None or lang == source_settings["LANG_DETAILS"][0:2] or lang == 'en':
|
|
filepath = item.get('url')
|
|
if filepath:
|
|
if tmdb_type.startswith('season'):
|
|
image_type = tmdb_type[6:]
|
|
for s in range(len(show_info.get('seasons', []))):
|
|
season_num = show_info['seasons'][s]['season_number']
|
|
artseason = item.get('season', '')
|
|
if not show_info['seasons'][s].get('images'):
|
|
show_info['seasons'][s]['images'] = {}
|
|
if not show_info['seasons'][s]['images'].get(image_type):
|
|
show_info['seasons'][s]['images'][image_type] = []
|
|
if artseason == '' or artseason == str(season_num):
|
|
show_info['seasons'][s]['images'][image_type].append(
|
|
{'file_path': filepath, 'type': 'fanarttv', 'iso_639_1': lang})
|
|
else:
|
|
show_info['images'][tmdb_type].append(
|
|
{'file_path': filepath, 'type': 'fanarttv', 'iso_639_1': lang})
|
|
api_utils.set_headers({})
|
|
return show_info
|
|
|
|
|
|
def trim_artwork(show_info):
|
|
# type: (InfoType) -> Optional[InfoType]
|
|
"""
|
|
Trim artwork to keep the text blob below 65K characters
|
|
|
|
:param show_info: the current show info
|
|
:return: show info
|
|
"""
|
|
image_counts = {}
|
|
image_total = 0
|
|
backdrops_total = 0
|
|
for image_type, image_list in show_info.get('images', {}).items():
|
|
total = len(image_list)
|
|
if image_type == 'backdrops':
|
|
backdrops_total = backdrops_total + total
|
|
else:
|
|
image_counts[image_type] = {'total': total}
|
|
image_total = image_total + total
|
|
for season in show_info.get('seasons', []):
|
|
for image_type, image_list in season.get('images', {}).items():
|
|
total = len(image_list)
|
|
thetype = '%s_%s' % (str(season['season_number']), image_type)
|
|
image_counts[thetype] = {'total': total}
|
|
image_total = image_total + total
|
|
if image_total <= settings.MAXIMAGES and backdrops_total <= settings.MAXIMAGES:
|
|
return show_info
|
|
if backdrops_total > settings.MAXIMAGES:
|
|
logger.error('there are %s fanart images' % str(backdrops_total))
|
|
logger.error('that is more than the max of %s, image results will be trimmed to the max' % str(
|
|
settings.MAXIMAGES))
|
|
reduce = -1 * (backdrops_total - settings.MAXIMAGES)
|
|
del show_info['images']['backdrops'][reduce:]
|
|
if image_total > settings.MAXIMAGES:
|
|
reduction = (image_total - settings.MAXIMAGES)/image_total
|
|
logger.error('there are %s non-fanart images' % str(image_total))
|
|
logger.error('that is more than the max of %s, image results will be trimmed by %s' % (
|
|
str(settings.MAXIMAGES), str(reduction)))
|
|
for key, value in image_counts.items():
|
|
image_counts[key]['reduce'] = -1 * \
|
|
int(floor(value['total'] * reduction))
|
|
logger.debug('%s: %s' % (key, pformat(image_counts[key])))
|
|
for image_type, image_list in show_info.get('images', {}).items():
|
|
if image_type == 'backdrops':
|
|
continue # already handled backdrops above
|
|
reduce = image_counts[image_type]['reduce']
|
|
if reduce != 0:
|
|
del show_info['images'][image_type][reduce:]
|
|
for s in range(len(show_info.get('seasons', []))):
|
|
for image_type, image_list in show_info['seasons'][s].get('images', {}).items():
|
|
thetype = '%s_%s' % (
|
|
str(show_info['seasons'][s]['season_number']), image_type)
|
|
reduce = image_counts[thetype]['reduce']
|
|
if reduce != 0:
|
|
del show_info['seasons'][s]['images'][image_type][reduce:]
|
|
return show_info
|
|
|
|
|
|
def _sort_image_types(imagelist):
|
|
# type: (Dict) -> Dict
|
|
"""
|
|
sort the images by language
|
|
|
|
:param imagelist:
|
|
:return: imagelist
|
|
"""
|
|
for image_type, images in imagelist.items():
|
|
imagelist[image_type] = _image_sort(images, image_type)
|
|
return imagelist
|
|
|
|
|
|
def _image_sort(images, image_type):
|
|
# type: (List, Text) -> List
|
|
"""
|
|
sort the images by language
|
|
|
|
:param images:
|
|
:param image_type:
|
|
:return: list of images
|
|
"""
|
|
source_settings = settings.getSourceSettings()
|
|
lang_pref = []
|
|
lang_null = []
|
|
lang_en = []
|
|
firstimage = True
|
|
for image in images:
|
|
image_lang = image.get('iso_639_1')
|
|
if image_lang == source_settings["LANG_DETAILS"][0:2]:
|
|
lang_pref.append(image)
|
|
elif image_lang == 'en':
|
|
lang_en.append(image)
|
|
else:
|
|
if firstimage:
|
|
lang_pref.append(image)
|
|
else:
|
|
lang_null.append(image)
|
|
firstimage = False
|
|
if image_type == 'posters':
|
|
return lang_pref + lang_en + lang_null
|
|
else:
|
|
return lang_pref + lang_null + lang_en
|