Twitter API v2を使って特定アカウントのツイートを取得

Twitter API v2の機能の中にUser Tweet timelineというものがあり、これによって任意のアカウントのタイムラインを取得できる。

これを利用して、コウメ太夫氏のTwitterアカウントから「#まいにちチクショー」を取得する。
以下に示す「コウメ太夫百人一首」を作ることが目的。

コウメ太夫百人一首リリース

コウメ太夫百人一首「コウメ太夫百人一首」とはコウメ太夫氏の「#まいにちチクショー」によって作られた百人一首を楽しめるサービス。コウメ太夫百人一首「コウメ太夫百人…

1. 開発者登録・環境構築など
2. コウメ太夫氏の全ツイート取得
- 2.1. ユーザーID取得
- 2.2. ツイート取得
3. まいにちチクショーの抽出
- 3.1. 必要なチクショー
- 3.2. koume_perser
4. 成果物
5. Twitter API リンク

1. 開発者登録・環境構築など

このあたりの作業は以下の記事通りにされている前提。

Twitter API v2の使い方1 (Twitterアカウント作成からツイート取得まで)

Twitterアカウントの作成からAPIを使用してツイート取得するまで。Twitter APIというものを使うとTwitter上の情報を集めたりできて便利。すご…

2. コウメ太夫氏の全ツイート取得

2.1. ユーザーID取得

Twitterには名前(name)、ユーザー名(username)に加えてプロフィールに表示されないユーザーID(id)が存在する。
この後の処理でユーザーID(id)が必要になるため事前に取得する。

種類	型	説明	例
id	string	ユーザーのID	4790312834
name	string	ユーザーの名前	KEN
username	string	ユーザーの名前(＠の後のやつ)	CircleKENdotNet

これについては以前の記事でも使用したUser lookupを用いる。

以前の記事↓

Twitter API v2の使い方4[ユーザー情報取得編]

Twitter API v2の使い方Twitter API v2は大きく分けると2つ、ツイートユーザー2つの情報が取得できる。以前はツイート検索をしたかったため…

コウメ太夫氏のusernameは@dayukoumeなのでusernames = ["dayukoume"]とする。

import requests
import os
import csv

##### FUNCTIONS
# usernames：ユーザー名　user_fields:取得データ　APIリクエスト用のURLを作成
def create_url(usernames, user_fields):
    if(any(usernames)):
        formatted_user_names = "usernames=" + ",".join(usernames)
    else:
        formatted_user_names = ""

    if(any(user_fields)):
        formatted_user_fields = "user.fields=" + ",".join(user_fields)
    else:
        formatted_user_fields = "user.fields=id,name,username"

    url = "https://api.twitter.com/2/users/by?{}&{}".format(formatted_user_names, formatted_user_fields)
    return url


# bearer_token：APIキー　APIリクエスト用のヘッダーを作成
def create_headers(bearer_token):
    headers = {"Authorization": "Bearer {}".format(bearer_token)}
    return headers

# url：APIリクエスト用のURL headers:APIリクエスト用のヘッダー APIリクエスト
def connect_to_endpoint(url, headers):
    response = requests.request("GET", url, headers=headers)
    print(response.status_code)
    if response.status_code != 200: # HTTPレスポンスステータスコードが200以外ならエラー
        raise Exception("Request returned an error: {} {}".format(response.status_code, response.text))
    return response.json()

# filename：出力ファイル名 header:ヘッダー contents_list:中身
def save_csvfile(filename, header, contents_list): 
    if(not os.path.isfile(filename)): # 既存の同名ファイルがなければ新規作成
        with open(filename, "w", newline="", encoding="utf-16") as f:
            w = csv.DictWriter(f, fieldnames=header, dialect="excel-tab", quoting=csv.QUOTE_ALL)
            w.writeheader()

    with open(filename, "a", newline="", encoding="utf-16") as f:
        w = csv.DictWriter(f, fieldnames=header, dialect="excel-tab", quoting=csv.QUOTE_ALL)
        for contents in contents_list:
            for k in contents.keys(): #### 文字列内のNullを除去
                if(type(contents[k]) is str): contents[k] = contents[k].replace("\0", "")
            w.writerow(contents)

#### MAIN
def main():
    #### VARIABLES
    # APIキー https://developer.twitter.com から取得
    BEARER_TOKEN = r"取得したBEARER_TOKEN"  
    # ユーザー名  Twitterアカウントの　@の後
    usernames = ["dayukoume"]

    # 取得データ  e.g. user_fields = ["id", "name", "username", "created_at", "description"]
    user_fields = ["id", "name", "username", "created_at","protected",
     "withheld", "location", "url", "description", "verified", "entities",
      "profile_image_url", "public_metrics", "pinned_tweet_id"]
    #user_fields = ["id", "name", "username", "created_at", "description"]
    # 保存するCSVファイルの名前
    save_csv_filename = "コウメtwitter-info.csv"

    #### データ取得
    url = create_url(usernames, user_fields)
    headers = create_headers(BEARER_TOKEN)
    json_response = connect_to_endpoint(url, headers)
    data =  json_response["data"]

    #### 取得データ保存
    save_csvfile(save_csv_filename, user_fields, data)


if __name__ == "__main__":
    main()

するとコウメ太夫氏のid、「391900115」が得られる。

2.2. ツイート取得

Twitter API v2のUser Tweet timeline機能を使って任意のユーザーIDのアカウントのツイートを取得できる。
さきほど入手したコウメ太夫氏のユーザーIDを代入してuser_id = 391900115とする。

欲しいのはツイート本文だけなのでtweet_fields = ["id", "text"]とする。
(idは必須)

他に取得したい情報があれば以下の記事の4. 取得できる属性にいろいろ説明が書いてある。

Twitter API v2の使い方3[Tweet情報取得編](検索オプション・属性のまとめ)

Twitter API v2の使い方Twitter API v2は下の表のようなことができる。機能説明Tweet lookupツイートのID からツイートを検索…

import requests
import os
import csv
import re

#### FUNCTIONS
# user_id：ユーザーID　tweet_fields:取得データ　APIリクエスト用のURLを作成
def create_url(user_id, tm, tweet_fields, next_token, max_results): 
    if(any(tweet_fields)):
        formatted_tweet_fields = "tweet.fields=" + ",".join(tweet_fields)
    else:
        formatted_tweet_fields = ""
    if(next_token is not None):
        formatted_next_token = "&pagination_token=" + (next_token)
    else:
        formatted_next_token = ""
    if(max_results > 100): max_results = 100

    url = "https://api.twitter.com/2/users/{}/{}?max_results={}{}&{}".format(user_id, tm, max_results, formatted_next_token, formatted_tweet_fields)
    return url

# bearer_token：APIキー　APIリクエスト用のヘッダーを作成
def create_headers(bearer_token): 
    headers = {"Authorization": "Bearer {}".format(bearer_token)}
    return headers

# url：APIリクエスト用のURL headers:APIリクエスト用のヘッダー APIリクエスト
def connect_to_endpoint(url, headers):
    response = requests.request("GET", url, headers=headers)
    print(response.status_code)
    if response.status_code != 200: # HTTPレスポンスステータスコードが200以外ならエラー
        raise Exception(response.status_code, response.text)
    return response.json()

# filename：出力ファイル名 header:ヘッダー contents_list:中身
def save_csvfile(filename, header, contents_list): 
    if(not os.path.isfile(filename)): # 既存の同名ファイルがなければ新規作成
        with open(filename, "w", newline="", encoding="utf-16") as f:
            w = csv.DictWriter(f, fieldnames=header, dialect="excel-tab", quoting=csv.QUOTE_ALL)
            w.writeheader()

    with open(filename, "a", newline="", encoding="utf-16") as f:
        w = csv.DictWriter(f, fieldnames=header, dialect="excel-tab", quoting=csv.QUOTE_ALL)
        for contents in contents_list:
            w.writerow(contents)

#### MAIN
def main():
    #### VARIABLES
    # APIキー https://developer.twitter.com から取得
    BEARER_TOKEN = r"取得したBEARER_TOKEN"  
    # ID
    user_id = 391900115
    # ユーザーのツイートは[0] ユーザーに対する@ツイートは[1]
    tm = ["tweets", "mentions"][0]
    # 取得データ  e.g. tweet_fields = ["id", "text", "created_at", "author_id"]
    # created_at(投稿時刻), author_id(アカウントID)などの情報が欲しい場合はtweet_fieldsに書く
    tweet_fields = ["id", "text"]
    # 取得したいデータ数
    max_results = 5000
    
    #### データ取得
    next_token = None
    data_len = 0
    while(True):
        url = create_url(user_id, tm, tweet_fields, next_token, max_results)
        headers = create_headers(BEARER_TOKEN)
        json_response = connect_to_endpoint(url, headers)
        data = json_response["data"]
        meta = json_response["meta"]

        #### 取得データ保存(user_id + [following or followers] + .csv)
        csv_file_name = re.sub(r'[\\/:*?"<>|.]+','',str(user_id) + "_" + tm +"_timeline" ) + ".csv" # ファイルに使えない文字削除
        save_csvfile(csv_file_name, tweet_fields, data)

        #### 終了チェック(欲しい数に達するか全て取り終わったら終了)
        data_len += len(data)
        if(max_results <= data_len): break
        if(not "next_token" in meta): break
        next_token = meta["next_token"]
    pass



if __name__ == "__main__":
    main()

これによって3249件のコウメ太夫氏のツイートが得られた。

3. まいにちチクショーの抽出

コウメ太夫氏のツイートのすべてが「#まいにちチクショー」というわけではない。
したがって得られたツイートの中から必要なものだけを抽出する必要がある。

3.1. 必要なチクショー

コウメ太夫氏の「#まいにちチクショー」を観察すると、例外はあるがおおむね以下の構文によって構成されていることが分かる。

1
2
3

〇〇たら～、
××した～。
チクショー！

ここで、〇〇たら～、部を上の句。
××した～。部を下の句とする。

今回の目的である成果物「コウメ太夫百人一首」では「対応する上の句と下の句のセット」が必要となるため、この構文を利用してあてはまるものを取得する。
必ずしも全ての「#まいにちチクショー」を網羅する必要はないため例外は無視する。

3.2. koume_perser

以下に示すkoume_perser.pyによって「#まいにちチクショー」から「対応する上の句と下の句のセット」を取得した。
その総数は2118件、ゲームに用いるには十分な量だ。

import csv
import json

#### 設定
csv_filepath = "391900115_tweets_timeline01.csv" # 読み込むCSVファイル

#### CSVファイルからテキスト取得
text_list = [] 
with open(csv_filepath, "r", encoding="utf-16", errors="ignore", newline="") as f:
    #reader = csv.reader( (line.replace('\0','') for line in f) )
    for row in csv.DictReader(f, dialect="excel-tab"):
        text_list.append(row["text"]) #取得したい列

chikusyo_list = []
for text in text_list:
    tara_id, sita_id = None, None
    lines = text.split()
    for id, line in enumerate(lines):
        if ("たら～、" in line and tara_id is None): tara_id = id
        elif ("した～。" in line and sita_id is None): sita_id = id
    if( tara_id is not None and sita_id is not None ):
        tara = "".join(lines[0:tara_id+1])
        sita = "".join(lines[tara_id+1:sita_id+1])
        chikusyo_list.append([tara, sita])

with open('chikusyo.json', 'w', encoding="utf-8") as fp:
    json.dump({"chikusyo":chikusyo_list}, fp, ensure_ascii=False)