TikTok data scraping

If you do not live under a cave, you sure must have heard of TikTok.
It allows users to create 15-second videos, sound tracked by music clips.
It sounds simple enough, but it is a wildly popular concept.

I am not a big fan but had to study it for one of our customers.

Here are some statistics. I’ll talk a little about this study then share how you can set up a flow to collect TikTok data. Click here to skip to code .

TikTok Stats at the time of writing-

Around 689 million active users.
It has over 2 billion downloads on the play store.
Over 1 billion total video views.
Available in 155 countries.
On average, a user spends 52 minutes per day!

source

It’s another decoy for getting meaningful things done in life . But I’ll try to hate it less for the sake of this post

Heisenbergers GIFs - Get the best GIF on GIPHY

So it seems very useful for the following purposes –

Ads (Snuck in your feed)

Influencer Marketing

Branding

Customer Engagement

TikTok ads cost more than other social media platforms, with CPM as high as 10$.

Influencer marketing is less annoying and yields good returns since they’ve created their niche audience who are faithful to them At least for the time being until they get bored the influencer runs out of content or people find someone more interesting But it seems relatively cheap and more effective.

So which data would you collect? It’s very subjective but, I think these could be useful

Follower count
Following count
# of views on a post
Likes on a post
Comments on a post
Shares of a post
Hearts received (not literally)
Delta counts per day of the above stats!
Posts liked by a user
Backing up user posts (Downloading)

To collect data from TikTok, they provide an API for content management and marketing. But we want to gather other user’s data, which they don’t allow

TikTok has a web app So we turn to scrape !!!!!!!!!!

This library provides an API to scrape TikTok data from their web app using Selenium or Playwright Web Automation.

https://github.com/davidteather/TikTok-Api

TikTok realizes this and blocks your IP if you try to scrape data.
So what do we do?

Just use a Proxy Server !

Let’s outline the steps –

Collect a list of TikTok handles you want to scrape.
1. It can be a list of suggested accounts on TikTok from a seed account.
2. It can be a list of Tiktok handles that users registered on your website.
3. It can be a list of handles you’re managing or want to monitor (or spy ? ).
Collect a list of stats required.
If storing in a database, design a schema.
Set up a script to loop over your list of users.
Create a proxy server / Buy a service
1. Buy a service ? (https://www.techradar.com/in/best/proxies) OR
2. To create your own proxy server-
  1. Subscribe to ADEO Anonymous HTTPS Proxy (https://aws.amazon.com/marketplace/pp/prodview-lo2helasbjorq)
  2. Create an AWS EC2 instance
  3. Attach an elastic IP
  4. That’s it. This elastic IP is your proxy server IP!
Create a headless browser instance using this proxy IP (The library has inbuilt functions for this)
Call the profile details functions from the library.
Parse each JSON record and store it as a row in a CSV
Import into your database, if applicable.
Run static DB queries to populate the delta counts.

It roughly looks like this-

import requests
from TikTokApi import TikTokApi
import json
 
import utilityFunctions #custom made helper functions for eg Json key value to csv data
import time
 
 
 
class Tiktok( utilityFunctions):
 
    def __init__(self,args):
        #initialize global vars
         
        utilityFunctions.__init__(self, args)       
         
        #Put your configurations in a JSON file
        self.data_call = self.extractor_config['data']['call'][0]
         
        #proxy ip for avoiding reCaptcha after too many API Calls, in this case I've setup a proxy server on EC2 that rotates IPs
        self.proxy = self._chainedGet(self.data_call,"proxy_ip")
         
        #api response data file dump path
        self.local_tmp_file = self._chainedGet(self.data_call,"local_tmp_file")
         
        #response file content type
        self.content_type = self._chainedGet(self.data_call,"content_type")
         
        #Delay in seconds before each API call
        self.delay = self._chainedGet(self.data_call,"scrape_delay")
 
        #List of static tiktok hanldes to be scrapped, if you don't use API or want to manually add tiktok handles
        self.static_tiktok_handles = self._chainedGet(self.data_call,"static_tiktok_handles")
         
        if self.delay is None:
            self.delay = 5
         
        #user listing API/ this will give list of users to get the TikTok handles for
        self.user_list_api_url = self._chainedGet(self.data_call,"user_list_api_url")
         
        if self.proxy is not None:
            self.api = TikTokApi.get_instance(proxy=self.proxy)
        else:
            self.api = TikTokApi.get_instance()
         
        self.userList = self.getUserList()
 
        if self.static_tiktok_handles is not None and len(self.static_tiktok_handles)>0:
            self.userList = self.static_tiktok_handles + self.userList
 
    def getTikTokUserLikeList(self,tiktokUsername):
        #To scrape and assemble a JSON of posts Liked by a tiktok handle
        try:
            self.logger.info("Fetching User Likes for : "+tiktokUsername)           
            try:
                likeListDataResponse = self.api.userLikedbyUsername(tiktokUsername,count=2000, language='en')
            except Exception as e:
                self.logger.exception("Cannot fetch user "+tiktokUsername+": "+str(e))
                return
            self.logger.info("Delaying API Call for "+str(self.delay)+" seconds")
            time.sleep(self.delay)   
            if likeListDataResponse is not None:
                likeListData=[]
                 
                for item in likeListDataResponse:
                    item["uniqueId"]=tiktokUsername
                    likeListData.append(item)
 
            return likeListData
        except Exception as e:
            self.logger.exception("Error in getTikTokUserLikeList: "+str(e))
 
    def getTikTokVideoData(self,transformerObj):
        #To loop over accounts and assemble all users posts details in a list of json objects
        try:
            local_tmp_file = self.local_tmp_file
            content_type = self.content_type
 
            if local_tmp_file is None:
                local_tmp_file = "/tmp/VideoData.json"
 
            if content_type is None:
                content_type = "json"
 
            self.transformerObj = transformerObj
             
            userList = self.userList
 
            TikTokData = []
 
            for idx,username in enumerate(userList):  
                self.logger.info("Total  Users: "+str(len(userList)))
                self.logger.info("Processing user no "+str(idx+1)+"  Username: "+username)
                try:
                    TikTokUsername = self.getTikTokHandle(username)[0] 
                    Id = self.getTikTokHandle(username)[1] 
                    if TikTokUsername is None and username in self.static_tiktok_handles:
                        TikTokUsername = username
                
                    if TikTokUsername is not None:
                        self.logger.info("TikTok handle: "+TikTokUsername)
                        self.logger.info("Delaying API Call for "+str(self.delay)+" seconds")
                        time.sleep(self.delay)    
                        TikTokUserProfileData = self.getTikTokProfileVideos(TikTokUsername)
                                 
                        TikTokData.extend(TikTokUserProfileData)
                except Exception as e:
                    self.logger.exception("Error in getTikTokVideoData: "+str(e))
             
            data_count = len(TikTokData)
 
            with open(local_tmp_file, 'w', encoding='utf8') as json_file:
                json.dump(TikTokData, json_file, ensure_ascii=False)
             
            data_config = {
            "content_type": content_type
            }
             
            aFileContents = self.extractFileData(data_config, local_tmp_file)
 
            rTransformer = self.transformerObj.transform_process( aFileContents )
 
            self.pagination["total"] += data_count
 
            if( rTransformer != True ):
                self.logger.exception("Transformation Error")
                raise Exception("Transformation Error")
 
            return True
        except Exception as e:
            self.logger.exception("Error in getTikTokVideoData :"+str(e))
         
 
    def getTikTokUserData(self,transformerObj):
        #To loop over accounts and assemble all users profile details in a list of json objects
        try:
            local_tmp_file = self.local_tmp_file
            content_type = self.content_type
 
            if local_tmp_file is None:
                local_tmp_file = "/tmp/UserData.json"
 
            if content_type is None:
                content_type = "json"
 
            self.transformerObj = transformerObj
             
            userList = self.userList
             
 
            TikTokData = []
            for idx,username in enumerate(userList):  
                self.logger.info("Total  Users: "+str(len(userList)))
                self.logger.info("Processing user no "+str(idx+1)+"  Username: "+username)
                try:
                    TikTokUsername = self.getTikTokHandle(username)[0]  
                    Id = self.getTikTokHandle(username)[1]            
                    if TikTokUsername is None and username in self.static_tiktok_handles:
                        TikTokUsername = username
 
                    if TikTokUsername is not None:
                        self.logger.info("TikTok handle: "+TikTokUsername)
                        self.logger.info("Delaying API Call for "+str(self.delay)+" seconds")
                        time.sleep(self.delay)    
                        TikTokUserProfileData = self.getTikTokProfile(TikTokUsername)
                             
                        TikTokData.extend(TikTokUserProfileData)
                except Exception as e:
                    self.logger.exception("Error in getTikTokUserData: "+str(e))
 
            data_count = len(TikTokData)
 
            with open(local_tmp_file, 'w', encoding='utf8') as json_file:
                json.dump(TikTokData, json_file, ensure_ascii=False)
             
            data_config = {
            "content_type": "json"
            }
 
            aFileContents = self.extractFileData(data_config, local_tmp_file)
 
            self.pagination["total"] += data_count
            rTransformer = self.transformerObj.transform_process( aFileContents )
 
            if( rTransformer != True ):
                self.logger.exception("Transformation Error")
                raise Exception("Transformation Error")
 
            return True
        except Exception as e:
            self.logger.exception("Error in getTikTokUserData :"+str(e))
         
    def getTikTokUserLikeListData(self,transformerObj):
        #To loop over accounts and assemble all users liked tiktok posts details in a list of json objects
        try:
             
            local_tmp_file = self.local_tmp_file
            content_type = self.content_type
             
 
            if local_tmp_file is None:
                local_tmp_file = "/tmp/UserLikeListData.json"
 
            if content_type is None:
                content_type = "json"
 
            self.transformerObj = transformerObj
             
            userList = self.userList
 
            TikTokData = []
            for idx,username in enumerate(userList):  
                self.logger.info("Total  Users: "+str(len(userList)))
                self.logger.info("Processing user no "+str(idx+1)+"  Username: "+username)
                try:
                    TikTokUsername = self.getTikTokHandle(username)[0]
                     
                    if TikTokUsername is None and username in self.static_tiktok_handles:
                        TikTokUsername = username
                
                    if TikTokUsername is not None:
                        self.logger.info("TikTok handle: "+TikTokUsername)
                        self.logger.info("Delaying API Call for "+str(self.delay)+" seconds")
                        time.sleep(self.delay)    
                        TikTokUserProfileLikeListData = self.getTikTokUserLikeList(TikTokUsername)
                        if TikTokUserProfileLikeListData is not None and len(TikTokUserProfileLikeListData)>0:
                            TikTokData.extend(TikTokUserProfileLikeListData)
                except Exception as e:
                    self.logger.exception("Error in getTikTokUserData: "+str(e))
 
            data_count = len(TikTokData)
 
            with open(local_tmp_file, 'w', encoding='utf8') as json_file:
                json.dump(TikTokData, json_file, ensure_ascii=False)
             
            data_config = {
            "content_type": "json"
            }
 
            aFileContents = self.extractFileData(data_config, local_tmp_file)
 
            self.pagination["total"] += data_count
            rTransformer = self.transformerObj.transform_process( aFileContents )
 
            if( rTransformer != True ):
                self.logger.exception("Transformation Error")
                raise Exception("Transformation Error")
 
            return True
        except Exception as e:
            self.logger.exception("Error in getTikTokUserData :"+str(e))
 
    def getUserList(self):
        #This will call an API to get the list of users 
        if self.user_list_api_url is None:
            self.user_list_api_url = "<YOU API URL that return USER>"
         
        UserListResponse = requests.get(self.user_list_api_url, params="", headers="",timeout=5)
        UserListResponseJson = json.loads(UserListResponse.text)
        userList = []
        for user in UserListResponseJson["data"]["data"]:
            userList.append(user["user"]["username"])
        return userList
 
    def getTikTokHandle(self,username):
        #This will call another API to get the tiktok handle for that user
        if username is None:
            self.logger.info("No Tiktok handle for user")
            return
        tiktok_handle = None
 
        try:
            UserProfileAPIURL = "YOUR API URL/get-user-details-endpoint/"+username
            UserProfileResponse = requests.get(UserProfileAPIURL, params="", headers="",timeout=5)
            UserProfileResponseJson = json.loads(UserProfileResponse.text)
             
            for social_network in UserProfileResponseJson["data"]["social_networks"]:        
                tiktok_handle = social_network["tiktok_handle"]
                         
            return [tiktok_handle]
        except Exception as e:
            self.logger.exception("Error in getTikTokHandle: "+str(e))
 
 
    def getTikTokProfileVideos(self,tiktokUsername):
        #To get the stats for tiktok posts / videos
        try:
            self.logger.info("Fetching Posts for user: "+tiktokUsername)
            try:
                profileGenerator = self.api.getUserPager(tiktokUsername, cursor=0)
            except Exception as e:
                self.logger.exception("Cannot fetch user "+tiktokUsername+": "+str(e))
                return
            tikTokList = []
            count = 0
            for item in profileGenerator:
                if item is not None:        
                    self.logger.info("Delaying API Call for "+str(self.delay)+" seconds")
                    time.sleep(self.delay)    
                    tikTokList.extend(item)
                    count = count + len(item)
                self.logger.info("Posts fetched:"+str(count))
            return tikTokList
        except Exception as e:
            self.logger.exception("Error in getTikTokProfileVideos: "+str(e))
 
    def getTikTokProfile(self,tiktokUsername):
        #To scrape and assemble a JSON of TikTok Profile Data of a tiktok handle
        try:
            self.logger.info("Fetching User Profile for : "+tiktokUsername)
            try:
                profileDataResponse = self.api.getUser(tiktokUsername, cursor=0)
            except Exception as e:
                self.logger.exception("Cannot fetch user "+tiktokUsername+": "+str(e))
                return
            self.logger.info("Delaying API Call for "+str(self.delay)+" seconds")
            time.sleep(self.delay)    
            profileData = profileDataResponse["userInfo"]
             
            return [profileData]
        except Exception as e:
            self.logger.exception("Error in getTikTokProfile: "+str(e))

If you’re going to store it in a database, you can create a schema like below-

/*Table for TikTok UserData*/
create table IF NOT EXISTS orgUserData(
    org_id bigint(20) NOT NULL,
    userId bigint(20) NOT NULL,
    uniqueId varchar(255) NOT NULL,
    nickname varchar(255) DEFAULT NULL,
    userCreatedDate datetime DEFAULT NULL,
    avatarThumb text DEFAULT NULL,
    secUid varchar(255) DEFAULT NULL,
    privateAccount tinyint(1) DEFAULT '0',
    secret tinyint(1) DEFAULT '0',
    unique (org_id)
);
 
/*Table for TikTok UserStats*/
create table IF NOT EXISTS orgUserStats(
    userId bigint(20) NOT NULL,
    followerCount bigint(20) DEFAULT NULL,
    followingCount bigint(20) DEFAULT NULL,
    heartCount bigint(20) DEFAULT NULL,
    videoCount bigint(20) DEFAULT NULL,
    diggCount bigint(20) DEFAULT NULL,
    unique(userId)
);
 
 
/*Table for TikTok Posts meta Info*/
create table IF NOT EXISTS orgContentData(
    contentId bigint(20) NOT NULL,
    contentTitle varchar(255) DEFAULT NULL,
    contentDescription text,
    contentCreatedDate datetime DEFAULT NULL,
    contentCover text DEFAULT NULL,
    contentOriginCover text DEFAULT NULL,
    contentType varchar(10) DEFAULT 'video',
    PRIMARY KEY (contentId)
);
 
/*Table for TikTok Posts & User Data Mapping with Content Stats*/
CREATE TABLE IF NOT EXISTS orgUserContentUploadData (
    userId bigint(20) NOT NULL,
    contentId bigint(20) NOT NULL,
    diggCount bigint(20) DEFAULT NULL,
    shareCount bigint(20) DEFAULT NULL,
    commentCount bigint(20) DEFAULT NULL,
    playCount bigint(20) DEFAULT NULL,
    PRIMARY KEY (contentId),
    CONSTRAINT orgUserContentUploadData_ibfk_1 FOREIGN KEY (contentId) REFERENCES orgContentData (contentId)
);
 
/*Table for TikTok Posts Stats*/
create table IF NOT EXISTS orgContentStats(
    contentId bigint(20) NOT NULL,
    diggCount bigint(20) DEFAULT NULL,
    shareCount bigint(20) DEFAULT NULL,
    commentCount bigint(20) DEFAULT NULL,
    playCount bigint(20) DEFAULT NULL,
    PRIMARY KEY (contentId),
    CONSTRAINT orgContentDataInStats_ibfk_1 FOREIGN KEY (contentId) REFERENCES orgContentData (contentId)
);
 
 
/*Table for TikTok User Likes*/
create table IF NOT EXISTS orgUserLikes(
    userId bigint(20) NOT NULL,
    contentId bigint(20) NOT NULL,
    UNIQUE KEY user_content_unqiue (userId, contentId)
);
 
/*For Delta counts*/
select
    distinct cd.userId,
    cd.contentId,
    tmp.diggCount - cd.diggCount as diggDiff,
    tmp.shareCount - cd.shareCount as diggDiff,
    tmp.commentCount - cd.commentCount as commentDiff,
    tmp.playCount - cd.playCount as playDiff
from
    orgContentDataTmp tmp
    join orgUserContentUploadData cd on cd.contentId = tmp.contentId;
 
/*Table to store stats delta per day*/
/*We use a different DB for this, can you guess ? */
CREATE TABLE orgUserContentUploadDataDeltas (
    userId int NOT NULL,
    contentId int NOT NULL,
    diggCount int DEFAULT NULL :: int,
    shareCount int DEFAULT NULL :: int,
    commentCount int DEFAULT NULL :: int,
    playCount int DEFAULT NULL :: int,
    date_time timestamp DEFAULT (now()) :: timestamptz(6)
);

In the End –

Now you can run an Analysis on this data!
Use it for machine learning, running campaigns with influencers,
monitoring account stats for optimizing engagement and lots more.

Want to get this setup ? with detailed reporting, analytic dashboards with useful KPIs, and tracking.
You can check out Catchmedia Inc (https://www.catchmedia.com/). We’ve got an awesome team