rumble.py

#!/usr/bin/python3

"""Module for handling Rumble channels."""

import logging
import datetime
import requests
import dateutil

from feedgen.feed import FeedGenerator
from bs4 import BeautifulSoup
from tornado import web

headers = {
    'User-Agent': 'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/94.0.4606.104 Safari/537.36'
}

class ChannelHandler(web.RequestHandler):

    def head(self, channel):
        self.set_header('Content-type', 'application/rss+xml')
        self.set_header('Accept-Ranges', 'bytes')

    def get(self, channel):
        logging.debug( "Got channel: %s", channel )

        url = "https://rumble.com/c/%s" % channel
        logging.info( "Rumble: Handling URL: %s", url )

        self.set_header('Content-type', 'application/rss+xml')
        feed = self.generate_rss( channel )
        self.write( feed )
        self.finish()

    def get_html( self, channel ):
        url = "https://rumble.com/c/%s" % channel
        r = requests.get( url, headers=headers )
        bs = BeautifulSoup( r.text, "lxml" )
        html = str( bs.find("main") )
        return html

    def generate_rss( self, channel ):
        logging.info("Rumble: Generating RSS for Channel: %s", channel)
        bs = BeautifulSoup( self.get_html( channel ), "lxml" )

        feed = FeedGenerator()
        feed.load_extension('podcast')

        ## Get Channel Info
        channel_title = bs.find("div", "channel-header--title")
        if channel_title:
            channel_title = channel_title.find("h1").text
            if channel_title:
                feed.title( channel_title )
            else:
                logging.error("Rumble: Failed to pull channel title. Using provided channel instead")
                feed.title( channel )
        else:
            logging.error("Rumble: Failed to pull channel title. Using provided channel instead")
            feed.title( channel )

        thumb = bs.find("img", "channel-header--img")
        if thumb:
            feed.image( thumb['src'] )
        else:
            logging.error("Rumble: Channel thumbnail not found")

        feed.description( "--" )
        feed.id( channel )
        feed.link(
            href = f'https://rumble.com/c/{channel}',
            rel = 'self'
        )
        feed.language('en')

        ## Assemble RSS items list
        videos = bs.find("ol", "thumbnail__grid")
        if videos:
            videos = videos.find_all("div", "videostream")
        else:
            logging.info("Failed to find video list")

        if videos:
            for video in videos:
                ## Check for and skip live videos and upcomming videos
                if video.find("span", "video-item--live") \
                    or video.find("span", "video-item--upcoming") \
                    or video.find("div", "videostream__status--live") \
                    or video.find("div", "videostream__footer--live"):
                    vidTitle = video.find("h3", "thumbnail__title")
                    logging.info("Rumble: Skipping live/upcoming video: %s", vidTitle.text.strip())
                    continue

                ## Check for and skip premium videos
                el = video.find("span", "text-link-green")
                if el and el.text == "Premium only":
                    vidTitle = video.find("h3", "thumbnail__title")
                    logging.info("Rumble: Skipping premium video: %s", vidTitle.text.strip())

                ## Gather channel information
                item = feed.add_entry()

                ## Gather video information
                vidtitle = video.find("h3", "thumbnail__title")
                if vidtitle:
                    item.title( vidtitle.text.strip() )
                else:
                    logging.info("Rumble: Failed to pull video thumbnail.")

                viddesc = video.find("div", "videostream__description")
                if viddesc:
                    item.description( viddesc.text.strip() )
                # else:
                #     logging.info("Failed to pull video description.")

                vid = video.find("a", "videostream__link")
                if vid:
                    vid = vid['href']
                else:
                    logging.info("Rumble: Failed to pull URL to video.")

                link = f'http://{self.request.host}/rumble/video' + vid
                item.link(
                    href = link,
                    title = item.title()
                )

                vidduration = video.find('div', 'videostream__status--duration')
                if vidduration:
                    item.podcast.itunes_duration( vidduration.text.strip() )
                else:
                    logging.info("Rumble: Failed to pull video duration.")

                viddatetime = video.find("time", "videostream__time")
                if viddatetime:
                    date = dateutil.parser.parse( viddatetime['datetime'] )
                    item.pubDate( date )
                else:
                    logging.info("Rumble: Failed to pull video date.")

                item.enclosure(
                    url = link,
                    type = "video/mp4"
                )
        return feed.rss_str( pretty=True )

class UserHandler(web.RequestHandler):
    def head(self, user):
        self.set_header('Content-type', 'application/rss+xml')
        self.set_header('Accept-Ranges', 'bytes')

    def get(self, user):
        logging.debug( "Got user: %s", user )

        url = f"https://rumble.com/user/{user}"
        logging.info( "Handling Rumble URL: %s", url )

        self.set_header('Content-type', 'application/rss+xml')
        feed = self.generate_rss( user )
        self.write( feed )
        self.finish()

    def get_html( self, user ):
        url = f"https://rumble.com/user/{user}"
        logging.info("Rumble URL: %s", url)
        r = requests.get( url, headers=headers )
        if r.status_code == 404:
            logging.error( "Rumble returned 404: Not found" )
            return None
        bs = BeautifulSoup( r.text, 'lxml' )
        html = str( bs.find("main") )
        return html

    def generate_rss( self, user ):
        logging.debug("User: %s", user)
        html = self.get_html( user )
        if html is None:
            logging.error("Rumble returned 404: Not found")
            self.set_status( 404 )
            return None
        bs = BeautifulSoup( html, 'lxml' )

        feed = FeedGenerator()
        feed.load_extension('podcast')

        ## Get User/Channel Info
        chantitle = bs.find("div", "channel-header--title")
        if chantitle:
            feed.title( chantitle.find("h1").text )
        else:
            logging.info("Failed to pull user channel title.")
            feed.title( user )

        chanheader = bs.find("img", "channel-header--img")
        if chanheader:
            feed.image( chanheader['src'] )
        else:
            logging.info("Failed to pull user channel thumbnail.")

        feed.description( "--" )
        feed.id( user )
        feed.link(
            href = f'https://rumble.com/user/{user}',
            rel = 'self'
        )
        feed.language('en')

        ## Assemble RSS items list
        videos = bs.find("ol", "thumbnail__grid")
        if videos:
            videos = videos.find_all("div", "videostream")
        else:
            logging.info("Failed to find video list")

        if videos:
            for video in videos:
                ## Filter out live and upcoming videos
                if video.find("div", "thumbnail__thumb--upcoming"):
                    logging.info("Found upcoming video, skipping")
                    continue
                if video.find("div", "thumbnail__thumb--live"):
                    logging.info("Found live video, skipping")
                    continue

                item = feed.add_entry()

                ## Gather channel information
                vidtitle = video.find("h3", "thumbnail__title")
                if vidtitle:
                    item.title( vidtitle.text.strip() )
                else:
                    logging.info("Failed to get video title")
                    item.title( 'N/A' )

                viddescription = video.find("div", "videostream__description")
                if viddescription:
                    item.description( viddescription.text )
                else:
                    logging.info("Failed to get video description")
                    item.description( 'N/A' )

                lnk = video.find("a", "videostream__link")
                if lnk:
                    vid = lnk['href']
                    link = f'http://{self.request.host}/rumble/video' + vid
                    icon = video.find("img", "thumbnail__image")['src']
                    item.podcast.itunes_image( icon )
                    item.link(
                        href = link,
                        title = item.title()
                    )
                else:
                    item.link(
                        href = "N/A",
                        title = item.title()
                    )

                vidduration = video.find('div', 'videostream__status--duration')
                if vidduration:
                    item.podcast.itunes_duration( vidduration.text.strip() )
                else:
                    logging.info("Failed to get video duration")

                viddatetime = video.find("time", "videostream__time")
                if viddatetime:
                    viddatetime = viddatetime['datetime']
                    date = dateutil.parser.parse( viddatetime )
                    item.pubDate( date )
                else:
                    logging.info("Failed to get video date")
                    item.pubDate( datetime.datetime.now() )
                item.enclosure(
                    url = link,
                    type = "video/mp4"
                )
        else:
            logging.error("Failed to find video list")
        return feed.rss_str( pretty=True )

class CategoryHandler(web.RequestHandler):
    def head(self, category):
        self.set_header('Content-type', 'application/rss+xml')
        self.set_header('Accept-Ranges', 'bytes')

    def get(self, category):
        logging.info( "Got category: %s", category )

        url = f"https://rumble.com/category/{category}"
        logging.info( "Handling Rumble URL: %s", url )

        self.set_header('Content-type', 'application/rss+xml')
        feed = self.generate_rss( category )
        self.write( feed )
        self.finish()

    def get_html(self, category):
        url = f"https://rumble.com/category/{category}/recorded"
        logging.info("Rumble URL: %s", url)

        r = requests.get( url, headers=headers )
        bs = BeautifulSoup( r.text, 'lxml' )
        html = str( bs.find("main") )
        return html

    def generate_rss( self, category ):
        logging.info( "Category: %s", category )
        bs = BeautifulSoup( self.get_html( category ), 'lxml' )

        feed = FeedGenerator()
        feed.load_extension('podcast')

        ## Get Channel Info
        feedtitle = bs.find("h1", "header__heading")
        if feedtitle:
            feed.title( f"Rumble: {feedtitle.text.strip()}" )
        else:
            logging.info( "Failed to pull category name" )
            feed.title( category )

        feed.description( f"New videos from Rumble's {category} category page" )

        feed.id( category )
        feed.link(
            href = f'https://rumble.com/category/{category}',
            rel = 'self'
        )
        feed.language('en')

        ## Assemble RSS items list
        videos = bs.find("ol", "thumbnail__grid")
        if videos:
            videos = videos.find_all("div", "videostream")
        else:
            logging.error("Failed to find video list")
            return

        videos = bs.find("ol", "thumbnail__grid")
        if videos:
            videos = videos.find_all("div", "videostream")
        else:
            logging.error("Failed to find video list")
            return

        if videos:
            for video in videos:
                item = feed.add_entry()

                vidtitle = video.find("h3", "thumbnail__title")
                if vidtitle:
                    item.title( vidtitle.text.strip() )
                else:
                    logging.info("Failed to get video title")
                    item.title( 'N/A' )

                viddescription = video.find("span", "channel__name")
                if viddescription:
                    item.description( viddescription.text.strip() )
                else:
                    logging.info("Failed to get video description/user name")
                    item.description( 'N/A' )


                lnk = video.find("a", "videostream__link")
                if lnk:
                    vid = lnk['href']
                else:
                    logging.info("Failed to get video link")
                    vid = None

                link = f'http://{self.request.host}/rumble/video' + vid

                icon = video.find( "img", "thumbnail__image" )
                if icon:
                    item.podcast.itunes_image( icon['src'] )
                else:
                    logging.info("Failed to get video icon")
                    item.podcast.itunes_image( None )

                item.link(
                    href = link,
                    title = item.title()
                )

                item.enclosure(
                    url = link,
                    type = "video/mp4"
                )

                viddatetime = video.find("time", "videostream__time")
                if viddatetime:
                    item.pubDate = viddatetime['datetime']
                else:
                    logging.info("Failed to get video date")
                    item.pubDate = None

                vidduration = video.find("div", "videostream__status--duration")
                if vidduration:
                    item.podcast.itunes_duration( vidduration.text.strip() )
                else:
                    logging.info("Failed to get video duration")

        return feed.rss_str( pretty=True )

def get_rumble_url( video, bitrate=None ):
    url = f"https://rumble.com/{video}"
    logging.debug( "Rumble: Getting: %s", url )

    ## first, we need to get the embed url from the data set
    # headers = { 'User-Agent': 'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/94.0.4606.104 Safari/537.36' }
    r = requests.get( url, headers=headers )

    # Check for errors from Rumble directly
    if r.status_code == 410:
        logging.error( "Rumble: Returned 410: Not found" )
        return url
    elif r.status_code == 403:
        logging.error( "Rumble: Returned 403: Forbidden" )
        return url

    bs = BeautifulSoup( r.text, 'lxml' )

    import json
    dat=json.loads(bs.find("script", type="application/ld+json").string)

    vidurl = dat[0]['embedUrl']
    logging.info( "Rumble: Found embedded URL: %s", vidurl )

    embed_vid_id = vidurl.rstrip('/').split('/')[-1]

    ## second, we get the url to the mp4 file
    ## tricky stuff that will likely break a lot
    ## but we need to parse out values within a javascript function
    ## and remove escape backslashes
    r = requests.get( vidurl, headers=headers )
    bs = BeautifulSoup( r.text, 'lxml' )
    el = bs.find("script").string

    import re
    vidurl = None

    # Using regex, grab the entire json data set from the javascript function.
    # Note: Expect this to break as Rumble makes more changes.
    regex_search = None
    try:
        regex_search = re.search( r';[b|f|h|v|m|y]\.f\["%s"\]=.*:[a|b|d|f|h|u]\(\)\}' % embed_vid_id, el ).group().replace( r';y.f["%s"]=' % embed_vid_id, '').replace( r';m.f["%s"]=' % embed_vid_id, '').replace( r';b.f["%s"]=' % embed_vid_id, '' ).replace( r';f.f["%s"]=' % embed_vid_id, '' ).replace( r';h.f["%s"]=' % embed_vid_id, '' ).replace( r';v.f["%s"]=' % embed_vid_id, '').replace( r',loaded:f()', '' ).replace( r',loaded:d()', '' ).replace( r',loaded:a()', '' ).replace( r',loaded:b()', '' ).replace( r',loaded:u()', '').replace( r',loaded:h()', '')
    except Exception as e:
        logging.error( "Rumble: Failed to parse JSON data:\n%s", e )

    # try: #again
    #     regexSearch = re.search( )
    vid_info = None

    if regex_search is not None:
        vid_info = json.loads( regex_search )
        logging.debug("Rumble: Successfully parsed JSON data")
        for thing in ('ua', 'u'):
            if thing in vid_info:
                if 'mp4' in vid_info[thing]:
                    if '360' in vid_info[thing]['mp4']:
                        logging.info('Rumble: Found 360p video')
                        vidurl = vid_info[thing]['mp4']['360']['url']
                    elif '480' in vid_info[thing]['mp4']:
                        logging.info('Rumble: Found 480p video')
                        vidurl = vid_info[thing]['mp4']['480']['url']
                if 'audio' in vid_info[thing]:
                    if 'url' in vid_info[thing]['audio']:
                        logging.info("Rumble: Found audio track")
                        vidurl = vid_info[thing]['audio']['url']
                    else:
                        if '192' in vid_info[thing]['audio']:
                            logging.info("Rumble: Found audio track")
                            vidurl = vid_info[thing]['audio']['192']['url']

    ## Fallback method, in case the above code failed to find anything
    if vidurl is None:
        logging.info("Rumble: Using fallback Rumble 'geturl' method")
        if bitrate is not None:
            # find the requested bitrate video
            for vid in vid_info[0]:
                ## handle bitrate requests
                if vid == bitrate:
                    vidurl = vid['url']
                    break
        else:
            ## First try to get video
            try:
                if 0 in vid_info and len(vid_info[0]) > 1:
                    # Resort to default bitrate video. 240p first, 360p second, anything at all third
                    for res in ('240', '360', '480'):
                        if res in vid_info[0]:
                            vid = vid_info[0].get(res)
                            if vid is not None:
                                logging.info("Rumble: Grabbing %sp video", res)
                                vidurl = vid['url']
                                break
                        else:
                            logging.info("Rumble: %s not found in video JSON", res)
                else:
                    logging.error( "Rumble: Video JSON does not have an mp4 link:\n%s", vid_info[0] )
                    logging.info( "Rumble: Will try to get audio only" )
                    if vid_info['u']['audio'] is not None:
                        vidurl = vid_info['u']['audio']['url']
                        logging.info( "Rumble: Found audio track at %s", vidurl )
                    else:
                        logging.error( "Rumble: Failed to find audio track: %s", video)

                    if vidurl is None:
                        logging.info( "Rumble: Last resort, trying to get anything" )
                        for vid in vid_info[0]:
                            if vid_info[0][vid]['url'] is not None:
                                logging.info("Rumble: Found %sp format", vid)
                                vidurl = vid_info[0][vid]['url']
                                logging.info("Rumble: Got something! Grabbing %s", vid)
                                break
                            else:
                                logging.error("Rumble: Failed to find anything for %s", vid)
            except Exception as e:
                logging.error( "Rumble: Failed to get video and audio: %s - %s", video, e )
    return vidurl

class VideoHandler(web.RequestHandler):
    def get(self, video):
        logging.debug("Rumble: Video: %s", video)
        bitrate = None
        if bitrate is not None:
            logging.info("Rumble: Requesting bitrate: %s", bitrate)
        vid = get_rumble_url(video, bitrate)
        if vid is not None:
            logging.info("Rumble: Redirecting to: %s", vid)
            self.redirect( vid )
        else:
            logging.error("Rumble: Failed to find video: %s", video)
            self.set_status( 404 )
            self.finish()