複数のSqlite dbファイルにある同じ構造のテーブルからデータを抽出するメモ
#!/usr/bin/env python # -*- coding: utf-8 -*- import sqlite3 import csv writer = csv.writer(file('result.csv','w'),lineterminator="\n",quoting = csv.QUOTE_ALL) con = sqlite3.connect(':memory:') cur = con.cursor() sql2 = "SELECT * FROM (SELECT * FROM tweet" for i in xrange(1,10): #dbファイルの数 sql = "ATTACH DATABASE 'tweet%d.db' as tweet%d" % (i,i) con.execute(sql) con.commit() sql2 = sql2 + """ UNION ALL SELECT * from tweet%d.tweet""" % (i) sql2 = sql2 + ") t WHERE created_at >= datetime(?) AND created_at < datetime(?)" for row in cur.execute(sql2,('2011-10-01','2011-10-03')): try: writer.writerow([col.encode('s-jis') if isinstance(col, unicode) else col for col in row]) except: pass cur.close() con.close()
tweet+番号.dbという名前の複数ファイルがあって
それぞれに同じ構造のテーブルtweetがある。
created_atカラム（日付）でのデータ絞込み。
lineterminator="\n"
はレコード毎に空白行ができてしまう現象を防ぐための措置。
sql文のLIMIT、OFFSET句とリスト内包表記のメモ
表データをN行ずつ分けて処理する。
#!/usr/bin/env python # -*- coding: utf-8 -*- import sqlite3 ↓↓↓↓↓↓ここから <test用DBの準備>↓↓↓↓↓↓ con = sqlite3.connect(":memory:") cur = con.cursor() sql = """CREATE TABLE IF NOT EXISTS user ([user_id] TEXT, [user_name] TEXT );""" cur.execute(sql) sql = "delete from user" cur.execute(sql) #test用データ tplist = ((1,'taro'),(2,'hanako'),(3,'becky'),(4,'alice'),(5,'ken'),(6,'jiro'),(7,'kaori')) print tplist ph = "?," * (len(tplist[0]) - 1) + "?" for t in tplist: sql = "INSERT INTO %s VALUES(%s)" % ('user',ph) con.execute(sql,t) con.commit() #↑↑↑↑↑↑ここまで <test用DBの準備>↑↑↑↑↑↑ limit = 2 """ sql文でLIMIT句OFFSET句を使う """ sql = "select count(user_name) from user" cur.execute(sql) rowcount = cur.fetchone()[0] for offset in range(int(rowcount)/limit+1): sql = "select * from user order by user_id LIMIT %d OFFSET %d" % \ (limit,limit*offset) cur.execute(sql) print cur.fetchall() """ リスト内包表記でスライス """ sql = "select * from user" cur.execute(sql) tplist = cur.fetchall() for row in [tplist[limit*offset:limit*(offset+1)] for offset in range(len(tplist)/limit+1)]: print row cur.close() con.close()
GAE/Pyを使ってTwitter本アカのふぁぼを自動ではてブ＋サブアカに公式RTする。
Google App Engine を使ってTwitter本アカのふぁぼったツイートに含まれるリンクを自動ではてブ＋サブアカに公式RTする。
※はてブからTwitterに流す設定をしている場合、AtomAPIからの投稿をTwitterに流さないようにしておく。ブコメに@を使っているので、ふぁぼったツイートの発言者さんに@飛ばすことを防ぐため。
ついでにGoogle Readerのスター付きアイテムを自動ではてブ＋Twitterサブ垢に流す。
oauth.py
GAEのpython用oauthライブラリ
https://github.com/mikeknapp/AppEngine-OAuth-Library
WSSE.py
WSSE認証を利用したAtom APIとか - YAMAGUCHI::weblog
http://d.hatena.ne.jp/ymotongpoo/20081201/1228115936
[詰まったところ]
・Google Readerスター付きアイテムのAtomフィードをElementTreeだけでパースするところ。
・Task queueの使い方。
app.yaml
application: XXXXXXXXXXX #アプリケーション名 version: XX runtime: python api_version: 1 handlers: - url: /.* script: main.py login: admin
cron.yaml
cron: - description: twitter favs url: /regist_favs schedule: every 10 minutes timezone: Asia/Tokyo - description: Google Reader Stars url: /regist_GStars schedule: every 10 minutes timezone: Asia/Tokyo
queue.yaml
queue: - name: default rate: 1/s - name: bookmarkGStar rate: 1/m retry_parameters: task_retry_limit: 7 task_age_limit: 2d - name: bookmarkTwFav rate: 1/m retry_parameters: task_retry_limit: 7 task_age_limit: 2d - name: retweet rate: 1/m retry_parameters: task_retry_limit: 7 task_age_limit: 2d - name: tweetGStar rate: 1/m retry_parameters: task_retry_limit: 7 task_age_limit: 2d
main.py
#!/usr/bin/env python # -*- coding: utf-8 -*- __author__ = "yadokari23" import oauth import os import urllib, urllib2 from google.appengine.ext import webapp from google.appengine.ext.webapp import util from google.appengine.ext import db from google.appengine.api import urlfetch from django.utils import simplejson from xml.etree.ElementTree import * import datetime from WSSE import * from google.appengine.api.labs import taskqueue #Twitter アプリの情報 CONSUMER_KEY = "XXXXXXXXXX" #Consumer key CONSUMER_SECRET = "XXXXXXXXX" #Consumer secret #投稿先(サブ垢)のもの ACCESS_TOKEN = "XXXXXXXXXX" #access token ACCESS_TOKEN_SECRET = "XXXXXXXXXX" #access token secret #ふぁぼを監視するアカウントのscreen_name(今回は本アカ) T_ACCOUNT = 'XXXXXXXXXXX' #はてなアカウント情報 HATENA_USERNAME='XXXXXXX' HATENA_PWD='XXXXXXX' #Google Readerのスター付きアイテム公開ページのURL GSTARURL='XXXXXXXXXXXXXXXXXXXXX' MaxPage = 1 #MaxPage*20件 class favorites(db.Model): status_id = db.StringProperty(required=True) text = db.TextProperty() created = db.DateTimeProperty() url = db.TextProperty() user_id = db.StringProperty() user_screen_name = db.StringProperty() task = db.BooleanProperty(default=False) class GStars(db.Model): id = db.StringProperty(required=True) link = db.TextProperty() title = db.TextProperty() source_title = db.TextProperty() author = db.TextProperty() task = db.BooleanProperty(default=False) def get_longURL(shortURL): #短縮URLの展開 try: longURL = urllib2.urlopen(shortURL).geturl() except: longURL = '' return longURL class MainHandler(webapp.RequestHandler): def get(self, mode=""): if mode=='regist_favs': page = 1 while page <= MaxPage: query = {'screen_name':T_ACCOUNT, 'include_entities':'True', 'page':page } t_url = 'http://api.twitter.com/1/favorites.xml?%s' % urllib.urlencode(query) results = urlfetch.fetch(url=t_url, method=urlfetch.GET, deadline=10) dom = fromstring(results.content) if results.status_code == 200: if dom.findall('.//status'): for e in dom.findall('.//status'): q = favorites.gql("WHERE status_id = :1", e.findtext('./id')) fav = q.get() if fav is None: tweet_created = e.findtext('./created_at') tweet_created = datetime.datetime.strptime(tweet_created, '%a %b %d %H:%M:%S +0000 %Y') \ + datetime.timedelta(hours=9) p = favorites( status_id=e.findtext('./id'), text=e.findtext('./text'), created=tweet_created, url=e.findtext('./entities/urls/url/expanded_url'), user_id=e.findtext('./user/id'), user_screen_name=e.findtext('./user/screen_name'), ) if p.url == None: p.task = True p.put() page +=1 #task queue未登録のレコードを取り出す。 fav = favorites.all().filter('task =', False).get() if fav: fav_values={'status_id':fav.status_id, 'user_screen_name':fav.user_screen_name, 'url':fav.url } #タスクbookmarkTwFavのインスタンスを作成し、タスクを追加 bookmark_task = taskqueue.Queue('bookmarkTwFav') next_task = taskqueue.Task(url='/_bookmarkTwFav', params=fav_values) bookmark_task.add(next_task) #タスクretweetのインスタンスを作成し、タスクを追加 retweet_task = taskqueue.Queue('retweet') next_task = taskqueue.Task(countdown='60',url='/_retweet', params=fav_values) retweet_task.add(next_task) fav.task = True db.put(fav) if mode=='regist_GStars': #Googleスター付きアイテムをはてブする。 results = urlfetch.fetch(url=GSTARURL, method=urlfetch.GET, deadline=10) dom = fromstring(results.content) atom_ns = "http://www.w3.org/2005/Atom" if results.status_code == 200: if dom.findall('.//{%s}entry' % atom_ns): for e in dom.findall('.//{%s}entry' % atom_ns): q = GStars.gql("WHERE id = :1", e.find('./{%s}id' % atom_ns).text) GStar = q.get() if GStar is None: for e2 in e.findall('.//{%s}source' % atom_ns): source_title = e2.find('./{%s}title' % atom_ns).text for e3 in e.findall('.//{%s}author' % atom_ns): author = e3.find('./{%s}name' % atom_ns).text p = GStars(id=e.find('./{%s}id' % atom_ns).text, title=e.find('./{%s}title' % atom_ns).text, link=e.find('./{%s}link' % atom_ns).attrib['href'], source_title=source_title, author=author ) if p.link == None: p.task = True p.put() #task queue未登録のレコードを取り出す。 GStar = GStars.all().filter('task =', False).get() if GStar: param={'title':GStar.title, 'source_title':GStar.source_title, 'author':GStar.author, 'link':GStar.link } #タスクbookmarkGStarのインスタンスを作成し、タスクを追加 bookmark_task = taskqueue.Queue('bookmarkGStar') next_task = taskqueue.Task(url='/_bookmarkGStar', params=param) bookmark_task.add(next_task) #タスクtweetGStarのインスタンスを作成し、タスクを追加 tweet_task = taskqueue.Queue('tweetGStar') next_task = taskqueue.Task(url='/_tweet', params=param) tweet_task.add(next_task) GStar.task = True db.put(GStar) def post(self, mode=""): if mode=='_bookmarkTwFav': status_id = self.request.get("status_id") user_screen_name = self.request.get("user_screen_name") url = self.request.get("url") #WSSE認証 hbc = HatenaBookmarkClient(HATENA_USERNAME,HATENA_PWD) hbc.getServiceURI() # サービスエンドポイントの取得（PostURIとFeedURI） bookmarkURL = get_longURL(url) if bookmarkURL <> '': status_url = 'http://twitter.com/%s/status/%s' % (user_screen_name,status_id) hbcomment = u'[☆]@%s %s' % (user_screen_name,status_url) hbc.postBookmark(bookmarkURL, hbcomment) # ブックマークの登録 if mode=='_bookmarkGStar': url = self.request.get("link") #WSSE認証 hbc = HatenaBookmarkClient(HATENA_USERNAME,HATENA_PWD) hbc.getServiceURI() # サービスエンドポイントの取得（PostURIとFeedURI） if url <> '': hbcomment = u'[☆]' hbc.postBookmark(url, hbcomment) # ブックマークの登録 if mode=='_retweet': status_id = self.request.get("status_id") #TwitterClientクラスの作成 twc = oauth.TwitterClient(CONSUMER_KEY, CONSUMER_SECRET, None) url='http://api.twitter.com/1/statuses/retweet/%s.json' % status_id twc.make_request(url=url, token=ACCESS_TOKEN, secret=ACCESS_TOKEN_SECRET, protected=True, method='POST') if mode=='_tweet': title = self.request.get("title") source_title = self.request.get("source_title") author = self.request.get("author") link = self.request.get("link") text = u'[☆] %s / %s (%s) %s' % (title,source_title,author,link) query={'status':text} #TwitterClientクラスの作成 twc = oauth.TwitterClient(CONSUMER_KEY, CONSUMER_SECRET, None) url='http://api.twitter.com/1/statuses/update.json' twc.make_request(url=url, token=ACCESS_TOKEN, secret=ACCESS_TOKEN_SECRET, protected=True, additional_params=query, method='POST') def main(): application = webapp.WSGIApplication([('/(.*)', MainHandler)], debug=True) util.run_wsgi_app(application) if __name__ == '__main__': main()
queue.yamlにリトライ制限を記述
queue.yamlにretry_parametersを追加する。
Retrying Tasks in Push Queues
In fooqueue, tasks are retried at least seven times and for up to two days from the first execution attempt. After both limits are passed, it fails permanently.
http://code.google.com/intl/ja/appengine/docs/python/config/queue.html#Setting_the_Storage_Limit_for_All_Queues
queue:
- name: fooqueue
rate: 1/s
retry_parameters:
task_retry_limit: 7
task_age_limit: 2d
PythonでTwitterのAtomフィードをパースする
from google.appengine.api import urlfetch #GAE/Py from xml.etree import ElementTree as etree #････(略)････ qstr = u'#albirex' query = {'q':qstr, 'lang':'ja', 'rpp' :'100' } url = ("http://search.twitter.com/search.atom?%s" % urllib.urlencode(query)) try: results = urlfetch.fetch(url=url, method=urlfetch.GET, deadline=10) dom = etree.fromstring(results.content) atom_ns ='http://www.w3.org/2005/Atom' if results.status_code == 200: self.response.headers['Content-Type'] = "text/html; charset=UTF-8" self.response.out.write('<html><body>') self.response.out.write('<h1>%s tweets</h1>' % qstr) self.response.out.write('<ul>') if dom.findall('.//{%s}entry' % atom_ns): for e in dom.findall('.//{%s}entry' % atom_ns): s_text = e.find('./{%s}title' % atom_ns).text s_text = ReplaceLink(s_text) self.response.out.write('<li>%s</li>' % s_text) self.response.out.write('<ul></body></html>') else: self.response.out.write('No Data') except urlfetch.DownloadError: self.response.out.write('urlfetch DownloadError!') self.response.out.write('http status ' + str(results.status_code)+ '\n')
参考：(何を参考にしたか忘れたのであとで)
はてなブックマークエントリー情報取得APIを使うついでに全体のカテゴリ名を取得する。
参考:Pythonで、はてなAPIを使ってみる(1)
はてなブックマークエントリー情報取得APIを使ってみた。
はてなブックマークエントリー情報取得APIでは項目がないっぽいので、
全体のカテゴリ名をHTMLから正規表現で取ってきて、それを辞書に付加する。
import re import urllib,urllib2 from django.utils import simplejson def get_categoryName(trgurl): q = {'url':trgurl} entryinfourl = 'http://b.hatena.ne.jp/entry?%s' % urllib.urlencode(q) result=urllib2.urlopen(entryinfourl) reg = re.compile(r'class="category-link">(.+)</a>') categoryName = reg.findall(result.read().decode('utf-8')) result.close return categoryName[0] def get_entryinfo(trgurl,cat): entryinfourl = 'http://b.hatena.ne.jp/entry/json/%s' % trgurl result =urllib2.urlopen(entryinfourl) entryinfo = simplejson.loads(result.read()) result.close if cat: catName = get_categoryName(trgurl) dict = {'categoryName':catName} entryinfo.update(dict) return entryinfo else: return entryinfo
使い方例 : get_entryinfo('http://d.hatena.ne.jp/yadokari23/20110414',True)
Google+をGoogle Readerで購読するためのBookmarklet
Google+ Profile https://plus.google.com/<プロフィールID>/ のページを開いてこのブックマークレットを使うと、
lifehakerの記事
(Google+のユーザーストリームをRSSリーダーでチェックできる「Google Plus Feed」 http://www.lifehacker.jp/2011/07/110720_google_rss.html)
で紹介されていたGoogle+の個人のユーザーストリームを、RSSフィードで取り込むWebアプリ「Google Plus Feed」を使ってRSSを生成→そのままGoogle Readerにジャンプして登録できるというものです。
×できない。Google Buzz Public Activities Feed for <ユーザ名>のFeedになってる。
○単純にこれだけでよかったみたい...
javascript: (function(){ if(location.host=='plus.google.com'&&location.href.match(/plus.google.com\/([0-9a-zA-Z_]+)/)){ location='http://www.google.com/reader/view/feed/'+encodeURIComponent('http://plusfeed.appspot.com/'+RegExp.$1); } } )();
RSS生成のみVersion
javascript: (function(){ if(location.host=='plus.google.com'&&location.href.match(/plus.google.com\/([0-9a-zA-Z_]+)/)){ location='http://plusfeed.appspot.com/'+RegExp.$1; } } )();
Google＋のページでRSS Subscription Extension（by Google）をただ使うだけとGoogle Buzz Public Activities Feed for <ユーザ名>のfeedになるんですね。
ユーザーストリームとの違いは何？？
※Google＋使ってないのでよくわかってません。
とりあえずこちら http://www.findpeopleonplus.com/ でも使って面白そうな人を探して、これを使ってGoogleReaderでROMろうかと思ってます。
Gナビの方がいいかな。
