Python 3.3 script to coerce Kiln Harmony into finishing conversion of a large git repository (e.g. the Linux kernel), simulating multiple presses of the Repair button until the job is done. Requires BeautifulSoup 4, Requests, and pointfree.

coerce_kiln.py

#!/usr/bin/env python3

"""Force Kiln Harmony to finish repairing a large repository

I'm currently seeing an issue with Kiln Harmony where an imported Linux
kernel git repository fails partway through translation to Mercurial. It
seems that I can keep clicking the "Repair Repository" button to prod Kiln
into finishing more of the translation, however this is time consuming.

This script automates the process of checking whether the translation of a
given repository has completed yet and, if not, asks Kiln to try repairing
the repo.

"""

__author__ = "Mark Shroyer"
__date__   = "2013-4-20"

from datetime import datetime
from bs4 import BeautifulSoup
from getpass import getpass
from pointfree import pfmap, pffilter, pfreduce
import re
import requests
from requests import ConnectionError, HTTPError, RequestException
from time import sleep

class KilnError(Exception):
    pass

class KilnAuthError(KilnError):
    pass

def merge_dicts(a, b):
    result = a.copy()
    result.update(b)
    return result

def form_params(form, replace={}):
    """Get request params from a BeautifulSoup form parse

    Given a BeautifulSoup form object, returns a dict of parameters from
    the form's default values, optionally replacing select values with
    those from the replace argument.

    """

    get_inputs = pffilter(lambda input: 'name' in input.attrs) \
                 >> pfmap(lambda input: {input.attrs['name']: \
                        input.attrs['value'] if 'value' in input.attrs else ''}) \
                 >> pfreduce(merge_dicts, initial={})
    inputs = get_inputs(form.find_all('input'))
    inputs.update(replace)
    return inputs

class KilnRepo:
    """Kiln repository wrapper"""

    def __init__(self, url, username, password):
        self.url = url
        self.username = username
        self.password = password

        result = re.match("^https?://([^./]+\.kilnhg\.com)/(.*?)/?$", self.url)
        if not result:
            raise ValueError("Invalid repository URL")
        self.domain = result.group(1)
        self.repo_path = result.group(2)

    def site_url(self, path=""):
        """Return absolute URL relative to this repo's Kiln site"""

        path = path.lstrip("/")
        return "https://{domain}/{path}".format(domain=self.domain,
                                                path=path)

    def repo_url(self, path=""):
        """Return absolute URL relative to this repo's base URL"""

        path = path.lstrip("/")
        return self.site_url("{repo}/{path}".format(repo=self.repo_path,
                                                    path=path))

    def get_cookies(self):
        """Return Kiln auth cookies, after logging in if necessary"""

        if hasattr(self, '_cookies'):
            return self._cookies

        resp = requests.get(self.site_url())
        if resp.status_code != 200:
            raise KilnError("Failed to get login page")

        page = BeautifulSoup(resp.content)
        params = form_params(page.form, {'sPerson':   self.username,
                                         'sPassword': self.password})

        resp = requests.post(page.form.attrs['action'], params=params)
        if resp.status_code != 200:
            raise KilnError("Failed to log in: Server returned {0}".format(
                resp.status_code
            ))

        page = BeautifulSoup(resp.content)
        if "Incorrect username or password" in page.text:
            raise KilnAuthError("Failed to log in: Incorrect username or password")

        self._cookies = resp.cookies
        return self._cookies

    def logoff(self):
        """Log out of Kiln"""

        if not hasattr(self, '_cookies'): return
        resp = requests.get(self.site_url("Auth/LogOff"), cookies=self._cookies)
        if resp.status_code != 200:
            raise KilnError("Error logging off, status {0}".format(resp.status_code))

    def latest_changeset_date(self, vcs="Hg"):
        """Returns the date of the latest changeset in the repo

        Returns information for either the Hg repository or the Git
        repository, depending on the value of the vcs parameter.

        """

        resp = requests.get(self.repo_url(), cookies=self.get_cookies(),
                            params={"vcs": vcs})
        if resp.status_code != 200:
            raise KilnError("Error getting repo history")

        page = BeautifulSoup(resp.content)
        changesetList = page.find("table", {"class": "changesetList"})
        lastChangeset = changesetList.find("tr")
        dateStr = lastChangeset.find("span", {"class": "autodt"}).attrs['date']

        return datetime.strptime(dateStr, "%a, %d %b %Y %H:%M:%S %Z")

    def check(self):
        """Checks the repository's health

        Returns a BeautifulSoup parse of the repository's Check page. The
        value of this response can subsequently be used to evaluate the
        repository's translation status or to submit a repair request.

        """

        resp = requests.get(self.repo_url("Check"), cookies=self.get_cookies())
        if resp.status_code != 200:
            raise KilnError("Error checking repo status")

        return BeautifulSoup(resp.content)

    def is_translated(self, check_page=None):
        """Determines whether the repository is fully translated

        The contents of a repository Check page can be provided to prevent
        extraneous requests to the server.

        """

        if not check_page:
            check_page = self.check()

        return "is not fully translated" not in check_page.text

    def repair(self, check_page=None):
        """Asks Kiln to repair the repository

        The contents of a repository Check page can be provided to prevent
        extraneous requests to the server. These contents are used to
        simulate manually clicking the Repair button on the check page.

        """

        if not check_page:
            check_page = self.check()

        form = next(filter(lambda form: 'action' in form.attrs \
                                    and '/Repair' in form.attrs['action'], \
                           check_page.find_all("form")))

        resp = requests.post(self.site_url(form.attrs['action']),
                             cookies=self.get_cookies(),
                             params=form_params(form))
        return resp

def coerce_repair(repo):
    """Keep asking Kiln to repair the repository until it's done"""

    while True:
        try:
            check_page = repo.check()

            if repo.is_translated(check_page):
                print("Repo is fully translated!")
                break

            print("Repository is not fully translated")

            latest_hg = repo.latest_changeset_date(vcs="Hg")
            print("Timestamp of latest hg changeset: " + str(latest_hg))

            latest_git = repo.latest_changeset_date(vcs="Git")
            print("Timestamp of latest git commit:   " + str(latest_git))

            print("Repairing...")
            resp = repo.repair(check_page)

            print("Repair returned status code {0}".format(resp.status_code))

        except KilnAuthError as e:
            # Exit if we have a bad username or password...
            raise e

        except (ConnectionError, HTTPError, RequestException, KilnError) as e:
            # ...but keep retrying for other network or application errors
            print("Error talking to Kiln: " + str(e))

        # Wait for Kiln to catch up so that we can get an accurate account
        # of the latest changeset date on the next iteration
        print("Waiting to retry...\n")
        sleep(120)

def main():
    url = input("Kiln repository URL: ")
    username = input("Username: ")
    password = getpass()

    print("\nLogging in...")
    repo = KilnRepo(url, username, password)
    try:
        coerce_repair(repo)
    finally:
        print("Logging off...")
        repo.logoff()

if __name__ == '__main__':
    main()