Як скрапити дані з Twitter за допомогою скрипта на Python

24.09.2024

Коментарі: 0

Зміст статті:

Крок 1: Імпорт необхідних бібліотек і пакетів
Крок 2: Інтеграція проксі
Крок 3: Авторизація в X/Twitter
Крок 4: Витяг результатів
Крок 5: Організація даних

Запис у файл JSON
Розбиття на сторінки

Повна версія коду
Фінальний результат

Створення скрипта для скрапінгу даних з Twitter на Python актуально для збору відгуків, аналізу обговорюваності тем, пов'язаних з бізнесом, та інших маркетингових досліджень. Автоматизація повторюваних рутинних процесів дає змогу швидко й ефективно зібрати необхідні дані та проаналізувати їх.

Крок 1: Імпорт необхідних бібліотек і пакетів

Насамперед потрібно встановити два пакети, для чого виконується наступна команда в інтерфейсі командного рядка (CLI):

pip install selenium-wire selenium undetected-chromedriver

Після завершення встановлення необхідно імпортувати ці пакети у файл Python, як показано нижче.

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: розширення Selenium, яке дає змогу налаштувати проксі, що критично для уникнення блокувань під час скрапінгу.
Selenium: надає інструменти для скрапінгу даних, включно з ActionChains і "Keys" для імітації дій у браузері, "By" для пошуку елементів на сторінці, "WebDriverWait" і "expected_conditions" для перевірки виконання певних умов перед виконанням дій.
Undetected Chromedriver: модифікований ChromeDriver для Selenium Wire, який допомагає уникнути механізмів виявлення ботів на веб-сайтах, тим самим мінімізуючи ризик блокування.
time, random, json: вбудовані пакети Python, які використовуються для управління часом виконання операцій і запису даних у формат JSON.

Крок 2: Інтеграція проксі

Використання проксі під час скрапінгу даних з Twitter є критично важливим, оскільки платформа не схвалює скрапінг і може блокувати доступ.

Щоб уникнути цієї проблеми, необхідно вказати дані проксі. Запуск браузера в режимі headless дає змогу працювати з ним без графічного інтерфейсу, що сприяє прискоренню процесу скрапінгу. Тому ми додали опцію --headless.

# Вкажіть адресу сервера проксі з ім'ям користувача та паролем у списку проксі
proxies = [
    "proxy_username:proxy_password@proxy_address:port_number",
]




# функція для отримання випадкового проксі
def get_proxy():
    return random.choice(proxies)


# Налаштування опцій Chrome з проксі та аутентифікацією
chrome_options = Options()
chrome_options.add_argument("--headless")


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

Крок 3: Авторизація в X/Twitter

Для успішного скрапінгу даних з Twitter, скрипт повинен мати доступ до акаунту. Для цього потрібно надати ім'я користувача та пароль від Twitter.

Крім того, необхідно надати ключове слово пошуку. Команда https://twitter.com/search?q={search_keyword}&src=typed_query&f=top використовується у скрипті для створення URL, який дає змогу виконати пошук за заданим ключовим словом на платформі X.

Після цього створюється екземпляр ChromeDriver, який приймає деталі проксі як опцію. Це повідомляє ChromeDriver, яку IP-адресу використовувати під час завантаження сторінки. Нарешті, ми завантажуємо створений URL пошуку, використовуючи всі ці конфігурації. Як тільки сторінка завантажена, ми повинні спочатку увійти в систему, перш ніж ми зможемо побачити результати пошуку. Використовуючи WebDriverWait, ми гарантуємо повне завантаження сторінки, перевіряючи наявність області для введення нашого імені користувача. Якщо ця область не завантажена, рекомендується закрити екземпляр.

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"


# вкажіть тут ваше ім'я користувача та пароль від X/Twitter
x_username = "" 
x_password = ""


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


# Створюємо екземпляр WebDriver із драйвером Chrome
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()


# Вхід у систему
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)

Крок 4: Витяг результатів

Створіть змінну списку results, щоб систематично зберігати всі отримані дані у форматі словників. Після цього створіть функцію під назвою scrape(), яка буде систематично збирати великі дані для кожного твіта. Ці дані включають відображуване ім'я, ім'я користувача, зміст поста і метрики, такі як лайки і покази.

Використання функції min() гарантує, що довжина кожного списку відповідає іншим. Дотримуючись цієї методології, ми забезпечуємо синхронізований і структурований підхід до збору та обробки даних Twitter.

Коли ми скрапуємо метрики, вони повертаються як рядки, а не як числа. Потім нам потрібно перетворити рядки в числа за допомогою функції convert_to_numeric(), щоб результати можна було організувати за показниками.

results = []


# Збір даних
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):
   # Сортуємо JSON список на місці за полем 'Impressions' у порядку убування
   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:
       # Обробка випадку, коли перетворення не вдалося
       return None

Крок 5: Організація даних

Для кращої організації даних ми створили функцію, яка приймає результати і сортує твіти в порядку убування, використовуючи кількість показів, зібраних кожним твітом. Таким чином, спочатку ми бачитимемо твіт із найбільшим числом показів.

def reorder_json_by_impressions(json_data):
    # Сортуємо JSON список на місці за полем 'Impressions' у порядку убування
    json_data.sort(key=lambda x:int(x['Impressions']), reverse=True)

Запис у файл JSON

Файл JSON надає всі зібрані дані в найоптимальнішому форматі візуалізації.

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}")

Розбиття на сторінки

Для початку виконання коду нам потрібно послідовно викликати наші функції, щоб почати збір даних. Ми створюємо посилання, використовуючи модуль ActionChains у Selenium, для виконання різних дій. Цей модуль особливо важливий для імітації прокрутки сторінки вниз.

Перший етап включає збір даних з уже завантаженої сторінки. Потім запускається цикл, що повторюється п'ять разів, під час якого сторінка прокручується вниз, після чого відбувається пауза на п'ять секунд перед наступною ітерацією скрапінгу.

Користувачі можуть налаштувати діапазон циклу, збільшуючи або зменшуючи його, щоб контролювати обсяг зібраних даних. Важливо враховувати, що якщо немає додаткового контенту для відображення, скрипт збиратиме одні й ті самі дані. Щоб уникнути цього, налаштуйте діапазон циклу відповідним чином, щоб запобігти запису надлишкових даних.

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()

Повна версія коду

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"

# Вкажіть тут ваше ім'я користувача та пароль від X/Twitter
x_username = ""
x_password = ""

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

# Вкажіть адресу сервера проксі з логіном і паролем у списку проксі
proxies = [
   "USERNAME:PASSWORD@IP:PORT",
]


# Функція для отримання проксі випадковим чином
def get_proxy():
   return random.choice(proxies)


# Налаштування опцій Chrome з проксі та аутентифікацією
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}",
   }
}

# Створюємо екземпляр WebDriver із драйвером Chrome
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()

# Вхід у систему
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 = []


# Збір даних
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):
   # Сортуємо JSON список на місці за полем 'Impressions' у порядку убування
   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:
       # Обробка випадку, коли перетворення не вдалося
       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()

Фінальний результат

Ось який вигляд має мати JSON-файл після завершення скрапінгу:

[
  {
    "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
  }
]

Розглянутий посібник можна використовувати для скрапінгу даних у темах, що цікавлять, і подальшого їх вивчення з метою аналізу настроїв громадськості, відстеження трендів, моніторингу та управління репутацією. Використання Python, своєю чергою, полегшує процес автоматичного збирання даних завдяки великій кількості вбудованих модулів і функцій, необхідних для налаштування проксі, прокрутки сторінок і правильної організації отриманої інформації.

Коментарії:

0 Коментаріїв

Попередня стаття

Наступна стаття