# configurations
.idea

# crawlee and apify storage folders
apify_storage
crawlee_storage
storage

# installed files
.venv

# git folder
.git


.dockerignore

root = true

[*]
indent_style = space
indent_size = 4
charset = utf-8
trim_trailing_whitespace = true
insert_final_newline = true
end_of_line = lf


.editorconfig

# This file tells Git which files shouldn't be added to source control

.idea
.DS_Store

apify_storage
storage

.venv/
.env/
__pypackages__
dist/
build/
*.egg-info/
*.egg

__pycache__

.mypy_cache
.dmypy.json
dmypy.json
.pytest_cache

.scrapy
*.log

# Added by Apify CLI
node_modules
.venv


.gitignore

# Add your dependencies here.
# See https://pip.pypa.io/en/latest/reference/requirements-file-format/
# for how to format them
apify ~= 1.1.1
beautifulsoup4 ~= 4.12.0
# requests ~= 2.31.0
httpx ~= 0.24.1
lxml ~= 4.9.2
tzdata == 2023.3


requirements.txt

Powerful API providing insights and statistics about Rumble channels and videos.

rumble-channel-analytics

Rumble Channel Analytics API

.actor/actor.json

# First, specify the base Docker image.
# You can see the Docker images from Apify at https://hub.docker.com/r/apify/.
# You can also use any other image from Docker Hub.
FROM apify/actor-python:3.11

# Second, copy just requirements.txt into the actor image,
# since it should be the only file that affects the dependency install in the next step,
# in order to speed up the build
COPY requirements.txt ./

# Install the packages specified in requirements.txt,
# Print the installed Python version, pip version
# and all installed packages with their versions for debugging
RUN echo "Python version:" \
 && python --version \
 && echo "Pip version:" \
 && pip --version \
 && echo "Installing dependencies:" \
 && pip install -r requirements.txt \
 && echo "All installed Python packages:" \
 && pip freeze

# Next, copy the remaining files and directories with the source code.
# Since we do this after installing the dependencies, quick build will be really fast
# for most source file changes.
COPY . ./

# Specify how to launch the source code of your actor.
# By default, the "python3 -m src" command is run
CMD ["python3", "-m", "src"]


.actor/Dockerfile

Provide one or more Rumble channel names or use the prefilled values for a first try.

Names of the Rumble channels you want to get the data for. Please note: a channel name is not the same as a user account name. A single user can have multiple channels. A Rumble channel has URL rumble.com/c/<channel name> while a Rumble user has URL rumble.com/user/<user name>.

Channel names

If set to 'yes', relevant video statistics will be provided in addition to the raw channel and video data. You get the minumum, maximum, mean, median and sum of the likes, dislikes, views and comments of the latest 3, 10 and 25 videos.

Include video analytics

Rumble Channel Analytics

.actor/input_schema.json

# import requests
import httpx

from statistics import fmean, median
from zoneinfo import ZoneInfo
from datetime import timezone, datetime
from apify import Actor
from bs4 import BeautifulSoup

def txt_to_int(txt:str) -> int:
    """
    Converts number string to int.
    For example, "11K" -> 11.000 or "2.76M" to 2760000.
    """
    d = {'K': 1000, 'M': 1000000, 'B': 1000000000}
    if txt[-1] in d:
        num, magnitude = txt[:-1], txt[-1]
        return int(float(num) * d[magnitude])
    else:
        return int(txt)
    
def rnd(num:float) -> int:
    """
    Rounds float to nearest integer.
    Deviates from Python's built-in round() behaviour.
    See https://stackoverflow.com/a/39304541.
    """
    return int(num + 0.5)

def get_stats(l: list) -> dict:
    """
    Returns summary statistics for a list of numbers.
    """
    return {
        'min': min(l),
        'max': max(l),
        'median': median(l),
        'mean': fmean(l),
        'mean_rounded': rnd(fmean(l)),
        'sum': sum(l)
    }

def get_videos_stats(videos:list) -> dict:
    """
    Returns summary statistics for a list of videos.
    """
    return {
        'n_videos': len(videos),
        'likes': get_stats([v.get('likes_int') for v in videos]),
        'dislikes': get_stats([v.get('dislikes_int') for v in videos]),
        'views': get_stats([v.get('views_int') for v in videos]),
        'comments': get_stats([v.get('comments_int') for v in videos])
    }

async def main():
    async with Actor:
        # Read the Actor input
        actor_input = await Actor.get_input() or {}
        channel_names = actor_input.get('channel_names', ['Bongino', 'StevenCrowder'])
        include_video_analytics = actor_input.get('include_video_analytics', True)
        start_urls = [
            {
                'url': f'https://rumble.com/c/{channel_name}',
                'userData': {
                    'channel_name': channel_name,
                    'include_video_analytics': include_video_analytics 
                }
            }
            for channel_name in channel_names
        ]

        if not start_urls:
            Actor.log.info('No start URLs specified in actor input, exiting...')
            await Actor.exit()

        # Enqueue the starting URLs in the default request queue
        default_queue = await Actor.open_request_queue()
        for start_url in start_urls:
            url = start_url.get('url')
            user_data = start_url.get('userData')
            Actor.log.info(f'Enqueuing {url} ...')
            await default_queue.add_request(
                {
                    'url': url,
                    'userData': user_data
                }
            )

        # Process the requests in the queue one by one
        while request := await default_queue.fetch_next_request():
            url = request['url']
            channel_name = request['userData'].get('channel_name')
            include_video_analytics = request['userData'].get('include_video_analytics')
            Actor.log.info(f'Scraping {url} ...')

            try:
                ### MAIN ACCOUNT PAGE ###

                # Brew soup.
                response = httpx.get(url)
                if response.status_code != 200:
                    raise ValueError(f'The URL "{url}" could not be processed. Are you sure "{channel_name}" is a Rumble channel?')
                html = response.text
                soup = BeautifulSoup(html, 'lxml')

                # Brew subsoups.
                backsplash_sp = soup.find(class_ = 'channel-header--backsplash-img')
                header_content_sp = soup.find(class_ = 'channel-header--content')
                header_title_sp = header_content_sp.find(class_ = 'channel-header--title')
                thumb_sp = header_content_sp.find(class_ = 'channel-header--thumb')
                verified_sp = header_content_sp.find(class_ = 'channel-header--verified')
                n_followers_sp = header_title_sp.find(class_ = 'channel-header--followers')
                locals_message_sp = soup.find(class_ = 'media-locals-message')
                locals_url_sp = locals_message_sp.find('a') if locals_message_sp else None

                videos_sp = soup.find_all(class_ = 'video-item', limit = 25)
                # print(videos_sp)
                videos = []
                if videos_sp:
                    for video_sp in videos_sp:
                        # Brew subsoups.
                        video_url_sp = video_sp.find(class_ = 'video-item--a')
                        video_thumb_sp = video_sp.find(class_ = 'video-item--img')
                        video_duration_sp = video_sp.find(class_ = 'video-item--duration')
                        video_time_posted_sp = video_sp.find(class_ = 'video-item--time')
                        video_title_sp = video_sp.find(class_ = 'video-item--title')
                        video_likes_sp = video_sp.find(class_ = 'rumbles-vote-up')
                        video_dislikes_sp = video_sp.find(class_ = 'rumbles-vote-down')
                        video_views_sp = video_sp.find(class_ = 'video-item--views')
                        video_comments_sp = video_sp.find(class_ = 'video-item--comments')

                        # Parse soups.
                        video_title = video_title_sp.getText() if video_title_sp else None
                        video_url = f'https://rumble.com{video_url_sp["href"]}' if video_url_sp else None
                        video_thumb_url = video_thumb_sp['src'] if video_thumb_sp else None
                        video_duration = video_duration_sp['data-value'] if video_duration_sp else None
                        video_upcoming = video_sp.find(class_ = 'video-item--img-wrapper--upcoming') is not None
                        video_status = 'posted' if not video_upcoming else 'upcoming'
                        video_time_posted = (
                            datetime.
                                fromisoformat(video_time_posted_sp['datetime']). # Convert timezone-aware ISO 8601 string to datetime object.
                                astimezone(ZoneInfo('UTC')). # Convert to UTC.
                                isoformat() # Convert to ISO 8601 string. 
                            if video_time_posted_sp else None
                        )
                        video_likes = video_likes_sp.getText() if video_likes_sp else '0'
                        video_dislikes = video_dislikes_sp.getText() if video_dislikes_sp else '0'
                        video_views = video_views_sp.getText() if video_views_sp else '0'
                        video_comments = video_comments_sp.getText() if video_comments_sp else '0'

                        # Create video entity.
                        video = {
                            'title': video_title,
                            'url': video_url,
                            'thumb_url': video_thumb_url,
                            'duration': video_duration,
                            'status': video_status,
                            'time_posted': video_time_posted,
                            'likes': video_likes,
                            'likes_int': txt_to_int(video_likes),
                            'dislikes': video_dislikes,
                            'dislikes_int': txt_to_int(video_dislikes),
                            'views': video_views,
                            'views_int': txt_to_int(video_views),
                            'comments': video_comments,
                            'comments_int': txt_to_int(video_comments)                            
                        }
                        videos.append(video)

                # Latest n videos stats.
                if include_video_analytics:
                    latest_3_videos_stats = get_videos_stats(videos[:3])
                    latest_10_videos_stats = get_videos_stats(videos[:10])
                    latest_25_videos_stats = get_videos_stats(videos[:25])
                else:
                    latest_3_videos_stats = None
                    latest_10_videos_stats = None
                    latest_25_videos_stats = None

                # Parse soups.
                display_name = soup.title.string if soup.title else None
                thumb_url = thumb_sp['src'] if thumb_sp else None
                backsplash_url = backsplash_sp['src'] if backsplash_sp else None
                verified = verified_sp is not None
                n_followers = n_followers_sp.getText().replace('Followers', '').strip() if n_followers_sp else None
                locals_url = locals_url_sp['href'] if locals_url_sp else None

                ### ABOUT SUB PAGE ###

                # Brew soup.
                about_url = f'{url}/about'
                response = httpx.get(about_url)
                html = response.text
                soup = BeautifulSoup(html, 'lxml')

                # Brew subsoups.
                main_sp = soup.find(class_='channel-about-description-socials')
                description_sp = main_sp.find(class_='channel-about--description') if main_sp else None
                sidebar_sp = soup.find(class_='channel-about-sidebar--inner')
                paragraphs_sp = sidebar_sp.find_all('p')

                # Parse soups.
                description = description_sp.find('p').getText() if description_sp else None

                if len(paragraphs_sp) == 2:
                    join_date_str_raw = paragraphs_sp[0].getText().replace('Joined', '').strip() # e.g. "Joined Sep 13, 2022" -> "Sep 13, 2022"
                    join_date = datetime.strptime(join_date_str_raw, '%b %d, %Y').strftime('%Y-%m-%d') # e.g. "2022-09-13"
                    n_videos = int(paragraphs_sp[1].getText().replace('videos', '').replace(',', '').strip())
                else:
                    join_date = None
                    n_videos = None

                ### DONE EXTRACTING ###

                # Create channel entity.
                channel = {
                    'channel_name': channel_name,
                    'display_name': display_name,
                    'url': url,
                    'thumb_url': thumb_url,
                    'backsplash_url': backsplash_url,
                    'n_followers': n_followers,
                    'n_followers_int': txt_to_int(n_followers),
                    'n_videos': n_videos,
                    'verified': verified,
                    'description': description,
                    'join_date': join_date,
                    'locals_url': locals_url,
                    'latest_25_videos': videos,
                    'latest_3_videos_stats': latest_3_videos_stats,
                    'latest_10_videos_stats': latest_10_videos_stats,
                    'latest_25_videos_stats': latest_25_videos_stats
                }

                # Push data to Actor.
                await Actor.push_data(channel)                    
            except:
                Actor.log.exception(f'Cannot extract data from {url}.')
            finally:
                # Mark the request as handled so it's not processed again
                await default_queue.mark_request_as_handled(request)

src/main.py

src/__init__.py

import asyncio
import logging

from apify.log import ActorLogFormatter

from .main import main

handler = logging.StreamHandler()
handler.setFormatter(ActorLogFormatter())

apify_client_logger = logging.getLogger('apify_client')
apify_client_logger.setLevel(logging.INFO)
apify_client_logger.addHandler(handler)

apify_logger = logging.getLogger('apify')
apify_logger.setLevel(logging.DEBUG)
apify_logger.addHandler(handler)

asyncio.run(main())


src/__main__.py

Powerful API providing data and statistics for Rumble channels and videos 🚀 Get all channel info plus analytics on channel and video performance 📈 Followers, likes, dislikes, views, comments, and much more 📝 Process multiple channels in batch. Fast and easy.  👌

Rumble Channel Analytics API · Apify

Input · Rumble Channel Analytics API · Apify

API · Rumble Channel Analytics API · Apify

Related Actors · Rumble Channel Analytics API · Apify

Powerful API providing data and statistics for Rumble channels and videos 🚀 Get all channel info plus analytics on channel and video performance 📈 Followers, likes, dislikes, views, comments, and...

Where next?