파이썬 스크립트를 사용하여 트위터 데이터를 스크랩하는 방법

24.09.2024

댓글: 0

기사 내용:

1단계: 설치 및 가져오기
2단계: 프록시 초기화
3단계: X/Twitter에 로그인하는 방법
4단계: 상위 결과 추출하기
5단계: 데이터 정리

JSON 파일에 쓰기
페이지 매김

전체 코드
최종 결과

트위터 데이터를 스크랩하기 위한 Python 스크립트를 만들면 사용자 리뷰나 특정 주제에 대한 토론과 같은 인사이트를 수집하는 데 유용하며, 이는 마케팅 및 조사에 큰 도움이 될 수 있습니다. 이러한 스크립트를 사용한 자동화는 수집 프로세스를 간소화하여 빠르고 효율적으로 만듭니다.

1단계: 설치 및 가져오기

실제 코드 작성을 시작하기 전에 설치해야 하는 패키지가 두 가지 있습니다. 이 패키지를 설치하려면 Python 패키지용 패키지 관리자(PIP)도 필요합니다. 다행히도 컴퓨터에 Python을 설치하면 PIP도 함께 설치됩니다. 이러한 패키지를 설치하려면 명령줄 인터페이스(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" 같은 도구를 사용하여 데이터 스크래핑을 용이하게 합니다.
탐지되지 않은 크롬 드라이버: 웹사이트의 봇 탐지 메커니즘을 우회하여 차단 위험을 줄이기 위해 Selenium Wire와 함께 사용하도록 크롬 드라이버를 변경합니다.
time, random, json: 작업 타이밍을 관리하고 JSON 형식의 데이터를 처리하기 위한 표준 Python 라이브러리.

2단계: 프록시 초기화

스크래핑 시 프록시를 사용하는 것이 중요하다는 사실은 여러 차례 밝혀진 바 있습니다. 트위터는 데이터 스크래핑을 금지하는 소셜 미디어 플랫폼 중 하나이며, 안전을 지키고 금지 조치를 피하려면 프록시를 사용해야 합니다.

프록시 주소, 프록시 사용자 아이디, 비밀번호만 입력하면 IP가 마스킹되어 보호됩니다. 기본적으로 인터페이스가 없는 브라우저를 실행하는 것과 동일한 헤드리스 브라우저를 실행하면 스크래핑 프로세스 속도가 빨라지므로 옵션에 헤드리스 플래그를 추가했습니다.

# 프록시 목록에서 사용자 아이디와 비밀번호로 프록시 서버 주소를 지정합니다.
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에 로그인하는 방법

Python을 사용하여 트위터 데이터를 효과적으로 스크랩하려면 스크립트에 사용자 아이디와 비밀번호를 포함한 트위터 계정에 대한 액세스 자격 증명이 필요합니다.

또한 검색 키워드를 지정해야 합니다. 이 스크립트는 https://twitter.com/search?q={search_keyword}&src=typed_query&f=top 명령을 사용하여 트위터에서 이 키워드를 검색할 수 있는 URL을 구성합니다.

다음 단계는 프록시 세부정보를 옵션으로 통합하여 ChromeDriver의 인스턴스를 생성하는 것입니다. 이 설정은 페이지를 로드할 때 특정 IP 주소를 사용하도록 ChromeDriver에 지시합니다. 이 설정이 완료되면 검색 URL이 이러한 구성으로 로드됩니다. 페이지가 로드되면 검색 결과에 액세스하려면 로그인해야 합니다. 스크립트는 WebDriverWait을 사용하여 사용자 이름 입력 영역이 있는지 확인하여 페이지가 완전히 로드되었는지 확인합니다. 이 영역이 로드되지 않으면 ChromeDriver 인스턴스를 종료하는 것이 좋습니다.

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 인스턴스 만들기
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를 만듭니다. 그런 다음 스크랩()이라는 함수를 설정하여 표시 이름, 사용자 아이디, 게시물 콘텐츠, 마음에 들어요 및 노출 수와 같은 메트릭 등 중요한 세부 정보를 포함하여 각 트윗에 대한 풍부한 데이터를 체계적으로 수집합니다.

목록 길이의 균일성을 보장하기 위해 사전 예방적 접근 방식이 채택되었습니다. min() 함수는 각 목록의 길이가 다른 목록과 일치하도록 합니다. 이 방법론을 준수함으로써 트위터 데이터를 수집하고 처리하는 데 있어 동기화되고 구조화된 접근 방식을 보장합니다.

허수 번호/메트릭을 스크랩하면 숫자가 아닌 문자열로 반환됩니다. 그런 다음 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 목록을 제자리에서 정렬합니다.
   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 목록을 제자리에서 정렬합니다.
    json_data.sort(key=lambda x:int(x['Impressions']), reverse=True)

JSON 파일에 쓰기

JSON 파일은 수집된 모든 데이터를 시각화하는 가장 좋은 방법입니다. JSON 파일에 쓰는 것은 Python의 다른 파일에 쓰는 것과 똑같습니다. 유일한 차이점은 파일에 쓰기 전에 데이터를 올바르게 포맷하기 위해 JSON 모듈이 필요하다는 것입니다.

코드가 올바르게 실행되었다면 파일 구조에 result.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 모듈을 사용하여 참조를 생성합니다. 이 모듈은 페이지에서 아래로 스크롤을 시뮬레이션하는 데 핵심적인 역할을 합니다.

첫 번째 라운드에서는 현재 로드된 페이지에서 데이터를 스크랩합니다. 이후 페이지가 아래로 스크롤되는 동안 5회 반복되는 루프가 시작되고 다음 스크래핑 반복 전에 5초 동안 일시 중지됩니다.

사용자는 루프의 범위를 늘리거나 줄여 스크랩하는 데이터의 양을 사용자 지정할 수 있습니다. 표시할 추가 콘텐츠가 없는 경우 스크립트가 동일한 데이터를 지속적으로 스크랩하여 중복이 발생할 수 있다는 점에 유의해야 합니다. 이를 방지하려면 루프 범위를 적절히 조정하여 중복 데이터 기록을 방지하세요.

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 인스턴스 만들기
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 목록을 제자리에서 정렬합니다.
   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 댓글

이전 기사