soundcloud-scrape/scrape.py

#!/usr/bin/python3

# It is assumed that you have installed the required packages in requirements.txt
# and have also installed ChromeDriver from your package manager.
# If you have installed ChromeDriver separately, change the line below to the
# path of the ChromeDriver executable.
CHROMEDRIVER_PATH = '/usr/bin/chromedriver'

# The number of steps outward from each starting user the script will scrape.
# This works for both who the user follows and who follows the user.
# For example:
#   User <-- Follower <-- Follower's Follower
#    ^0         ^1                ^2
# User is zero hops away from itself, so it's degree is zero.
# Follower is one hop away from User, so it's degree is one, and so on.
MAX_DEGREE = 2

# The list of starting users. Can be as many or as few as you would like.
# Bear in mind that each additional user will greatly increase run time, depending
# on how high the max degree is set to (see above).
STARTING_USERS = ['thomotron']

# The below settings determine how many of each item will be collected per user.
# Adjust these as you see fit. The higher they are, the larger the graph and the
# longer the scrape time.
# Be careful when changing the following and followers settings. These will, in
# combination with MAX_DEGREE, determine how long the scraping process will take.
USER_MAX_TRACKS = 100
USER_MAX_LIKES = 100
USER_MAX_FOLLOWING = 25
USER_MAX_FOLLOWERS = 25

##### Import all the things #################################

import time
import shelve
import re
import networkx as nx
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions
from selenium.webdriver.chrome.options import Options
from bs4 import BeautifulSoup
from collections import Counter

##### Set up NetworkX #######################################

graph = nx.DiGraph()

print('NetworkX initialised')

##### Set up WebDriver with Chrome ##########################

chrome_options = Options()

# Start headless so we can run on a server overnight
chrome_options.add_argument('--headless')

# Disable image loading, courtesy of rocky qi
# Found at https://stackoverflow.com/a/31581387, accessed on 2018/09/17 at 12:54 UTC
chrome_options.add_experimental_option("prefs", {"profile.managed_default_content_settings.images":2})

driver = webdriver.Chrome(executable_path=CHROMEDRIVER_PATH, chrome_options=chrome_options)

print('WebDriver initialised')

##### Define some classes to hold our data ##################

# User class, holds all the data we will be collecting from user pages
# It's being used here as a more structured alternative to a dictionary
class User:
    url_username = None
    username = None
    track_count = None
    like_count = None
    following_count = None
    follower_count = None

    tracks = None
    likes = None

    followers = None
    following = None

    processed = False

    def __init__(self):
        self.tracks = []
        self.likes = []
        self.followers = []
        self.following = []

class Track:
    url = None
    title = None
    artist = None
    date = None
    tag = None # We'll only use the first tag for now, saves opening each track

    plays = None
    likes = None
    reposts = None
    comments = None

##### Define some methods to do our scraping ################

# Gets a user's username, track count, following count, follower count, and like count
def Get_Basic_Info(user):
    # Get the user's profile page and soup it
    print('GET: https://soundcloud.com/' + user.url_username)
    driver.get('https://soundcloud.com/' + user.url_username)
    soup = BeautifulSoup(driver.page_source, 'lxml')

    print('  Basic Info:')

    # Grab their username
    # Use stripped_strings generator as workaround for users with premium badge
    username = next(soup.find(class_='profileHeaderInfo__userName').stripped_strings)
    user.username = username
    print('    Username: ' + user.username)

    # Grab their track count
    track_count_elem = soup.find('a', href='/' + user.url_username + '/tracks', class_='infoStats__statLink')
    if track_count_elem:
        user.track_count = Human_Str_To_Int(str(track_count_elem.div.string))
    print('    Tracks: ' + str(user.track_count))

    # Grab their following count
    following_count_elem = soup.find('a', href='/' + user.url_username + '/following', class_='infoStats__statLink')
    if following_count_elem:
        user.following_count = Human_Str_To_Int(str(following_count_elem.div.string))
    print('    Following: ' + str(user.following_count))

    # Grab their follower count
    follower_count_elem = soup.find('a', href='/' + user.url_username + '/followers', class_='infoStats__statLink')
    if follower_count_elem:
        user.follower_count = Human_Str_To_Int(str(follower_count_elem.div.string))
    print('    Followers: ' + str(user.follower_count))

    # Grab their like count
    like_count_elem = soup.find(class_='sidebarHeader__actualTitle', text=re.compile(r'^[\S]*?\ likes$'))
    if like_count_elem:
        user.like_count = Human_Str_To_Int(str(like_count_elem.string.split(' ')[0]))
    print('    Likes: ' + str(user.like_count))

# Gets a list of fully-populated Track objects
def Get_Tracks_Info(url, limit = 100):
    # Get the page and soup it
    print('GET: ' + url)
    driver.get(url)

    # Loop until we reach the bottom of the page so all tracks are loaded
    soup = None
    while True:
        print('Scrolling...')
        driver.execute_script('window.scrollTo(0, document.body.scrollHeight);')  # Scroll to the bottom of the page
        time.sleep(1) # Wait a second for the page to load

        # Soup what we have
        soup = BeautifulSoup(driver.page_source, 'lxml')
        loading_elem = soup.find(class_='loading')
        track_elems = soup.find_all(class_='soundList__item')

        # Check if there are more items or we have enough
        if not loading_elem:
            print('Reached end of page')
            break
        elif len(track_elems) >= limit:
            print('Reached track limit')
            break

    print('  Tracks:')

    tracks = {}

    # Get all of the track elements up to the given limit
    track_elems = soup.find_all('li', class_='soundList__item')[:limit]
    for track_elem in track_elems:
        # Initialise a track object to hold our data
        track = Track()

        # Ignore collections
        if track_elem.find(class_='sound__trackList'):
            continue

        # Grab the title and url
        title_elem = track_elem.find(class_='soundTitle__title')
        if title_elem:
            track.title = str(title_elem.span.string.strip())
            track.url = 'https://soundcloud.com' + title_elem['href']

        # Check if we have already gotten this track before
        if track.url in tracks.keys():
            continue

        # Grab who uploaded it (usually the artist unless it's a label)
        artist_elem = track_elem.find(class_='soundTitle__usernameText')
        if artist_elem:
            track.artist = str(artist_elem.string.strip())

        # Grab when it was uploaded
        date_elem = track_elem.find(class_='soundTitle__uploadTime')
        if date_elem:
            track.date = str(date_elem.time['datetime'])

        # Grab the first tag featured on the list item
        tag_elem = track_elem.find(class_='soundTitle__tagContent')
        if tag_elem:
            track.tag = str(tag_elem.string).lower()

        # Grab the play and comment counts (either of these may or may not be present)
        play_comment_elems = track_elem.find_all(class_='sc-ministats-item')
        if play_comment_elems:
            for elem in play_comment_elems:
                num, unit = str(elem['title']).split(' ')
                if unit == 'plays':
                    track.plays = Human_Str_To_Int(num)
                elif unit == 'comments':
                    track.comments = Human_Str_To_Int(num)

        # Grab the like count
        like_elem = track_elem.find(class_='sc-button-like')
        if like_elem:
            like_elem_text = str(like_elem.string.strip())
            if 'like' in like_elem_text.lower():
                track.likes = 0
            else:
                track.likes = Human_Str_To_Int(like_elem_text)

        # Grab the repost count
        repost_elem = track_elem.find(class_='sc-button-repost')
        if repost_elem:
            repost_elem_text = str(repost_elem.string.strip())
            if 'repost' in repost_elem_text.lower():
                track.reposts = 0
            else:
                track.reposts = Human_Str_To_Int(repost_elem_text)

        # Finally add the track to the dictionary
        tracks[track.url] = track

        print('    ' + track.artist + ' - ' + track.title)

    # Strip off keys and return a track list
    return tracks.values()

# Gets a list of User objects from a badge list containing only their url_username
def Get_Follows(url, limit = 100):
    # Get the page
    print('GET: ' + url)
    driver.get(url)

    # Loop until we reach the bottom of the page so all badges are loaded
    soup = None
    while True:
        print('Scrolling...')
        driver.execute_script('window.scrollTo(0, document.body.scrollHeight);')  # Scroll to the bottom of the page
        time.sleep(1) # Wait a second for the page to load

        # Soup what we have
        soup = BeautifulSoup(driver.page_source, 'lxml')
        loading_elem = soup.find(class_='loading')
        track_elems = soup.find_all(class_='badgeList__item')

        # Check if there are more items or we have enough
        if not loading_elem:
            print('Reached end of page')
            break
        elif len(track_elems) >= limit:
            print('Reached user limit')
            break

    print('  Users:')

    users = {}

    # Get all of the user badges up to the given limit
    badge_elems = soup.find_all(class_='badgeList__item')[:limit]
    for badge_elem in badge_elems:
        # Grab their URL username
        username_elem = badge_elem.find('a', class_='userBadgeListItem__heading')
        if username_elem:
            user = User()
            user.url_username = username_elem['href'].strip('/')
            users[user.url_username] = user

            print('    ' + user.url_username)

    # Return just the values to make adding to the list easier
    return users.values()

# Converts string representations of numbers such as '1,234' or '1.6M' to integers
def Human_Str_To_Int(string):
    num = None
    string = string.lower().replace(',', '')
    if 'k' in string:
        string = string.replace('k', '')
        num = int(float(string) * 1000)
    elif 'm' in string:
        string = string.replace('m', '')
        num = int(float(string) * 1000000)
    else:
        num = int(string)

    return num

# Iterates through a set of track URLs and finds the modal tag and artist
def Favourite_Tag_And_Artist(track_urls, track_db):
    tags = []
    artists = []
    for track in track_urls:
        if not track in track_db.keys():
            continue

        tags.append(track_db[track].tag)
        artists.append(track_db[track].artist)

    # Return the first mode of the list, originally by Christian Witts and Rory Daulton
    # Found at https://stackoverflow.com/a/10797913, accessed on 2018/09/30 at 01:14 UTC
    return [Counter(tags).most_common(1)[0][0], Counter(artists).most_common(1)[0][0]]

##### Scrape ################################################

user_dict = {}
track_dict = {}
users_to_process = STARTING_USERS
users_to_process_next = []
iterator = 0

with shelve.open('shelf.db') as shelf:
    if 'user_dict' in shelf.keys():
        user_dict = shelf['user_dict']
    if 'track_dict' in shelf.keys():
        track_dict = shelf['track_dict']
    if 'users_to_process' in shelf.keys():
        users_to_process = shelf['users_to_process']
    if 'users_to_process_next' in shelf.keys():
        users_to_process_next = shelf['users_to_process_next']
    if 'iterator' in shelf.keys():
        iterator = shelf['iterator']


while iterator <= MAX_DEGREE:
    for user_str in users_to_process:
        print('Processing ' + user_str + ' (distance ' + str(iterator) + ')')

        # Skip any users that have already been passed over to avoid infinite loops
        if user_str in user_dict.keys():
            if user_dict[user_str].processed:
                print('  Already processed, skipping')
                continue

        # Initialise our user object to store our values in
        user = User()
        user.url_username = user_str
        print('  URL Username: ' + user.url_username)

        # Get their basic info
        Get_Basic_Info(user)

        # Get their tracks
        for track in Get_Tracks_Info('https://soundcloud.com/' + user.url_username + '/tracks', USER_MAX_TRACKS):
            if not track.url in track_dict.keys():
                track_dict[track.url] = track
            user.tracks.append(track.url)

        # Get their likes
        for like in Get_Tracks_Info('https://soundcloud.com/' + user.url_username + '/likes', USER_MAX_LIKES):
            if not like.url in track_dict.keys():
                track_dict[like.url] = like
            user.likes.append(like.url)

        # Get who they follow
        for following in Get_Follows('https://soundcloud.com/' + user.url_username + '/following', USER_MAX_FOLLOWING):
            if not following.url_username in user_dict.keys():
                user_dict[following.url_username] = following
            user.following.append(following.url_username)
            users_to_process_next.append(following.url_username)

        # Get who follows them
        for follower in Get_Follows('https://soundcloud.com/' + user.url_username + '/followers', USER_MAX_FOLLOWERS):
            if not follower.url_username in user_dict.keys():
                user_dict[follower.url_username] = follower
            user.followers.append(follower.url_username)
            users_to_process_next.append(follower.url_username)

        # Finally add the user to the dictionary and mark them as processed
        user.processed = True
        user_dict[user.url_username] = user

        # Update the shelf with the new dictionaries
        with shelve.open('shelf.db') as shelf:
            shelf['user_dict'] = user_dict
            shelf['track_dict'] = track_dict
            shelf['users_to_process'] = users_to_process
            shelf['users_to_process_next'] = users_to_process_next
            shelf['iterator'] = iterator

    iterator += 1
    users_to_process = users_to_process_next
    users_to_process_next = []

driver.quit()

##### Process what was scraped ##############################

print('Done scraping, here\'s what we got')
print('==================================')

for _, track in track_dict.items():
    graph.add_node( \
        track.url, \
        type='track', \
        label=str(track.title), \
        artist=str(track.artist), \
        title=str(track.title), \
        date=str(track.date), \
        tag=str(track.tag), \
        plays=str(track.plays), \
        likes=str(track.likes), \
        reposts=str(track.reposts), \
        comments=str(track.comments) \
    )

for _, user in user_dict.items():
    if not user.processed:
        continue

    favourite_tag, favourite_artist = Favourite_Tag_And_Artist(user.likes, track_dict)

    graph.add_node( \
        user.url_username, \
        type='user', \
        label=str(user.username), \
        track_count=str(user.track_count), \
        like_count=str(user.like_count), \
        following_count=str(user.following_count), \
        follower_count=str(user.follower_count) \
    )

    print(user.username)
    print('  ' + str(user.track_count) + ' tracks')
    print('  ' + str(user.like_count) + ' likes')
    print('  Following ' + str(user.following_count))
    print('  Followed by ' + str(user.follower_count))

    print('  Tracks:')
    for track in user.tracks:
        print('    ' + str(track))
        graph.add_edge( \
            user.url_username, \
            track, \
            label='created'
        )

    print('  Likes:')
    for track in user.likes:
        print('    ' + str(track))
        graph.add_edge( \
            user.url_username, \
            track, \
            label='likes'
        )

    print('  Followers:')
    for follower in user.followers:
        if follower in user_dict.keys():
            if not user_dict[follower].processed:
                continue

            print('    ' + str(follower))
            graph.add_edge( \
                follower, \
                user.url_username, \
                label='follows'
            )

    print('  Following:')
    for following in user.following:
        if following in user_dict.keys():
            if not user_dict[following].processed:
                continue

            print('    ' + str(following))
            graph.add_edge( \
                user.url_username, \
                following, \
                label='follows'
            )

print('Making graph...')
nx.write_gexf(graph, 'graph.gexf')
print('Graph written to \'graph.gexf\'')