Skip to content

Instantly share code, notes, and snippets.

@mikhail

mikhail/publish.py

Created Aug 27, 2015
Embed
What would you like to do?
Git -> WordPress draft publisher
#!/usr/bin/env python
import logging
import md5
import optparse
import os
import re
import subprocess
import sys
from wordpress_xmlrpc import Client, WordPressPost
from wordpress_xmlrpc.methods.posts import GetPost, GetPosts, NewPost, EditPost
import markdown
import xmlrpclib
import yaml
"""Create new Wordpress Post from meta.yml files.
Recursively walk the current_dir directory looking for meta.yml
When found - check the specified `slug` and create a post if it's not created.
"""
log = logging.getLogger('WPPost')
logging.basicConfig(stream=sys.stdout, level=logging.DEBUG)
META_FILE = 'meta.yaml'
# Initial option handler to set up the basic application environment.
usage = 'usage: %prog <options>'
parser = optparse.OptionParser(usage=usage, add_help_option=True)
parser.set_defaults(verbose=True)
# Job Configuration
parser.add_option('-d', '--dry', dest='dry', action='store_true',
help='Executes a dry run only.')
# Logging Configuration
parser.add_option('-l', '--level', dest='level', default='info',
help='Set logging level (INFO|WARN|DEBUG|ERROR)')
parser.add_option('', '--debug', dest='level_debug', default=False,
action='store_true', help='Equivalent to --level=DEBUG')
(options, args) = parser.parse_args()
def get_hashsum(directory):
"""Get a unique hash for the entire directory."""
total_bits = subprocess.Popen(
'tar c --exclude .git --exclude "*.log" %s' % directory,
shell=True,
stdout=subprocess.PIPE).stdout.read()
return md5.new(total_bits).hexdigest()
def run():
log.info('Connecting to WordPress RPC...')
wp = Client('https://engblog.nextdoor.com/xmlrpc.php',
os.getenv('WP_USER'),
os.getenv('WP_PASS'))
try:
all_posts = wp.call(GetPosts())
except xmlrpclib.Fault as e:
log.critical('Could not fetch WP posts. Reason: %s' % e)
return False
log.info('Success!')
statuses = dict([(p.slug, p) for p in all_posts])
log.info('Finding meta files')
exit_status = 0
for current_dir, dirs, files in os.walk('.'):
if META_FILE not in files:
continue
log.debug('Found meta file in %s' % current_dir)
meta = yaml.load(open('%s/%s' % (current_dir, META_FILE)).read())
log.debug('Checking content file...')
content_file = '%s/%s' % (current_dir, meta['contents'])
try:
md_content = open(content_file).read()
except IOError as e:
log.error('Path %s cannot be opened. Reason: %s' % (
content_file, e))
exit_status = exit_status + 1
continue
slug = meta['slug']
orig_post = statuses.get(slug)
status = None
if orig_post:
status = orig_post.post_status
if status == 'publish':
log.info('Slug "%s" is already published. Skipping' % slug)
continue
if status == 'draft':
log.warning('Slug "%s" exists. Editing it.' % slug)
post = orig_post
elif status is None:
log.info('Creating a WP Post for "%s"' % slug)
post = WordPressPost()
post.custom_fields = [{'key': 'cpa_author',
'value': meta['author']}]
metaversion = get_hashsum(current_dir)
online_version = [str(pair['value'])
for pair in post.custom_fields
if pair['key'] == 'version']
if len(online_version) == 1 and online_version[0] == metaversion:
log.info('Meta version not updated. Skipping.')
continue
log.info('Version changed to %s. Updating.' % metaversion)
# Update the "version" field
for f in post.custom_fields:
if f['key'] == 'version':
f['value'] = metaversion
break
else:
post.custom_fields.extend([{'key': 'version', 'value': metaversion}])
post.title = meta['title']
post.excerpt = meta['summary']
# Custom Post Author plugin
post.terms_names = { 'post_tag': meta['tags'] }
post.slug = slug
# 1) Escape non-unicode. 2) Convert from Markdown 3) Unescape non-unicode
html_content = markdown.markdown(
md_content.encode('string-escape').replace(r'\n', '\n')
).decode('string-escape')
# Remove single new lines
minified_html = re.sub("([^\n])\n([^\n])", r'\1 \2', html_content)
# Convert more than 2 new lines into just 2
minified_html = re.sub('\n{3,}', '\n\n', minified_html)
post.content = minified_html
log.info('Uploading post "%s" to WordPress...' % slug)
log.debug('Checking if post needs to be created or updated.')
log.debug(post.__dict__)
if hasattr(post, 'id'):
log.info('Updating...')
if not options.dry:
wp.call(EditPost(post.id, post))
else:
log.warning('Dry run. Not doing anything.')
log.info('Done!')
else:
log.info('POSTING NEW BLOG!')
if not options.dry:
pid = wp.call(NewPost(post))
log.debug('Fetching post %s' % pid)
post = wp.call(GetPost(pid))
log.info('Created %s' % post.guid)
else:
log.warning('Dry run. Did not do anything.')
exit(exit_status)
if __name__ == '__main__':
run()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment