Merge pull request #146 from TotallyNotRobots/gonzobot+expand-yt-errors

Expand youtube.py error info

Author: linuxdaemon

CHANGELOG.md

@@ -13,6 +13,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Change link_announcer.py to only warn on connection errors
 - Change user lookup logic in last.fm plugin
 - Refactor minecraft_ping plugin for updated mcstatus library
+- Expand youtube.py error information
 ### Fixed
 - Fix matching exception in horoscope test
 - Fix youtube.py ISO time parse

plugins/youtube.py

@@ -1,79 +1,147 @@
 import re
+from typing import Iterable, Mapping, Match, Optional, Union
 
 import isodate
 import requests
 
 from cloudbot import hook
 from cloudbot.bot import bot
 from cloudbot.util import colors, timeformat
-from cloudbot.util.formatting import pluralize_auto
+from cloudbot.util.formatting import pluralize_suffix
+
+youtube_re = re.compile(
+    r'(?:youtube.*?(?:v=|/v/)|youtu\.be/|yooouuutuuube.*?id=)([-_a-zA-Z0-9]+)', re.I
+)
+ytpl_re = re.compile(
+    r'(.*:)//(www.youtube.com/playlist|youtube.com/playlist)(:[0-9]+)?(.*)', re.I
+)
 
-youtube_re = re.compile(r'(?:youtube.*?(?:v=|/v/)|youtu\.be/|yooouuutuuube.*?id=)([-_a-zA-Z0-9]+)', re.I)
 
 base_url = 'https://www.googleapis.com/youtube/v3/'
-api_url = base_url + 'videos?part=contentDetails%2C+snippet%2C+statistics&id={}&key={}'
-search_api_url = base_url + 'search?part=id&maxResults=1'
-playlist_api_url = base_url + 'playlists?part=snippet%2CcontentDetails%2Cstatus'
-video_url = "http://youtu.be/%s"
-err_no_api = "The YouTube API is off in the Google Developers Console."
 
 
 class APIError(Exception):
-    def __init__(self, message, response=None):
+    def __init__(self, message: str, response: Optional[str] = None) -> None:
         super().__init__(message)
         self.message = message
         self.response = response
 
 
-def get_video_description(video_id):
-    dev_key = bot.config.get_api_key("google_dev_key")
-    request = requests.get(api_url.format(video_id, dev_key))
-    json = request.json()
+class NoApiKeyError(APIError):
+    def __init__(self) -> None:
+        super().__init__("Missing API key")
+
+
+class NoResultsError(APIError):
+    def __init__(self) -> None:
+        super().__init__("No results")
+
+
+def raise_api_errors(response: requests.Response) -> None:
+    try:
+        response.raise_for_status()
+    except requests.RequestException as e:
+        try:
+            data = response.json()
+        except ValueError:
+            raise e
+
+        errors = data.get('errors')
+        if not errors:
+            return
+
+        first_error = errors[0]
+        domain = first_error['domain']
+        reason = first_error['reason']
+        raise APIError("API Error ({}/{})".format(domain, reason), data) from e
+
+
+def make_short_url(video_id: str) -> str:
+    return "http://youtu.be/{}".format(video_id)
+
+
+ParamValues = Union[int, str]
+ParamMap = Mapping[str, ParamValues]
+Parts = Iterable[str]
+
+
+def do_request(
+    method: str, parts: Parts, params: Optional[ParamMap] = None, **kwargs: ParamValues
+) -> requests.Response:
+    api_key = bot.config.get_api_key("google_dev_key")
+    if not api_key:
+        raise NoApiKeyError()
+
+    if params:
+        kwargs.update(params)
+
+    kwargs['part'] = ','.join(parts)
+    kwargs['key'] = api_key
+    return requests.get(base_url + method, kwargs)
+
+
+def get_video(video_id: str, parts: Parts) -> requests.Response:
+    return do_request('videos', parts, params={'maxResults': 1, 'id': video_id})
+
+
+def get_playlist(playlist_id: str, parts: Parts) -> requests.Response:
+    return do_request('playlists', parts, params={'maxResults': 1, 'id': playlist_id})
+
 
-    if json.get('error'):
-        if json['error']['code'] == 403:
-            raise APIError(err_no_api, json)
+def do_search(term: str, result_type: str = 'video') -> requests.Response:
+    return do_request(
+        'search', ['snippet'], params={'maxResults': 1, 'q': term, 'type': result_type}
+    )
 
-        raise APIError("Unknown error", json)
+
+def get_video_description(video_id: str) -> str:
+    parts = ['statistics', 'contentDetails', 'snippet']
+    request = get_video(video_id, parts)
+    raise_api_errors(request)
+
+    json = request.json()
 
     data = json['items']
     if not data:
-        return None
+        raise NoResultsError()
 
-    snippet = data[0]['snippet']
-    statistics = data[0]['statistics']
-    content_details = data[0]['contentDetails']
+    item = data[0]
+    snippet = item['snippet']
+    statistics = item['statistics']
+    content_details = item['contentDetails']
 
     out = '\x02{}\x02'.format(snippet['title'])
 
     if not content_details.get('duration'):
         return out
 
     length = isodate.parse_duration(content_details['duration'])
-    out += ' - length \x02{}\x02'.format(timeformat.format_time(int(length.total_seconds()), simple=True))
+    out += ' - length \x02{}\x02'.format(
+        timeformat.format_time(int(length.total_seconds()), simple=True)
+    )
     try:
         total_votes = float(statistics['likeCount']) + float(statistics['dislikeCount'])
     except (LookupError, ValueError):
         total_votes = 0
 
     if total_votes != 0:
         # format
-        likes = pluralize_auto(int(statistics['likeCount']), "like")
-        dislikes = pluralize_auto(int(statistics['dislikeCount']), "dislike")
+        likes = pluralize_suffix(int(statistics['likeCount']), "like")
+        dislikes = pluralize_suffix(int(statistics['dislikeCount']), "dislike")
 
         percent = 100 * float(statistics['likeCount']) / total_votes
-        out += ' - {}, {} (\x02{:.1f}\x02%)'.format(likes,
-                                                    dislikes, percent)
+        out += ' - {}, {} (\x02{:.1f}\x02%)'.format(likes, dislikes, percent)
 
     if 'viewCount' in statistics:
         views = int(statistics['viewCount'])
-        out += ' - \x02{:,}\x02 view{}'.format(views, "s"[views == 1:])
+        out += ' - \x02{:,}\x02 view{}'.format(views, "s"[views == 1 :])
 
     uploader = snippet['channelTitle']
 
     upload_time = isodate.parse_datetime(snippet['publishedAt'])
-    out += ' - \x02{}\x02 on \x02{}\x02'.format(uploader,
-                                                upload_time.strftime("%Y.%m.%d"))
+    out += ' - \x02{}\x02 on \x02{}\x02'.format(
+        uploader, upload_time.strftime("%Y.%m.%d")
+    )
 
     try:
         yt_rating = content_details['contentRating']['ytRating']
@@ -86,119 +154,100 @@ def get_video_description(video_id):
     return out
 
 
-def get_video_id(reply, text):
-    dev_key = bot.config.get_api_key('google_dev_key')
-    if not dev_key:
-        return None, "This command requires a Google Developers Console API key."
-
+def get_video_id(text: str) -> str:
     try:
-        request = requests.get(search_api_url, params={'q': text, 'key': dev_key, 'type': 'video'})
-        request.raise_for_status()
-    except Exception:
-        reply("Error performing search.")
-        raise
+        request = do_search(text)
+    except requests.RequestException as e:
+        raise APIError("Unable to connect to API") from e
 
+    raise_api_errors(request)
     json = request.json()
 
-    if json.get('error'):
-        if json['error']['code'] == 403:
-            return None, err_no_api
-
-        return None, "Error performing search."
-
     if not json.get('items'):
-        return None, "No results found."
+        raise NoResultsError()
 
-    video_id = json['items'][0]['id']['videoId']
-    return video_id, None
+    video_id = json['items'][0]['id']['videoId']  # type: str
+    return video_id
 
 
 @hook.regex(youtube_re)
-def youtube_url(match):
+def youtube_url(match: Match[str]) -> str:
     return get_video_description(match.group(1))
 
 
 @hook.command("youtube", "you", "yt", "y")
-def youtube(text, reply):
+def youtube(text: str, reply) -> str:
     """<query> - Returns the first YouTube search result for <query>."""
-    video_id, err = get_video_id(reply, text)
-    if err:
-        return err
-
     try:
-        return get_video_description(video_id) + " - " + video_url % video_id
+        video_id = get_video_id(text)
+        return get_video_description(video_id) + " - " + make_short_url(video_id)
+    except NoResultsError as e:
+        return e.message
     except APIError as e:
         reply(e.message)
         raise
 
 
 @hook.command("youtime", "ytime")
-def youtime(text, reply):
+def youtime(text: str, reply) -> str:
     """<query> - Gets the total run time of the first YouTube search result for <query>."""
-    video_id, err = get_video_id(reply, text)
-    if err:
-        return err
-
-    dev_key = bot.config.get_api_key('google_dev_key')
-    request = requests.get(api_url.format(video_id, dev_key))
-    request.raise_for_status()
+    parts = ['statistics', 'contentDetails', 'snippet']
+    try:
+        video_id = get_video_id(text)
+        request = get_video(video_id, parts)
+        raise_api_errors(request)
+    except NoResultsError as e:
+        return e.message
+    except APIError as e:
+        reply(e.message)
+        raise
 
     json = request.json()
 
-    if json.get('error'):
-        return
     data = json['items']
-    snippet = data[0]['snippet']
-    content_details = data[0]['contentDetails']
-    statistics = data[0]['statistics']
+    item = data[0]
+    snippet = item['snippet']
+    content_details = item['contentDetails']
+    statistics = item['statistics']
 
-    if not content_details.get('duration'):
-        return
+    duration = content_details.get('duration')
+    if not duration:
+        return "Missing duration in API response"
 
-    length = isodate.parse_duration(content_details['duration'])
+    length = isodate.parse_duration(duration)
     l_sec = int(length.total_seconds())
     views = int(statistics['viewCount'])
     total = int(l_sec * views)
 
     length_text = timeformat.format_time(l_sec, simple=True)
     total_text = timeformat.format_time(total, accuracy=8)
 
-    return 'The video \x02{}\x02 has a length of {} and has been viewed {:,} times for ' \
-           'a total run time of {}!'.format(snippet['title'], length_text, views,
-                                            total_text)
-
-
-ytpl_re = re.compile(r'(.*:)//(www.youtube.com/playlist|youtube.com/playlist)(:[0-9]+)?(.*)', re.I)
+    return (
+        'The video \x02{}\x02 has a length of {} and has been viewed {:,} times for '
+        'a total run time of {}!'.format(
+            snippet['title'], length_text, views, total_text
+        )
+    )
 
 
 @hook.regex(ytpl_re)
-def ytplaylist_url(match, reply):
+def ytplaylist_url(match: Match[str]) -> str:
     location = match.group(4).split("=")[-1]
-    dev_key = bot.config.get_api_key("google_dev_key")
-    try:
-        request = requests.get(playlist_api_url, params={"id": location, "key": dev_key})
-        request.raise_for_status()
-    except Exception:
-        reply("Error looking up playlist.")
-        raise
+    request = get_playlist(location, ['contentDetails', 'snippet'])
+    raise_api_errors(request)
 
     json = request.json()
 
-    if json.get('error'):
-        if json['error']['code'] == 403:
-            return err_no_api
-
-        return 'Error looking up playlist.'
-
     data = json['items']
     if not data:
-        return "No results found."
+        raise NoResultsError()
 
-    snippet = data[0]['snippet']
-    content_details = data[0]['contentDetails']
+    item = data[0]
+    snippet = item['snippet']
+    content_details = item['contentDetails']
 
     title = snippet['title']
     author = snippet['channelTitle']
     num_videos = int(content_details['itemCount'])
-    count_videos = ' - \x02{:,}\x02 video{}'.format(num_videos, "s"[num_videos == 1:])
+    count_videos = ' - \x02{:,}\x02 video{}'.format(num_videos, "s"[num_videos == 1 :])
     return "\x02{}\x02 {} - \x02{}\x02".format(title, count_videos, author)