Twitter

Twitter を Python から使うライブラリは Twitter libraries の Python の項にいくつかリストされている。どれがいいかわからないが,とりあえず一番上の python-twitter を試してみる(そのうち TwitterAPItweepytwython も試してみよう)。インストールは

pip install python-twitter

でできる。

まずは Twitter のアプリケーション登録をする。方法は python-twitter's documentation にも書かれているし,日本語の解説もたくさんある(例えば Python3でTwitterAPIを使う方法をどのサイトよりも丁寧に解説する の前半)。

登録をしたら,Consumer Key,Consumer Secret,Access Token,Access Token Secret がもらえる。これらを次のコードの ... の部分にコピペする。

試してみよう:

import twitter

api = twitter.Api(consumer_key='...',
                  consumer_secret='...',
                  access_token_key='...',
                  access_token_secret='...',
                  tweet_mode='extended',
                  sleep_on_rate_limit=True)

api.VerifyCredentials() # 自分が正しく認識できているかチェック

home = api.GetHomeTimeline() # ホームタイムラインを取得(デフォルト20件)

これで home を表示すればだいたいの感じがわかる。このクラスの詳細は ??twitter.models.Status と打ち込めば出てくる。この中の必要な項目を自分好みに出力すればよい。簡単な例を挙げておく:

for t in home:
    if (t.retweeted_status):
        t = t.retweeted_status
    for u in t.urls:
        t.full_text = t.full_text.replace(u.url, u.expanded_url)
    print(t.id, t.created_at, t.user.screen_name, t.full_text)

これを少しいじれば,自家製Twitterクライアントができる。これ以外に,位置情報 t.place,正確な位置情報 t.geo などが得られる。普段はEmacsのtwittering-modeを使っているが,自動処理にはコマンドラインで使えるほうが便利だ。

書き込みはこんな感じ(通常はツイート本文だけでよい):

api.PostUpdate("テスト",  # ツイート本文(これだけ必須)
               media="hoge.jpg",  # 添付画像ファイル
               in_reply_to_status_id=1140050903828819970, # リプライ先のツイートID
               auto_populate_reply_metadata=True, # リプライ先の @... を自動挿入
               attachment_url="https://twitter.com/.../status/...", # 引用リツイート
               latitude=34.7468, longitude=136.5248, # 位置情報
               display_coordinates=True) # 正確な位置情報にするか

引用リツイートは従来のようにツイート本文末尾にURLを書いてもいいが,attachment_url="..." で指定すると文字数カウントに入らない。リプライの先頭に付ける @... も,手で書く代わりに auto_populate_reply_metadata=True にすれば自動で入り,文字数カウントに入らない。

r = api.PostUpdate(...) のように戻り値を取得すれば,例えば r.id でツイートのIDが調べられる。ツイートをつなげていくときに便利。


毎時cronで起動して日毎のホームタイムラインをファイルにするコード例:

#! /usr/local/bin/python3

import twitter
import time
from dateutil.parser import parse
from pytz import timezone

api = twitter.Api(consumer_key='...',
                  consumer_secret='...',
                  access_token_key='...',
                  access_token_secret='...',
                  tweet_mode='extended',
                  sleep_on_rate_limit=True)

try:
    with open("maxid") as f:
        sid = int(f.read())
except:
    sid = 1

home = []
mid = None
while True:
    h = api.GetHomeTimeline(count=200, since_id=sid, max_id=mid)
    home.extend(h)
    if len(h) < 150:
        break
    mid = home[-1].id - 1

if not home:
    exit()

with open("maxid", "w") as f:
    f.write(str(home[0].id))

home.reverse()

with open(time.strftime("%Y%m%d"), "a", newline="\n") as f:
    for t in home:
        if (t.retweeted_status):
            t = t.retweeted_status
        s = t.full_text
        for u in t.urls:
            s = s.replace(u.url, u.expanded_url)
        c = parse(t.created_at).astimezone(timezone('Asia/Tokyo')).strftime("%Y-%m-%d %H:%M:%S")
        print(t.id, c, t.user.screen_name, s.replace("\n", "\r"), sep="\t", file=f)

grepしやすいように1ツイート1行にするために,ツイート中の改行は "\r" に変換している。

稀に h = api.GetHomeTimeline(...) が失敗することがある。ゆっくり何度かやれば成功するみたいなので,この行を次で置き換えてみた:

    cnt = 0
    while cnt < 5:
        time.sleep(30)
        try:
            h = api.GetHomeTimeline(count=200, since_id=sid, max_id=mid)
            cnt = 10
        except:
            h = []
            cnt += 1

検索は python-twitter のドキュメントや,Twitter API の Search TweetsSearch Tweet Guides 参照。例:

import urllib.parse

query = {"q": "ほげ ふが",        # 検索文字列(スペース区切り)
         "result_type": "recent", # 新しいもの順に取得
         "count": 100}            # 最大100

results = []
while True:
    r = api.GetSearch(raw_query=urllib.parse.urlencode(query))
    print('Retrieved', len(r), 'tweets')
    if len(r) == 0:
        break
    results.extend(r)
    query['max_id'] = results[-1].id - 1
    time.sleep(1)

この方法で取得できるものは,検索語によっては,大部分が重複(リツイート)である。

結果をMeCabで形態素解析し,名詞の出現頻度をグラフにしてみよう:

import MeCab
import collections

text = ""
for t in results:
    if (t.retweeted_status):
        t = t.retweeted_status
    s = t.full_text
    for u in t.urls:
        s = s.replace(u.url, ' ')
    text += s + '\n'

mecab = MeCab.Tagger()
nodes = mecab.parseToNode(text)
s = []
while nodes:
    if nodes.feature[:2] == '名詞':
        s.append(nodes.surface)
    nodes = nodes.next

c = collections.Counter(s)

stopwords={"こと","もの","さん","それ","たち","よう","そう",
           "ため","これ","どこ","ほう","とき","みたい","そこ",
           "of","the","RT","うち","あと","こちら","あたり","あら",
           "ところ","わけ","はず","たくさん","ほんと","すべて",
           "ツイッター","Twitter"}

i = 0
mc = {}
for k,v in c.most_common():
    if len(k) > 1 and k not in stopwords:
        print(k, v)
        mc[k] = v
        i += 1
        if i >= 30:
            break

plt.figure(figsize=[5, 5])
plt.barh(range(len(mc),0,-1), mc.values())
plt.yticks(range(len(mc),0,-1), mc.keys())
plt.savefig('hist.png', bbox_inches="tight")

Last modified: