Source

scripts / check-hg-repos.py

Full commit
#!/usr/bin/env python
# -*- coding: utf-8 -*-
# 2011-09-03

import os
import mercurial.ui
import mercurial.hg
import mercurial.commands
import mercurial.error
import colorama
from colorama import Fore, Back, Style
from argh import *


ui = mercurial.ui.ui()
colorama.init()


def get_repo(path):
    return mercurial.hg.repository(ui, path)


@command
def check_repo(path, **kwargs):
    "Checks if given repo has uncommitted/outgoing changes."
    status = check_repo_status(path, **kwargs)
    incoming = check_repo_incoming(path)
    outgoing = check_repo_outgoing(path)
    return ', '.join(unicode(x) for x in (status, incoming, outgoing) if x)


@command
def check_repo_status(path, skip_modified=False, skip_added=False,
                      skip_removed=False, skip_missing=False,
                      skip_unknown=False):
    "Checks if given repo has uncommitted changes."
    repo = get_repo(path)

    repo.ui.pushbuffer()
    mercurial.commands.status(repo.ui, repo)
    status = repo.ui.popbuffer()

    prefixes_ordered = 'MAR!?'
    prefixes = {
        'M': ('modified', skip_modified, Fore.BLUE),
        'A': ('to add', skip_added, Fore.GREEN),
        'R': ('to remove', skip_removed, Fore.RED),
        '!': ('missing', skip_missing, Fore.CYAN),
        '?': ('stray file(s)', skip_unknown, Fore.MAGENTA),
    }

    # count items that match each prefix (if any)
    found = {}
    items = (x for x in status.splitlines() if x)
    for item in items:
        prefix = item[0]
        found.setdefault(prefix, 0)
        found[prefix] += 1

    # compile a list of nicely formatted messages per prefix
    messages = []
    for prefix in prefixes_ordered:
        label, skip, colour = prefixes[prefix]
        if skip:
            continue
        if not skip and prefix in found:
            msg = '{0} {1}'.format(found[prefix], label)
            messages.append(colour + msg + Fore.RESET)

    return ', '.join(messages)


@command
def check_repo_incoming(path):
    "Checks if given repo has incoming changes."
    repo = get_repo(path)

    repo.ui.pushbuffer()
    try:
        mercurial.commands.incoming(repo.ui, repo, bundle='hg.bundle',
                                    force=False)
    except mercurial.error.RepoError as e:
        return e
    incoming = repo.ui.popbuffer()

    incoming_cnt = incoming.count('changeset')
    if incoming_cnt:
        msg = Fore.RED + 'OUT OF DATE ({0} changesets)' + Fore.RESET
        return msg.format(incoming_cnt)


@command
def check_repo_outgoing(path):
    "Checks if given repo has outgoing changes."
    repo = get_repo(path)

    repo.ui.pushbuffer()
    try:
        mercurial.commands.outgoing(repo.ui, repo)
    except mercurial.error.RepoError as e:
        return e
    outgoing = repo.ui.popbuffer()

    outgoing_cnt = outgoing.count('changeset')
    if outgoing_cnt:
        msg = Fore.GREEN + 'has {0} outgoing changesets' + Fore.RESET
        return msg.format(outgoing_cnt)


@command
def check_repos(path, **kwargs):
    "Checks all repos in given path."
    print u'Checking repositories inside "{0}"...'.format(path)
    for repo_name in sorted(os.listdir(path)):
        repo_path = os.path.join(path, repo_name)
        if os.path.isdir(repo_path):
            yield u'* {0}'.format(Style.BRIGHT + repo_name + Style.NORMAL)
            yield ''
            try:
                messages = check_repo(repo_path, **kwargs)
            except mercurial.error.RepoError as e:
                yield '      (not a Hg repo) {0}'.format(e)
            else:
                if messages:
                    yield u'      {0}'.format(messages)
            yield ''

if __name__ == '__main__':
    parser = ArghParser()
    parser.add_commands([
        check_repo,
        check_repo_status,
        check_repo_incoming,
        check_repo_outgoing,
        check_repos
    ])
    parser.dispatch()