PK! 9*9*libtvdb/__init__.py"""libtvdb is a wrapper around the TVDB API (https://api.thetvdb.com/swagger). """ import json from typing import Any, ClassVar, Dict, List, Optional import urllib.parse import deserialize import keyper import requests from libtvdb.exceptions import TVDBException, NotFoundException, TVDBAuthenticationException from libtvdb.model.actor import Actor from libtvdb.model.episode import Episode from libtvdb.model.show import Show from libtvdb.utilities import Log class TVDBClient: """The main client wrapper around the TVDB API. Instantiate a new one of these to use a new authentication session. """ class Constants: """Constants that are used elsewhere in the TVDBClient class.""" AUTH_TIMEOUT: ClassVar[float] = 3 MAX_AUTH_RETRY_COUNT: ClassVar[int] = 3 _BASE_API: ClassVar[str] = "https://api.thetvdb.com" def __init__(self, *, api_key: Optional[str] = None, user_key: Optional[str] = None, user_name: Optional[str] = None) -> None: """Create a new client wrapper. If any of the supplied parameters are None, they will be loaded from the keychain if possible. If not possible, an exception will be thrown. """ if api_key is None: api_key = keyper.get_password(label="libtvdb_api_key") if api_key is None: raise Exception("No API key was supplied or could be found in the keychain") if user_key is None: user_key = keyper.get_password(label="libtvdb_user_key") if user_key is None: raise Exception("No user key was supplied or could be found in the keychain") if user_name is None: user_name = keyper.get_password(label="libtvdb_user_name") if user_name is None: raise Exception("No user name was supplied or could be found in the keychain") self.api_key = api_key self.user_key = user_key self.user_name = user_name self.auth_token = None #pylint: disable=no-self-use def _expand_url(self, path: str) -> str: """Take the path from a URL and expand it to the full API path.""" return f"{TVDBClient._BASE_API}/{path}" #pylint: enable=no-self-use #pylint: disable=no-self-use def _construct_headers(self, *, additional_headers: Optional[Any] = None) -> Dict[str, str]: """Construct the headers used for all requests, inserting any additional headers as required.""" headers = { "Accept": "application/json" } if self.auth_token is not None: headers["Authorization"] = f"Bearer {self.auth_token}" if additional_headers is None: return headers for header_name, header_value in additional_headers.items(): headers[header_name] = header_value return headers #pylint: enable=no-self-use def authenticate(self): """Authenticate the client with the API. This will exit early if we are already authenticated. It does not need to be called. All calls requiring that the client is authenticated will call this. """ if self.auth_token is not None: Log.debug("Already authenticated, skipping") return Log.info("Authenticating...") login_body = { "apikey": self.api_key, "userkey": self.user_key, "username": self.user_name, } for i in range(0, TVDBClient.Constants.MAX_AUTH_RETRY_COUNT): try: response = requests.post( self._expand_url("login"), json=login_body, headers=self._construct_headers(), timeout=TVDBClient.Constants.AUTH_TIMEOUT ) # Since we authenticated successfully, we can break out of the # retry loop break except requests.exceptions.Timeout: will_retry = i < (TVDBClient.Constants.MAX_AUTH_RETRY_COUNT - 1) if will_retry: Log.warning("Authentication timed out, but will retry.") else: Log.error("Authentication timed out maximum number of times.") raise Exception("Authentication timed out maximum number of times.") if response.status_code < 200 or response.status_code >= 300: Log.error(f"Authentication failed withs status code: {response.status_code}") raise TVDBAuthenticationException(f"Authentication failed with status code: {response.status_code}") content = response.json() token = content.get("token") if token is None: Log.error("Failed to get token from login request") raise TVDBAuthenticationException("Failed to get token from login request") self.auth_token = token Log.info("Authenticated successfully") def get(self, url_path: str, *, timeout: float) -> Any: """Search for shows matching the name supplied. If no matching show is found, a NotFoundException will be thrown. """ if url_path is None or url_path == "": raise AttributeError("An invalid URL path was supplied") self.authenticate() Log.info(f"GET: {url_path}") response = requests.get( self._expand_url(url_path), headers=self._construct_headers(), timeout=timeout ) TVDBClient._check_errors(response) content = response.json() data = content.get('data') if data is None: raise NotFoundException(f"Could not get data for path: {url_path}") return data def get_paged(self, url_path: str, *, timeout: float) -> List[Any]: """Get paged data.""" if url_path is None or url_path == "": raise AttributeError("An invalid URL path was supplied") self.authenticate() page = 0 all_results: List[Any] = [] while True: if page != 0: url_path += f"?page={page}" Log.info(f"GET: {url_path}") response = requests.get( self._expand_url(url_path), headers=self._construct_headers(), timeout=timeout ) TVDBClient._check_errors(response) content = response.json() data = content.get('data') if data is None: raise NotFoundException(f"Could not get data for path: {url_path}") all_results += data links = content.get('links') if links is None: break if links.get('next'): Log.debug("Fetching next page") page = links["next"] else: break return all_results def search_show(self, show_name: str, *, timeout: float = 10.0) -> List[Show]: """Search for shows matching the name supplied. If no matching show is found, a NotFoundException will be thrown. """ if show_name is None or show_name == "": return [] encoded_name = urllib.parse.quote(show_name) Log.info(f"Searching for show: {show_name}") shows_data = self.get(f"search/series?name={encoded_name}", timeout=timeout) shows = [] for show_data in shows_data: show = deserialize.deserialize(Show, show_data) shows.append(show) return shows def show_info(self, show_identifier: int, *, timeout: float = 10.0) -> Optional[Show]: """Get the full information for the show with the given identifier.""" Log.info(f"Fetching data for show: {show_identifier}") show_data = self.get(f"series/{show_identifier}", timeout=timeout) return deserialize.deserialize(Show, show_data) def actors_from_show_id(self, show_identifier: int, timeout: float = 10.0) -> List[Actor]: """Get the actors in the given show.""" Log.info(f"Fetching actors for show id: {show_identifier}") actor_data = self.get(f"series/{show_identifier}/actors", timeout=timeout) actors: List[Actor] = [] for actor_data_item in actor_data: actors.append(deserialize.deserialize(Actor, actor_data_item)) return actors def actors_from_show(self, show: Show, timeout: float = 10.0) -> List[Actor]: """Get the actors in the given show.""" return self.actors_from_show_id(show.identifier, timeout=timeout) def episodes_from_show_id(self, show_identifier: int, timeout: float = 10.0) -> List[Episode]: """Get the episodes in the given show.""" Log.info(f"Fetching episodes for show id: {show_identifier}") episode_data = self.get_paged(f"series/{show_identifier}/episodes", timeout=timeout) episodes: List[Episode] = [] for episode_data_item in episode_data: episodes.append(deserialize.deserialize(Episode, episode_data_item)) return episodes def episodes_from_show(self, show: Show, timeout: float = 10.0) -> List[Episode]: """Get the episodes in the given show.""" return self.episodes_from_show_id(show.identifier, timeout=timeout) def episode_by_id(self, episode_identifier: int, timeout: float = 10.0) -> Episode: """Get the episode information from its ID.""" Log.info(f"Fetching info for episode id: {episode_identifier}") episode_data = self.get(f"episodes/{episode_identifier}", timeout=timeout) print(episode_data) return deserialize.deserialize(Episode, episode_data) @staticmethod def _check_errors(response: requests.Response) -> Any: """Check an API response for errors.""" if response.status_code >= 200 and response.status_code < 300: return Log.error(f"Bad response code from API: {response.status_code}") # Try and read the JSON. If we don't have it, we return the generic # exception type try: data = response.json() except json.JSONDecodeError: raise TVDBException(f"Could not decode error response: {response.text}") # Try and get the error message so we can use it error = data.get('Error') # If we don't have it, just return the generic exception type if error is None: raise TVDBException(f"Could not get error information: {response.text}") if error == "Resource not found": raise NotFoundException(f"Could not find resource: {response.url}") else: raise TVDBException(f"Unknown error: {response.text}") PK!!,VVlibtvdb/exceptions.py"""Custom exception types.""" class TVDBException(Exception): """Thrown when we can't get a more specific exception type.""" class NotFoundException(TVDBException): """Thrown when a show is not found after a search.""" pass class TVDBAuthenticationException(TVDBException): """Thrown on authentication error.""" pass PK! ..libtvdb/model/__init__.py"""All the types that are used in the API.""" PK!%libtvdb/model/actor.py"""All the types that are used in the API.""" import datetime from typing import Optional import deserialize from libtvdb.utilities import parse_datetime def datetime_parser(value: Optional[str]) -> Optional[datetime.datetime]: """Parser method for parsing datetimes to pass to deserialize.""" if value is None: return None if value == '0000-00-00 00:00:00': return None return parse_datetime(value) @deserialize.key("identifier", "id") @deserialize.key("series_identifier", "seriesId") @deserialize.key("sort_order", "sortOrder") @deserialize.key("image_author", "imageAuthor") @deserialize.key("image_added", "imageAdded") @deserialize.key("last_updated", "lastUpdated") @deserialize.parser("imageAdded", datetime_parser) @deserialize.parser("lastUpdated", datetime_parser) class Actor: """Represents an actor on a show.""" identifier: int series_identifier: int name: str role: str sort_order: int image: str image_author: int image_added: Optional[datetime.datetime] last_updated: Optional[datetime.datetime] def __str__(self): return f"{self.name} ({self.role}, {self.identifier})" PK!`]{libtvdb/model/enums.py"""All the enums that are used in the API.""" import enum class ShowStatus(enum.Enum): """Represents the status of a show.""" continuing = 'Continuing' ended = 'Ended' unknown = 'Unknown' class AirDay(enum.Enum): """Represents when a show airs.""" monday = 'Monday' tuesday = 'Tuesday' wednesday = 'Wednesday' thursday = 'Thursday' friday = 'Friday' saturday = 'Saturday' sunday = 'Sunday' PK!}y}libtvdb/model/episode.py"""All the types that are used in the API.""" import datetime from typing import List, Optional import deserialize from libtvdb.utilities import parse_date def date_parser(value: Optional[str]) -> Optional[datetime.date]: """Parser method for parsing dates to pass to deserialize.""" if value is None: return None if value in ['', '0000-00-00']: return None return parse_date(value) def timestamp_parser(value: Optional[int]) -> Optional[datetime.datetime]: """Parser method for parsing datetimes to pass to deserialize.""" if value is None: return None return datetime.datetime.fromtimestamp(value) def optional_float(value: Optional[int]) -> Optional[float]: """Parser for optional ints to floats.""" if value is None: return None return float(value) def optional_empty_str(value: Optional[str]) -> Optional[str]: """Parser for empty strs to None.""" if value is None: return None if value == '': return None return value @deserialize.key("absolute_number", "absoluteNumber") @deserialize.key("aired_episode_number", "airedEpisodeNumber") @deserialize.key("aired_season", "airedSeason") @deserialize.key("aired_season_id", "airedSeasonID") @deserialize.key("airs_after_season", "airsAfterSeason") @deserialize.key("airs_before_season", "airsBeforeSeason") @deserialize.key("airs_before_episode", "airsBeforeEpisode") @deserialize.key("dvd_chapter", "dvdChapter") @deserialize.key("dvd_disc_id", "dvdDiscid") @deserialize.key("dvd_episode_number", "dvdEpisodeNumber") @deserialize.key("dvd_season", "dvdSeason") @deserialize.key("episode_name", "episodeName") @deserialize.key("file_name", "filename") @deserialize.key("first_aired", "firstAired") @deserialize.key("guest_stars", "guestStars") @deserialize.key("identifier", "id") @deserialize.key("imdb_id", "imdbId") @deserialize.key("last_updated", "lastUpdated") @deserialize.key("last_updated_by", "lastUpdatedBy") @deserialize.key("production_code", "productionCode") @deserialize.key("series_id", "seriesId") @deserialize.key("show_url", "showUrl") @deserialize.key("site_rating", "siteRating") @deserialize.key("site_rating_count", "siteRatingCount") @deserialize.key("thumb_added", "thumbAdded") @deserialize.key("thumb_author", "thumbAuthor") @deserialize.key("thumb_height", "thumbHeight") @deserialize.key("thumb_width", "thumbWidth") @deserialize.parser("director", optional_empty_str) @deserialize.parser("dvdDiscid", optional_empty_str) @deserialize.parser("dvdEpisodeNumber", optional_float) @deserialize.parser("filename", optional_empty_str) @deserialize.parser("firstAired", date_parser) @deserialize.parser("imdbId", optional_empty_str) @deserialize.parser("lastUpdated", timestamp_parser) @deserialize.parser("productionCode", optional_empty_str) @deserialize.parser("showUrl", optional_empty_str) @deserialize.parser("siteRating", float) @deserialize.parser("thumbAdded", optional_empty_str) class Episode: """Represents an episode of a show.""" @deserialize.key("episode_name", "episodeName") class LanguageCode: """Represents the language that an episode is in.""" episode_name: str overview: str absolute_number: Optional[int] aired_episode_number: int aired_season: int aired_season_id: Optional[int] airs_after_season: Optional[int] airs_before_episode: Optional[int] airs_before_season: Optional[int] director: Optional[str] directors: List[str] dvd_chapter: Optional[int] dvd_disc_id: Optional[str] dvd_episode_number: Optional[float] dvd_season: Optional[int] episode_name: str file_name: Optional[str] first_aired: Optional[datetime.date] guest_stars: List[str] identifier: int imdb_id: Optional[str] language: Optional[LanguageCode] last_updated: datetime.datetime last_updated_by: int overview: Optional[str] production_code: Optional[str] series_id: int show_url: Optional[str] site_rating: float site_rating_count: int thumb_added: Optional[str] thumb_author: int thumb_height: Optional[str] thumb_width: Optional[str] writers: List[str] def __str__(self): return f"{self.identifier})" PK!,   libtvdb/model/show.py"""All the types that are used in the API.""" import datetime from typing import List, Optional import deserialize from libtvdb.model.enums import AirDay, ShowStatus from libtvdb.utilities import parse_date, parse_datetime def date_parser(value: Optional[str]) -> Optional[datetime.date]: """Parser method for parsing dates to pass to deserialize.""" if value is None: return None if value in ['', '0000-00-00']: return None return parse_date(value) def datetime_parser(value: Optional[str]) -> Optional[datetime.datetime]: """Parser method for parsing datetimes to pass to deserialize.""" if value is None: return None if value in ['', '0000-00-00 00:00:00']: return None return parse_datetime(value) def timestamp_parser(value: Optional[int]) -> Optional[datetime.datetime]: """Parser method for parsing datetimes to pass to deserialize.""" if value is None: return None return datetime.datetime.fromtimestamp(value) def status_parser(value: Optional[str]) -> str: """Parser method for cleaning up statuses to pass to deserialize.""" if value is None or value == '': return ShowStatus.unknown.value return value @deserialize.key("identifier", "id") @deserialize.key("name", "seriesName") @deserialize.key("first_aired", "firstAired") @deserialize.key("series_identifier", "seriesId") @deserialize.key("network_identifier", "networkId") @deserialize.key("genres", "genre") @deserialize.key("last_updated", "lastUpdated") @deserialize.key("air_day", "airsDayOfWeek") @deserialize.key("air_time", "airsTime") @deserialize.key("imdb_id", "imdbId") @deserialize.key("zap2it_id", "zap2itId") @deserialize.key("added_by", "addedBy") @deserialize.key("site_rating", "siteRating") @deserialize.key("site_rating_count", "siteRatingCount") @deserialize.parser("status", status_parser) @deserialize.parser("firstAired", date_parser) @deserialize.parser("lastUpdated", timestamp_parser) @deserialize.parser("added", datetime_parser) class Show: """Represents a single show.""" identifier: int name: str slug: str status: ShowStatus first_aired: Optional[datetime.date] aliases: List[str] network: str overview: Optional[str] banner: Optional[str] # These properties are only populated on a specific query (i.e. not a search) series_identifier: Optional[str] network_identifier: Optional[str] runtime: Optional[str] genres: Optional[List[str]] last_updated: Optional[datetime.datetime] air_day: Optional[AirDay] air_time: Optional[str] rating: Optional[str] imdb_id: Optional[str] zap2it_id: Optional[str] added: Optional[datetime.datetime] added_by: Optional[int] site_rating: Optional[float] site_rating_count: Optional[int] PK!L)Hlibtvdb/utilities.py"""Utility classes and methods for working with the TVDB API.""" import datetime def parse_date(input_string: str) -> datetime.date: """Parse a date string from the API in YYYY-MM-DD format into a date object.""" if input_string is None: raise ValueError("The input string should not be none.") if input_string == "": raise ValueError("The input string should not be empty.") components = input_string.split("-") if len(components) != 3: raise ValueError("The input string should be of the format YYYY-MM-DD.") for component in components: try: _ = int(component) except ValueError: raise ValueError("The input string should be of the format YYYY-MM-DD, where each date component is an integer.") year = int(components[0]) month = int(components[1]) day = int(components[2]) return datetime.date(year=year, month=month, day=day) def parse_datetime(input_string: str) -> datetime.datetime: """Parse a datetime string from the API in 'YYYY-MM-DD HH:MM:SS' format into a datetime object.""" if input_string is None: raise ValueError("The input string should not be none.") if input_string == "": raise ValueError("The input string should not be empty.") if input_string == "0000-00-00 00:00:00": raise ValueError("Invalid date time") return datetime.datetime.strptime(input_string, '%Y-%m-%d %H:%M:%S') class Log: """Fake log class that will be used until we implement logging.""" @staticmethod def info(message): """Log an info level log message.""" print("INFO: " + message) @staticmethod def debug(message): """Log a debug level log message.""" print("DEBUG: " + message) @staticmethod def warning(message): """Log a warning level log message.""" print("WARNING: " + message) @staticmethod def error(message): """Log an error level log message.""" print("ERROR: " + message) PK!WI::libtvdb-0.3.0.dist-info/LICENSEMIT License Copyright (c) Dale Myers. All rights reserved. Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions: The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software. THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWAREPK!HnHTUlibtvdb-0.3.0.dist-info/WHEEL A н#Z;/"d&F[xzw@Zpy3Fv]\fi4WZ^EgM_-]#0(q7PK!H{ libtvdb-0.3.0.dist-info/METADATASn@}߯$R R% BDD[&BƞKmﲻNjF(`˙3g^aP {kSFp&kU`U:T_d_LP`kEʙL!dLxg Z$,^oCVeb^r,jB\KOWxڸGNX Lt#!\,k8.qa?:5<$h-vO O'H'%]M-1p$!| )cnft٩_L. o9oƝChG6Er;ѵ]&I2E!(|`H`W5 o]ɂk0V*'/m5*p|4Ĕ>_i:!7恨QkCnϋȍ8C%Tm)Cz2Gjn`M/PK!H<libtvdb-0.3.0.dist-info/RECORDuK@444,[T亱i3eevgVH:͛;?o2LW˧Uc9"*|3nmMZŏNILYU3 O'3|N$Fg_R߾M FXnPcR9pO@ o^uQF~+C-uw+(xIAƥj/+r4քaX\xFlzLݠ" dߵt_NjWq&7+w8V!Ȭh^O$T$I(,7E[-.ʄʮ/0CyjM" ?D1{IIz2k\%fYp̪ a4،szIH?L=]WCZc/ّmMn 0E6qMuYߏTOպUx&eft]{Kg1R&Vt,۹t^ÒBqϵ.ZcZg٢:4-|@uXR/PK! 9*9*libtvdb/__init__.pyPK!!,VVj*libtvdb/exceptions.pyPK! ..+libtvdb/model/__init__.pyPK!%X,libtvdb/model/actor.pyPK!`]{(1libtvdb/model/enums.pyPK!}y}3libtvdb/model/episode.pyPK!,   Dlibtvdb/model/show.pyPK!L)HOOlibtvdb/utilities.pyPK!WI::~Wlibtvdb-0.3.0.dist-info/LICENSEPK!HnHTU[libtvdb-0.3.0.dist-info/WHEELPK!H{ \libtvdb-0.3.0.dist-info/METADATAPK!H<_libtvdb-0.3.0.dist-info/RECORDPK Pb