Cara mengikis data Twitter menggunakan skrip python

Membuat skrip Python untuk mengikis data Twitter memang berguna untuk mengumpulkan wawasan, seperti ulasan pengguna atau diskusi seputar topik tertentu, yang dapat sangat membantu dalam pemasaran dan penelitian. Otomatisasi menggunakan skrip semacam itu menyederhanakan proses pengumpulan, membuatnya cepat dan efisien.

Langkah 1: Instalasi dan impor

Ada 2 paket yang harus Anda instal sebelum Anda mulai menulis kode yang sebenarnya. Anda juga membutuhkan manajer paket untuk paket Python (PIP) untuk menginstal paket-paket ini. Untungnya, setelah Anda menginstal Python di mesin Anda, PIP juga sudah terinstal. Untuk menginstal paket-paket ini, Anda hanya perlu menjalankan perintah di bawah ini pada Command Line Interface (CLI).

pip install selenium-wire selenium undetected-chromedriver

Setelah instalasi selesai, Anda harus mengimpor paket-paket ini ke dalam berkas Python Anda seperti yang ditunjukkan di bawah ini.

from seleniumwire import webdriver as wiredriver
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.support.wait import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from selenium.webdriver.common.by import By
from selenium.webdriver.common.keys import Keys
from selenium.webdriver.common.action_chains import ActionChains
from time import sleep
import json
import undetected_chromedriver as uc
import random
Import ssl

• Seleniumwire: meningkatkan Selenium dengan menambahkan kemampuan untuk mengonfigurasi proksi secara langsung, sangat penting untuk menghindari penyumbatan selama aktivitas scraping.
• Selenium: memfasilitasi pengikisan data dengan alat bantu seperti ActionChains dan "Keys" untuk mensimulasikan tindakan peramban, "By" untuk pencarian elemen, "WebDriverWait", dan "expected_conditions" untuk eksekusi berbasis kondisi.
• Undetected Chromedriver: mengubah ChromeDriver untuk digunakan dengan Selenium Wire guna menghindari mekanisme deteksi bot di situs web, sehingga mengurangi risiko pemblokiran.
• time, random, json: pustaka Python standar untuk mengelola waktu operasi dan menangani data dalam format JSON.

Langkah 2: Inisialisasi proxy

Telah ditetapkan beberapa kali bahwa menggunakan proxy selama scraping adalah penting. Twitter adalah salah satu platform media sosial yang tidak setuju dengan scraping data dan agar aman dan terhindar dari pelarangan, Anda harus menggunakan proxy.

Yang harus Anda lakukan adalah memberikan alamat proxy, nama pengguna dan kata sandi proxy Anda dan IP Anda sekarang akan disamarkan dan dilindungi. Menjalankan peramban tanpa kepala, pada dasarnya sama dengan menjalankan peramban tanpa antarmuka, membantu mempercepat proses pengikisan, itulah sebabnya kami menambahkan bendera tanpa kepala pada opsi.

# Tentukan alamat server proxy dengan nama pengguna dan kata sandi dalam Daftar proxy
proxies = [
    "proxy_username:proxy_password@proxy_address:port_number",
]




# berfungsi untuk mendapatkan proxy acak
def get_proxy():
    return random.choice(proxies)


# Mengatur opsi Chrome dengan proxy dan autentikasi
chrome_options = Options()
chrome_options.add_argument("--headless")


proxy = get_proxy()
proxy_options = {
    "proxy": {
        "http": f"http://{proxy}",
        "https": f"https://{proxy}",
    }
}

Langkah 3: Cara Masuk ke X/Twitter

Untuk mengikis data Twitter secara efektif menggunakan Python, skrip memerlukan kredensial akses untuk akun Twitter, termasuk nama pengguna dan kata sandi.

Selain itu, Anda harus menentukan kata kunci pencarian. Skrip menggunakan perintah https://twitter.com/search?q={search_keyword}&src=typed_query&f=top untuk membuat URL yang memungkinkan pencarian kata kunci ini di Twitter.

Langkah selanjutnya adalah membuat instance ChromeDriver, dengan memasukkan detail proxy sebagai opsi. Penyiapan ini mengarahkan ChromeDriver untuk menggunakan alamat IP tertentu saat memuat halaman. Setelah penyiapan ini, URL pencarian dimuat dengan konfigurasi ini. Setelah halaman dimuat, Anda harus masuk untuk mengakses hasil pencarian. Dengan menggunakan WebDriverWait, skrip memverifikasi bahwa halaman telah dimuat sepenuhnya dengan memeriksa keberadaan area entri nama pengguna. Jika area ini gagal dimuat, disarankan untuk menghentikan instans ChromeDriver.

search_keyword = input("What topic on X/Twitter would you like to gather data on?\n").replace(' ', '%20')
constructed_url = f"https://twitter.com/search?q={search_keyword}&src=typed_query&f=top"


# berikan nama pengguna dan kata sandi X/Twitter Anda di sini
x_username = "" 
x_password = ""


print(f'Opening {constructed_url} in Chrome...')


# Membuat instance WebDriver dengan driver chrome yang tidak terdeteksi
driver = uc.Chrome(options=chrome_options, seleniumwire_options=proxy_options)


driver.get(constructed_url)


try:
    element = WebDriverWait(driver, 20).until(
        EC.presence_of_element_located((By.XPATH, "//div[@class='css-175oi2r r-1mmae3n r-1e084wir-13qz1uu']"))
    )
except Exception as err:
    print(f'WebDriver Wait Error: Most likely Network TimeOut: Details\n{err}')
    driver.quit()


#Masuk
if element:
    username_field = driver.find_element(By.XPATH, "//input[@class='r-30o5oe r-1dz5y72 r-13qz1uu r-1niwhzg r-17gur6a r-1yadl64 r-deolkf r-homxoj r-poiln3 r-7cikom r-1ny4l3l r-t60dpp r-fdjqy7']")
    username_field.send_keys(x_username)
    username_field..send_keys(Keys.ENTER)


    password_field = WebDriverWait(driver, 10).until(
        EC.presence_of_element_located((By.XPATH, "//input[@class='r-30o5oe r-1dz5y72 r-13qz1uu r-1niwhzg r-17gur6a r-1yadl64 r-deolkf r-homxoj r-poiln3 r-7cikom r-1ny4l3l r-t60dpp r-fdjqy7']"))
    )
    password_field.send_keys(x_password)
    password_field.send_keys(Keys.ENTER)


    print("Sign In Successful...\n")


    sleep(10)

Langkah 4: Mengekstrak hasil teratas

Buat variabel daftar, hasil, untuk menyimpan semua data yang dikumpulkan secara sistematis dalam format kamus. Setelah itu, buat fungsi bernama scrape() untuk secara sistematis mengumpulkan banyak data untuk setiap tweet, yang mencakup detail penting seperti nama tampilan, nama pengguna, konten posting, dan metrik seperti suka dan tayangan.

Pendekatan proaktif telah diadopsi untuk menjamin keseragaman dalam panjang daftar. Fungsi min() memastikan bahwa panjang setiap daftar selaras dengan yang lain. Dengan mengikuti metodologi ini, kami memastikan pendekatan yang tersinkronisasi dan terstruktur untuk mengumpulkan dan memproses data Twitter.

Ketika kita mengikis angka/metrik vanity, mereka dikembalikan sebagai string, bukan angka. Kemudian, kita perlu mengubah string menjadi angka menggunakan convert_to_numeric() agar hasilnya dapat diatur berdasarkan tayangan.

results = []


# Mengikis
def scrape():
   display_names = driver.find_elements(By.XPATH,
                                        '//*[@class="css-175oi2r r-1wbh5a2 r-dnmrzs r-1ny4l3l r-1awozwy r-18u37iz"]/div[1]/div/a/div/div[1]/span/span')
   usernames = driver.find_elements(By.XPATH,
                                    '//*[@class="css-175oi2r r-1wbh5a2 r-dnmrzs r-1ny4l3l r-1awozwy r-18u37iz"]/div[2]/div/div[1]/a/div/span')
   posts = driver.find_elements(By.XPATH,
                                '//*[@class="css-146c3p1 r-8akbws r-krxsd3 r-dnmrzs r-1udh08x r-bcqeeo r-1ttztb7 r-qvutc0 r-37j5jr r-a023e6 r-rjixqe r-16dba41 r-bnwqim"]/span')
   comments = driver.find_elements(By.XPATH,
                                   '//*[@class="css-175oi2r r-1kbdv8c r-18u37iz r-1wtj0ep r-1ye8kvj r-1s2bzr4"]/div[1]/button/div/div[2]/span/span/span')
   retweets = driver.find_elements(By.XPATH,
                                   '//*[@class="css-175oi2r r-1kbdv8c r-18u37iz r-1wtj0ep r-1ye8kvj r-1s2bzr4"]/div[2]/button/div/div[2]/span/span/span')
   likes = driver.find_elements(By.XPATH,
                                '//*[@class="css-175oi2r r-1kbdv8c r-18u37iz r-1wtj0ep r-1ye8kvj r-1s2bzr4"]/div[3]/button/div/div[2]/span/span/span')
   impressions = driver.find_elements(By.XPATH,
                                      '//*[@class="css-175oi2r r-1kbdv8c r-18u37iz r-1wtj0ep r-1ye8kvj r-1s2bzr4"]/div[4]/a/div/div[2]/span/span/span')

   min_length = min(len(display_names), len(usernames), len(posts), len(comments), len(retweets), len(likes),
                    len(impressions))

   for each in range(min_length):
       results.append({
           'Username': usernames[each].text,
           'displayName': display_names[each].text,
           'Post': posts[each].text.rstrip("Show more"),
           'Comments': 0 if comments[each].text == "" else convert_to_numeric(comments[each].text),
           'Retweets': 0 if retweets[each].text == "" else convert_to_numeric(retweets[each].text),
           'Likes': 0 if likes[each].text == "" else convert_to_numeric(likes[each].text),
           'Impressions': 0 if impressions[each].text == "" else convert_to_numeric(impressions[each].text)
       })


def reorder_json_by_impressions(json_data):
   # Urutkan daftar JSON di tempat berdasarkan 'Tayangan' dalam urutan menurun
   json_data.sort(key=lambda x: int(x['Impressions']), reverse=True)


def organize_write_data(data: dict):
   output = json.dumps(data, indent=2, ensure_ascii=False).encode("ascii", "ignore").decode("utf-8")
   try:
       with open("result.json", 'w', encoding='utf-8') as file:
           file.write(output)
   except Exception as err:
       print(f"Error encountered: {err}")


def convert_to_numeric(value):
   multipliers = {'K': 10 ** 3, 'M': 10 ** 6, 'B': 10 ** 9}

   try:
       if value[-1] in multipliers:
           return int(float(value[:-1]) * multipliers[value[-1]])
       else:
           return int(value)
   except ValueError:
       # Menangani kasus di mana konversi gagal
       return None

Langkah 5: Mengatur data

Untuk mengatur data dengan lebih baik, kami membuat fungsi yang mengambil hasil dan mengurutkan tweet dalam urutan menurun menggunakan jumlah tayangan yang dikumpulkan oleh setiap tweet. Secara logika, kita ingin melihat tweet dengan jumlah tayangan tertinggi terlebih dahulu sebelum yang lain.

def reorder_json_by_impressions(json_data):
    # Urutkan daftar JSON di tempat berdasarkan 'Tayangan' dalam urutan menurun
    json_data.sort(key=lambda x:int(x['Impressions']), reverse=True)

Menulis ke file JSON

File JSON adalah cara terbaik untuk memvisualisasikan semua data yang dikumpulkan. Menulis ke file JSON sama seperti menulis ke file lain di Python. Satu-satunya perbedaan adalah kita membutuhkan modul JSON untuk memformat data dengan benar sebelum ditulis ke file.

Jika kode berjalan dengan benar, Anda akan melihat file result.json dalam struktur file dan di dalamnya akan ada hasil seperti yang ditunjukkan pada bagian di bawah ini.

def organize_write_data(data:dict):
    output = json.dumps(data, indent=2, ensure_ascii=False).encode("ascii", "ignore").decode("utf-8")
    try:
        with open("result.json", 'w', encoding='utf-8') as file:
            file.write(output)
    except Exception as err:
        print(f"Error encountered: {err}") 

Penomoran halaman

Untuk memulai eksekusi kode, kita perlu memanggil fungsi-fungsi kita secara berurutan untuk memulai penggalian data. Kita membuat sebuah referensi menggunakan modul ActionChains di dalam Selenium untuk memfasilitasi berbagai aksi Selenium. Modul ini terbukti sangat penting untuk mensimulasikan pengguliran ke bawah pada halaman.

Putaran pertama melibatkan pengikisan data dari halaman yang sedang dimuat. Selanjutnya, satu putaran dimulai, diulang sebanyak lima kali, di mana halaman digulir ke bawah, diikuti dengan jeda lima detik sebelum pengulangan pengikisan berikutnya.

Pengguna dapat menyesuaikan rentang perulangan, baik menambah atau mengurangi untuk menyesuaikan volume data yang di-scraping. Sangat penting untuk diperhatikan, bahwa jika tidak ada konten tambahan untuk ditampilkan, skrip akan terus-menerus mengikis data yang sama, yang mengakibatkan redundansi. Untuk mencegah hal ini, sesuaikan rentang perulangan yang sesuai untuk menghindari perekaman data yang berlebihan.

actions = ActionChains(driver)
for i in range(5):
    actions.send_keys(Keys.END).perform()
    sleep(5)
    scrape()


reorder_json_by_impressions(results)
organize_write_data(results)


print(f"Scraping Information on {search_keyword} is done.")


driver.quit()

Kode lengkap

from selenium.webdriver.chrome.options import Options
from selenium.webdriver.support.wait import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from selenium.webdriver.common.by import By
from selenium.webdriver.common.keys import Keys
from selenium.webdriver.common.action_chains import ActionChains
from time import sleep
import json
import undetected_chromedriver as uc
import random
import ssl

ssl._create_default_https_context = ssl._create_stdlib_context


search_keyword = input("What topic on X/Twitter would you like to gather data on?\n").replace(' ', '%20')
constructed_url = f"https://twitter.com/search?q={search_keyword}&src=typed_query&f=top"

# berikan nama pengguna dan kata sandi X/Twitter Anda di sini
x_username = ""
x_password = ""

print(f'Opening {constructed_url} in Chrome...')

# Tentukan alamat server proxy dengan nama pengguna dan kata sandi dalam Daftar proxy
proxies = [
   "USERNAME:PASSWORD@IP:PORT",
]


# berfungsi untuk mendapatkan proxy acak
def get_proxy():
   return random.choice(proxies)


# Mengatur opsi Chrome dengan proxy dan autentikasi
chrome_options = Options()
chrome_options.add_argument("--headless")
chrome_options.add_argument('--ignore-certificate-errors')
chrome_options.add_argument('--ignore-ssl-errors')

proxy = get_proxy()
proxy_options = {
   "proxy": {
       "http": f"http://{proxy}",
       "https": f"https://{proxy}",
   }
}

# Membuat instance WebDriver dengan driver chrome yang tidak terdeteksi
driver = uc.Chrome(options=chrome_options, seleniumwire_options=proxy_options)

driver.get(constructed_url)

try:
   element = WebDriverWait(driver, 20).until(
       EC.presence_of_element_located((By.XPATH, "//div[@class='css-175oi2r r-1mmae3n r-1e084wi r-13qz1uu']"))
   )
except Exception as err:
   print(f'WebDriver Wait Error: Most likely Network TimeOut: Details\n{err}')
   driver.quit()

# Masuk
if element:
   username_field = driver.find_element(By.XPATH,
                                        "//input[@class='r-30o5oe r-1dz5y72 r-13qz1uu r-1niwhzg r-17gur6a r-1yadl64 r-deolkf r-homxoj r-poiln3 r-7cikom r-1ny4l3l r-t60dpp r-fdjqy7']")
   username_field.send_keys(x_username)
   username_field.send_keys(Keys.ENTER)

   password_field = WebDriverWait(driver, 10).until(
       EC.presence_of_element_located((By.XPATH,
                                       "//input[@class='r-30o5oe r-1dz5y72 r-13qz1uu r-1niwhzg r-17gur6a r-1yadl64 r-deolkf r-homxoj r-poiln3 r-7cikom r-1ny4l3l r-t60dpp r-fdjqy7']"))
   )
   password_field.send_keys(x_password)
   password_field.send_keys(Keys.ENTER)

   print("Sign In Successful...\n")

   sleep(10)

results = []


# Mengikis
def scrape():
   display_names = driver.find_elements(By.XPATH,
                                        '//*[@class="css-175oi2r r-1wbh5a2 r-dnmrzs r-1ny4l3l r-1awozwy r-18u37iz"]/div[1]/div/a/div/div[1]/span/span')
   usernames = driver.find_elements(By.XPATH,
                                    '//*[@class="css-175oi2r r-1wbh5a2 r-dnmrzs r-1ny4l3l r-1awozwy r-18u37iz"]/div[2]/div/div[1]/a/div/span')
   posts = driver.find_elements(By.XPATH,
                                '//*[@class="css-146c3p1 r-8akbws r-krxsd3 r-dnmrzs r-1udh08x r-bcqeeo r-1ttztb7 r-qvutc0 r-37j5jr r-a023e6 r-rjixqe r-16dba41 r-bnwqim"]/span')
   comments = driver.find_elements(By.XPATH,
                                   '//*[@class="css-175oi2r r-1kbdv8c r-18u37iz r-1wtj0ep r-1ye8kvj r-1s2bzr4"]/div[1]/button/div/div[2]/span/span/span')
   retweets = driver.find_elements(By.XPATH,
                                   '//*[@class="css-175oi2r r-1kbdv8c r-18u37iz r-1wtj0ep r-1ye8kvj r-1s2bzr4"]/div[2]/button/div/div[2]/span/span/span')
   likes = driver.find_elements(By.XPATH,
                                '//*[@class="css-175oi2r r-1kbdv8c r-18u37iz r-1wtj0ep r-1ye8kvj r-1s2bzr4"]/div[3]/button/div/div[2]/span/span/span')
   impressions = driver.find_elements(By.XPATH,
                                      '//*[@class="css-175oi2r r-1kbdv8c r-18u37iz r-1wtj0ep r-1ye8kvj r-1s2bzr4"]/div[4]/a/div/div[2]/span/span/span')

   min_length = min(len(display_names), len(usernames), len(posts), len(comments), len(retweets), len(likes),
                    len(impressions))

   for each in range(min_length):
       results.append({
           'Username': usernames[each].text,
           'displayName': display_names[each].text,
           'Post': posts[each].text.rstrip("Show more"),
           'Comments': 0 if comments[each].text == "" else convert_to_numeric(comments[each].text),
           'Retweets': 0 if retweets[each].text == "" else convert_to_numeric(retweets[each].text),
           'Likes': 0 if likes[each].text == "" else convert_to_numeric(likes[each].text),
           'Impressions': 0 if impressions[each].text == "" else convert_to_numeric(impressions[each].text)
       })


def reorder_json_by_impressions(json_data):
   # Urutkan daftar JSON di tempat berdasarkan 'Tayangan' dalam urutan menurun
   json_data.sort(key=lambda x: int(x['Impressions']), reverse=True)


def organize_write_data(data: dict):
   output = json.dumps(data, indent=2, ensure_ascii=False).encode("ascii", "ignore").decode("utf-8")
   try:
       with open("result.json", 'w', encoding='utf-8') as file:
           file.write(output)
   except Exception as err:
       print(f"Error encountered: {err}")


def convert_to_numeric(value):
   multipliers = {'K': 10 ** 3, 'M': 10 ** 6, 'B': 10 ** 9}

   try:
       if value[-1] in multipliers:
           return int(float(value[:-1]) * multipliers[value[-1]])
       else:
           return int(value)
   except ValueError:
       # Menangani kasus di mana konversi gagal
       return None


actions = ActionChains(driver)
for i in range(5):
   actions.send_keys(Keys.END).perform()
   sleep(5)
   scrape()

reorder_json_by_impressions(results)
organize_write_data(results)

print(f"Scraping Information on {search_keyword} is done.")

driver.quit()

Hasil akhir

Berikut adalah tampilan file JSON setelah pengikisan dilakukan:

[
  {
    "Username": "@LindaEvelyn_N",
    "displayName": "Linda Evelyn Namulindwa",
    "Post": "Still getting used to Ugandan local foods so I had Glovo deliver me a KFC Streetwise Spicy rice meal (2 pcs of chicken & jollof rice at Ugx 18,000)\n\nNot only was it fast but it also accepts all payment methods.\n\n#GlovoDeliversKFC\n#ItsFingerLinkingGood",
    "Comments": 105,
    "Retweets": 148,
    "Likes": 1500,
    "Impressions": 66000
  },
  {
    "Username": "@GymCheff",
    "displayName": "The Gym Chef",
    "Post": "Delicious High Protein KFC Zinger Rice Box!",
    "Comments": 1,
    "Retweets": 68,
    "Likes": 363,
    "Impressions": 49000
  }
]

Panduan yang diuraikan dapat digunakan untuk mengikis data tentang berbagai topik yang menarik, memfasilitasi studi dalam analisis sentimen publik, pelacakan tren, pemantauan, dan manajemen reputasi. Python, pada gilirannya, menyederhanakan proses pengumpulan data otomatis dengan beragam modul dan fungsi bawaannya. Alat-alat ini sangat penting untuk mengonfigurasi proksi, mengelola pengguliran halaman, dan mengatur informasi yang dikumpulkan secara efektif.