Wie man Twitter-Daten mit einem Python-Skript ausliest

24.09.2024

Bemerkungen: 0

Inhalt des Artikels:

Schritt 1: Installationen und Importe
Schritt 2: Initialisierung des Proxys
Schritt 3: So melden Sie sich bei X/Twitter an
Schritt 4: Extrahieren der besten Ergebnisse
Schritt 5: Organisieren der Daten

Schreiben in eine JSON-Datei
Paginierung

Vollständiger Code
Endgültige Ergebnisse

Die Erstellung eines Python-Skripts für das Scraping von Twitter-Daten ist in der Tat nützlich, um Erkenntnisse zu sammeln, z. B. Nutzerbewertungen oder Diskussionen zu bestimmten Themen, was für Marketing und Forschung sehr hilfreich sein kann. Die Automatisierung mit solchen Skripten rationalisiert den Erfassungsprozess und macht ihn schnell und effizient.

Schritt 1: Installationen und Importe

Es gibt 2 Pakete, die Sie installieren müssen, bevor Sie mit dem Schreiben des eigentlichen Codes beginnen. Außerdem benötigen Sie einen Paketmanager für Python-Pakete (PIP), um diese Pakete zu installieren. Glücklicherweise ist PIP bereits installiert, sobald Sie Python auf Ihrem Rechner installiert haben. Um diese Pakete zu installieren, müssen Sie nur den folgenden Befehl in Ihrer Befehlszeilenschnittstelle (CLI) ausführen.

pip install selenium-wire selenium undetected-chromedriver

Sobald die Installation abgeschlossen ist, müssen Sie diese Pakete wie unten gezeigt in Ihre Python-Datei importieren.

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: Erweitert Selenium um die Möglichkeit, Proxys direkt zu konfigurieren, was für die Vermeidung von Blockaden bei Scraping-Aktivitäten entscheidend ist.
Selenium: erleichtert das Data Scraping mit Tools wie ActionChains und "Keys" zur Simulation von Browser-Aktionen, "By" für die Elementsuche, "WebDriverWait" und "expected_conditions" für die bedingungsbasierte Ausführung.
Undetected Chromedriver: Ändert ChromeDriver für die Verwendung mit Selenium Wire, um Bot-Erkennungsmechanismen auf Websites zu umgehen und so das Blockierungsrisiko zu verringern.
time, random, json: Standard-Python-Bibliotheken für die Verwaltung des Operations-Timings und den Umgang mit Daten im JSON-Format.

Schritt 2: Initialisierung des Proxys

Es wurde bereits mehrfach festgestellt, dass die Verwendung eines Proxys beim Scraping wichtig ist. Twitter ist eine der Social-Media-Plattformen, die Daten-Scraping missbilligt, und um sicher zu gehen und eine Sperre zu vermeiden, sollte man einen Proxy verwenden.

Alles, was Sie tun müssen, ist, Ihre Proxy-Adresse, Ihren Proxy-Benutzernamen und Ihr Passwort anzugeben, und Ihre IP sollte nun maskiert und geschützt sein. Die Verwendung eines Headless-Browsers, also eines Browsers ohne Benutzeroberfläche, beschleunigt den Scraping-Prozess, weshalb wir das Headless-Flag in den Optionen hinzugefügt haben.

# Geben Sie die Adresse des Proxy-Servers mit Benutzernamen und Kennwort in einer Liste von Proxys an
proxies = [
    "proxy_username:proxy_password@proxy_address:port_number",
]




# Funktion, um einen zufälligen Proxy zu erhalten
def get_proxy():
    return random.choice(proxies)


# Einrichten der Chrome-Optionen mit dem Proxy und der Authentifizierung
chrome_options = Options()
chrome_options.add_argument("--headless")


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

Schritt 3: So melden Sie sich bei X/Twitter an

Um Twitter-Daten mit Python effektiv auslesen zu können, benötigt das Skript Zugangsdaten für das Twitter-Konto, einschließlich Benutzername und Passwort.

Außerdem müssen Sie ein Suchwort angeben. Das Skript verwendet den Befehl https://twitter.com/search?q={search_keyword}&src=typed_query&f=top, um eine URL zu konstruieren, die die Suche nach diesem Stichwort auf Twitter ermöglicht.

Der nächste Schritt besteht darin, eine Instanz von ChromeDriver zu erstellen und dabei die Proxy-Details als Option einzubinden. Durch diese Einrichtung wird ChromeDriver angewiesen, beim Laden der Seite eine bestimmte IP-Adresse zu verwenden. Nach dieser Einrichtung wird die Such-URL mit diesen Konfigurationen geladen. Sobald die Seite geladen ist, müssen Sie sich anmelden, um auf die Suchergebnisse zuzugreifen. Mithilfe von WebDriverWait prüft das Skript, ob die Seite vollständig geladen ist, indem es das Vorhandensein des Bereichs für die Eingabe des Benutzernamens kontrolliert. Wenn dieser Bereich nicht geladen wird, ist es ratsam, die ChromeDriver-Instanz zu beenden.

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"


# Geben Sie hier Ihren X/Twitter-Benutzernamen und Ihr Passwort ein
x_username = "" 
x_password = ""


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


# Erstellen einer WebDriver-Instanz mit unerkanntem Chrome-Treiber
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()


#Eintragen
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)

Schritt 4: Extrahieren der besten Ergebnisse

Erstellen Sie eine Listenvariable, results, um systematisch alle gesammelten Daten im Format von Wörterbüchern zu speichern. Anschließend erstellen Sie eine Funktion namens scrape(), um systematisch eine Fülle von Daten für jeden Tweet zu sammeln, die wichtige Details wie den Anzeigenamen, den Benutzernamen, den Inhalt des Beitrags und Metriken wie Likes und Impressionen umfassen.

Es wurde ein proaktiver Ansatz gewählt, um eine einheitliche Länge der Listen zu gewährleisten. Die Funktion min() sorgt dafür, dass die Länge der einzelnen Listen mit den anderen übereinstimmt. Durch die Einhaltung dieser Methodik wird ein synchronisierter und strukturierter Ansatz für die Erfassung und Verarbeitung von Twitter-Daten gewährleistet.

Wenn wir die Vanity-Nummern/Metriken scrapen, werden sie als Strings und nicht als Zahlen zurückgegeben. Dann müssen wir die Strings mit convert_to_numeric() in Zahlen umwandeln, damit das Ergebnis nach Impressionen geordnet werden kann.

results = []


# Kratzen
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):
   # Sortieren Sie die JSON-Liste an Ort und Stelle auf der Grundlage von "Impressionen" in absteigender Reihenfolge
   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:
       # Behandeln Sie den Fall, dass die Konvertierung fehlschlägt
       return None

Schritt 5: Organisieren der Daten

Um die Daten besser zu organisieren, haben wir eine Funktion erstellt, die die Ergebnisse nimmt und die Tweets in absteigender Reihenfolge nach der Anzahl der Eindrücke sortiert, die jeder Tweet gesammelt hat. Logischerweise wollen wir den Tweet mit der höchsten Eitelkeitszahl vor den anderen sehen.

def reorder_json_by_impressions(json_data):
    # Sortieren Sie die JSON-Liste an Ort und Stelle auf der Grundlage von "Impressionen" in absteigender Reihenfolge
    json_data.sort(key=lambda x:int(x['Impressions']), reverse=True)

Schreiben in eine JSON-Datei

Eine JSON-Datei ist die beste Möglichkeit, alle gesammelten Daten zu visualisieren. Das Schreiben in eine JSON-Datei ist genau wie das Schreiben in jede andere Datei in Python. Der einzige Unterschied ist, dass wir das JSON-Modul benötigen, um die Daten richtig zu formatieren, bevor sie in die Datei geschrieben werden.

Wenn der Code korrekt ausgeführt wurde, sollte in der Dateistruktur eine result.json-Datei zu sehen sein, in der sich das Ergebnis befindet, wie im folgenden Abschnitt dargestellt.

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

Paginierung

Um mit der Ausführung des Codes zu beginnen, müssen wir unsere Funktionen nacheinander aufrufen, um mit dem Data Scraping zu beginnen. Wir erstellen eine Referenz mit dem ActionChains-Modul in Selenium, um verschiedene Selenium-Aktionen zu ermöglichen. Dieses Modul erweist sich als entscheidend für die Simulation des Abwärtsscrollens auf der Seite.

In der ersten Runde werden Daten von der aktuell geladenen Seite abgefragt. Anschließend wird eine Schleife gestartet, die fünfmal durchlaufen wird, während der die Seite nach unten gescrollt wird, gefolgt von einer fünfsekündigen Pause vor der nächsten Iteration des Scrapings.

Der Benutzer kann den Bereich der Schleife anpassen, indem er ihn entweder vergrößert oder verkleinert, um das Volumen der gescrapten Daten anzupassen. Es ist wichtig zu beachten, dass das Skript, wenn kein zusätzlicher Inhalt angezeigt werden soll, ständig dieselben Daten abfragt, was zu Redundanz führt. Um dies zu verhindern, passen Sie den Schleifenbereich entsprechend an, um eine redundante Datenerfassung zu vermeiden.

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

Vollständiger Code

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"

# Geben Sie hier Ihren X/Twitter-Benutzernamen und Ihr Passwort ein
x_username = ""
x_password = ""

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

# Geben Sie die Adresse des Proxy-Servers mit Benutzernamen und Kennwort in einer Liste von Proxys an
proxies = [
   "USERNAME:PASSWORD@IP:PORT",
]


# Funktion, um einen zufälligen Proxy zu erhalten
def get_proxy():
   return random.choice(proxies)


# Einrichten der Chrome-Optionen mit dem Proxy und der Authentifizierung
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}",
   }
}

# Erstellen einer WebDriver-Instanz mit unerkanntem Chrome-Treiber
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()

# Eintragen
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 = []


# Kratzen
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):
   # Sortieren Sie die JSON-Liste an Ort und Stelle auf der Grundlage von "Impressionen" in absteigender Reihenfolge
   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:
       # Behandeln Sie den Fall, dass die Konvertierung fehlschlägt
       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()

Endgültige Ergebnisse

So sollte die JSON-Datei aussehen, nachdem das Scraping abgeschlossen ist:

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

Mit Hilfe des Leitfadens können Daten zu verschiedenen Themen von Interesse gesammelt werden, was Studien zur Analyse der öffentlichen Meinung, Trendverfolgung, Überwachung und Reputationsmanagement erleichtert. Python wiederum vereinfacht den Prozess der automatischen Datenerfassung mit seinem umfangreichen Angebot an integrierten Modulen und Funktionen. Diese Tools sind für die Konfiguration von Proxys, die Verwaltung des Seiten-Scrollens und die effektive Organisation der gesammelten Informationen unerlässlich.

Bemerkungen:

0 Bemerkungen

Vorheriger Artikel

Nächster Artikel