Twitter verilerini kazımak için bir Python betiği oluşturmak, pazarlama ve araştırmaya büyük ölçüde yardımcı olabilecek kullanıcı yorumları veya belirli konulardaki tartışmalar gibi içgörüleri toplamak için gerçekten yararlıdır. Bu tür komut dosyalarını kullanan otomasyon, toplama sürecini kolaylaştırarak hızlı ve verimli hale getirir.
Gerçek kodu yazmaya başlamadan önce yüklemeniz gereken 2 paket vardır. Bu paketleri yüklemek için Python paketleri için bir paket yöneticisine (PIP) de ihtiyacınız var. Neyse ki, Python'u makinenize yüklediğinizde PIP de yüklenir. Bu paketleri yüklemek için aşağıdaki komutu Komut Satırı Arayüzünüzde (CLI) çalıştırmanız yeterlidir.
PHP
Node.js
Python
Java
Golang
tr 
English 
Español 
中國人 
Tiếng Việt 
Deutsch 
Українська 
Português 
Français 
भारतीय 
한국인 
Italiano 
Gaeilge 
اردو 
Indonesia 
Polski pip install selenium-wire selenium undetected-chromedriverKurulum tamamlandıktan sonra, bu paketleri aşağıda gösterildiği gibi Python dosyanıza aktarmanız gerekir.
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
Kazıma sırasında bir proxy kullanmanın önemli olduğu birkaç kez tespit edilmiştir. Twitter, veri kazımayı hoş karşılamayan sosyal medya platformlarından biridir ve güvende olmak ve yasaklanmamak için bir proxy kullanmalısınız.
Tek yapmanız gereken proxy adresinizi, proxy kullanıcı adınızı ve şifrenizi girmek ve IP'niz artık maskelenmiş ve korunuyor olmalı. Temelde arayüzü olmayan bir tarayıcı çalıştırmakla aynı olan başsız bir tarayıcı çalıştırmak, kazıma işlemini hızlandırmaya yardımcı olur, bu nedenle seçeneklere başsız bayrağını ekledik.
# Proxy sunucu adresini kullanıcı adı ve parola ile birlikte proxy listesinde belirtin
proxies = [
"proxy_username:proxy_password@proxy_address:port_number",
]
# rastgele bir proxy almak için işlev
def get_proxy():
return random.choice(proxies)
# Proxy ve kimlik doğrulama ile Chrome seçeneklerini ayarlama
chrome_options = Options()
chrome_options.add_argument("--headless")
proxy = get_proxy()
proxy_options = {
"proxy": {
"http": f"http://{proxy}",
"https": f"https://{proxy}",
}
}
Python kullanarak Twitter verilerini etkili bir şekilde kazımak için komut dosyası, kullanıcı adı ve şifre dahil olmak üzere Twitter hesabının erişim kimlik bilgilerine ihtiyaç duyar.
Ek olarak, bir arama anahtar sözcüğü belirtmeniz gerekir. Kod, Twitter'da bu anahtar kelimenin aranmasını sağlayan bir URL oluşturmak için https://twitter.com/search?q={search_keyword}&src=typed_query&f=top komutunu kullanır.
Bir sonraki adım, proxy ayrıntılarını bir seçenek olarak içeren bir ChromeDriver örneği oluşturmayı içerir. Bu kurulum, ChromeDriver'ı sayfayı yüklerken belirli bir IP adresini kullanmaya yönlendirir. Bu kurulumun ardından, arama URL'si bu yapılandırmalarla yüklenir. Sayfa yüklendikten sonra, arama sonuçlarına erişmek için oturum açmanız gerekir. WebDriverWait kullanan kod, kullanıcı adı giriş alanının varlığını kontrol ederek sayfanın tamamen yüklendiğini doğrular. Bu alan yüklenmezse ChromeDriver örneğinin sonlandırılması önerilir.
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 kullanıcı adınızı ve şifrenizi buraya girin
x_username = ""
x_password = ""
print(f'Opening {constructed_url} in Chrome...')
# Algılanmamış chrome sürücüsü ile bir WebDriver örneği oluşturma
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()
#Oturum Aç
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)Toplanan tüm verileri sözlük biçiminde sistematik olarak depolamak için sonuçlar adında bir liste değişkeni oluşturun. Bunun ardından, her tweet için görünen ad, kullanıcı adı, gönderi içeriği ve beğeniler ve gösterimler gibi metrikler gibi önemli ayrıntıları kapsayan zengin verileri sistematik olarak toplamak için scrape() adlı bir işlev oluşturun.
Listelerin uzunluklarında tekdüzeliği garanti etmek için proaktif bir yaklaşım benimsenmiştir. min() işlevi, her listenin uzunluğunun diğerleriyle aynı hizada olmasını sağlar. Bu metodolojiye bağlı kalarak, Twitter verilerinin toplanması ve işlenmesi için senkronize ve yapılandırılmış bir yaklaşım sağlıyoruz.
Kibir numaralarını/metrikleri kazdığımızda, bunlar sayı olarak değil dizeler olarak döndürülür. Ardından, convert_to_numeric() kullanarak dizeleri sayılara dönüştürmemiz gerekir, böylece sonuç gösterimlere göre düzenlenebilir.
results = []
# Kazımak
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 listesini azalan sırada 'Gösterimler' temelinde yerinde sıralayın
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:
# Dönüştürmenin başarısız olduğu durumları ele alın
return NoneVerileri daha iyi düzenlemek için, sonuçları alan ve her tweetin topladığı gösterim sayısını kullanarak tweetleri azalan sırada sıralayan bir işlev oluşturduk. Mantıksal olarak, diğerlerinden önce en yüksek gösteriş sayısına sahip tweet'i görmek istiyoruz.
def reorder_json_by_impressions(json_data):
# JSON listesini azalan sırada 'Gösterimler' temelinde yerinde sıralayın
json_data.sort(key=lambda x:int(x['Impressions']), reverse=True)Bir JSON dosyası, toplanan tüm verileri görselleştirmenin en iyi yoludur. Bir JSON dosyasına yazmak Python'daki herhangi bir dosyaya yazmak gibidir. Tek fark, dosyaya yazılmadan önce verileri düzgün bir şekilde biçimlendirmek için JSON modülüne ihtiyacımız olmasıdır.
Kod doğru çalıştıysa, dosya yapısında bir result.json dosyası görmelisiniz ve içinde aşağıdaki bölümde gösterildiği gibi sonuç olmalıdır.
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}") Kodun yürütülmesine başlamak için, veri kazıma işlemini başlatmak üzere fonksiyonlarımızı sırayla çağırmamız gerekir. Çeşitli Selenium eylemlerini kolaylaştırmak için Selenium içindeki ActionChains modülünü kullanarak bir referans oluşturuyoruz. Bu modül, sayfada aşağı kaydırmayı simüle etmek için çok önemlidir.
İlk tur, o anda yüklü olan sayfadan veri kazımayı içerir. Daha sonra, sayfanın aşağı kaydırıldığı beş kez yinelenen bir döngü başlatılır ve ardından bir sonraki kazıma yinelemesinden önce beş saniyelik bir duraklama yapılır.
Kullanıcılar, kazınan veri hacmini özelleştirmek için döngünün aralığını artırarak veya azaltarak ayarlayabilirler. Görüntülenecek ek içerik yoksa, kodun sürekli olarak aynı verileri kazıyacağını ve bunun da fazlalığa neden olacağını unutmamak çok önemlidir. Bunu önlemek için, gereksiz veri kaydını önlemek amacıyla döngü aralığını buna göre ayarlayın.
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 kullanıcı adınızı ve şifrenizi buraya girin
x_username = ""
x_password = ""
print(f'Opening {constructed_url} in Chrome...')
# Proxy sunucu adresini kullanıcı adı ve parola ile birlikte proxy listesinde belirtin
proxies = [
"USERNAME:PASSWORD@IP:PORT",
]
# rastgele bir proxy almak için işlev
def get_proxy():
return random.choice(proxies)
# Proxy ve kimlik doğrulama ile Chrome seçeneklerini ayarlama
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}",
}
}
# Algılanmamış chrome sürücüsü ile bir WebDriver örneği oluşturma
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()
# Oturum Aç
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 = []
# Kazımak
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 listesini azalan sırada 'Gösterimler' temelinde yerinde sıralayın
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:
# Handle the case where the conversion fails
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()Kazıma işlemi tamamlandıktan sonra JSON dosyası şu şekilde olmalıdır:
[
{
"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
}
]
Ana hatlarıyla belirtilen kılavuz, kamuoyu duyarlılık analizi, trend izleme, izleme ve itibar yönetimi çalışmalarını kolaylaştırarak çeşitli ilgi alanlarına ilişkin verileri kazımak için kullanılabilir. Python ise geniş kapsamlı yerleşik modülleri ve işlevleriyle otomatik veri toplama sürecini basitleştirir. Bu araçlar, proxy'leri yapılandırmak, sayfa kaydırmayı yönetmek ve toplanan bilgileri etkili bir şekilde düzenlemek için gereklidir.
Yorumlar: 0