matteo
/
feedleware
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Add youtube endpoint

This commit is contained in:
Mattéo Delabre 2021-09-13 00:00:24 +02:00
parent 33d10c20a1
commit 527077fab4
Signed by: matteo
GPG Key ID: AE3FBD02DC583ABB
8 changed files with 260 additions and 40 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
feedleware/__init__.py
View File

@ -3,12 +3,13 @@ import logging
import sys
from os import environ
from flask import Flask
from . import twitch
from . import twitch, youtube
logger = logging.getLogger(__name__)
blueprints = {
"twitch": twitch,
"youtube": youtube,
}

2
feedleware/twitch/__init__.py
View File

@ -9,7 +9,7 @@ def create_blueprint(config):
twitch = Blueprint("twitch", __name__)
@twitch.route("/<string:login>", methods=["GET", "HEAD"])
def vod(login: str):
def get(login: str):
try:
return (
construct_rss(client, login),

10
feedleware/twitch/feed.py
View File

@ -1,12 +1,8 @@
from datetime import datetime
from ..feedformatter import Feed
from ..util import parse_iso
from .twitch import APIClient
def parse_iso(iso: str) -> datetime:
return datetime.fromisoformat(iso.replace("Z", "+00:00"))
def construct_rss(client: APIClient, login: str) -> str:
"""
Build a RSS stream for a Twitch user.
@ -28,9 +24,9 @@ def construct_rss(client: APIClient, login: str) -> str:
# Set the feed/channel level properties
feed.feed["title"] = user_info["display_name"]
feed.feed["link"] = user_url
feed.feed["author"] = "Twitch RSS Generated"
feed.feed["author"] = "Feedleware"
feed.feed["description"] = user_info["description"]
feed.feed["ttl"] = '10'
feed.feed["ttl"] = "10"
if stream is not None:
item = {}

43
feedleware/twitch/twitch.py
View File

@ -8,6 +8,7 @@ import urllib
import urllib.request
from typing import Any, Dict, Optional
from cachetools import cached, TTLCache
from ..util import send_with_retry
HTTPRequest = urllib.request.Request
@ -43,40 +44,17 @@ class APIClient:
"""
self.client_id: str = client_id
self.secret: str = secret
self.retries: int = 3
self.retries = retries
self.oauth_token: str = ""
self.oauth_expire_epoch: int = 0
def _send_with_retry(self, request: HTTPRequest) -> HTTPResponse:
"""
Send an HTTP request and retry in case of failure.
The number of retries is configured by `self.retries`.
:param request: request to try sending
:returns: response sent by the server
:throws HTTPException: if the number of retries is exceeded
"""
retries = self.retries
last_err = HTTPException()
while retries:
try:
return urllib.request.urlopen(request, timeout=3)
except HTTPException as err:
logger.warning("HTTP error: %s", err)
retries -= 1
last_err = err
raise last_err
def authorize(self) -> bool:
"""
Refresh the current OAuth app access token if needed.
This uses the Twitch OAuth client credentials flow. See:
This uses the Twitch OAuth client credentials flow. See
<https://dev.twitch.tv/docs/authentication/getting-tokens-oauth/#oauth-client-credentials-flow>
for details.
:returns: true if the OAuth app access token was refreshed,
false if the existing one was still valid
@ -99,7 +77,7 @@ class APIClient:
method="POST",
)
http_response = self._send_with_retry(request)
http_response = send_with_retry(request, self.retries)
response = json.loads(http_response.read().decode("utf-8"))
self.oauth_token = response["access_token"]
@ -138,7 +116,7 @@ class APIClient:
method=method,
)
http_response = self._send_with_retry(request)
http_response = send_with_retry(request, self.retries)
if http_response.info().get("Content-Encoding") == "gzip":
return json.loads(gzip.decompress(http_response.read()))
@ -150,6 +128,8 @@ class APIClient:
"""
Get information about a user.
See <https://dev.twitch.tv/docs/api/reference#get-users> for details.
:param login: user login
:returns: user information
:throws HTTPException: if the query fails
@ -163,13 +143,10 @@ class APIClient:
method="GET",
data={"login": login},
)
assert type(response) == dict
if "data" not in response or not response["data"]:
raise NoSuchUser(f"User '{login}' does not exist")
assert type(response["data"]) == list
assert type(response["data"][0]) == dict
return response["data"][0]
@cached(cache=TTLCache(maxsize=1000, ttl=10 * 60))
@ -177,6 +154,8 @@ class APIClient:
"""
Get the list of videos from a channel.
See <https://dev.twitch.tv/docs/api/reference#get-videos> for details.
:param channel_id: channel ID
:returns: list of videos
:throws HTTPException: if the query fails
@ -198,6 +177,8 @@ class APIClient:
"""
Get the information about the stream currently active on a channel.
See <https://dev.twitch.tv/docs/api/reference#get-streams> for details.
:param channel_id: channel ID
:returns: stream information or None
:throws HTTPException: if the query fails

33
feedleware/util.py Normal file
View File

@ -0,0 +1,33 @@
from datetime import datetime
import http
import urllib
import urllib.request
HTTPRequest = urllib.request.Request
HTTPResponse = http.client.HTTPResponse
HTTPException = http.client.HTTPException
def parse_iso(iso: str) -> datetime:
return datetime.fromisoformat(iso.replace("Z", "+00:00"))
def send_with_retry(request: HTTPRequest, retries: int) -> HTTPResponse:
"""
Send an HTTP request and retry in case of failure.
:param request: request to try sending
:param retries: number of retries
:returns: response sent by the server
:throws HTTPException: if the number of retries is exceeded
"""
last_err = HTTPException()
while retries:
try:
return urllib.request.urlopen(request, timeout=3)
except HTTPException as err:
retries -= 1
last_err = err
raise last_err

21
feedleware/youtube/__init__.py Normal file
View File

@ -0,0 +1,21 @@
from flask import abort, Blueprint
from .youtube import APIClient, NoSuchChannel
from .feed import construct_rss
def create_blueprint(config):
"""Create a YouTube endpoint blueprint."""
client = APIClient(config["key"])
twitch = Blueprint("youtube", __name__)
@twitch.route("/<string:login>", methods=["GET", "HEAD"])
def vod(login: str):
try:
return (
construct_rss(client, login),
{"Content-Type": "application/rss+xml"},
)
except NoSuchChannel:
abort(404)
return twitch

47
feedleware/youtube/feed.py Normal file
View File

@ -0,0 +1,47 @@
from ..feedformatter import Feed
from ..util import parse_iso
from .youtube import APIClient
def construct_rss(client: APIClient, channel_id: str) -> str:
"""
Build a RSS stream for a YouTube channel.
:param client: YouTube API client
:param channel_id: channel ID
:returns: RSS stream
:raises HTTPException: if one of the requests fail
:raises NoSuchChannel: if the channel does not exist
"""
channel_info = client.channel(channel_id)
videos = client.playlist(channel_info["playlist"])
feed = Feed()
channel_url = f"https://www.youtube.com/channel/{channel_id}"
# Set the feed/channel level properties
feed.feed["title"] = channel_info["title"]
feed.feed["link"] = channel_url
feed.feed["author"] = "Feedleware"
feed.feed["description"] = channel_info["description"]
feed.feed["ttl"] = "30"
for video in videos:
item = {}
video_id = video["resourceId"]["videoId"]
link = f"https://www.youtube.com/watch?v={video_id}"
thumbnail = video["thumbnails"]["standard"]["url"]
item["guid"] = video["resourceId"]["videoId"]
item["title"] = video.get("title", "Untitled Video")
item["link"] = link
item["description"] = (
f'<a href="{link}"><img src="{thumbnail}" /></a><br><br>'
+ video["description"]
)
item["pubDate"] = parse_iso(video["publishedAt"]).timetuple()
feed.items.append(item)
return feed.format_rss2_string()

141
feedleware/youtube/youtube.py Normal file
View File

@ -0,0 +1,141 @@
import gzip
import http
import json
import logging
import urllib
import urllib.request
from typing import Any, Iterable, Optional, Tuple
from cachetools import cached, TTLCache
from ..util import send_with_retry
HTTPRequest = urllib.request.Request
HTTPResponse = http.client.HTTPResponse
HTTPException = http.client.HTTPException
logger = logging.getLogger(__name__)
class NoSuchChannel(Exception):
"""Raised when an unknown channel is queried."""
class APIClient:
"""Client for the YouTube Data API."""
def __init__(self, key: str = "", retries: int = 3):
"""
Create a YouTube Data API client.
See <https://developers.google.com/youtube/v3/docs> for details.
:param key: YouTube API key
:param retries: number of times to retry each request in case of failure
"""
self.key = key
self.retries = retries
def _query(
self,
url: str,
method: str = "GET",
data: Iterable[Tuple[str, str]] = []
) -> Any:
"""
Low-level method to query the API.
:param url: URL to query
:param method: HTTP method to use
:param data: payload dictionary to send
:returns: JSON data
:throws HTTPException: if the query fails
"""
logger.debug("Querying %s %s %s", method, url, data)
headers = {
"Accept": "application/json",
"Accept-Encoding": "gzip",
}
payload = (
*data,
("key", self.key),
)
request = HTTPRequest(
url=f"{url}?{urllib.parse.urlencode(payload)}",
headers=headers,
method=method,
)
http_response = send_with_retry(request, self.retries)
if http_response.info().get("Content-Encoding") == "gzip":
return json.loads(gzip.decompress(http_response.read()))
else:
return json.loads(http_response.read())
@cached(cache=TTLCache(maxsize=1000, ttl=24 * 60 * 60))
def channel(self, channel_id: str) -> Any:
"""
Get information about a channel.
See <https://developers.google.com/youtube/v3/docs/channels>
for details.
:param channel_id: channel ID
:returns: channel information
:throws HTTPException: if the query fails
:throws NoSuchChannel: if the channel doesn’t exist
"""
response = self._query(
url="https://youtube.googleapis.com/youtube/v3/channels",
method="GET",
data=(
("part", "id"),
("part", "snippet"),
("part", "contentDetails"),
("id", channel_id),
("maxResults", 1),
)
)
if response["pageInfo"]["totalResults"] == 0:
raise NoSuchChannel(f"Channel '{channel_id}' does not exist")
data = response["items"][0]
return {
"id": data["id"],
"playlist": data["contentDetails"]["relatedPlaylists"]["uploads"],
**response["items"][0]["snippet"],
}
@cached(cache=TTLCache(maxsize=1000, ttl=30 * 60))
def playlist(self, playlist_id: str) -> Any:
"""
Get the latest videos from a playlist.
See <https://developers.google.com/youtube/v3/docs/playlistItems>
for details.
:param playlist_id: channel ID
:returns: list of latest videos
:throws HTTPException: if the query fails
"""
response = self._query(
url="https://youtube.googleapis.com/youtube/v3/playlistItems",
method="GET",
data=(
("part", "snippet"),
("part", "status"),
("playlistId", playlist_id),
("maxResults", 50),
)
)
return [
item["snippet"]
for item in response["items"]
if item["status"]["privacyStatus"] == "public"
and item["snippet"]["resourceId"]["kind"] == "youtube#video"
]