Merging Tumblr blogs Python script.

merge_tumblr_blogs.py

# -*- coding: utf-8 -*-
#
# No Rights Reserved
# http://creativecommons.org/publicdomain/zero/1.0/

"""Merge Tumblr blogs helper.

Load posts from one ("old") blog and copy them to another ("new") one.
The script also tries to prevent duplicates if it will run multiple
times.

Usage:

1. Install dependencies: pip install pytumblr
2. Set auth parameters: consumer_key, consumer_secret, oauth_token, oauth_secret
3. Set source and destination blogs: from_blog, to_blog
4. Run: python merge_blogs.py
"""
import json
import pytumblr

# Setup keys here: https://api.tumblr.com/console
consumer_key = '***'
consumer_secret = '***'
oauth_token = '***'
oauth_secret = '***'

# Source and destination blogs
from_blog = 'from-my-blog' # Read posts from
to_blog = 'to-my-blog' # Save posts to


def extract_from_to(from_, to_, fields):
    """Extract fields from one dict to another."""
    for f in fields:
        if f in from_ and from_[f]:
            if isinstance(from_[f], (list, dict, int, float)):
                to_[f] = from_[f]
            else:
                to_[f] = from_[f].encode('utf-8')


def post_key(p):
    """Post key for preventing duplicates."""    
    title = p.get('title', '') or ''
    return '.'.join((title, p['date'], p['format'], p['slug'], p['state']))


if __name__ == '__main__':
    client = pytumblr.TumblrRestClient(consumer_key, consumer_secret, oauth_token, oauth_secret)
    from_posts = client.posts(from_blog)
    to_posts = client.posts(to_blog)
    to_posts_keys = set(map(post_key, to_posts['posts']))

    print('Merging blogs: rudyryk-en (total: %s) => rudyryk (total: %s)' %
        (from_posts['total_posts'], to_posts['total_posts']))

    counter = 0

    for p in from_posts['posts']:
        # Naive preventing duplicates
        if post_key(p) in to_posts_keys:
            continue

        # Prepare post data
        data = {}
        extract_from_to(p, data, ('state', 'tags', 'tweet', 'date', 'format', 'slug'))

        if p['type'] == 'text':
            extract_from_to(p, data, ('title', 'body'))
            api_meth = client.create_text
        if p['type'] == 'photo':
            extract_from_to(p, data, ('caption', 'link', 'source'))
            api_meth = client.create_photo
        elif p['type'] == 'link':
            extract_from_to(p, data, ('title', 'url', 'description'))
            api_meth = client.create_link
        elif p['type'] == 'quote':
            extract_from_to(p, data, ('quote', 'source'))
            api_meth = client.create_quote
        elif p['type'] == 'chat':
            extract_from_to(p, data, ('title', 'conversation'))
            api_meth = client.create_chat
        elif p['type'] == 'audio':
            extract_from_to(p, data, ('caption', 'external_url'))
            api_meth = client.create_audio
        elif p['type'] == 'video':
            extract_from_to(p, data, ('caption', 'embed'))
            api_meth = client.create_video

        # Create new post
        counter += 1
        api_meth(to_blog, **data)
        print("Post created:", data)

    print("Done. Posts created: %s" % counter)