blog

#!/usr/bin/env python

import datetime
import glob
import io
import json
import os
import re
import shutil
import signal
import socket
import subprocess
import sys
import tempfile
import time
import webbrowser
from collections import Counter
from os.path import *
from typing import Tuple
from urllib.error import URLError
from urllib.request import urlopen

import PIL.Image
import PIL.ImageGrab
import cairosvg
import click
import iso8601
import psutil

THIS_DIR = abspath(dirname(__file__))
PUBLIC_DIR = normpath(join(THIS_DIR, 'public'))
BASE_DIR = normpath(join(THIS_DIR, 'emptysquare'))
CONTENT_DIR = join(BASE_DIR, 'content')
STATIC_DIR = join(BASE_DIR, 'static')

FULL_WIDTH = 2400
THUMB_WIDTH = 240


def get_image_dims(fpath: str) -> Tuple[int, int]:
    img = PIL.Image.open(fpath)
    return img.size


def computed_height(width, actual_width, actual_height):
    return int(float(actual_height) * (float(width) / float(actual_width)))


def get_quality(source_filename):
    ext = source_filename.rsplit('.', 1)[-1].lower()
    if ext.lower() == 'png':
        return 75
    return 85


def img_size_to(source_img, dst, width, suffix=''):
    root, ext = splitext(source_img)
    thumbnail_filename = '%s%s%s' % (basename(root), suffix, ext)
    target_img = join(dst, thumbnail_filename)
    if exists(target_img):
        os.remove(target_img)

    w, h = get_image_dims(source_img)
    if w <= width:
        # Don't resize.
        shutil.copy2(source_img, target_img)
    else:
        height = computed_height(width, w, h)
        cmdline = ['magick', source_img,
                   '-resize', '%dx%d' % (width, height),
                   '-auto-orient',
                   '-quality', str(get_quality(source_img)),
                   target_img]

        subprocess.check_call(cmdline)

    return thumbnail_filename


def read(path):
    with open(path, 'r') as f:
        return f.read()


def write(path, contents):
    with open(path, 'w') as f:
        f.write(contents)


@click.group()
def cli():
    """emptysqua.re blog utility."""
    pass


def is_image(filename):
    ext = splitext(filename)[-1]
    return ext.lower() in ('.png', '.jpg', '.jpeg')


def is_resized(filename):
    return re.match(r'.*?@\d+\.(png|jpg|jpeg)', filename, re.IGNORECASE)


def localpath(ctx, param, where):
    if where:
        return normpath(expanduser(where))


def wherepath(ctx, param, where):
    if not where:
        return

    _, ext = splitext(where)
    if ext:
        raise click.BadParameter("specify location without extension")

    return localpath(ctx, param, join(CONTENT_DIR, where))


@cli.command('draft')
@click.argument('where', type=click.Path(), callback=wherepath)
@click.argument('images', type=click.Path(), callback=localpath, required=False)
def blog_draft(where, images):
    made_dir = False

    if exists(where):
        raise click.BadParameter("%s already exists!" % where)

    fpath = where + ".md"
    if exists(fpath):
        raise click.BadParameter("%s already exists!" % fpath)

    if images:
        images = normpath(expanduser(images))
        if isdir(images):
            filenames = [join(images, filename)
                         for filename in os.listdir(images)]
        elif isfile(images):
            filenames = [images]
        else:
            raise click.FileError('"%s" does not exist!' % images)

        for filename in filenames:
            if not is_image(filename):
                continue

            print(filename)

            if not made_dir:
                os.makedirs(where)
                made_dir = True

            img_size_to(filename, where, FULL_WIDTH)

        plural = "s" if len(images) > 1 else ""
        images_front_matter = 'thumbnail = "%s"' % (basename(min(filenames)),)

        # Use "gallery" shortcode, in themes/hugo_theme_emptysquare/shortcodes/.
        images_markdown = """{{< gallery path="%s" >}}

<span style="color: gray">Image%s &copy; A. Jesse Jiryu Davis</span>
""" % (basename(where), plural)

    else:
        images_front_matter = None
        images_markdown = ""

    write(fpath, """+++
type = "post"
title = ""
description = ""
category = []
tag = []
draft = true
enable_lightbox = true%s
+++

%s
""" % ("\n" + images_front_matter if images_front_matter else "",
       images_markdown))

    print(fpath)
    subprocess.call(['pycharm', fpath])


@cli.command('mv')
@click.option('--no-redirect')
@click.argument('from_name', metavar='from', type=click.Path(),
                callback=wherepath)
@click.argument('to_name', metavar='to', type=click.Path(), callback=localpath)
def blog_move(no_redirect, from_name, to_name):
    from_fpath = from_name + '.md'
    if not exists(from_fpath):
        raise click.FileError(from_fpath)

    if exists(to_name):
        raise click.BadParameter("%s already exists!" % to_name)

    to_fpath = to_name + '.md'
    if exists(to_fpath):
        raise click.BadParameter("%s already exists!" % to_fpath)

    os.rename(from_fpath, to_fpath)
    if isdir(from_name):
        shutil.copytree(from_name, to_name)

    if not no_redirect:
        redirects_path = normpath(join(STATIC_DIR, '_redirects'))

        redirects = read(redirects_path)

        from_path = '/blog/' + basename(from_name)
        to_path = '/blog/' + basename(to_name)
        if not re.search(r'^' + from_path, redirects):
            write(redirects_path, redirects.strip() + """
{:<35}{}
""".format(from_path, to_path))


def parse_post(content):
    state = "init"
    front_matter_lines = []
    content_lines = []
    for line in content.split('\n'):
        if line.strip() == '+++':
            if state == "init":
                state = "front matter"
            else:
                state = "content"

        elif state == "front matter":
            front_matter_lines.append(line.strip())

        else:
            content_lines.append(line)

    parsed = {}
    for l in front_matter_lines:
        key, value = l.split('=', 1)
        value = value.strip()
        if value == 'true':
            value = True
        elif value == 'false':
            value = False

        try:
            value = iso8601.parse_date(value)
        except iso8601.iso8601.ParseError:
            try:
                value = eval(str(value))
            except:
                pass

        parsed[key.strip()] = value

    return parsed, "\n".join(content_lines)


def unparse(post, contents):
    def to_str(value):
        if isinstance(value, datetime.datetime):
            return value.isoformat()
        else:
            return json.dumps(value)

    return """+++
%s
+++

%s
""" % (
        "\n".join(
            "%s = %s" % (k, to_str(v)) for k, v in sorted(post.items())),
        contents.strip())


def _gen_thumbnail(where, post, dirpath):
    images = [fname for fname in os.listdir(dirpath)
              if is_image(fname) and not is_resized(fname)]

    thumb_file = None
    if len(images) > 1:
        # Require author to choose one image as the thumbnail.
        if not post.get('thumbnail'):
            raise click.BadParameter('"%s" no thumbnail!' % where)

        thumb_file = join(dirpath, post['thumbnail'])
    elif len(images) == 1:
        thumb_file = join(dirpath, images[0])
        post['thumbnail'] = images[0]

    if thumb_file:
        if not isfile(thumb_file):
            raise click.BadParameter('thumbnail "%s" not found!'
                                     % post['thumbnail'])

        print(img_size_to(thumb_file, dirpath, THUMB_WIDTH,
                          suffix='@%d' % THUMB_WIDTH))


def _pngs_from_svgs(where):
    for fname in os.listdir(where):
        name, ext = os.path.splitext(fname)
        if not ext.lower() == '.svg':
            continue

        dst = os.path.join(where, f'{name}.png')
        if os.path.exists(dst):
            continue

        print(f'Converting {fname} -> {fname}.png')
        png_data = cairosvg.svg2png(url=os.path.join(where, fname))
        image = PIL.Image.open(io.BytesIO(png_data))
        if image.width > FULL_WIDTH:
            ratio = FULL_WIDTH / float(image.width)
            new_height = int(float(image.height) * ratio)
            image = image.resize((FULL_WIDTH, new_height), PIL.Image.LANCZOS)

        # Save the resized image to the specified path
        image.save(dst, format='PNG')


@cli.command('replace-quotes')
@click.argument('where', type=click.Path(), callback=wherepath)
def blog_replace_quotes(where):
    fpath = where + ".md"
    if not exists(fpath):
        raise click.FileError(fpath)

    post, contents = parse_post(read(fpath))
    for smart, dumb in [("\u2018", "'"), ("\u2019", "'"),
                        ("\u201c", '"'), ("\u201d", '"')]:
        contents = contents.replace(smart, dumb)

    write(fpath, unparse(post, contents))


@cli.command('publish')
@click.argument('where', type=click.Path(), callback=wherepath)
def blog_publish(where):
    fpath = where + ".md"
    if not exists(fpath):
        raise click.FileError(fpath)

    post, contents = parse_post(read(fpath))

    if not post.get('draft', False):
        raise click.BadParameter('"%s" already published!' % where)

    if not post.get('description'):
        raise click.BadParameter('"%s" missing description!' % where)

    description_len = len(post['description'])
    if description_len > 150:
        raise click.BadParameter(
            '"%s" description is too long: %d characters, aim for 150' %
            (where, description_len))

    if not post.get('title'):
        raise click.BadParameter('"%s" missing title!' % where)

    if not post.get('category') and post['type'] == 'post':
        raise click.BadParameter('"%s" no categories!' % where)

    for c in post['category']:
        if not (len(c) and c[0].upper() == c[0]):
            raise click.BadParameter('category "%s" should be title-cased' % c)

    if isdir(where):
        _pngs_from_svgs(where)
        _gen_thumbnail(where, post, where)

    post['date'] = datetime.datetime.now().isoformat()
    post['draft'] = False
    write(fpath, unparse(post, contents))


@cli.command('thumbnail')
@click.argument('where', type=click.Path(), callback=wherepath)
def thumbnail(where):
    fpath = where + ".md"
    if not exists(fpath):
        raise click.FileError(fpath)

    post, contents = parse_post(read(fpath))

    if not isdir(where):
        raise click.FileError(where)

    _gen_thumbnail(where, post, where)
    write(fpath, unparse(post, contents))


@cli.command('media')
@click.argument('where', type=click.Path(), callback=wherepath)
def media(where):
    fpath = where + ".md"
    if not exists(fpath):
        raise click.FileError(fpath)

    if not isdir(where):
        os.mkdir(where)

    subprocess.check_call(['open', where])


def _add_image(where, image):
    fpath = where + ".md"
    if not exists(fpath):
        raise click.FileError(fpath)

    if not isdir(where):
        os.mkdir(where)

    shutil.copy(image, where)
    dest_path = os.path.join(where, os.path.split(image)[-1])
    # Shrink if necessary to 2000px wide, using ImageMagick's "convert".
    subprocess.check_call(
        ['magick', dest_path, '-resize', '2000x>', dest_path])

    # Copy the image Markdown to the clipboard so user can paste it wherever.
    image_filename = split(image)[1]
    md = f'![]({image_filename})'
    process = subprocess.Popen('pbcopy', env={'LANG': 'en_US.UTF-8'},
                               stdin=subprocess.PIPE)
    process.communicate(md.encode('utf-8'))
    print('The Markdown code for the image is in your clipboard')


@cli.command('add-image')
@click.argument('where', type=click.Path(), callback=wherepath)
@click.argument('image', type=click.Path(), callback=localpath)
def add_image(where, image):
    _add_image(where, image)


@cli.command('paste-image')
@click.argument('where', type=click.Path(), callback=wherepath)
@click.argument('name')
def paste_image(where, name):
    img = PIL.ImageGrab.grabclipboard()
    assert img
    with tempfile.TemporaryDirectory() as tmpdir:
        image_path = os.path.join(tmpdir, f'{name}.png')
        img.save(image_path, format='PNG')
        _add_image(where, image_path)

    subprocess.check_call(['osascript', '-e', '''
tell application "PyCharm" to activate

tell application "System Events"
    tell application process "PyCharm"
        tell menu bar 1
            tell menu bar item "Edit"
                tell menu "Edit"
                    tell menu item "Paste"
                        tell menu "Paste"
                            click menu item "Paste"
                        end tell
                    end tell
                end tell
            end tell
        end tell
    end tell
end tell'''])


@cli.command('pngs-from-svgs')
@click.argument('where', type=click.Path(), callback=wherepath)
def pngs_from_svgs(where):
    if not isdir(where):
        raise click.FileError(where)

    _pngs_from_svgs(where)

def parsed_posts():
    for name in glob.glob(CONTENT_DIR + '/*.md'):
        post, contents = parse_post(read(join(CONTENT_DIR, name)))
        yield post, contents


def counts(field):
    counter = Counter()
    for post, _ in parsed_posts():
        counter.update(post.get(field, []))

    sorted_cnts = sorted([(value, cnt) for value, cnt in counter.items()],
                         key=lambda pair: -pair[1])

    print('\n'.join('{:<23}{:>3}'.format(value, cnt)
                    for value, cnt in sorted_cnts))


@cli.command('tag')
@click.argument('tag')
def tag(tag):
    for post, _ in parsed_posts():
        if tag in post.get('tag', []):
            print(post['title'])


@cli.command('tags')
def tags():
    counts('tag')


@cli.command('categories')
def categories():
    counts('category')


@cli.command('drafts')
def categories():
    for post, _ in parsed_posts():
        if post.get('draft'):
            print(post['title'])


SUPERVISOR_CONF = 'supervisord.conf'
BINDIR = os.path.dirname(sys.executable)


def start_supervisord():
    subprocess.check_output(['%s/supervisord' % BINDIR, '-c', SUPERVISOR_CONF])


def supervisorctl(command):
    subprocess.check_call(
        ['%s/supervisorctl' % BINDIR, '-c', SUPERVISOR_CONF] + command)


@cli.command('server')
@click.argument('action', type=click.Choice(['start', 'stop', 'restart']))
def server(action):
    # Assume Python and Supervisor are both in the same virtualenv.
    if not os.path.exists('supervisord.pid'):
        start_supervisord()
    else:
        # Check that supervisord.pid represents a running process.
        pid = int(open('supervisord.pid').read().strip())
        try:
            psutil.Process(pid)
        except psutil.NoSuchProcess:
            start_supervisord()

    supervisorctl([action, 'all'])
    if action in ('start', 'restart'):
        # Wait for startup.
        for _ in range(10):
            try:
                urlopen('https://localhost:1313/blog/').read()
            except (URLError, socket.error) as exc:
                print(exc)
                supervisorctl(['tail', 'hugo'])
                time.sleep(1)

        # Sometimes needs another second.
        time.sleep(1)
    elif action == 'stop':
        supervisorctl(['shutdown'])


@cli.command('preview')
@click.argument('where', type=click.Path(), callback=wherepath)
@click.pass_context
def preview(ctx, where):
    fpath = where + ".md"
    if not exists(fpath):
        raise click.FileError(fpath)

    ctx.invoke(server, action='start')
    path = where.split('content/')[-1]
    webbrowser.open_new_tab('https://localhost:1313/blog/%s' % path)


@cli.command('build')
def build():
    if os.path.exists('supervisord.pid'):
        os.kill(int(open('supervisord.pid').read().strip()), signal.SIGTERM)

    subprocess.check_call(['hugo', '-v', '-d', join(PUBLIC_DIR, 'blog')],
                          cwd=BASE_DIR)

    shutil.copytree(STATIC_DIR, PUBLIC_DIR, dirs_exist_ok=True)


@cli.command('deploy')
@click.pass_context
def deploy(ctx):
    ctx.invoke(build)
    subprocess.check_call(
        ['netlify', 'deploy', '-s', 'emptysquare', '-p', 'public'])


cli()