Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
Simple bottle.py application to play twitch.tv streams through HTML5 video
# uses bottle.py as web framework, streamlink for getting the stream URL
# and hls.js to play the HLS stream in an HTML5 video tag
from bottle import *
from streamlink import Streamlink
from urllib.request import urlopen, Request
import sys, re, json
def u(s):
if isinstance(s, str):
return s
else:
return s.decode()
# public url of the rewrite route
LOCAL_SERVER_PREFIX = 'http://localhost:8080/twitch-redirect/rewrite/'
# get oauth token from "streamlink --twitch-oauth-authenticate"
TWITCH_OAUTH_TOKEN = '_insert_token_here_'
TWITCH_API_URL = 'https://api.twitch.tv/kraken/streams/followed?limit=100&oauth_token={}'
DEFAULT_QUALITY = '480p'
SANITIZE_RE = re.compile('[^a-zA-Z0-9_]')
TWITCH_SERVER_RE = re.compile(r'^https://[^/]+\.hls\.ttvnw\.net/')
HTML_STREAMS_TEMPLATE = """
<html>
<head>
<title>twitch-redirect</title>
</head>
<body>
<h2>currently online</h2>
%for stream in streams:
<a href="{{stream['name']}}">{{stream['display_name']}}</a>
<br/>
<span>{{stream['status']}}</span>
<br/>
<span>(playing {{stream['game']}} for {{stream['viewers']}} viewers)</span>
<br/><br/>
%end
</body>
</html>
"""
HTML_PLAYER_TEMPLATE = """
<html>
<head>
<title>twitch-redirect {{channel}}</title>
</head>
<body>
%for q in qualities:
<a href="?quality={{q}}">{{q}}</a>
%end
<br><br>
<video id="video" controls></video>
<script src="https://cdn.jsdelivr.net/npm/hls.js@latest"></script>
<script>
if (Hls.isSupported()) {
var url = '{{url}}';
var video = document.getElementById('video');
var hls = new Hls();
hls.loadSource(url);
hls.attachMedia(video);
hls.on(Hls.Events.MANIFEST_PARSED,function() {
video.play();
});
} else {
var newNode = document.createElement('h1');
newNode.appendChild(document.createTextNode('Error: hls.js does not support your browser!'));
document.body.appendChild(newNode);
}
</script>
</body>
</html>
"""
def get_online_streams():
"""return a lazy list of [name, displayname, status, game]
for each online followed stream"""
url = TWITCH_API_FOLLOWED_URL.format(TWITCH_OAUTH_TOKEN)
req = Request(url, headers={'Accept': 'application/vnd.twitchtv.v5+json'})
resp = urlopen(req)
data = json.loads(u(resp.read()))
fields = ('name', 'display_name', 'status', 'game')
for stream in data[u('streams')]:
ch = stream[u('channel')]
item = {u(f): ch[u(f)] for f in fields}
item[u('viewers')] = stream[u('viewers')]
yield item
def get_stream_url(channel, quality):
"""return an HLS url for the given channel and quality, or None"""
sl = Streamlink()
sl.set_plugin_option("twitch", "oauth_token", TWITCH_OAUTH_TOKEN)
# sanitize both params
channel = SANITIZE_RE.sub('', channel)
quality = SANITIZE_RE.sub('', quality)
url = 'twitch.tv/' + channel
streams = sl.streams(url)
if not streams:
return
# fall back to worst quality
stream = streams.get(quality) or streams.get("worst")
if not stream:
return
qualities = streams.keys()
return (stream.url, qualities)
@route('/twitch-redirect')
@route('/')
def redirect_to_index():
# we need trailing slash for relative urls
return redirect('/twitch-redirect/')
@route('/twitch-redirect/')
def index():
streams = get_online_streams()
return template(HTML_STREAMS_TEMPLATE, streams=streams)
@route('/twitch-redirect/rewrite/<url:path>')
def rewrite_stream(url):
# sanity check, only take twitch video server urls
url = 'https://' + url
if not TWITCH_SERVER_RE.match(url):
abort(404, "Invalid url")
resp = urlopen(url)
response.content_type = resp.getheader('Content-Type')
# rewrite playlists to redirect them through this server
if url.endswith('.m3u8'):
data = u(resp.read())
new_data = data.replace('https://', LOCAL_SERVER_PREFIX)
return new_data
else:
return resp
@route('/twitch-redirect/<channel>')
def stream_player(channel):
quality = request.query.quality or DEFAULT_QUALITY
direct = bool(request.query.direct)
url_only = bool(request.query.url)
(url, qualities) = get_stream_url(channel, quality)
if not url:
abort(404, "Unknown channel or quality")
if url_only:
return url
else:
if not direct:
# redirect through this server
url = url.replace('https://', LOCAL_SERVER_PREFIX)
return template(HTML_PLAYER_TEMPLATE, url=url, channel=channel, qualities=qualities)
if __name__ == '__main__':
run(host='localhost', port=8080)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.