github-notifier

#! /usr/bin/env python
#
# Needs http://jacquev6.github.io/PyGithub.

import optparse
import time
import os
import shutil
import subprocess
import sys
import github

try:
    # Python 2
    from ConfigParser import ConfigParser, NoSectionError, NoOptionError
except ImportError:
    # Python 3
    from configparser import NoSectionError, NoOptionError
    import configparser

    class ConfigParser(configparser.ConfigParser):
        def __init__(self, *args, **kw):
            kw['interpolation'] = None
            super(ConfigParser, self).__init__(*args, **kw)


VERSION   = "0.7-40"  # Filled in automatically.

try:
    # Find Git binary in a cross-platform way
    # One-liner based on https://stackoverflow.com/a/28909933/2883579
    GitBinary = next(os.path.join(path, "git") for path in os.environ["PATH"].split(os.pathsep) if os.access(os.path.join(path, "git"), os.X_OK))
except StopIteration:
    raise RuntimeError("Git binary not found in PATH")

Name       = "github-notifier"
ConfigFile = "./%s.cfg" % Name

Config = None
Options = None

def log(msg):
    assert Options
    Config.log.write("%s - %s\n" % (time.asctime(), msg))

def error(msg):
    log("Error: %s" % msg)
    sys.exit(1)

def getOption(section, key, default):
    try:
        return Config.get(section, key)
    except (NoSectionError, NoOptionError):
        return default

DirectoryStack = []

def pushDirectory(dir):
    DirectoryStack.append(os.getcwd())
    os.chdir(dir)

def popDirectory():
    os.chdir(DirectoryStack.pop())

def runCommand(args, stdout=None, stderr=None):
    environment = { "PATH": os.environ["PATH"], "GIT_ASKPASS": "echo" }

    if Options.debug:
        sys.stderr.write("> %s\n" % (" ".join(args)))

    try:
        child = subprocess.Popen(args, stdin=None, stdout=stdout, stderr=stderr, env=environment)
        (stdout, stderr) = child.communicate()
    except OSError as e:
        error("cannot run command '%s': %s" % (" ".join(args), e))

    if child.returncode != 0:
        error("command failed with exit code %d" % child.returncode)

def gitClone(repo):
    assert not os.path.exists(repo.path)

    log("creating %s" % repo)

    try:
        os.makedirs(repo.path)
    except IOError as e:
        error("cannot create %s: %s" % (repo.path, e))

    # We don't actually clone here to avoid storing the token in git's
    # configuration. Instead we just use fetch on update.
    # See https://github.com/blog/1270-easier-builds-and-deployments-using-git-over-https-and-oauth

def gitUpdate(repo):
    log("updating %s" % repo)

    pushDirectory(repo.path)
    runCommand([GitBinary, "--bare", "init", "--quiet"])
    runCommand([GitBinary, "--bare", "fetch", "--prune", "--quiet", repo.url(True), "+refs/heads/*:refs/heads/*"])
    runCommand([GitBinary, "--bare", "fetch", "--prune", "--quiet", repo.url(True), "+refs/tags/*:refs/tags/*"])
    runCommand([GitBinary, "remote", "update"], stdout=open(os.devnull, 'wb'), stderr=subprocess.STDOUT)

    # git-notifier picks this up.
    fname = open("repo-name.dat", "w")
    fname.write('%s\n' % repo.name)
    fname.close()

    popDirectory()

def runNotifier(repo):
    log("running git-notifier for %s" % repo)

    opts = []
    opts += ["--repouri=%s" % repo.url(False)]
    opts += ["--link=%s" % ("%s/commit/%%s" % repo.url(False))]

    if Options.debug:
        opts += ["--debug"]
        opts += ["--noupdate"]

    for (key, value) in repo.rset.notifier_options.items():
        public_option = key.endswith("-public")
        private_option = key.endswith("-private")

        if public_option:
            key = key[:len(key) - len("-public")]
        elif private_option:
            key = key[:len(key) - len("-private")]

        use_option = True

        if private_option:
            use_option = repo.priv
        if public_option:
            use_option = not repo.priv

        if use_option:
            if value:
                opts += ["--%s=%s" % (key, value)]
            else:
                opts += ["--%s" % key]

    gn = os.path.abspath(os.path.join(os.path.dirname(sys.argv[0]), "git-notifier"))

    pushDirectory(repo.path)
    runCommand([gn] + opts)
    popDirectory()

def gitRepositories(rset, gh, org = None):
    try:
        repos = [Repository(rset, org, repo.name, repo.private) for repo in gh.get_user(org).get_repos()]
        repos += [Repository(rset, org, repo.name, repo.private) for repo in gh.get_organization(org).get_repos()]
    except github.GithubException as e:
        error("GitHub exception: %s" % e._GithubException__data["message"])

    if not repos:
        log("warning: no GitHub repositories found for set %s" % rset.name)

    return repos

# Information about one repository.
class Repository:
    def __init__(self, rset, org, name, priv):
        if name.endswith(".git"):
            name = name[:-4]

        self.priv = priv
        self.name = name
        self.org = org
        self.rset = rset
        self.path = os.path.join(Options.base_directory, "%s-%s-%s" % (self.rset.name, self.org, self.name))
        self.path = os.path.abspath(self.path)

    def url(self, auth):
        if self.rset.token and auth:
            return "https://%s:x-oauth-basic@github.com/%s/%s" % (self.rset.token, self.org, self.name)
        else:
            return "https://github.com/%s/%s" % (self.org, self.name)

    def printDebug(self):
        sys.stderr.write("  %s/%s (path: %s)\n" % (self.org, self.name, self.path))

    def update(self):
        if not os.path.exists(self.path):
            gitClone(self)

        gitUpdate(self)

        if not Options.update_only:
            runNotifier(self)

    def __str__(self):
        return self.path

    def __equal__(self, other):
        return self.name == other.name and self.org == other.org

# A RepositorySet corresponds to one section in the configuration file.
class RepositorySet:
    def __init__(self, section):
        self.name = section
        self.user = getOption(section, "user", None)
        self.token = getOption(section, "token", None)

        self.notifier_options = {}

        for (key, value) in Config.items(section):
            if key.startswith("notifier-"):
                self.notifier_options[key[9:]] = value

        repos = getOption(section, "repositories", "")

        repos = [r.strip() for r in repos.split(",")]

        all = {}
        include = set()
        exclude = set()

        if repos and self.user and self.token:
            gh = github.Github(self.user, self.token)
        else:
            gh = github.Github()

        for r in repos:

            if r.startswith("-"):
                negate = True
                r = r[1:]
            else:
                negate = False

            m = r.split("/")

            if len(m) == 1:
                org = self.user
                name = m[0]

                if not self.user:
                    error("no user or organisation given for repository '%s'" % name)

            elif len(m) == 2:
                org = m[0]
                name = m[1]

            else:
                error("can't parse '%s'" % r)

            if name == "*":
                srepos = gitRepositories(self, gh, org)
            else:
                gh_name = name[:-5] if name.endswith('.wiki') else name
                priv = gh.get_repo(org + '/' + gh_name).private
                srepos = [Repository(self, org, name, priv)]

            for repo in srepos:
                all[str(repo)] = repo

                if negate:
                    exclude.add(str(repo))
                else:
                    include.add(str(repo))

        self.repositories = [all[r] for r in (include - exclude)]

    def update(self):
        for rset in self.repositories:
            rset.update()

    def printDebug(self):
        sys.stderr.write("Set:%s\n" % self.name)
        sys.stderr.write("  User %s\n" % self.user)
        sys.stderr.write("  Token %s\n" % self.token)

        for (key, value) in self.notifier_options.items():
            sys.stderr.write("  Notifier: %s=%s\n" % (key, value))

        for r in self.repositories:
            r.printDebug()

# Main

optparser = optparse.OptionParser(usage="%prog [options]", version=VERSION)
optparser.add_option("-c", "--config", action="store", dest="config", default=ConfigFile,
                     help="specify alternative configuration file to use")
optparser.add_option("-d", "--debug", action="store_true", dest="debug", default=False,
                     help="enable debug mode, logs to stderr")
optparser.add_option("-l", "--log", action="store", dest="log", default="%s.log" % Name,
                     help="specify an alternative log file to use")
optparser.add_option("-u", "--update-only", action="store_true", dest="update_only", default=False,
                     help="update the local clones only, do not run git-notifier")

(Options, args) = optparser.parse_args()

if len(args) > 1:
    optparser.error("wrong number of arguments")

if not os.path.exists(Options.config):
    sys.stderr.write("configuration file '%s' not found\n" % Options.config)
    sys.exit(1)

Config = ConfigParser()
Config.read(Options.config)

if Options.debug:
    Config.log = sys.stderr
else:
    Config.log = open(Options.log, "a")

(basedir, fname) = os.path.split(Options.config)

Options.base_directory = basedir if basedir else os.getcwd()

sets = []

for section in Config.sections():
    sets += [RepositorySet(section)]

if Options.debug:
    for rset in sets:
        rset.printDebug()

for rset in sets:
    rset.update()