Tạo tập lệnh Python để quét dữ liệu Twitter thực sự hữu ích để thu thập những hiểu biết, chẳng hạn như đánh giá hoặc thảo luận của người dùng xung quanh các chủ đề cụ thể, có thể hỗ trợ rất nhiều trong tiếp thị và nghiên cứu. Tự động hóa bằng cách sử dụng các tập lệnh như vậy hợp lý hóa quá trình thu thập, làm cho nó nhanh chóng và hiệu quả.
Có 2 gói bạn phải cài đặt trước khi bắt đầu viết mã thực tế. Bạn cũng cần một trình quản lý gói cho các gói Python (PIP) để cài đặt các gói này. May mắn thay, một khi bạn cài đặt Python trên máy, PIP cũng được cài đặt. Để cài đặt các gói này, bạn chỉ cần chạy lệnh bên dưới trong giao diện dòng lệnh (CLI).
PHP
Node.js
Python
Java
Golang
vn 
English 
Español 
中國人 
Deutsch 
Українська 
Português 
Français 
भारतीय 
Türkçe 
한국인 
Italiano 
Gaeilge 
اردو 
Indonesia 
Polski pip install selenium-wire selenium undetected-chromedriverSau khi cài đặt hoàn tất, bạn phải nhập các gói này vào tệp Python của bạn như được hiển thị bên dưới.
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
Nó đã được thiết lập nghiêm trọng rằng sử dụng proxy trong quá trình cạo là rất quan trọng. Twitter là một trong những nền tảng truyền thông xã hội cau mày khi lấy dữ liệu và an toàn và tránh lệnh cấm, bạn nên sử dụng proxy.
Tất cả những gì bạn phải làm là cung cấp địa chỉ proxy, tên người dùng và mật khẩu proxy và IP của bạn bây giờ nên được che dấu và bảo vệ. Chạy trình duyệt không đầu, về cơ bản giống như chạy trình duyệt mà không có giao diện, giúp tăng tốc quá trình cạo, đó là lý do tại sao chúng tôi đã thêm cờ không đầu vào các tùy chọn.
# Chỉ định địa chỉ máy chủ proxy với tên người dùng và mật khẩu trong danh sách proxy
proxies = [
"proxy_username:proxy_password@proxy_address:port_number",
]
# chức năng để có được một proxy ngẫu nhiên
def get_proxy():
return random.choice(proxies)
# Thiết lập các tùy chọn Chrome với proxy và xác thực
chrome_options = Options()
chrome_options.add_argument("--headless")
proxy = get_proxy()
proxy_options = {
"proxy": {
"http": f"http://{proxy}",
"https": f"https://{proxy}",
}
}
Để xóa dữ liệu Twitter một cách hiệu quả bằng Python, tập lệnh yêu cầu thông tin đăng nhập truy cập cho tài khoản Twitter, bao gồm tên người dùng và mật khẩu.
Ngoài ra, bạn phải chỉ định một từ khóa tìm kiếm. Script sử dụng lệnh https://twitter.com/search?q={search_keyword}&src=typed_query&f=top để tạo URL cho phép tìm kiếm từ khóa này trên Twitter.
Bước tiếp theo liên quan đến việc tạo một thể hiện của Chromedriver, kết hợp các chi tiết proxy như một tùy chọn. Thiết lập này chỉ đạo Chromedriver sử dụng một địa chỉ IP cụ thể khi tải trang. Sau thiết lập này, URL tìm kiếm được tải với các cấu hình này. Khi trang được tải, bạn phải đăng nhập để truy cập kết quả tìm kiếm. Sử dụng WebDriverWait, tập lệnh xác minh rằng trang được tải đầy đủ bằng cách kiểm tra sự hiện diện của khu vực nhập tên người dùng. Nếu khu vực này không tải, nên chấm dứt thể hiện crômedriver.
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"
# Cung cấp tên người dùng và mật khẩu X/Twitter của bạn tại đây
x_username = ""
x_password = ""
print(f'Opening {constructed_url} in Chrome...')
# Tạo một thể hiện webdriver với trình điều khiển Chrome không bị phát hiện
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()
#Đăng nhập
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)Tạo một biến danh sách, kết quả, để lưu trữ một cách có hệ thống tất cả các dữ liệu được lượm lặt theo định dạng từ điển. Sau đó, thiết lập một hàm có tên Scrape () để thu thập một cách có hệ thống vô số dữ liệu cho mỗi tweet, bao gồm các chi tiết quan trọng như tên hiển thị, tên người dùng, nội dung bài đăng và số liệu như lượt thích và ấn tượng.
Một cách tiếp cận chủ động đã được áp dụng để đảm bảo tính đồng nhất trong độ dài của danh sách. Hàm Min () đảm bảo rằng độ dài của mỗi danh sách phù hợp với các hàm khác. Bằng cách tuân thủ phương pháp này, chúng tôi đảm bảo cách tiếp cận được đồng bộ hóa và có cấu trúc để thu thập và xử lý dữ liệu Twitter.
Khi chúng tôi xóa các số/số liệu phù hợp, chúng được trả về dưới dạng chuỗi không phải là số. Sau đó, chúng ta cần chuyển đổi các chuỗi thành các số bằng convert_to_numeric () để kết quả có thể được tổ chức bằng các ấn tượng.
results = []
# Cạo
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):
# Sắp xếp danh sách JSON tại chỗ dựa trên 'ấn tượng' theo thứ tự giảm dầ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:
# Xử lý trường hợp chuyển đổi không thành công
return NoneĐể tổ chức tốt hơn dữ liệu, chúng tôi đã tạo một chức năng lấy kết quả và sắp xếp các tweet theo thứ tự giảm dần bằng cách sử dụng số lần hiển thị được thu thập bởi mỗi tweet. Về mặt logic, chúng tôi muốn xem tweet với số vanity cao nhất trước tiên trước những người khác.
def reorder_json_by_impressions(json_data):
# Sắp xếp danh sách JSON tại chỗ dựa trên 'ấn tượng' theo thứ tự giảm dần
json_data.sort(key=lambda x:int(x['Impressions']), reverse=True)Tệp JSON là cách tốt nhất để trực quan hóa tất cả các dữ liệu được thu thập. Viết vào tệp JSON cũng giống như viết vào bất kỳ tệp nào khác trong Python. Sự khác biệt duy nhất là chúng tôi cần mô -đun JSON để định dạng dữ liệu đúng trước khi nó được ghi vào tệp.
Nếu mã chạy chính xác, bạn sẽ thấy tệp result.json trong cấu trúc tệp và trong đó sẽ là kết quả như trong phần bên dưới.
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}") Để bắt đầu thực thi mã, chúng ta cần gọi các chức năng của mình một cách tuần tự để bắt đầu quét dữ liệu. Chúng tôi tạo một tài liệu tham khảo bằng mô -đun ActionChains trong Selenium để tạo điều kiện cho các hành động selen khác nhau. Mô -đun này chứng minh then chốt để mô phỏng cuộn xuống trên trang.
Vòng đầu tiên liên quan đến dữ liệu cạo từ trang hiện đang tải. Sau đó, một vòng lặp được bắt đầu, lặp lại năm lần, trong đó trang được cuộn xuống, sau đó là một khoảng dừng năm giây trước khi lần lặp lại lần lặp tiếp theo.
Người dùng có thể điều chỉnh phạm vi của vòng lặp, tăng hoặc giảm nó để tùy chỉnh khối lượng dữ liệu được cạo. Điều quan trọng cần lưu ý là nếu không có nội dung bổ sung để hiển thị, tập lệnh sẽ liên tục xóa cùng một dữ liệu, dẫn đến dự phòng. Để ngăn chặn điều này, điều chỉnh phạm vi vòng lặp phù hợp để tránh ghi dữ liệu dự phòng.
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"
# Cung cấp tên người dùng và mật khẩu X/Twitter của bạn tại đây
x_username = ""
x_password = ""
print(f'Opening {constructed_url} in Chrome...')
# Chỉ định địa chỉ máy chủ proxy với tên người dùng và mật khẩu trong danh sách proxy
proxies = [
"USERNAME:PASSWORD@IP:PORT",
]
# chức năng để có được một proxy ngẫu nhiên
def get_proxy():
return random.choice(proxies)
# Thiết lập các tùy chọn Chrome với proxy và xác thực
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}",
}
}
# Tạo một thể hiện webdriver với trình điều khiển Chrome không bị phát hiện
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()
# Đăng nhập
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 = []
# Cạo
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):
# Sắp xếp danh sách JSON tại chỗ dựa trên 'ấn tượng' theo thứ tự giảm dầ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:
# Xử lý trường hợp chuyển đổi không thành công
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()Ở đây, những gì tệp JSON nên thích sau khi thực hiện cạo:
[
{
"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
}
]
Hướng dẫn được phác thảo có thể được sử dụng để cạo dữ liệu về các chủ đề quan tâm khác nhau, tạo điều kiện cho các nghiên cứu trong phân tích tình cảm công khai, theo dõi xu hướng, giám sát và quản lý danh tiếng. Đến lượt, Python đơn giản hóa quá trình thu thập dữ liệu tự động với mảng các mô-đun và chức năng tích hợp rộng rãi. Các công cụ này rất cần thiết để định cấu hình proxy, quản lý cuộn trang và tổ chức thông tin được thu thập một cách hiệu quả.
Bình luận: 0