Tweet

かつて、Streaming APIで大量のつぶやきをリアルタイムに「保存する方法」を紹介しました（cURL編・Python編）。つぶやきはJSON形式で保存していましたが、そこから何かを見出すためにはまず、JSON形式のデータを処理できなければなりません。ここではその方法を確認します。

Pythonで簡単に処理する方法を、@nokunoさんが紹介しています（Twitter Streaming APIの使い方）。cURLで取得したつぶやきは、1行に1つずつファイルに格納されるので、次のように1行ずつ取り出してJSONデータをパースすればいいようです（ここではtweet['text']、つまりつぶやきの本文だけを取り出しています）。

#!/usr/bin/env python
import sys, json
for line in sys.stdin:
    tweet = json.loads(line)
    if 'text' in tweet:
        print(tweet['text'])

cURL編やPython編でつぶやきを保存したファイルがresult.datだとすると、「python3 parse.py < ressult.dat」などとして本文だけを取り出すことができそうです。

Streaming APIのsample.jsonならこれでもいいのですが、filter.jsonを使うときには、条件に合うつぶやきが途絶えることがあります。そういうときには、30秒ごとにCR LFが送信されてきて、それがファイルに書かれることになるので、上のように1行ずつ処理しようとすると、ValueErrorが発生して処理が途中で止まってしまいます（cURLの場合）。ですから、次のように例外を無視します。

#!/usr/bin/env python
import sys, json
#from pprint import pprint
 
for line in sys.stdin:
    try:
        tweet = json.loads(line)
        #pprint(tweet)
        print(tweet['text'])
    except:
        pass

つぶやかれた日時はtweet['created_at']で取得できるのですが（このあたりのことはpprint()で確認できます）、この値は「Thu Dec 29 04:27:28 +0000 2011」のように標準時なので、日本ではちょっと使いにくいです。そこで、日本時間の年月日時分秒（20111231000000）のような表現に変換する関数を用意します。

#!/usr/bin/env python
import sys, json, time, calendar
#from pprint import pprint

def YmdHMS(created_at):
    time_utc = time.strptime(created_at, '%a %b %d %H:%M:%S +0000 %Y')
    unix_time = calendar.timegm(time_utc)
    time_local = time.localtime(unix_time)
    return int(time.strftime("%Y%m%d%H%M%S", time_local))

for line in sys.stdin:
    try:
        tweet = json.loads(line)
        #pprint(tweet)
        print(YmdHMS(tweet['created_at']), tweet['text'])
    except:
        pass

リツイートを除外したいときは、次のようにすればいいでしょう。

for line in sys.stdin:
    try:
        tweet = json.loads(line)
        #pprint(tweet)
        if 'retweeted_status' not in tweet:
            print(YmdHMS(tweet['created_at']), tweet['text'])
    except:
        pass

cURL編で書いたように、こういうことをしているのは、Ustreamなどで配信された動画の整理のためなので、保存したつぶやきは、時間を指定してとりだせるようにしておきましょう。

以下のようなスクリプト（parse.py）を使えば、「python3 parse.py 20111230150000 20111230174600 < result.dat」などとして、日本時間の2011年12月30日15時から2011年12月30日17時46分までのつぶやきだけを取得できます（日時の指定は省略してもかまいません）。

#!/usr/bin/env python
import sys, json, time, calendar
#from pprint import pprint

def YmdHMS(created_at):
    time_utc = time.strptime(created_at, '%a %b %d %H:%M:%S +0000 %Y')
    unix_time = calendar.timegm(time_utc)
    time_local = time.localtime(unix_time)
    return int(time.strftime("%Y%m%d%H%M%S", time_local))

argv = sys.argv
start_time = 0
end_time = 99999999999999
if 1 < len(argv):
    start_time = int(argv[1])
    end_time = int(argv[2])

for line in sys.stdin:
    try:
        tweet = json.loads(line)
        #pprint(tweet)
        if 'retweeted_status' not in tweet:
            tweet_time = YmdHMS(tweet['created_at'])
            if start_time <= tweet_time and tweet_time <= end_time:
                tweet_sec = tweet_time-start_time
                screen_name = tweet['user']['screen_name']
                text = tweet['text']
                url = "https://twitter.com/#!/%s/status/%s" % (screen_name, tweet['id_str'])
                #print tweet_sec, url, text
                #print text
                t = time.strptime(str(tweet_time), "%Y%m%d%H%M%S")
                print(time.strftime("%H:%M:%S", t), text)
    except:
        pass

どのような形式で出力するかは、後で何に使いたいかで決まります。たとえばyouTube上でキャプションを付けたいなら、YouTubeがサポートする形式（参考）にあわせるといいでしょう。

実際に、「#iwakamiyasumi」というハッシュタグで検索し保存したつぶやきのうち、「年末特番 ジャーナリスト休業直前！上杉隆氏ラストインタビュー」の配信時間中（2011年12月30日15:00から17:45）のものを抜き出すと下のようになります（スクロールバーを表示できない環境では閲覧できないかもしれません）。

Tweet

前提：Twitter Developersでアプリのconsumer_key等を取得しておく必要があります。詳しく知りたい人は「Twitter OAuth」などを調べるといいでしょう。

なぜこんな技術が必要なのかは、cURL編で書きました。

Streaming APIを使うときは、接続が切れてしまったときにいかに接続し直すかというのがキモなわけですが、cURLで簡単に実現する方法がわからなかったので、240秒ごとに強制的に接続し直すという方法を採用しました。この方法には、(1)再接続のときにつぶやきを取りこぼす危険と重複して取得する可能性があること、(2)接続が切れてから再接続を試みるまでに平均で120秒かかる、という問題がありました。

Streaming APIは、つぶやきが無いときには「CR LF」を30秒おきに送信してくるので、これをチェックすれば切断を検出できるのですが、検出と再接続の機能を自分で実装するのは大変なので、ライブラリに頼りましょう。

ここでは、PythonでTwitterを利用するためのライブラリ、Tweepyを使います。Ubuntuなら次のようにインストールできます。

sudo apt-get update
sudo apt-get install python3-setuptools python3-pip
sudo easy_install3 tweepy

CentOSなら、suで管理者になってから、

yum install python-setuptools
easy_install tweepy

Tweepyは切断の検出と再接続の機能を持っているので、Streaming APIで受け取ったデータを出力するプログラム（stream.py）は、サンプルを参考にして次のように簡単に書けます。StreamListenerで、つぶやきに対応するon_status()ではなくon_data()をオーバーライドしているのは、データをそのまま（JSON形式で）出力したいからです。とりあえず全部保存しておいて、必要なものをあとで取り出すというわけです。JSON以外のものがdataとして来る場合もあるので、最初が「{」のものだけをJSONデータと見なして出力するようにしています（ちょっとかっこわるい）。

# -*- coding: utf-8 -*-

from tweepy.streaming import StreamListener
from tweepy import OAuthHandler
from tweepy import Stream

consumer_key = ""#引用符の中にconsumer_keyの情報を記述する
consumer_secret = ""#引用符の中にconsumer_secretの情報を記述する

access_token = ""#引用符の中にaccess_tokenの情報を記述する
access_token_secret = ""#引用符の中にaccess_token_secretの情報を記述する

class StdOutListener(StreamListener):
    def on_data(self, data):
        if data.startswith("{"):
            print(data, flush=True)
        return True

    def on_error(self, status):
        print(status, flush=True)

if __name__ == '__main__':
    l = StdOutListener()
    auth = OAuthHandler(consumer_key, consumer_secret)
    auth.set_access_token(access_token, access_token_secret)

    stream = Stream(auth, l)
    stream.filter(track = [keyword])#検索する場合
#    stream.sample()#ツイートのランダムサンプリングを取得する場合
#    stream.userstream()#タイムラインを取得する場合

consumer_keyとconsumer_secret、access_token、access_token_secret、つぶやきを絞り込むためのキーワード（例：「track = ["UK USA"]」、「track = ["日本"]」。OR条件はカンマ、AND条件はスペースで連結）を入力して、「python3 stream.py」で実行します。うまく動いていたら，「python3 stream.py >> result.dat」などとして結果をresult.datに保存します。

「せっかくPython使っているんだから、この場でパースして・・・」と思うかもしれませんが、後で何が必要になるかわからないので、とりあえず生データを保存しておくのがいいでしょう（リアルタイムでアウトプットしたいという場合は別です）。ファイルではなくデータベースに保存してもいいのですが、話が複雑になるのでやめておきます（データベースを使う場合でも、後で何が必要になるかわからないので、全データを保存しておくのがいいでしょう）。

再接続できるからといって、通信の不安定な場所での利用はおすすめできません。通信の安定した環境に置いたサーバで動かしっぱなしにするのがいいでしょう。何らかの原因でプロセス自体が落ちたときのために、以下のようなスクリプトを使うといいかもしれません（通常の切断ではプロセスは落ちません）。

#!/usr/bin/env bash

while :
do
  python3 stream.py >> result.dat
  sleep 240
done

Streaming APIで取得したつぶやきの処理方法