git-deblog

#!/usr/bin/env python3

# Copyright 2014-2016 dunnhumby Germany GmbH.
# Distributed under the Boost Software License, Version 1.0.
# (See accompanying file LICENSE or copy at http://www.boost.org/LICENSE_1_0.txt)

import sys
from datetime import datetime
from subprocess import CalledProcessError, PIPE


def main():
    setup_decoder()

    parser, args = parse_args()

    # Git assumes UTF-8 for commit messages, unless there is a config option
    # overriding it. Even if the config option is not used, Git just emits
    # warnings, so users could use any encoding, so we allow overriding this
    # via command-line options.
    # See: https://git-scm.com/docs/git-commit#_discussion
    commit_encoding = None
    try:
        commit_encoding = run('git config i18n.logOutputEncoding'.split())
    except CalledProcessError as e:
        if e.returncode != 1: # Config not defined
            raise e
    if not commit_encoding:
        commit_encoding = 'UTF-8'

    git_log = get_log(commit_encoding, args.git_log_args)
    if not git_log:
        return

    first = git_log[0]
    if isinstance(first, Commit):
        tag = Tag(first.hash, args.initial_version, get_git_author(),
                datetime.now())
    else:
        tag = first
    if args.initial_version:
        tag.name = args.initial_version
    if tag.name is None:
        parser.error('{}: no tag for the last commit, you must '
                'specify an initial version (--initial-version)')
    if tag is not first:
        git_log.insert(0, tag)

    for i, obj in enumerate(git_log):
        if isinstance(obj, Tag):
            if i > 0:
                print()
                print(' -- {}  {}'.format(obj.author, obj.date.isoformat()))
                print()

            print('{} ({}) {}; urgency={}'.format(args.pkg_name, obj.name,
                    args.dist_name, args.urgency))
            print()
        else:
            print('  * {} ({})'.format(obj.message, obj.hash[:7]))


def run(*args, **kwargs):
    from locale import getpreferredencoding
    from subprocess import check_output
    encoding = getpreferredencoding()
    dbg('running command: {} (using {} encoding)', args[0], encoding)
    output = check_output(*args, **kwargs).strip()
    (output, errors) = decode_count(output, encoding)
    if errors:
        warn('found {} errors when trying to decode the output of "{}" '
                '(using encoding {})', errors, ' '.join(args[0]), encoding)
    return output


def get_dist():
    return run('lsb_release -cs'.split())


def get_git_author():
    try:
        name = run('git config user.name'.split())
    except CalledProcessError as e:
        warn("can't get author name from `git config user.name` ({}): ", e)
        name = 'Unknown'
    try:
        email = run('git config user.email'.split())
    except CalledProcessError as e:
        warn("can't get author e-mail from `git config user.email` ({})", e)
        email = 'unknown@example.com'
    return '{} <{}>'.format(name, email)


class Tag:
    def __init__(self, hash, name, author, date):
        self.hash = hash
        self.name = name
        self.author = author
        self.date = date
    def __str__(self):
        return self.name
    def __repr__(self):
        return 'Tag({!r}, {!r}, {!r}, {!r})'.format(
                self.hash, self.name, self.author, self.date)


class Commit:
    def __init__(self, hash, message):
        self.hash = hash
        self.message = message
    def __str__(self):
        return self.message
    def __repr__(self):
        return 'Commit({!r}, {!r})'.format(self.hash, self.message)


def get_log(encoding, opts=()):
    from subprocess import Popen

    cmd = ('git', 'log') + tuple(opts) + ('--format=%h%x1f%D%x1f%s%x1e',)
    dbg('running command: {} (using {} encoding)', cmd, encoding)
    p = Popen(cmd, stdout=PIPE)

    log = []
    row = ''
    # Reset decoding errors
    decoding_error_count = 0
    for l in p.stdout.readlines():
        l = l.decode(encoding, errors="replacecount")
        # accumulate if the end of record is not found
        if not l.endswith('\x1e\n'):
            row += l
            continue
        # full record read
        row += l[:-2] # remove record and line terminator
        hash, refs, message = row.split("\x1f")
        tag = get_tag(refs)
        if tag is not None:
            log.append(tag)
        log.append(Commit(hash, message))
        row = ''

    status = p.wait()
    assert status == 0

    if decoding_error_count:
        warn('found {} errors when trying to decode the output of "git log" '
                '(using encoding {})', errors, encoding)

    return log


def get_tag(refs):
    import re
    from datetime import timezone, timedelta

    tag_re = re.compile(r'(?:^|, )tag: (?P<tag>.*?)(?:,|$)')
    tag = None
    tags = set(tag_re.findall(refs))
    for t in sorted(set(tags), reverse=True):
        try:
            ti = run('git cat-file tag'.split() + [t], stderr=PIPE)
            if tag is None:
                tag = t
                tag_info = ti.splitlines()
        except CalledProcessError as e:
            warn("not using tag {} because is not annotated", t)
            tags.remove(t)
    if not tags:
        return None
    def get_field(field):
        hdr = field + ' '
        r = [l for l in tag_info if l.startswith(hdr)]
        return [l for l in tag_info if l.startswith(hdr)][0][len(hdr):]
    hash = get_field('object')
    if len(tags) > 1:
        warn('more than one tag ({}) found for commit {}, using the last '
                'one ({})', ', '.join(tags), hash, tag)
    author, timestamp, offset = get_field('tagger').rsplit(' ', 2)
    tz = timezone(timedelta(minutes=int(offset[0]+'1') * (
            int(offset[1:2])*60 + int(offset[3:4]))))
    date = datetime.fromtimestamp(int(timestamp), tz)
    return Tag(hash, tag, author, date)


def parse_args():
    from argparse import ArgumentParser, REMAINDER

    parser = ArgumentParser(description='Formats git log as a Debian changelog')
    parser.add_argument('-v', '--verbose', action='count', default=1,
            help="be more verbose (can be specified multiple times to get "
            "extra verbosity)")
    parser.add_argument('-i', '--initial-version',
            help="specifies an initial version (by default the tag of the last "
            "commit is used if present)")
    parser.add_argument('-d', '--dist-name',
            help="specifies the distribution name (by default obtained from "
            "lsb_release)")
    parser.add_argument('-u', '--urgency', default='medium',
            choices='low medium high emergency critical'.split(),
            help="specifies the urgency of the update, in Debian terms")
    parser.add_argument('pkg_name', help="package name")
    parser.add_argument('git_log_args', nargs=REMAINDER,
            help="extra arguments to be passed to git log")
    args = parser.parse_args()

    global VERBOSE
    VERBOSE = args.verbose

    if args.dist_name is None:
        args.dist_name = get_dist()

    return parser, args


decoding_error_count = 0
def setup_decoder():
    # Extend backslashreplace_errors to handle decode errors too.
    # Borrowed from:https://gist.github.com/ynkdir/867347
    # But also store if we had problems decoding
    import codecs
    _backslashreplace_errors = codecs.backslashreplace_errors
    def replacecount_errors(exc):
        if isinstance(exc, UnicodeDecodeError):
            global decoding_error_count
            decoding_error_count += 1
            tohex = lambda c: "\\x{0:02x}".format(c)
            u = "".join(tohex(c) for c in exc.object[exc.start:exc.end])
            return (u, exc.end)
        return _backslashreplace_errors(exc)

    codecs.register_error("replacecount", replacecount_errors)


def decode_count(s, encoding):
    prev_count = decoding_error_count
    return (s.decode(encoding, errors="replacecount"),
            decoding_error_count - prev_count)


def dbg(fmt, *args, **kwargs):
    if VERBOSE > 1:
        sys.stderr.write(('{}: debug: ' + fmt).format(
                sys.argv[0], *args, **kwargs) + '\n')


def warn(fmt, *args, **kwargs):
    if VERBOSE > 0:
        sys.stderr.write(('{}: warning: ' + fmt).format(
                sys.argv[0], *args, **kwargs) + '\n')


def err(fmt, *args, **kwargs):
    sys.stderr.write(('{}: error: ' + fmt).format(
            sys.argv[0], *args, **kwargs) + '\n')


if __name__ == '__main__':
    main()