Skip to content

Commit

Permalink
Add type annotations - WIP2
Browse files Browse the repository at this point in the history
  • Loading branch information
romanvm committed May 15, 2020
1 parent ba0c952 commit ce697a9
Show file tree
Hide file tree
Showing 6 changed files with 74 additions and 33 deletions.
25 changes: 12 additions & 13 deletions metadata.tvmaze/libs/actions.py
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,7 @@

"""Plugin route actions"""

from __future__ import absolute_import
from __future__ import absolute_import, unicode_literals
import sys
from six import itervalues
from six.moves import urllib_parse
Expand All @@ -27,15 +27,15 @@
from .utils import logger

try:
from typing import Optional
from typing import Optional, Text
except ImportError:
pass

HANDLE = int(sys.argv[1]) # type: int


def find_show(title, year=None):
# type: (str, Optional[str]) -> None
# type: (str, Optional[Text]) -> None
"""Find a show by title"""
logger.debug('Searching for TV show {} ({})'.format(title, year))
search_results = tvmaze.search_show(title)
Expand All @@ -45,7 +45,7 @@ def find_show(title, year=None):
for search_result in search_results:
show_name = search_result['show']['name']
if search_result['show']['premiered']:
show_name += u' ({})'.format(search_result['show']['premiered'][:4])
show_name += ' ({})'.format(search_result['show']['premiered'][:4])
list_item = xbmcgui.ListItem(show_name, offscreen=True)
list_item = data_utils.add_main_show_info(list_item, search_result['show'], False)
# Below "url" is some unique ID string (may be an actual URL to a show page)
Expand All @@ -59,7 +59,7 @@ def find_show(title, year=None):


def get_show_id_from_nfo(nfo):
# type: (str) -> None
# type: (Text) -> None
"""
Get show ID by NFO file contents
Expand All @@ -70,7 +70,7 @@ def get_show_id_from_nfo(nfo):
"""
if isinstance(nfo, bytes):
nfo = nfo.decode('utf-8', 'replace')
logger.debug(u'Parsing NFO file:\n{}'.format(nfo))
logger.debug('Parsing NFO file:\n{}'.format(nfo))
parse_result = data_utils.parse_nfo_url(nfo)
if parse_result:
if parse_result.provider == 'tvmaze':
Expand All @@ -93,7 +93,7 @@ def get_show_id_from_nfo(nfo):


def get_details(show_id):
# type: (str) -> None
# type: (Text) -> None
"""Get details about a specific show"""
logger.debug('Getting details for show id {}'.format(show_id))
show_info = tvmaze.load_show_info(show_id)
Expand All @@ -106,7 +106,7 @@ def get_details(show_id):


def get_episode_list(show_id):
# type: (str) -> None
# type: (Text) -> None
logger.debug('Getting episode list for show id {}'.format(show_id))
if not show_id.isdigit():
# Kodi has a bug: when a show directory contains an XML NFO file with
Expand Down Expand Up @@ -144,12 +144,12 @@ def get_episode_list(show_id):


def get_episode_details(encoded_ids):
# type: (str) -> None
# type: (Text) -> None
encoded_ids = urllib_parse.unquote(encoded_ids)
decoded_ids = dict(urllib_parse.parse_qsl(encoded_ids))
logger.debug('Getting episode details for {}'.format(decoded_ids))
episode_info = tvmaze.load_episode_info(
int(decoded_ids['show_id']), int(decoded_ids['episode_id'])
decoded_ids['show_id'], decoded_ids['episode_id']
)
if episode_info:
list_item = xbmcgui.ListItem(episode_info['name'], offscreen=True)
Expand All @@ -160,7 +160,7 @@ def get_episode_details(encoded_ids):


def get_artwork(show_id):
# type: (str) -> None
# type: (Text) -> None
"""
Get available artwork for a show
Expand All @@ -177,12 +177,11 @@ def get_artwork(show_id):


def router(paramstring):
# type: (str) -> None
# type: (Text) -> None
"""
Route addon calls
:param paramstring: url-encoded query string
:type paramstring: str
:raises RuntimeError: on unknown call action
"""
params = dict(urllib_parse.parse_qsl(paramstring))
Expand Down
8 changes: 4 additions & 4 deletions metadata.tvmaze/libs/cache.py
Original file line number Diff line number Diff line change
Expand Up @@ -17,18 +17,18 @@

"""Cache-related functionality"""

from __future__ import absolute_import
from __future__ import absolute_import, unicode_literals
import os
from datetime import datetime, timedelta
from six.moves import cPickle as pickle
from .utils import get_cache_directory, logger

try:
from typing import Optional
from typing import Optional, Text
except ImportError:
pass

CACHE_DIR = get_cache_directory() # type: str
CACHE_DIR = get_cache_directory() # type: Text
CACHING_DURATION = timedelta(hours=3) # type: timedelta


Expand All @@ -47,7 +47,7 @@ def cache_show_info(show_info):


def load_show_info_from_cache(show_id):
# type: (str) -> Optional[dict]
# type: (Text) -> Optional[dict]
"""
Load show info from a local cache
Expand Down
22 changes: 15 additions & 7 deletions metadata.tvmaze/libs/data_utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,7 @@
from .utils import safe_get

try:
from typing import Optional
from typing import Optional, Text
from xbmcgui import ListItem
except ImportError:
pass
Expand Down Expand Up @@ -68,7 +68,7 @@ def process_episode_list(show_info, episode_list):


def _clean_plot(plot):
# type: (str) -> str
# type: (Text) -> Text
"""Replace HTML tags with Kodi skin tags"""
for repl in CLEAN_PLOT_REPLACEMENTS:
plot = plot.replace(repl[0], repl[1])
Expand Down Expand Up @@ -109,6 +109,7 @@ def _get_credits(show_info):


def _set_unique_ids(show_info, list_item):
# type: (dict, ListItem) -> ListItem
"""Extract unique ID in various online databases"""
unique_ids = {'tvmaze': str(show_info['id'])}
for key, value in six.iteritems(safe_get(show_info, 'externals', {})):
Expand All @@ -120,6 +121,7 @@ def _set_unique_ids(show_info, list_item):


def _set_rating(show_info, list_item):
# type: (dict, ListItem) -> ListItem
"""Set show rating"""
if show_info['rating'] is not None and show_info['rating']['average'] is not None:
rating = float(show_info['rating']['average'])
Expand All @@ -128,18 +130,20 @@ def _set_rating(show_info, list_item):


def _extract_artwork_url(resolutions):
# type: (dict) -> Text
"""Extract image URL from the list of available resolutions"""
url = u''
url = ''
for image_size in IMAGE_SIZES:
url = safe_get(resolutions, image_size, u'')
url = safe_get(resolutions, image_size, '')
if not isinstance(url, six.text_type):
url = safe_get(url, 'url', u'')
url = safe_get(url, 'url', '')
if url:
break
return url


def _add_season_info(show_info, list_item):
# type: (dict, ListItem) -> ListItem
"""Add info for show seasons"""
for season in show_info['_embedded']['seasons']:
list_item.addSeason(season['number'], safe_get(season, 'name', ''))
Expand All @@ -152,6 +156,7 @@ def _add_season_info(show_info, list_item):


def set_show_artwork(show_info, list_item):
# type: (dict, ListItem) -> ListItem
"""Set available images for a show"""
fanart_list = []
for item in show_info['_embedded']['images']:
Expand All @@ -167,6 +172,7 @@ def set_show_artwork(show_info, list_item):


def add_main_show_info(list_item, show_info, full_info=True):
# type: (dict, ListItem, bool) -> ListItem
"""Add main show info to a list item"""
plot = _clean_plot(safe_get(show_info, 'summary', ''))
video = {
Expand All @@ -182,15 +188,15 @@ def add_main_show_info(list_item, show_info, full_info=True):
}
if show_info['network'] is not None:
country = show_info['network']['country']
video['studio'] = u'{0} ({1})'.format(show_info['network']['name'], country['code'])
video['studio'] = '{0} ({1})'.format(show_info['network']['name'], country['code'])
video['country'] = country['name']
elif show_info['webChannel'] is not None:
video['studio'] = show_info['webChannel']['name']
# Global Web Channels do not have a country specified
if show_info['webChannel']['country'] is not None:
country = show_info['webChannel']['country']
video['country'] = country['name']
video['studio'] += u' ({})'.format(country['code'])
video['studio'] += ' ({})'.format(country['code'])
if show_info['premiered'] is not None:
video['year'] = int(show_info['premiered'][:4])
video['premiered'] = show_info['premiered']
Expand All @@ -212,6 +218,7 @@ def add_main_show_info(list_item, show_info, full_info=True):


def add_episode_info(list_item, episode_info, full_info=True):
# type: (ListItem, dict, bool) -> ListItem
"""Add episode info to a list item"""
video = {
'title': episode_info['name'],
Expand Down Expand Up @@ -239,6 +246,7 @@ def add_episode_info(list_item, episode_info, full_info=True):


def parse_nfo_url(nfo):
# type: (Text) -> Optional[UrlParseResult]
"""Extract show ID from NFO file contents"""
for regexp in SHOW_ID_REGEXPS:
show_id_match = re.search(regexp, nfo, re.I)
Expand Down
9 changes: 8 additions & 1 deletion metadata.tvmaze/libs/debugger.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,7 @@
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.

from __future__ import absolute_import
from __future__ import absolute_import, unicode_literals
import inspect
from contextlib import contextmanager
from platform import uname
Expand All @@ -24,8 +24,14 @@
import xbmc
from .utils import logger

try:
from typing import Text, Generator, Callable
except ImportError:
pass


def _format_vars(variables):
# type: (dict) -> Text
"""
Format variables dictionary
Expand All @@ -45,6 +51,7 @@ def _format_vars(variables):

@contextmanager
def debug_exception(logger_func=logger.error):
# type: (Callable[[Text], None]) -> Generator[None]
"""
Diagnostic helper context manager
Expand Down
23 changes: 18 additions & 5 deletions metadata.tvmaze/libs/tvmaze.py
Original file line number Diff line number Diff line change
Expand Up @@ -17,24 +17,31 @@

"""Functions to interact with TVmaze API"""

from __future__ import absolute_import
from __future__ import absolute_import, unicode_literals
from pprint import pformat
from requests.exceptions import HTTPError
from . import cache
from .utils import get_requests_session, get_cache_directory, logger, safe_get
from .data_utils import process_episode_list

try:
from typing import Text, Optional, Union, List
import requests
except ImportError:
pass

SEARCH_URL = 'http://api.tvmaze.com/search/shows'
SEARCH_BU_EXTERNAL_ID_URL = 'http://api.tvmaze.com/lookup/shows'
SHOW_INFO_URL = 'http://api.tvmaze.com/shows/{}'
EPISODE_LIST_URL = 'http://api.tvmaze.com/shows/{}/episodes'
EPISODE_INFO_URL = 'http://api.tvmaze.com/episodes/{}'

SESSION = get_requests_session()
CACHE_DIR = get_cache_directory()
SESSION = get_requests_session() # type: requests.Session
CACHE_DIR = get_cache_directory() # type: Text


def _load_info(url, params=None):
# type: (Text, Optional[dict]) -> Union[dict, list]
"""
Load info from TVmaze
Expand All @@ -53,6 +60,7 @@ def _load_info(url, params=None):


def search_show(title):
# type: (Text) -> list
"""
Search a single TV show
Expand All @@ -63,10 +71,11 @@ def search_show(title):
return _load_info(SEARCH_URL, {'q': title})
except HTTPError as exc:
logger.error('TVmaze returned an error: {}'.format(exc))
return ()
return []


def filter_by_year(shows, year):
# type: (List[dict], Text) -> Optional[dict]
"""
Filter a show by year
Expand All @@ -82,16 +91,18 @@ def filter_by_year(shows, year):


def load_episode_list(show_id):
# type: (Text) -> list
"""Load episode list from TVmaze API"""
episode_list_url = EPISODE_LIST_URL.format(show_id)
try:
return _load_info(episode_list_url, {'specials': '1'})
except HTTPError as exc:
logger.error('TVmaze returned an error: {}'.format(exc))
return ()
return []


def load_show_info(show_id):
# type: (Text) -> Optional[dict]
"""
Get full info for a single show
Expand All @@ -117,6 +128,7 @@ def load_show_info(show_id):


def load_show_info_by_external_id(provider, show_id):
# type: (Text, Text) -> Optional[dict]
"""
Load show info by external ID (TheTVDB or IMDB)
Expand All @@ -133,6 +145,7 @@ def load_show_info_by_external_id(provider, show_id):


def load_episode_info(show_id, episode_id):
# type: (Text, Text) -> Optional[dict]
"""
Load episode info
Expand Down

0 comments on commit ce697a9

Please sign in to comment.