Создание скрипта для скрапинга данных с Twitter на Python актуально для сбора отзывов, анализа обсуждаемости тем, связанных с бизнесом, и других маркетинговых исследований. Автоматизация повторяющихся рутинных процессов позволяет быстро и эффективно собрать необходимые данные и проанализировать их.
Прежде всего, нужно установить два пакета, для чего выполняется следующая команда в интерфейсе командной строки (CLI):
PHP
Node.js
Python
Java
Golang
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
Использование прокси при скрапинге данных с 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}",
}
}
Для успешного скрапинга данных с 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)Создайте переменную списка 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Для лучшей организации данных мы создали функцию, которая принимает результаты и сортирует твиты в порядке убывания, используя количество показов, собранных каждым твитом. Таким образом, сначала мы будем видеть твит с наибольшим числом показов.
def reorder_json_by_impressions(json_data):
# Сортируем JSON список на месте по полю 'Impressions' в порядке убывания
json_data.sort(key=lambda x:int(x['Impressions']), reverse=True)Файл 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