Google App EngineによるTumblrのバックアップ

久しぶりのGoogle App Engine関連の投稿です。先日の料金体系の大幅な変更などの影響で、最近はGoogle App Engineで新しく何かを作るのを避けていたのですが、用途を限定すればまだまだ使えると考えました。特に、自分専用のバックアップアプリなんかは相性がいいと思い、Tumblrの引っ越しのついでに投稿バックアップアプリを作ってみました。この記事では、Google App Engine(Python)を使ってTumblrのテキスト形式の投稿をバックアップする方法について書いています。

目次

  1. 下準備
  2. Tumblrの投稿取得
  3. Tumblrバックアップアプリ
  4. Cronの設定

下準備

Google App Engineの準備

まずはGoogle App Engineを準備です。もちろんGoogle App Engineを既に利用できる状態になっている場合は必要ありません。初めて利用する場合のセットアップには、「Google App Engine - Google Code」の「スタートガイド」なんかが参考になると思います。

TumblrのConsumer Keyの発行

次にTumblrの投稿を取得するAPIを呼ぶために、Tumblrにアプリケーションを登録して、「Consumer Key」を発行しておく必要があります。。これは以前書いた記事、「PythonとOAuthを使ってTumblrに投稿する方法」の「アプリケーションの登録」が参考になると思います。※ 必要なのはConsumer Keyなので、同投稿の「リクエストトークンの取得」など、それ以降の手順をやるは必要ありません。

以上で下準備が整いました。

↑ 目次に戻る

Tumblrの投稿取得

「Consumer Key」を取得したら、TumblrのAPIを使って投稿を取得できるようになります。例えば、テキスト形式の投稿を最新20件は、以下のURLで取得することができます。

api.tumblr.com/v2/blog/{base-hostname}/posts/text?api_key={key}&offset=0&limit=20

※ {}のパラメータは以下のように置き換えてください
base-hostname: Tumblrのホスト名 例) yusukemurayama.tumblr.com
key: 下準備で用意した「Consumer Key」

このAPIを使って、テキスト形式の投稿をすべて取得するプログラムは、Pythonを使って以下のように書くことができます。

※ 以下のプログラムはGoogle App Engineで使うバージョンではありません。あくまでコマンドラインで実行して動作を確認するためのものです。Google App Engineで利用するバージョンは、Tumblrバックアップアプリに載せています。また、「with」ステートメントや「json」モジュールを使っているため、Python2.5では動きません。

#!/opt/local/bin/python
#coding: utf-8

import json
import urllib
import time
from contextlib import closing

CONSUMER_KEY = '[Consumer Key]'
BASE_HOSTNAME = '[Base Hostname]'

def main():
    for post in post_iterator():
        print '''id: %(id)d
post_url: %(post_url)s
title: %(title)s
body: %(body)s''' % post
        print '-' * 80

def post_iterator():
    # API毎にスリープする秒数
    sleep_sec_per_api = 3 

    # 一度に取得する件数(Tumblrの仕様で最大20件)
    limit = 20
    pos = 0
    has_more_posts = True
    api_url_template = 'http://api.tumblr.com/v2/blog/%s/posts/text?api_key=%s&offset=%s&limit=%s'
    while has_more_posts:
        api_url = api_url_template % (BASE_HOSTNAME, CONSUMER_KEY, pos, limit)
        with closing(urllib.urlopen(api_url)) as f:
            response = json.load(f)['response']
            if response['total_posts'] <= pos + limit:
                # すべての投稿を取得した場合
                has_more_posts = False
            for post in response['posts']:
                yield post
        pos = pos + limit
        time.sleep(sleep_sec_per_api)

if __name__ == '__main__':
    main()

上記プログラムの[Consumer Key]と[Base Hostname]を修正して実行すると、Tumblrの投稿取得を確認できるかと思います。

--- 実行結果 ---

id: ...
post_url: ...
title: ...
body: ...
----------------------------------------------------------------------
id: ...
post_url: ...
...

↑ 目次に戻る

Tumblrバックアップアプリ

これまでの作業をふまえて、Google App Engineで動く、Tumblrのテキスト形式の投稿をバックアップするアプリを完成させます。以下の説明では、「tumblr_backup」というディレクトリにすべてのファイルを保存したとして説明しております。

実際に使用される場合は「app.yaml」の[Application Identifier]と、「tumblr_utils.py」の[Consumer Key]と[Base Hostname]を修正する必要があります。なお、このアプリでは1日1回起動のみを想定しているため、2重起動のロック処理などは実装しておりません。

tumblr_backup下に置くファイル

app.yamlの作成

application: [Application Identifier]
version: 1
runtime: python
api_version: 1

handlers:
- url: /.*
  script: main.py
#  login: admin

builtins:
- deferred: on

main.pyの作成

#coding: utf-8

import os

from google.appengine.ext import deferred
from google.appengine.ext import webapp
from google.appengine.ext.webapp import util

import models
import queue_logics
import tumblr_utils


class BackupTextPosts(webapp.RequestHandler):

    def get(self):
        for post in tumblr_utils.post_iterator('text'):
            ds_post = models.TextPost.get_or_insert(key_name=str(post['id']))
            if not ds_post.equals(post):
                deferred.defer(queue_logics.update_text_post, post)

def main():
    routes = [
        ('/backup/text', BackupTextPosts),
    ]
    application = webapp.WSGIApplication(routes,
        debug=os.environ.get('SERVER_SOFTWARE', '').startswith('Dev'))
    util.run_wsgi_app(application)

if __name__ == "__main__":
    main()

models.pyの作成

#coding: utf-8

from google.appengine.ext import db
from google.appengine.ext.db import polymodel


class Post(polymodel.PolyModel):
    post_url = db.StringProperty()
    post_type = db.StringProperty()
    timestamp = db.IntegerProperty()
    post_date = db.StringProperty()
    post_format = db.StringProperty()
    reblog_key = db.StringProperty()
    tags = db.StringListProperty()
    bookmarklet = db.BooleanProperty()
    mobile = db.BooleanProperty()
    source_url = db.StringProperty()
    source_title = db.StringProperty()

    def equals(self, post):
        '''
        引数で与えられた投稿と等しいかを判定します。
        Args:
            post: 比較対象の投稿(辞書型)
        Returns:
            True: 一致
            False: 不一致
        '''
        if self.post_url != post['post_url']:
            return False
        if self.post_type != post['type']:
            return False
        if self.timestamp != post['timestamp']:
            return False
        if self.post_date != post['date']:
            return False
        if self.post_format  != post['format']:
            return False
        if self.reblog_key != post['reblog_key']:
            return False
        if self.tags != post['tags']:
            return False
        if self.bookmarklet != 'bookmarklet' in post:
            return False
        if self.mobile != 'mobile' in post:
            return False
        if self.source_url != post['source_url'] if 'source_url' in post else u'':
            return False
        if self.source_title != post['source_title'] if 'source_title' in post else u'':
            return False
        return True


class TextPost(Post):
    title = db.StringProperty()
    body = db.TextProperty()

    def equals(self, post):
        if not super(TextPost, self).equals(post):
            return False
        if self.title != post['title']:
            return False
        if self.body != post['body']:
            return False
        return True

queue_logics.pyの作成

#coding: utf-8

import models

def update_text_post(post):
    ds_post = models.TextPost.get_by_key_name(str(post['id']))
    ds_post.post_url = post['post_url']
    ds_post.post_type = post['type']
    ds_post.timestamp = post['timestamp']
    ds_post.post_date = post['date']
    ds_post.post_format = post['format']
    ds_post.reblog_key = post['reblog_key']
    ds_post.tags = post['tags']
    ds_post.bookmarklet = 'bookmarklet' in post
    ds_post.mobile = 'mobile' in post
    ds_post.source_url = post['source_url'] if 'source_url' in post else u''
    ds_post.source_title = post['source_title'] if 'source_title' in post else u''
    ds_post.title = post['title']
    ds_post.body = post['body']
    ds_post.put()

tumblr_utils.pyの作成

#coding: utf-8

import time

from google.appengine.api import urlfetch

from django.utils import simplejson as json

CONSUMER_KEY = '[Consumer Key]'
BASE_HOSTNAME = '[Base Hostname]'

def post_iterator(post_type='text'):
    # API毎にスリープする秒数
    sleep_sec_per_api = 10

    # 一度に取得する件数(Tumblrの仕様で最大20件)
    limit = 20
    pos = 0
    has_more_posts = True
    url_template = 'http://api.tumblr.com/v2/blog/%(base_hostname)s/posts/' 
            '%(post_type)s?api_key=%(api_key)s&offset=%(offset)s&limit=%(limit)s'
    while has_more_posts:
        url = url_template % dict(
            base_hostname=BASE_HOSTNAME, post_type=post_type,
            api_key=CONSUMER_KEY, offset=pos, limit=limit)
        result = urlfetch.fetch(url)
        if not result.status_code == 200:
            # 200が返ってこなかったら中止して、次回実施時に任せます。
            has_more_posts = False
            continue
        response = json.loads(result.content)['response']
        if response['total_posts'] <= pos + limit:
            # すべての投稿を取得した場合は継続フラグを落とします。
            has_more_posts = False
        for post in response['posts']:
            yield post
        pos = pos + limit
        time.sleep(sleep_sec_per_api)

これらファイルをGoogle App Engineにアップするには以下のようにします。

Tumblrバックアップアプリのデプロイ

appcfg.py update tumblr_backup

これでアプリをデプロイすることができました。バックアップを起動するには以下のURLを叩きます。

http://[Application Identifier].appspot.com/backup/text

↑ 目次に戻る

Cronの設定

Tumblrの投稿をバックアップするのに毎回URLを叩くのは面倒なので、Cronを登録して自動的にバックアップするようにします。まずは、プログラム類と同じディレクトリに「cron.py」というファイルを作成します。

<

h3 id="file_cron">cron.pyの作成

cron:
- description: backup tumblr test posts
  url: /backup/text
  schedule: every day 12:20
  timezone: Asia/Tokyo

これは毎日12時30分に「/backup/text」を実施するように設定してあります。この辺の設定は、以前書いた「GAE/PのCronで定期的に処理を実施する方法」という投稿が参考になるかもしれません。

Cronのファイルをアップロードするには以下のコマンドをうちます。

Cronファイルのアップロード

appcfg.py update_cron tumblr_backup

最後にCronが登録されているかを確認します。

Cronの確認

appcfg.py cron_info tumblr_backup

↑ 目次に戻る

この記事が役に立った場合、シェアしていただけると励みになります!!