検索結果をCouchDBにとっておく。

CouchDB misc

昨日、twitter のTLで流れてはっとしたのですが、確かにtwitterの検索結果、履歴がさかのぼれなくなることがあって、これはこれで不便だなぁ、というときがあります。ということで、例によってCouchDBに放り込んじゃいましょう。今回は年も変わったということで、今後触る機会が減るであろうPythonで書きました(去年はPython結構仕事で書いた...)。

twitter の search API は twitter.com/search.json?q=XXX でとれて、認証がいらない、その上、必ず次のページへのリンクを"next_page"というフィールドに含めてくれるので非常に使いやすかったです。pagingをするAPIを提供する場合は、next/prev はHTMLだろうがJSONだろうがXMLだろうがちゃんと提供すべき。ハイパーリンク重要だっていってるじゃん。ちょうどCouchDB本のページングのレシピの部分を翻訳しているんだが、こんなのレシピでも何でもない、Viewの結果に next: "startkey=XXXX&...", prev:"startkey=XXX" というフィールドいれればいいだけじゃん、と一昨年の年末ぼやいてたっけか。変わってない。

検索結果は認証がいらないので、後日キーワード監視?サービスとして公開することにしようかと思います。これで、

  • キーワードを登録して保管しておける。
  • サービスとしてどんな検索キーワードがモニターされているのかがわかる(タグクラウドっぽく)
  • あるtweetに対して、どんな検索キーワードを引っかけられているのかがわかる。

かなぁ。似たようなのありそうな気もしますが、ひとまず自分がほしいのと、探して試すより作った方が早そうなので。

あとは、ネガティブtweetが多いのかポジティブtweetが多いのか、とか盛り上がり具合とか。。その辺は 2004年頃のBlogの付加価値サービスとかわらんですね。そこまではやるかどうかしりませんが、githubにおいとくことにします。

とりあえず以下は書いたPythonスクリプト。Pythonは素人です。こんなのを書いていたら除夜の鐘が鳴っていました。

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# 
# Name        : search_crawler.py
# Description : Get the search result from twitter.com and push it to CouchDB.
# Usage       :
#  search_crawler.py -d couchdb_url -q search_word [-s since_id]
#

import getopt, sys
import urllib, urllib2
import urlparse
import httplib
import json
import datetime

SEARCH_EP = "http://twitter.com/search.json"


def usage():
    sys.stderr.write("search_crawler.py -d couch_db_url -q search_word  -s since_id \n")

def get_bulk_docs_url(dst):
    path = urlparse.urlparse(dst).path
    dbinfo = json.loads(urllib2.urlopen(dst).read())
    if dbinfo.has_key("db_name") == False:
        sys.stderr.write("Error: Invalid CouchDB URL.\n")
        usage()
        sys.exit(2)
    return urlparse.urljoin(dst, dbinfo["db_name"] + "/_bulk_docs")


def push(list, dst):
    paths = dst.split("/")
    url = urlparse.urljoin(dst, "./_bulk_docs")
    print "%s docs are pushed to CouchDB(%s)" % (len(list), dst)
    request = urllib2.Request(url, json.dumps({"docs": list}))
    return json.loads(urllib2.urlopen(request).read())

def crawl(next_page, list):
    url = SEARCH_EP + next_page
    print "Endpoint url: %s" % url
    f = urllib2.urlopen(url=url, timeout=30)
    response = json.loads(f.read())
    if response.has_key("results"):
        list = list + response["results"]
        if response.has_key("next_page"):
            return crawl(response["next_page"], list)
    return list

def main():
    print "** check arguments ..."
    try:
        opts, args = getopt.getopt(sys.argv[1:], "q:d:hs:")
    except getopt.GetOptError:
        usage();
        sys.exit(2);
        
    q        = None
    since_id = None
    dst      = None
    for o, a in opts:
        if o == "-q":
            q = a
        if o == "-d":
            dst = a
        if o == "-s":
            since_id = a

    if q == None:
        usage();
        sys.exit(1)
    if dst == None:
        usage()
        sys.exit(1)

    bulk_url = get_bulk_docs_url(dst)
    params = {'q': q }
    if since_id != None:
        params["since_id"] = since_id

    print "** start crawling ..."
    list = crawl("?" + urllib.urlencode(params), [])
    #list = build_docs(list)
    print "** push docs to CouchDB ..."
    now = datetime.datetime.utcnow()
    push([{"tweet"      : doc, 
           "type"       : "TStore::Tweet", 
           "created_at" : now.strftime("%Y-%m-%d %H:%M:%S +0000"),
           "updated_at" : now.strftime("%Y-%m-%d %H:%M:%S +0000"),
           "keyword": q } for doc in list], 
         bulk_url)

if __name__ == "__main__":
    main()