Twitter Streaming API User Streamsを利用したリアルタイム検索 (C#)

前回作成したTwitter Search APIによるつぶやき検索 (C#)では、TweetDeckとの検索結果と比較して、情報鮮度がいまいちでした。。。ショック。

私が勝手にライバル視しているTweetDeckはTwitter Streaming APIなるものを使用しているらしく、Twitter側から検索結果がプッシュされるので、Twitter APIより新鮮なデータが取得できるようです。

そこで、今回は、Twitter Streaming APIを利用した検索クライアントを作成してみました。
TweetDeckと同じタイミングでつぶやきデータが受信できます。

2011/01/20 22:00 本文全体を加筆

Twitter Streaming API

---

Twitter Search APIによるつぶやき検索 (C#)では、クライアント側からデータを要求した分だけデータが返ってきますが、こちらはつぶやきが発生次第Twitterサーバからデータが届きます。
Streaming API Documentationに本家のドキュメントがありますが、私はTwitter API 仕様書 (勝手に日本語訳シリーズ)を活用させていただきましたw
　
・αテスト中につき、仕様変更される可能性あり　
・BASIC認証、HTTP POST/GET使用
・APIによって取得件数等の制限のある/なしが決まっている
・ツイートのデータは、XML or JSON形式
　

filter API 概要

---

今回は任意のワードで検索できるfilter APIを使用しました。
URL：http://stream.twitter.com/1/statuses/filter.json
検索ワードは、track=検索ワード1,検索ワード2,,,,のように指定します。
Twitter Serch APIとは異なり、日本語(マルチバイト文字)の指定ができません(残念)。
返ってくるデータ形式はJSONのみとのこと。

json形式つぶやきデータ

---

受信したつぶやきデータの例です。

key:valueの形になっています。userとentitiesのデータは入れ子になっています。
つぶやき時刻は”created_at”、ユーザ名は”user”の中の”name”、つぶやきデータは”text”です。

{
	"place":null,
	"user":
	{
		"profile_background_tile":true,
		"time_zone":"Hawaii",
		"location":"",
		"contributors_enabled":false,
		"verified":false,
		"listed_count":71,
		"profile_link_color":"0084B4",
		"show_all_inline_media":false,
		"geo_enabled":false,
		"profile_sidebar_border_color":"C0DEED",
		"id_str":"0123456789",
		"url":"http://hogehoge.com/",
		"notifications":null,
		"profile_use_background_image":true,
		"created_at":"Sat Jan 09 15:11:04 +0000 2010",
		"description":"hogehoge",
		"profile_background_color":"C0DEED",
		"profile_background_image_url":"http;//hogehoge.com/hoge.jpg",
		"favourites_count":64,
		"friends_count":919,
		"followers_count":840,
		"protected":false,
		"lang":"ja",
		"statuses_count":18866,
		"profile_image_url":"http://twi.com/hoge.jpg",
		"name":"hogename",
		"follow_request_sent":null,
		"following":null,
		"profile_text_color":"333333",
		"id":0123456789,
		"is_translator":false,
		"utc_offset":-36000,
		"profile_sidebar_fill_color":"DDEEF6",
		"screen_name":"karimas_P"
	},
	"in_reply_to_user_id":null,
	"favorited":false,
	"in_reply_to_status_id_str":null,
	"contributors":null,
	"in_reply_to_screen_name":null,
	"text":"hogehoge",
	"in_reply_to_user_id_str":null,
	"id_str":"hogehoge",
	"geo":null,
	"retweeted":false,
	"retweet_count":0,
	"source":"u003Ca href="http://hogehoge.com" rel="hogehoge",
	"created_at":"Thu Jan 20 12:18:01 +0000 2011",
	"coordinates":null,
	"truncated":false,
	"id":01234567890,
	
	"entities":{
		"urls":[{
			"indices":[37,62],
			"expanded_url":null,
			"url":"http://hogehoge.com"
		}],
		"hashtags":[{
		"indices":[63,74],
		"text":"sm13357394"}],
		"user_mentions":[]
	},
	
	"in_reply_to_status_id":null
}

C#でJSONデータの解析

---

JavaScriptSerializer .Deserialize()メソッドを使って解析しています。
時刻データは実際にはクライアント側のローカル時刻に変換してやる必要があります(最終コードを参考にして下さい)。

string jsonData;//受信文字列を格納

//解析
JavaScriptSerializer serializer = new JavaScriptSerializer();
Dictionary<String, Object> dic = serializer.Deserialize<Dictionary<String, Object>>(jsonData);
Dictionary<String, Object> usr = (Dictionary<string, Object>)dic["user"];

//データ取り出し
string[] data = new string[3];
data[0] = (string)dic["created_at"];//時刻
data[1] = (string)usr["name"];//ユーザ名
data[2] = (string)dic["text"];//つぶやき本文

　

ソースコード

---

全体のコードです。
つぶやきデータのReadはスレッドtrd1で行なっています。
また、特にマイナーな検索ワードだと、ReadLine()メソッドがずっと
ブロッキングされるため、タイムアウト時間を指定しています。

using System;
using System.Collections.Generic;
using System.Text;
using System.Windows.Forms;
using System.Text.RegularExpressions;
using System.Threading;
using System.Net;
using System.IO;
using System.Web.Script.Serialization;
using System.Globalization;

namespace StreamingAPITest
{
    public partial class Form1 : Form
    {
        string _SearchWord = "";    //検索ワード
        bool _StateStarted = false;//アプリの状態(停止中/動作中)
        bool _ReadTweetFlag;//true：つぶやきデータをReadし続ける

        //通信用
        Thread trd1 = null;
        HttpWebRequest _Request;
        WebResponse _Response;
        StreamReader _StreamReader;
        Stream _ReqStream;
        Stream _ResStream;

        //スレッド側からGUIコントロール更新用のdelegate
        delegate void UpdateString(string text);
        delegate void UpdateBool(bool value);
        delegate void SetListView(string[] data);

        public Form1()
        {
            InitializeComponent();

            //dataGridViewの行ヘッダーの表示は不要
            dataGridView1.RowHeadersVisible = false;
        }

        private void button1_Click(object sender, EventArgs e)
        {
            if (_StateStarted == false)
            {
                //停止中なので開始処理を行う
                //検索ワードチェック
                if (CheckWord(textBox1.Text) == false)
                {
                    return;
                }

                //スレッド起動
                _ReadTweetFlag = true;
                trd1 = new Thread(new ThreadStart(ReadTweet));
                trd1.IsBackground = true;
                trd1.Start();

                label1.Text = "接続しています...";
                textBox1.Enabled = false;
                button1.Enabled = false;
            }
            else
            {
                //動作中なので停止処理を行う
                //スレッド停止
                //ReadLine()処理後にスレッドを停止
                _ReadTweetFlag = false;

                //停止移行にする
                label1.Text = "切断しています...";
                textBox1.Enabled = true;
                button1.Enabled = false;
            }
        }
        //検索ワードチェック
        bool CheckWord(string str)
        {
            try
            {
                if (str == "")
                {
                    return false;//検索ワードなし
                }

                //半角英数か
                Match result = Regex.Match(str, "^[a-zA-Z0-9!-~ ]+$");
                if (result.Success == false)
                {
                    MessageBox.Show("半角英数字で入力してください");
                    return false;
                }

                //前回から変更されたか
                if (_SearchWord != str)
                {
                    dataGridView1.Rows.Clear();//ビューをクリア
                }

                _SearchWord = str;
            }
            catch (Exception)
            {
                return false;
            }
            return true;
        }

        //つぶやきの取得、表示
        void ReadTweet()
        {
            //ラベル更新関数
            UpdateString updateLabel = (text) =>
            {
                label1.Text = text;
            };
            //ボタン更新関数
            UpdateString updateButton = (text) =>
            {
                button1.Text = text;
                button1.Enabled = true;
            };
            //テキストボックス更新関数
            UpdateBool UpdateTextBox = (state) =>
            {
                textBox1.Enabled = (bool)state;
            };

            //POST
            this.Invoke(updateLabel, "POSTしています...");
            if (Post() == false)
            {
                this.Invoke(updateLabel, "POSTに失敗しました");
                return;
            }

            //GET
            this.Invoke(updateLabel, "GETしています...");
            if (Get() == false)
            {
                this.Invoke(updateLabel, "GETに失敗しました");
                return;
            }
            _StateStarted = true;
            //GUI更新
            this.Invoke(updateLabel, "接続しました");
            this.Invoke(updateButton, "STOP");

            //READ
            ReadLoop();

            //切断
            this.Invoke(updateLabel, "切断しています...");
            CloseConnect();
            _StateStarted = false;
            //GUI更新
            this.Invoke(updateLabel, "切断しました");
            this.Invoke(updateButton, "START");
            this.Invoke(UpdateTextBox, true);
        }

        //POST
        bool Post()
        {
            try
            {
                string id = "Your_Twitter_ID";
                string password = "Your_Twitter_PASS";
                string url = "http://stream.twitter.com/1/statuses/filter.json";

                //HTTP POSTリクエストの作成
                //trackによる検索(半角英数のみ対応。マルチバイト文字、記号は不可。コンマ区切り。)
                string param = String.Format("{0}={1}", "track", textBox1.Text.Replace(' ', ','));
                byte[] data = Encoding.UTF8.GetBytes(param);
                _Request = (HttpWebRequest)WebRequest.Create(url);
                _Request.Method = "POST";
                _Request.ContentType = "application/x-www-form-urlencoded";
                _Request.ContentLength = data.Length;
                _Request.Credentials = new NetworkCredential(id, password);

                //POSTを実行
                _ReqStream = _Request.GetRequestStream();
                _ReqStream.Write(data, 0, data.Length);
                _ReqStream.Close();
            }
            catch (WebException e)
            {
                MessageBox.Show("Status Code : " + ((HttpWebResponse)e.Response).StatusCode);
                MessageBox.Show("Status Description : " + ((HttpWebResponse)e.Response).StatusDescription);
                return false;
            }
            catch (Exception e)
            {
                MessageBox.Show("Message    :n" + e.Message);
                MessageBox.Show("Type       :n" + e.GetType().FullName);
                MessageBox.Show("StackTrace :n" + e.StackTrace.ToString());
                return false;
            }
            return true;
        }

        //GET
        bool Get()
        {
            try
            {
                //GET実行
                _Response = _Request.GetResponse();
                _ResStream = _Response.GetResponseStream();
                bool timeout = _ResStream.CanTimeout;
                //注)Timeout値を設定しとかないと、
                //   つぶやきを受信するまでReadLineはブロッキングされる
                _ResStream.ReadTimeout = 3 * 1000;
                _StreamReader = new StreamReader(_ResStream, Encoding.UTF8);
            }
            catch (WebException e)
            {
                MessageBox.Show("Status Code : " + ((HttpWebResponse)e.Response).StatusCode);
                MessageBox.Show("Status Description : " + ((HttpWebResponse)e.Response).StatusDescription);
                return false;
            }
            catch (Exception e)
            {
                MessageBox.Show("Message    :n" + e.Message);
                MessageBox.Show("Type       :n" + e.GetType().FullName);
                MessageBox.Show("StackTrace :n" + e.StackTrace.ToString());
                return false;
            }
            return true;
        }

        //READ
        void ReadLoop()
        {
            //ビューへ追加用関数
            SetListView setListView = (string[] data) =>
            {
                DataGridViewRow dgvr = new DataGridViewRow();
                dgvr.CreateCells(dataGridView1);

                dgvr.Cells[0].Value = data[0];
                dgvr.Cells[1].Value = data[1];
                dgvr.Cells[2].Value = data[2];

                dataGridView1.Rows.Add(dgvr);

                //最後の行までスクロール
                dataGridView1.FirstDisplayedScrollingRowIndex = dataGridView1.RowCount - 1;
            };

            //受信ループ(json形式データ)
            while (_ReadTweetFlag)
            {
                try
                {
                    string result = _StreamReader.ReadLine();
                    if (result.Length <= 0)
                    {
                        continue;
                    }

                    //jsonデータをデコード
                    JavaScriptSerializer serializer = new JavaScriptSerializer();
                    Dictionary<String, Object> dic = serializer.Deserialize<Dictionary<String, Object>>(result);

                    //データチェック
                    if ((dic["created_at"] == null) || (dic["user"] == null) || (dic["text"] == null))
                    {
                        //情報が欠けている場合はスキップ
                        continue;
                    }
                    Dictionary<String, Object> usr = (Dictionary<string, Object>)dic["user"];

                    //データ取り出し
                    string[] data = new string[3];
                    //日付："Thu Jan 20 10:45:54 +0000 2011"という形の文字列。
                    //DateTime.ParseExactでstring型からDateTime型に変換し、
                    //ToLocalTime()で自分の地域の日付とする。
                    //dddやMMMの変換のためCultureInfoも必要。
                    data[0] = DateTime.ParseExact((string)dic["created_at"], "ddd MMM dd HH:mm:ss +0000 yyyy", new CultureInfo("en-US")).ToLocalTime().ToString();
                    data[1] = (string)usr["name"];//ユーザ名
                    data[2] = (string)dic["text"];//つぶやき本文

                    //ビューに追加
                    this.Invoke(setListView, new object[] { data });
                }
                catch (Exception e)
                {

                    if ((e.Message.Contains("タイムアウト") || e.Message.Contains("応答"))//タイムアウトっぽい例外
                     || (e.Message.Contains("キー")))//情報欠けっぽい例外
                    {
                        //再びReadLine()する
                    }
                    else
                    {
                        //ループを抜ける
                        MessageBox.Show("Message    :n" + e.Message);
                        MessageBox.Show("Type       :n" + e.GetType().FullName);
                        MessageBox.Show("StackTrace :n" + e.StackTrace.ToString());
                        break;
                    }
                }
            }
        }

        //Close
        void CloseConnect()
        {
            try
            {
                //通信を止める
                if (_Request != null)
                {
                    _Request.Abort();
                }
                if (_Response != null)
                {
                    _Response.Close();
                }
                //ストリームを閉じる
                if ((_ReqStream != null) && (_ReqStream.CanRead == true))
                {
                    _ReqStream.Close();
                }
                if ((_ResStream != null) && (_ResStream.CanRead == true))
                {
                    _ResStream.Close();
                }
                if (_StreamReader != null)
                {
                    _StreamReader.Close();
                }
            }
            catch (Exception e)
            {
                MessageBox.Show("Message    :n" + e.Message);
                MessageBox.Show("Type       :n" + e.GetType().FullName);
                MessageBox.Show("StackTrace :n" + e.StackTrace.ToString());
                return;
            }
        }
        private void Form1_FormClosing(object sender, FormClosingEventArgs e)
        {
            try
            {
                if (_ReadTweetFlag == true)
                {
                    //スレッド停止
                    //ReadLine()処理後にスレッドを停止
                    _ReadTweetFlag = false;
                    label1.Text = "切断しています...";
                }
            }
            catch (Exception exp)
            {
                MessageBox.Show("Message    :n" + exp.Message);
                MessageBox.Show("Type       :n" + exp.GetType().FullName);
                MessageBox.Show("StackTrace :n" + exp.StackTrace.ToString());
            }
        }

    }
}