データ収集のためにTwitterの検索結果をスクレイピングする【Python】

データ収集のためにTwitterの検索結果をスクレイピングする【Python】 プログラミング

Twitterのデータが必要なケースは多いでしょう。
今なら、テキストマイニング(機械学習)が多いかもしれません。
そこまで本格的でなくても、形態素解析してタグクラウドを表示したい場合もあるでしょう。

そのような場合、通常はTwitterのAPIを利用します。
でも、正直言ってこのAPIは使えません。

だから、Twitterをスクレイピングします。
そして、今回はTwitterの検索結果をスクレイピングします。

サンプルコードも最後に載せています。
興味があれば、参考にしてみてください。

本記事の内容

  • Twitterをスクレイピングする理由
  • Twitterの検索結果をスクレイピングする【仕様みたいなもの】
  • Twitterの検索結果をスクレイピングするサンプルコード

まずは、Twitterをスクレイピングする理由から説明していきます。

Twitterをスクレイピングする理由

スクレイピングは違反でも何でもない

まず、最初に言っておきます。
Twitterは、利用規約でスクレイピングを禁止しています。
「でも、そんなの関係ねぇ」と言うことを次の記事で解説しています。

そもそも、スクレイピングが禁止される理由がわかりません。
スクレイピングが問題ではなく、大量アクセスが問題のはずです。

世の中的にそこが勘違いされているように感じます。
だから、私はこのブログ内で何度も言っています。
「スクレイピング自体は悪くない」と。

だから、利用規約に書いてあろうとスクレイピングはやりますよと言うことです。

TwitterのAPIが使えない

冒頭でも書きましたが、TwitterはAPIを提供しています。
無料版(Standard)と有料版(Premium、Enterprize)があります。

そして、無料版には厳しい利用制限があります。
それに直近1週間のデータしか取得できません。
さらに、TwitterのAPI利用申請も本当に面倒です。

最後に私が申請した時は、申請理由をそこそこ長文で書きました。
日本語ではなく、英語です。

Twitterは、マジでAPIを使わせる気はないでしょう。
実際、無料での利用はTwitterからしたら無駄コストですからね。

Twitterの知名度が低いときは、API戦略は必要だったのでしょう。
でも、今のTwitterにとって、無料版のAPIは邪魔者でしかありません。

だから、私はTwitterのAPIを使いません。
でも、Twitterのデータは欲しいです。
そうなると、残された手段はスクレイピングしかありません。

実際のスクレイピングのやりかたは、上記の記事で解説しています。
Twitter社の鉄壁の防御が、垣間見えます。

しかし、Webで公開している以上はスクレイピングは可能なのです。
今後もいたちごっこのようになるでしょうけどね。

Twitterの検索結果をスクレイピングする【仕様みたいなもの】

基本的な仕様は、上記の記事を参考にしてください。
その際は、アカウント毎のツイートをスクレイピングしました。

今回は、検索結果が対象となります。
と言っても、大きな変更はすることなく対応できました。

では、検索画面の仕様を確認します。
ポイントはURLです。

「感情分析」というキーワードで検索するとします。

キーワードを入力してEnter(PC)を押すと、以下のURLに遷移します。
https://twitter.com/search?q=%E6%84%9F%E6%83%85%E5%88%86%E6%9E%90&src=typed_query

q=%E6%84%9F%E6%83%85%E5%88%86%E6%9E%90
これは、「感情分析」をURLエンコードしたモノです。

src=typed_query
これは検索した方法を表現しています。

他には「src=recent_search_click」を確認できました。
これは不要です。

https://twitter.com/search?q=%E6%84%9F%E6%83%85%E5%88%86%E6%9E%90

これだけで「感情分析」で検索した一覧を表示できます。
上記URLに実際にアクセスすればわかりますが、「話題のツイート」の検索結果です。

では、ここでTwitterの検索の種類を確認しましょう。
Twitterの検索には、以下の種類があります。

  • 話題のツイート
  • 最新
  • アカウント
  • 画像
  • 動画

そして、それぞれにはフラグ(検索種別)が用意されています。

話題のツイート
最新live
アカウントuser
画像image
動画video

「最新」で検索した場合のURLは以下のようになります。
https://twitter.com/search?q=%E6%84%9F%E6%83%85%E5%88%86%E6%9E%90&f=live

f=liveが付加されています。
「アカウント」なら、f=userとなります。
「話題のツイート」は、何も付加されません。

これ以外は、ほぼ過去記事の内容と同じです。
若干、改善したり、不具合を修正しています。

Twitterの検索結果をスクレイピングするサンプルコード

以下の記事でコードの詳細を説明しています。

2020年08月04日時点では、元気に動いています。

from selenium import webdriver
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from selenium.webdriver.common.by import By
from selenium.webdriver.common.action_chains import ActionChains
from bs4 import BeautifulSoup
import time
import pandas as pd
import urllib.parse


search_keyword = "感情分析"
chromedriver = "chromedriver.exeのパス"
file_path = "./twitter_data/" + search_keyword + ".json"
scroll_count = 10
scroll_wait_time = 2

headers = {
    'User-Agent': 'Mozilla/5.0 (Windows NT 6.1; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/84.0.4147.105 Safari/537.36'
}

# tweet
def get_tweet(keyword):
    
    # urlencode
    keyword = urllib.parse.quote(keyword)
    
    url = 'https://twitter.com/search?q=' + keyword + '&f=live'
    
    id_list = []
    tweet_list = []
    # ヘッドレスモードでブラウザを起動
    options = Options()
    options.add_argument('--headless')
    
    # ブラウザーを起動
    driver = webdriver.Chrome(chromedriver, options=options)
    driver.get(url)

    # articleタグが読み込まれるまで待機(最大15秒)
    WebDriverWait(driver, 15).until(EC.visibility_of_element_located((By.TAG_NAME, 'article')))
    
    # 指定回数スクロール
    for i in range(scroll_count):
        id_list, tweet_list = get_article(url, id_list, tweet_list, driver)
        
        # スクロール=ページ移動
        scroll_to_elem(driver)
        
        # ○秒間待つ(サイトに負荷を与えないと同時にコンテンツの読み込み待ち)
        time.sleep(scroll_wait_time)  
    
    # ブラウザ停止
    driver.quit()
    
    return tweet_list

def scroll_to_elem(driver):
    
    # 最後の要素の一つ前までスクロール
    elems_article = driver.find_elements_by_tag_name('article')
    last_elem = elems_article[-2]
    
    actions = ActionChains(driver);
    actions.move_to_element(last_elem);
    actions.perform();
    
def get_info_of_article(data):
    
    soup = BeautifulSoup(data, features='lxml')
    elems_a = soup.find_all("a")
    
    # 名前
    name_str = elems_a[1].text
    
    id = ''
    name = ''
    if name_str:
        name_list = name_str.split('@')

        id = name_list[-1]
        r_name_str = name_str[::-1]
        r_name = r_name_str.replace(id[::-1] + "@","",1)
        name = r_name[::-1]
        
    # リンク
    link = elems_a[2].get("href")
    
    # 投稿日時
    datetime = elems_a[2].find("time").get("datetime")
    
    # 投稿
    base_elem = soup.find("div", role="group").parent
    
    # 投稿の要素をすべて取得
    tweet = ""
    elems_tweet = base_elem.find_all("div", recursive=False, limit=2)

    counter = 0
    for elem_tweet in elems_tweet:
        counter = counter + 1
        tweet = tweet + elem_tweet.text
        
    info = {}
    info["user_id"] = id
    info["user_name"] = name
    info["link"] = link
    info["datetime"] = datetime
    info["tweet"] = tweet
    
    return info

def get_article(url, id_list, tweet_list, driver):

    elems_article = driver.find_elements_by_tag_name('article')

    for elem_article in elems_article:
        tag = elem_article.get_attribute('innerHTML')
        elems_a = elem_article.find_elements_by_tag_name('a')
        
        try:
            href_tweet = ""

            # tweet
            href_tweet = elems_a[2].get_attribute("href")
            
            if href_tweet in id_list:
                print("重複")
            else:
                # tweet情報取得
                info = get_info_of_article(tag)
                
                id_list.append(href_tweet)
                tweet_list.append(info)
        except:
            print("except")
            #traceback.print_exc()        
        
    return id_list, tweet_list

if __name__ == '__main__':
    # tweet情報をlist型で取得
    tweet_list = get_tweet(search_keyword)
    # データフレームに変換
    df = pd.DataFrame(tweet_list)
    # jsonとして保存
    df.to_json(file_path, orient='records')
タイトルとURLをコピーしました