symfonyからTwitter Streaming APIを使ってつぶやきを保存してみる

TwitterのStreaming APIを利用する機会があったのでメモしておきます。
（symfonyのバージョンは1.4.8、ORMはPropelを用いました）

参考URL:Twtter Streaming API(filter)を使ったサンプルプログラム

上記ブログに記載されていたサンプルプログラムを参考にさせていただいて、symfonyでつぶやきを保存する機能を作ってみました。

Streaming APIについて

Twitter API 仕様書日本語訳第五十版 (2010年8月12日版)によると、現在のところ、誰でも使えるStreamingAPIとしては

①public_timelineを取得し続けるsample
http://stream.twitter.com/1/statuses/sample.format
（format = json or xml）

②フィルターを使って絞り込んだつぶやきを取得できるfilter
http://stream.twitter.com/1/statuses/filter.json
（こちらはjsonのみ）

の2種類が利用できます。

今回はfilterを使って特定のハッシュタグ（#twitter）を含むつぶやきを取得するプログラムを作成しました。

つぶやきを保存するためのTweetモデル作成

まずはschema.ymlにDBのテーブル構造を記述します。

今回はステータスID、本文、ユーザ名、つぶやき時刻を保存してみます。

vi config/schema.yml

propel:
  tweet:
    id: ~
    id_str: { type: varchar(64), index: unique }  #ステータスID
    text: { type: varchar(140) }                      #本文
    screen_name: { type: varchar(15) }                #ユーザ名
    created_at: { type: TIMESTAMP }                   #つぶやき時刻

DB構築＆モデルの作成

symfony propel:build --all --no-confirmation

これでつぶやき保存用のモデルが作成されました。

Streaming APIを利用するタスクを作成

つぶやき取得用のタスクを作成

symfony generate:task tweet:get

lib/task/tweetGetTask.class.phpが作成されます。

参考URL:symfonyからmailを送る

sf_root_dir/configディレクトリにapp.ymlを作成し、twitterに関する設定項目を記述しておきます。

vi config/app.yml
all:
  twitter_id: "TwitterID"
  twitter_pw: "Password"
  twitter_stream_url: "http://stream.twitter.com/1/statuses/filter.json"
  twitter_keyword: "twitter"

作成されたタスクを編集します。

vi lib/task/terrtGetTask.class.php

<?php

class tweetGetTask extends sfBaseTask
{
  protected function configure()
  {
    // // add your own arguments here
    // $this->addArguments(array(
    //   new sfCommandArgument('my_arg', sfCommandArgument::REQUIRED, 'My argument'),
    // ));

    $this->addOptions(array(
      new sfCommandOption('application', null, sfCommandOption::PARAMETER_REQUIRED, 'The application name', 'frontend'),
      new sfCommandOption('env', null, sfCommandOption::PARAMETER_REQUIRED, 'The environment', 'dev'),
      new sfCommandOption('connection', null, sfCommandOption::PARAMETER_REQUIRED, 'The connection name', 'propel'),
      // add your own options here
    ));

    $this->namespace        = 'tweet';
    $this->name             = 'get';
    $this->briefDescription = '';
    $this->detailedDescription = <<<EOF
The [tweet:get|INFO] task does things.
Call it with:

  [php symfony tweet:get|INFO]
EOF;
  }

  protected function execute($arguments = array(), $options = array())
  {
    // initialize the database connection
    $databaseManager = new sfDatabaseManager($this->configuration);
    $connection = $databaseManager->getDatabase($options['connection'])->getConnection();

    // add your code here
    $ctx = stream_context_create(
        array(
            'http' => array(
                'method' => 'POST',
                'header' => "Authorization: Basic " . base64_encode(sfConfig::get("app_twitter_id").':'.sfConfig::get("app_twitter_pw")) . "\r\n" .
                            "Content-type: application/x-www-form-urlencoded\r\n",
                'content' => http_build_query(array('track' => sfConfig::get("app_twitter_keyword")))
            )
        )
    );

    $stream = fopen(sfConfig::get("app_twitter_stream_url"), 'r', false, $ctx);

    while ($json = fgets($stream)) {
        $tweet = json_decode($json, true);
        
        if(!isset($tweet['entities']['hashtags'][0])) continue;   //ハッシュタグがない場合は次のつぶやきへ

        //ハッシュタグがtwitterかチェック
        if(strcmp($tweet['entities']['hashtags'][0]['text'], sfConfig::get("app_twitter_keyword")) == 0) {
            //テーブルに登録済みかチェック
            if(TweetPeer::getTweetByStatusId($tweet['id_str'])) continue;
            
            //つぶやきを保存
            $tw = new Tweet();
            $tw->setIdStr($tweet['id_str']);
            $tw->setText($tweet['text']);
            $tw->setScreenName($tweet['user']['screen_name']);
            $tw->setCreatedAt(date("Y-m-d H:i:s", strtotime($tweet['created_ at'])));
            $tw->save();
        }
    }

    fclose($stream);
  }
}

モデルにstatusIdが登録済みかをチェックするメソッドを作成

vi lib/model/TweetPeer.php

<?php
class TweetPeer extends BaseTweetPeer {

    public static function getTweetByStatusId($statusId)
    {
        $c = new Criteria();
        $c->add(self::ID_STR, $statusId);
        return self::doSelectOne($c);
    }

}

タスクを実行

symfony tweet:get

タスクを実行すると、Streaming APIを用いて「#twitter」に関するつぶやきをリアルタイムに保存し続けます。

つまづいた点など

タスクからsfConfig::get()を利用する方法

最初、twitterの設定をapp.ymlに記述して、タスクからsfConfig::get()で読み込ませようとしたところ、うまく読み込みができませんでした。
タスクの初期設定の際に、application名を指定する必要があるようです。

new sfCommandOption('application'〜)の第5引数でapplication名を指定

    $this->addOptions(array(
      new sfCommandOption('application', null, sfCommandOption::PARAMETER_REQUIRED, 'The application name', 'frontend'),
      new sfCommandOption('env', null, sfCommandOption::PARAMETER_REQUIRED, 'The environment', 'dev'),
      new sfCommandOption('connection', null, sfCommandOption::PARAMETER_REQUIRED, 'The connection name', 'propel'),
      // add your own options here
    ));

参考URL:Taskにも時々application名を指定してあげる

created_atの保存方法

created_atをそのまま保存する場合、日本時間になっていないため9時間ずれた値が保存されてしまいます。
strtotimeで時刻変換をすることで正しく登録されます。

$tw->setCreatedAt(date("Y-m-d H:i:s", strtotime($tweet['created_ at'])));

参考URL:
日時表示を日本時間(JST)に変更

PHPでTwitterの時刻を変換する方法

filterに関して

今回、filterの引数として利用したtrackに関する注意事項（仕様書から引用）

track=キーワードのリスト
指定したキーワードを含む public な情報を取得する
この API を使うことで、(契約なしで)最大 200 キーワードを追いかけることができる
[最大 10000 キーワードを追いかけ可能な "restricted track" 契約、最大 200000 キーワードを追いかけ可能な "partner track" 契約も用意されている]
(キーワードの大文字、小文字は区別しない。複数のキーワードを指定した場合は、指定したキーワードがどれか1つでも含まれていれば、取得対象になる)
例えば、キーワードに twitter を指定した場合、前後が空白文字で区切られた
TWITTER
twitter
"Twitter"
twitter.
#twitter
@twitter
といった「語(token)」を含む発言を取得することができる。ただし、
TwitterTracker
http://www.twitter.com
のようにキーワードの前後に別の文字が存在する「語」に関しては、取得対象(キーワードマッチング対象)にはならない
追跡対象のキーワードをコンマ区切りで指定する。指定するキーワードは1バイト以上30バイト以内とする
(訳者注: キーワードに指定可能な文字は英数字のみ。記号やマルチバイト文字には未対応)

ハッシュタグ検索の場合、#をキーワードに含める必要はありません。