PythonでTwitterの1日分の投稿を取得するサンプル
bonlifeです。ノリで誰かさんのTwitterの1日分の発言を取得するスクリプトを書いてみました。Twitter APIだと20件までしか取得できないという噂を聞いたので、確かめもせずにゴリゴリとスクレイピングしてみましたよ。lxmlとdateutilを使ってますので、事前にインストールしておいてくださいませ。
使い方は以下のような感じです。
>python twitterscraper.py Enter Twitter user id : takets Enter target date (YYYYMMDD) : 20080311 2008-03-11 23:28:33 : 数理生態学者の話を聞きたかった。 2008-03-11 22:58:15 : ニッポンの教養を見る (後略)
>python twitterscraper.py -u takets -d 20080311 2008-03-11 23:28:33 : 数理生態学者の話を聞きたかった。 2008-03-11 22:58:15 : ニッポンの教養を見る (後略)
In [1]: from twitterscraper import TwitterScraper In [2]: import datetime In [3]: ts = TwitterScraper('takets') In [4]: ts.get_entries(datetime.date(2008,3,11)) In [6]: for i in ts.entries: ...: print i.get('permalink') ...: ...: http://twitter.com/takets/statuses/769854615 http://twitter.com/takets/statuses/769840277 (後略)
この3つ目のやり方を活用すれば他のスクリプトとの連携も楽なはず。
(結構汚い)ソースコードは以下の通り。
- twitterscraper.py
import sys import re import datetime from lxml import etree from dateutil import parser as dateparser class TwitterScraper(object): def __init__(self,user): self.user = user self.entries = list() self._xpath_dict_for_latest = { 'top' : '//div[@class="desc hentry"]', 'permalink' : 'p[@class="meta entry-meta"]/a', 'datetime' : 'p[@class="meta entry-meta"]/a/abbr', 'comments' : 'p[@class="entry-title entry-content"]' } self._xpath_dict_for_others = { 'top' : '//tr[@class="hentry"]', 'permalink' : 'td/span[@class="meta entry-meta"]/a', 'datetime' : 'td/span[@class="meta entry-meta"]/a/abbr', 'comments' : 'td/span[@class="entry-title entry-content"]'} def latest_check(self, target_date): url = 'http://twitter.com/%s' % (self.user) try: self.check(url, target_date, self._xpath_dict_for_latest) except DateOutOfRangeException: return def others_check(self, target_date): # try 100 pages to find entries posted on target date for page in range(1,101): url = 'http://twitter.com/%s?page=%d' % (self.user, page) try: self.check(url, target_date, self._xpath_dict_for_others) except DateOutOfRangeException: return def check(self, url, target_date, xpath_dict): try : et = etree.parse(url,etree.HTMLParser()) except IOError, e: print "ERR : %s " % e sys.exit(1) ts = et.xpath(xpath_dict.get('top')) for i in ts: # get post datetime from abbr tag's title attribute # add 9 hours (for 'JST') permalink = i.xpath(xpath_dict.get('permalink'))[0].get('href') entry_datetime = dateparser.parse(i.xpath(xpath_dict.get('datetime'))[0].get('title')) + datetime.timedelta(hours=9) entry_date = entry_datetime.date() if entry_date < target_date: raise DateOutOfRangeException() elif entry_date == target_date: e = i.xpath(xpath_dict.get('comments'))[0] self.entries.append({ 'permalink' : permalink, 'posted_datetime' : entry_datetime, # 'comments' : "".join([ x for x in e.itertext() ]).strip() }) 'comments' : self.format_comments(e) }) def format_comments(self, e): comments = list() for i in e.iter(): # if the hyperlink starts with 'http', get full url if i.tag == 'a' and i.get('href').startswith('http'): comments.append(i.get('href')) if i.tail.strip() != "": comments.append(i.tail) else: comments.append(i.text) if i.tail.strip() != "": comments.append(i.tail) comments[0] = comments[0].lstrip() comments[-1] = comments[-1].rstrip() return "".join([ re.sub("\s+"," ", x) for x in comments ]) def get_entries(self,target_date=None): if target_date == None: target_date = (datetime.datetime.today() - datetime.timedelta(days=1)).date() self.latest_check(target_date) self.others_check(target_date) class DateOutOfRangeException(Exception): pass if __name__ == "__main__": # simple usage is like below: # # tw = TwitterScraper('bonlife') # tw.get_entries() # get Twitter entries posted on yesterday # for i in tw.entries: # print "%s : %s " % (i.get('posted_datetime').strftime('%Y-%m-%d %H:%M:%S'), # i.get('comments')) from optparse import OptionParser usage = "usage: %prog [options] (no arguments needed)" version = "%prog 0.1" optionparser = OptionParser(usage=usage,version=version) optionparser.add_option("-u","--user",action="store",type="string", dest="user_id",help="twitter's user id") optionparser.add_option("-d","--date",action="store",type="string", dest="date_str",help="target date (YYYYMMDD)") (options, args) = optionparser.parse_args() if args: optionparser.print_help() sys.exit(1) if options.user_id == None and options.date_str == None: user_id = raw_input("Enter Twitter user id : ") date_str = raw_input("Enter target date (YYYYMMDD) : ") else: user_id = options.user_id date_str = options.date_str try: target_date = dateparser.parse(date_str).date() except: target_date = None tw = TwitterScraper(user_id) tw.get_entries(target_date) # just print Twitter entries for i in tw.entries: print "%s : %s " % (i.get('posted_datetime').strftime('%Y-%m-%d %H:%M:%S'), i.get('comments'))
反省点をいくつか。
- 最新の投稿だけが上の方で「オレサマ別格！」って感じでのけぞってるので、そこを取得をするためにちょっとややこしいことに
- 時差の考慮が適当(マジメにやろうとすると結構面倒っぽい)
- 投稿本文の文字列をそのまま取得するとハイパーリンクが途中で切れてしまうので、無理矢理ハイパーリンクを取得する処理が相当苦しい
でも、なんだかんだで動いたので満足だよ、ママン！
