celery/extra/release/bump_version.py

#!/usr/bin/env python

from __future__ import absolute_import

import errno
import os
import re
import shlex
import subprocess
import sys

from contextlib import contextmanager
from tempfile import NamedTemporaryFile

rq = lambda s: s.strip("\"'")

str_t = str if sys.version_info[0] >= 3 else basestring


def cmd(*args):
    return subprocess.Popen(args, stdout=subprocess.PIPE).communicate()[0]


@contextmanager
def no_enoent():
    try:
        yield
    except OSError as exc:
        if exc.errno != errno.ENOENT:
            raise


class StringVersion(object):

    def decode(self, s):
        s = rq(s)
        text = ""
        major, minor, release = s.split(".")
        if not release.isdigit():
            pos = release.index(re.split("\d+", release)[1][0])
            release, text = release[:pos], release[pos:]
        return int(major), int(minor), int(release), text

    def encode(self, v):
        return ".".join(map(str, v[:3])) + v[3]
to_str = StringVersion().encode
from_str = StringVersion().decode


class TupleVersion(object):

    def decode(self, s):
        v = list(map(rq, s.split(", ")))
        return (tuple(map(int, v[0:3])) +
                tuple(["".join(v[3:])]))

    def encode(self, v):
        v = list(v)

        def quote(lit):
            if isinstance(lit, str_t):
                return '"{0}"'.format(lit)
            return str(lit)

        if not v[-1]:
            v.pop()
        return ", ".join(map(quote, v))


class VersionFile(object):

    def __init__(self, filename):
        self.filename = filename
        self._kept = None

    def _as_orig(self, version):
        return self.wb.format(version=self.type.encode(version),
                              kept=self._kept)

    def write(self, version):
        pattern = self.regex
        with no_enoent():
            with NamedTemporaryFile() as dest:
                with open(self.filename) as orig:
                    for line in orig:
                        if pattern.match(line):
                            dest.write(self._as_orig(version))
                        else:
                            dest.write(line)
                os.rename(dest.name, self.filename)

    def parse(self):
        pattern = self.regex
        gpos = 0
        with open(self.filename) as fh:
            for line in fh:
                m = pattern.match(line)
                if m:
                    if "?P<keep>" in pattern.pattern:
                        self._kept, gpos = m.groupdict()["keep"], 1
                    return self.type.decode(m.groups()[gpos])


class PyVersion(VersionFile):
    regex = re.compile(r'^VERSION\s*=\s*\((.+?)\)')
    wb = "VERSION = ({version})\n"
    type = TupleVersion()


class SphinxVersion(VersionFile):
    regex = re.compile(r'^:[Vv]ersion:\s*(.+?)$')
    wb = ':Version: {version}\n'
    type = StringVersion()


class CPPVersion(VersionFile):
    regex = re.compile(r'^\#\s*define\s*(?P<keep>\w*)VERSION\s+(.+)')
    wb = '#define {kept}VERSION "{version}"\n'
    type = StringVersion()


_filetype_to_type = {"py": PyVersion,
                     "rst": SphinxVersion,
                     "txt": SphinxVersion,
                     "c": CPPVersion,
                     "h": CPPVersion}


def filetype_to_type(filename):
    _, _, suffix = filename.rpartition(".")
    return _filetype_to_type[suffix](filename)


def bump(*files, **kwargs):
    version = kwargs.get("version")
    before_commit = kwargs.get("before_commit")
    files = [filetype_to_type(f) for f in files]
    versions = [v.parse() for v in files]
    current = list(reversed(sorted(versions)))[0]  # find highest
    current = current.split()[0]  # only first sentence

    if version:
        next = from_str(version)
    else:
        major, minor, release, text = current
        if text:
            raise Exception("Can't bump alpha releases")
        next = (major, minor, release + 1, text)

    print("Bump version from {0} -> {1}".format(to_str(current), to_str(next)))

    for v in files:
        print("  writing {0.filename!r}...".format(v))
        v.write(next)

    if before_commit:
        cmd(*shlex.split(before_commit))

    print(cmd("git", "commit", "-m", "Bumps version to {0}".format(
        to_str(next)), *[f.filename for f in files]))
    print(cmd("git", "tag", "v{0}".format(to_str(next))))


def main(argv=sys.argv, version=None, before_commit=None):
    if not len(argv) > 1:
        print("Usage: distdir [docfile] -- <custom version>")
        sys.exit(0)

    args = []
    for arg in argv:
        if arg.startswith("--before-commit="):
            _, before_commit = arg.split('=')
        else:
            args.append(arg)

    if "--" in args:
        c = args.index('--')
        version = args[c + 1]
        argv = args[:c]
    bump(*args[1:], version=version, before_commit=before_commit)

if __name__ == "__main__":
    main()