FacebookScrapper.py

import pandas as pd
from bs4 import BeautifulSoup
import json
import time
import numpy as np
from tqdm import tqdm
from typing import *
import re
import os
import glob
from IPython.display import clear_output

# Scrapping and crawling modules
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.support.wait import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from user_agent import generate_user_agent
from selenium.webdriver.common.keys import Keys
from selenium.webdriver.common.action_chains import ActionChains
from datetime import datetime, timedelta
from urllib.parse import quote
import base64
import json

import locale
locale.setlocale(locale.LC_TIME, 'id_ID.UTF-8')

def encode_params(params: str) -> str:
    '''
    Encode the parameters to base64 as facebook url uses it for its search parameter
    '''
    return base64.b64encode(params.encode('utf-8')).decode('utf-8')

def betterWait(value: int):
    '''
    Just a glorified time.sleep() function
    '''
    
    for i in tqdm(range(value), desc="Waiting"):
        time.sleep(1)
    clear_output()

def generateDailyBackwards(startDate: str, endDate: str) -> List[str]:
    '''
    PARAMS:
    - startDate: str, format 'YYYY-MM-DD'
    - endDate: str, format 'YYYY-MM-DD'
    RETURN:
    - list of date strings from startDate to endDate in reverse order
    '''
    startDate = datetime.strptime(startDate, '%Y-%m-%d')
    endDate = datetime.strptime(endDate, '%Y-%m-%d')
    current_date = startDate
    day_ranges = []

    while current_date <= endDate:
        day_ranges.append(current_date.strftime('%Y-%m-%d'))
        current_date += timedelta(days=1)

    return day_ranges[::-1]

def emptyCSV():
    '''
    This function will check whether there are empty csv files in the savepoints folder
    '''
    csvFiles = [i for i in os.listdir("savepoints") if i.endswith(".csv")]
    temp = []
    for i in csvFiles:
        with open(f'savepoints/{i}', "r", encoding='utf-8', errors='ignore') as f:
            if len(f.read().split("\n")) == 2:
                temp.append(i)
    return temp

# Credentials
with open('Credentials/Facebook.json') as f:
    credentials = json.load(f)
    EMAIL = credentials['Email']
    PASSWORD = credentials['Pass']

# TODO: Simplify it into OOP and handle if tab reaches its end and can't be found. Lol!
class facebookScraper:
    def __init__(self):
        self.theEntireDataset = pd.DataFrame({"User": [], "Post": [], "Time": [],
                           "Likes": [], "Comments": [], "Shares": [] })
            
    def startAndLogin(self):
        '''
        This function will start the driver and login to facebook
        '''
        loginURL = 'https://www.facebook.com/'
        
        # Initialize the driver
        usergAgent = generate_user_agent(device_type="desktop", os="win", navigator="chrome", platform="win")
        options = Options()
        options.add_argument(f'user-agent={usergAgent}')
        self.driver = webdriver.Chrome()
        self.driver.get(loginURL)
        
        WebDriverWait(self.driver, 20).until(EC.presence_of_element_located((By.XPATH, '//form[@data-testid="royal_login_form"]')))

        self.driver.find_element(By.XPATH, '//input[@data-testid="royal_email"]').send_keys(EMAIL)
        self.driver.find_element(By.XPATH, '//input[@name="pass"]').send_keys(PASSWORD)
        time.sleep(2)
        self.driver.find_element(By.XPATH, '//button[@data-testid="royal_login_button"]').click()
        betterWait(10)
        print("Logged in")
        print("Please disallow notification :)")
    
    def searchQuery(self, search: str, startDate: str, endDate: str, recent: Literal[True, False]):
        '''
        This function will search the query and filter the date
        PARAMS
        - search: str - The search query
                
        - startDate: str - The start date of the search
        `YYYY-MM-DD`
        
        - endDate: str - The end date of the search
        `YYYY-MM-DD`
        '''
        URL = f'https://www.facebook.com/search/posts?q={search}&filters='
        
        # This will crawl the search query + encode it before hand
        if any ([startDate, endDate]):
            year = startDate.split('-')[0]
            if recent:
                URL += encode_params(f'{{"rp_creation_time:0":"{{\\"name\\":\\"creation_time\\",\\"args\\":\\"{{\\\\\\"start_year\\\\\\":\\\\\\"{year}\\\\\\",\\\\\\"start_month\\\\\\":\\\\\\"{year}-1\\\\\\",\\\\\\"end_year\\\\\\":\\\\\\"{year}\\\\\\",\\\\\\"end_month\\\\\\":\\\\\\"{year}-12\\\\\\",\\\\\\"start_day\\\\\\":\\\\\\"{startDate}\\\\\\",\\\\\\"end_day\\\\\\":\\\\\\"{endDate}\\\\\\"}}\\"}}", "recent_posts:0":"{{\\"name\\":\\"recent_posts\\",\\"args\\":\\"\\"}}"}}')
            else:
                URL+= encode_params(f'{{"rp_creation_time:0":"{{\\"name\\":\\"creation_time\\",\\"args\\":\\"{{\\\\\\"start_year\\\\\\":\\\\\\"{year}\\\\\\",\\\\\\"start_month\\\\\\":\\\\\\"{year}-1\\\\\\",\\\\\\"end_year\\\\\\":\\\\\\"{year}\\\\\\",\\\\\\"end_month\\\\\\":\\\\\\"{year}-12\\\\\\",\\\\\\"start_day\\\\\\":\\\\\\"{startDate}\\\\\\",\\\\\\"end_day\\\\\\":\\\\\\"{endDate}\\\\\\"}}\\"}}"}}')
        elif recent:
            URL += encode_params('{"recent_posts:0":"{\\"name\\":\\"recent_posts\\",\\"args\\":\\"\\"}"}')
        
        self.driver.get(URL)
        time.sleep(10)
        
    def reelHandling(self):
        # This will handle reel post
        theReel = self.post.find_element(By.XPATH, f'.//a[@aria-label="Buka reel di Reels Viewer"]')

        # This to get user info and time of the post
        try:
            # This checks if it's sent from individual account
            upperInfos = theReel.find_element(By.XPATH, './div[1]/div[3]/div/div/div[1]/div/div/div[2]/div')
            User = upperInfos.find_element(By.XPATH, './div[1]/div[1]/h4').text
            hover = ActionChains(self.driver).move_to_element(upperInfos.find_element(By.XPATH, './div[2]/span/span/span/span[2]'))
            for i in range(50):
                hover.perform()
            theTime = self.driver.find_element(By.XPATH, '//div[@class="__fb-dark-mode"]//span').text
            theTime = datetime.strptime(theTime, '%A, %d %B %Y pada %H.%M')
            
            # This to get ammount of like, comment, and share
            lowerInfps = theReel.find_element(By.XPATH, './div[2]/div/div/div')
            like = lowerInfps.find_element(By.XPATH, './div[3]').text if lowerInfps.find_element(By.XPATH, './div[3]').text != '' else 0
            comment = lowerInfps.find_element(By.XPATH, './div[4]').text if lowerInfps.find_element(By.XPATH, './div[4]').text != '' else 0
            share = lowerInfps.find_element(By.XPATH, './div[5]').text if lowerInfps.find_element(By.XPATH, './div[5]').text != '' else 0
            
            try:
                # If it's nested
                theReel.find_element(By.XPATH, './div[1]/div[2]/div/div/div[2]/span//object[@type = "nested/pressable"]/div[@role = "button"]').click()
            except:
                pass

            # This to get the post
            theText = theReel.find_element(By.XPATH, './div[1]/div[2]/div/div/div[2]/span').text

        except:
            # This checks if it's sent from a group
            upperInfos = theReel.find_element(By.XPATH, './div[1]/div[3]/div/div/div[1]/div/div/div/div/div')
            User = upperInfos.find_element(By.XPATH, './div[1]').text
            hover = ActionChains(self.driver).move_to_element(upperInfos.find_element(By.XPATH, './div[2]/span/span/span/span[4]'))
            for i in range(50):
                hover.perform()
            theTime = self.driver.find_element(By.XPATH, '//div[@class="__fb-dark-mode"]//span').text
            theTime = datetime.strptime(theTime, '%A, %d %B %Y pada %H.%M')
            
            try:
                # If it's nested
                theReel.find_element(By.XPATH, './div[1]/div[2]/div/div/div[2]/span//object[@type = "nested/pressable"]/div[@role = "button"]').click()
            except:
                pass
            
            # This to get the post
            theText = theReel.find_element(By.XPATH, './div[1]/div[2]/div/div/div[2]/span').text
            
            # This to get ammount of like, comment, and share
            lowerInfos = self.post.find_element(By.XPATH, './div/div/div/div/div/div/div/div/div/div/div[13]/div/div/div[3]/div/div/div/div/div[1]/div')
            like = lowerInfos.find_element(By.XPATH, './div[1]/div/span/div/span').text if lowerInfos.find_element(By.XPATH, './div[1]/div/span/div/span').text != '' else 0
            comment = lowerInfos.find_element(By.XPATH, './div[2]/div[2]').text.split(" ")[0] if lowerInfos.find_element(By.XPATH, './div[2]/div[2]').text != '' else 0
            share = lowerInfos.find_element(By.XPATH, './div[2]/div[3]').text.split(" ")[0] if lowerInfos.find_element(By.XPATH, './div[2]/div[3]').text != '' else 0
        
        if theTime in self.theDict['Time'] and theText in self.theDict['Post']:
            return None
                    
        self.theDict['User'].append(User)
        self.theDict['Time'].append(f'{theTime}')
        self.theDict['Likes'].append(like)
        self.theDict['Comments'].append(comment)
        self.theDict['Shares'].append(share)
        self.theDict['Post'].append(theText)
    
    def articleHandling(self):
        theArticle = self.post.find_element(By.XPATH, './div/div/div/div/div/div/div/div/div/div/div[13]/div/div')
        
        # Get upperInfos (User and Time)
        upperInfos = theArticle.find_element(By.XPATH, './div[2]/div/div[2]/div')
        try:
            # Individual account
            User = upperInfos.find_element(By.XPATH, './div[1]//strong[1]').text

            hover = ActionChains(self.driver).move_to_element(upperInfos.find_element(By.XPATH, './div[2]/span/div/span[1]'))
            for i in range(50):
                hover.perform()
                
            theTime = self.driver.find_element(By.XPATH, '//div[@class="__fb-dark-mode"]//span').text
            theTime = datetime.strptime(theTime, '%A, %d %B %Y pada %H.%M')
            
        except:
            # If it's from a group
            groupName = upperInfos.find_element(By.XPATH, './div[1]/span').text
            User = upperInfos.find_element(By.XPATH, './div[2]/span/div/span[1]').text
            User = f'{groupName}; {User}'
            
            hover = ActionChains(self.driver).move_to_element(upperInfos.find_element(By.XPATH, './div[2]/span/div/span[3]'))
            for i in range(50):
                hover.perform()
                
            theTime = self.driver.find_element(By.XPATH, '//div[@class="__fb-dark-mode"]//span').text
            theTime = datetime.strptime(theTime, '%A, %d %B %Y pada %H.%M')
        
        # Get lowerInfos (Like, Comment, Share)
        lowerInfos = theArticle.find_element(By.XPATH, './div[4]/div/div/div/div/div[1]/div')
        like = lowerInfos.find_element(By.XPATH, './div[1]/div/span/div/span').text if lowerInfos.find_element(By.XPATH, './div[1]/div/span/div/span').text != '' else 0
        # This will check if there's no comment and share
        if lowerInfos.find_element(By.XPATH, './div[2]').text == '':
            comment = 0
            share = 0
        else:
            try: 
                # Check if it's not a live stream, what the literal fuck? Why do people use this dogshit feature for this specific tag
                lowerInfos.find_element(By.XPATH, './div[2]/div[3]')
                comment = lowerInfos.find_element(By.XPATH, './div[2]/div[2]').text.split(" ")[0] if lowerInfos.find_element(By.XPATH, './div[2]/div[2]').text != '' else 0
                share = lowerInfos.find_element(By.XPATH, './div[2]/div[3]').text.split(" ")[0] if lowerInfos.find_element(By.XPATH, './div[2]/div[3]').text != '' else 0
            except:
                comment = lowerInfos.find_element(By.XPATH, './div[2]/div[1]').text.split(" ")[0] if lowerInfos.find_element(By.XPATH, './div[2]/div[1]').text != '' else 0
                share = lowerInfos.find_element(By.XPATH, './div[2]/div[2]').text.split(" ")[0] if lowerInfos.find_element(By.XPATH, './div[2]/div[2]').text != '' else 0
        try:
            # If it's nested
            theArticle.find_element(By.XPATH, './div[3]/div[1]/div/div/div/div/span//div[@role="button"]').click()
        except:
            pass

        # This to get the post
        theText = theArticle.find_element(By.XPATH, './div[3]/div[1]/div/div').text
        
        if theTime in self.theDict['Time'] and theText in self.theDict['Post']:
            return None
        
        self.theDict['User'].append(User)
        self.theDict['Time'].append(f'{theTime}')
        self.theDict['Likes'].append(like)
        self.theDict['Comments'].append(comment)
        self.theDict['Shares'].append(share)
        self.theDict['Post'].append(theText)
    
    def startScrape(self, search: str, startDate: str = "", endDate: str = "", recent: bool = True, savepoint = False, customDayList: List["str"] = None):
        '''
        PARAMS
        - search: str - The search query
        - startDate: str - The start date of the search
        `YYYY-MM-DD`
        - endDate: str - The end date of the search
        `YYYY-MM-DD`
        - recent: bool - If it's True, it will only scrape the recent post
        - savepoint: bool - If it's True, it will save the progress every day
        - customDayList: List[str] - If you want to scrape specific date, you can input the list of the date
        '''
        
        # Error handling, check parameters
        if startDate == "" and endDate == "" and customDayList == None:
            raise ValueError("Please input the start date and end date or custom day list")
        if startDate != "" and endDate != "" and customDayList != None or startDate != "" and endDate == "" and customDayList != None or startDate == "" and endDate != "" and customDayList != None:
            raise ValueError("Please input either start date and end date or custom day list")
        if startDate != "" and endDate == "" and customDayList == None:
            raise ValueError("Please input the end date")
        if startDate == "" and endDate != "" and customDayList == None:
            raise ValueError("Please input the start date")
        if startDate != "" and endDate != "" and customDayList == None:
            dayLists = generateDailyBackwards(startDate, endDate)
        elif startDate == "" and endDate == "" and customDayList != None:
            dayLists = customDayList
            
        with tqdm(dayLists, ncols=100) as pbar:
            for day in pbar:
                self.theDict = {"User": [], "Post": [], "Time": [],
                    "Likes": [], "Comments": [], "Shares": [] }
                pbar.set_description(f"Current Date: {day}")
                
                self.searchQuery(search, day, day, recent)
                index = 0
                reachesEnd = False
                while True:
                    # Break the loop if it reaches the end
                    if reachesEnd:
                        if savepoint:
                            self.savepoint(day)
                        self.theEntireDataset = pd.concat([self.theEntireDataset, pd.DataFrame(self.theDict)])
                        break
                    # This will check if there's a post on this specific date.
                    try:
                        feed = self.driver.find_element(By.XPATH, '//div[@role="feed"]')
                        posts = WebDriverWait(feed, 3).until(EC.presence_of_all_elements_located((By.XPATH, "./div")))
                    except:
                        self.theEntireDataset = pd.concat([self.theEntireDataset, pd.DataFrame(self.theDict)])
                        self.savepoint(day)
                        break
                    
                    for i in range(index, len(posts)):
                        try:
                            # This is necessary, we need to know how many posts are there every loop. If not found, break and reaches the end
                            feed = self.driver.find_element(By.XPATH, '//div[@role="feed"]')
                            posts = WebDriverWait(feed, 3).until(EC.presence_of_all_elements_located((By.XPATH, "./div")))
                            print(f"Current Index: {index}/{len(posts)}")
                            # Get the post
                            if customDayList != None:
                                self.post = posts[i]
                            else:
                                self.post = posts[i+1]
                        except:
                            reachesEnd = True
                            break
                        
                        # Scroll to the post
                        self.driver.execute_script("arguments[0].scrollIntoView();", self.post)
                        self.driver.execute_script("window.scrollBy(0, -100);")
                        time.sleep(5)

                        #Just gonna ignore if it's and ad or the post is error or something
                        try:
                            # Check whether it's a article or a reel.
                            try:
                                self.reelHandling()
                            except:
                                # If it's an article
                                self.articleHandling()
                        except:
                            pass
                        index+=1
                        # This to adress an issue, where sometimes for loop doesn't break when it reaches the end
                        if index == len(posts):
                            reachesEnd = True
                            break
                        
        self.theEntireDataset.to_csv(f'{search}_{startDate}_{endDate}.csv', index=False)
        
    def savepoint(self,currentDate):
        '''
        This will save the current progress
        '''
        df = pd.DataFrame(self.theDict)
        os.makedirs('savepoints', exist_ok=True)
        df.to_csv(f'savepoints/{currentDate}.csv', index=False)

if __name__ == "__main__":
    session = facebookScraper()
    session.startAndLogin()
    session.startScrape('terimakasihjokowi', '2024-1-1', '2024-10-29', recent=True, savepoint=True)