svn-import python script meant as a replacement for svn_load_dirs.pl (from http://svn.haxx.se/users/archive-2006-10/0857.shtml) - changed to prevent window popups on windows + performance optimisation for 'add' phase

svn-import.py

#!/usr/bin/env python
# -*-mode: python; coding: utf-8 -*-
#
# svn-import - Import a new release, such as a vendor drop.
#
# The "Vendor branches" chapter of "Version Control with Subversion"
# describes how to do a new vendor drop with:
#
# >The goal here is to make our current directory contain only the
# >libcomplex 1.1 code, and to ensure that all that code is under version
# >control. Oh, and we want to do this with as little version control
# >history disturbance as possible.
#
# This utility tries to take you to this goal - automatically. Files
# new in this release is added to version control, and files removed
# in this new release are removed from version control. It can
# operate on a working copy or a repository URL by automatically
# checking out a working copy. 
#
# Compared to svn_load_dirs.pl, this utility:
#
# * Does not hard-code commit messages
# * Is much less complicated
# * Allows you to fine-tune the import before commit, which
#   allows you to turn adds+deletes into moves. 
#
# TODO:
# Consider not using chdir
# Verify symlink support
# Perhaps support --username and --password
# Perhaps automatically create import dir, if necessary.
# Automatic detection of moved files by comparing basenames. 

import os
import re
import sys
import getopt
import atexit
import shutil
import urlparse
import platform
import tempfile
import subprocess

class _VerboseWriter:
    def __init__(self, verbose=0):
        self.verbose = verbose
    
    def write(self, data):
        if self.verbose:
            sys.stderr.write(data)


def del_temp_tree(tmpdir):
    """Delete tree, standring in the root"""
    os.chdir("/")
    shutil.rmtree(tmpdir)


def copy2_symlinks(src, dst):
    """Just like shutil.copy2, but copy symbolic links"""
    if os.path.islink(src):
        os.symlink(os.readlink(src), dst)
    else:
        shutil.copy2(src, dst)


def url_join_dir(base, url, using_url):
    """Join local path or URL"""
    if using_url:
        # We must add a trailing slash, to indicate the the URL is a
        # directory.
        if not base.endswith("/"):
            base = base + "/"
        # Quick compensation for the fact that Python 2.4 and older does
        # not recoqnize svn:// URLs.
        base = base.replace("svn://", "http://")
        result = urlparse.urljoin(base, url)
        result = result.replace("http://", "svn://")
        return result
    else:
        return os.path.normpath(os.path.join(base, url))


def get_repo_root(path_or_url, using_url):
    """Get repository root"""
    ok = None
    while svn_call(["svn", "proplist", path_or_url], stdout=DEVNULL, stderr=subprocess.STDOUT)==0:
        ok = path_or_url
        path_or_url = url_join_dir(path_or_url, "..", using_url)
        if path_or_url == ok:
            # We have reached the top
            break
    return ok


def removeprefix(path, prefix):
    """Remove prefix from path, which makes it possible to turn an
    absolute path into a relative one. Example:

    /path/to/libcomplex-1.0/doc, /path/to => libcomplex-1.0/doc
    """
    path = os.path.normpath(path)
    prefix = os.path.normpath(prefix)

    if not path.startswith(prefix):
        raise Exception("%s is not a prefix of %s" % (path, prefix))
    
    path_comps = path.split(os.sep)
    prefix_comps = prefix.split(os.sep)
    return os.sep.join(path_comps[len(prefix_comps):])


def get_versioned_files(top):
    """Get versioned files in directory top"""
    files = []
    svnls = svn_popen(["svn", "ls", top], stdout=subprocess.PIPE)
    for line in svnls.stdout:
        # Remove trailing newline
        line = line.rstrip('\r\n')
        # Remove trailing slash for directories
        line = line.replace("/", "")
        files.append(line)
    return files


def svn_call(*args, **kwargs):
  """Launches 'command' windowless"""
  if platform.system() == "Windows":
    startupinfo = subprocess.STARTUPINFO()
    startupinfo.dwFlags |= subprocess.STARTF_USESHOWWINDOW
    kwargs["startupinfo"] = startupinfo
  return subprocess.call(*args, **kwargs)

def svn_popen(*args, **kwargs):
  """Launches 'command' windowless"""
  if platform.system() == "Windows":
    startupinfo = subprocess.STARTUPINFO()
    startupinfo.dwFlags |= subprocess.STARTF_USESHOWWINDOW
    kwargs["startupinfo"] = startupinfo
  return subprocess.Popen(*args, **kwargs)


def file_is_versioned(file):
    """Check if file is under version control"""
    return not svn_call(["svn", "ls", file], stdout=DEVNULL, stderr=subprocess.STDOUT)


def walk_versioned(top):
    """Like os.walk, but only for svn versioned files, without onerror
    support and always topdown"""
    names = get_versioned_files(top)

    dirs, nondirs = [], []
    for name in names:
        if os.path.isdir(os.path.join(top, name)):
            dirs.append(name)
        else:
            nondirs.append(name)

    yield top, dirs, nondirs

    for name in dirs:
        path = os.path.join(top, name)
        if not os.path.islink(path):
            for x in walk_versioned(path):
                yield x


def delete_removed_files(newtree):
    """Loop over versioned files in current dir. If files are not
    found in newtree, do svn delete"""
    for root, dirs, files in walk_versioned("."):
        print >>verbosew, "listing", root
        for name in files + dirs:
            wc_name = os.path.join(root, name)
            newtree_name = os.path.join(newtree, wc_name)
            if not os.path.exists(newtree_name):
                print >>verbosew, "  deleting", wc_name
                svn_call(["svn", "delete", wc_name], stdout=DEVNULL)
                # Prune tree; sufficient to remove the directory
                if name in dirs:
                    dirs.remove(name)
            

def add_new_files(newtree):
    """Copy all files from newtree. For files not versioned in working
    copy, add with svn add. For new directories, do svn mkdir."""
    for root, dirs, files in os.walk(newtree):
        print >>verbosew, "listing", root
        rel_dir = removeprefix(root, newtree)
        wc_content = set( get_versioned_files(rel_dir) )
        for name in files:
            wc_name = os.path.join(rel_dir, name)
            newtree_name = os.path.join(root, name)
            copy2_symlinks(newtree_name, wc_name)
            is_versioned = name in wc_content
            if not is_versioned:
                print >>verbosew, "  adding", wc_name
                svn_call(["svn", "add", wc_name], stdout=DEVNULL)
            
        for name in dirs:
            wc_name = os.path.join(rel_dir, name)
            is_versioned = name in wc_content
            if not is_versioned:
                print >>verbosew, "  mkdir", wc_name
                svn_call(["svn", "mkdir", wc_name], stdout=DEVNULL)


def usage():
    """Print usage message and exit"""
    print >>sys.stderr, """%s: Import a new release, such as a vendor drop.
usage: 1. %s [options] NEW_RELEASE PATH 
       2. %s [options] NEW_RELEASE URL 

  1. The directory specified by the working copy PATH is adapted to the
     directory NEW_RELEASE. Example:
     %s /path/to/libcomplex-1.0 .

  2. The repository directory specified by the URL is adapted to the
     directory NEW_RELEASE. Example:
     %s /path/to/libcomplex-1.0 http://svn.example.com/repos/vendor/libcomplex/current 

  This command executes these steps:

  1. Check out directory specified by URL in a temporary directory (only form 2)
  2. Adapt to the directory NEW_RELEASE
  3. Allow user to fine-tune import. (only form 2, unless overridden)
  4. Commit. (only form 2)
  5. Optionally tag new release.
  6. Delete the temporary directory (only form 2)

Valid options:
  -h [--help]              : show this usage
  -t [--tag] arg           : copy new release to directory ARG, relative to PATH/URL,
                             using automatic commit message. Example:
                             -t ../0.42
  --non-interactive        : do no interactive prompting, do not allow manual fine-tune
  -m [--message] arg       : specify commit message ARG
  -v [--verbose]           : verbose mode
    """ % ((os.path.basename(sys.argv[0]),) * 5)
    sys.exit(1)


def main():
    tag = None
    message = None
    interactive = 1
    global verbosew
    verbosew = _VerboseWriter()

    try:
        opts, args = getopt.gnu_getopt(sys.argv[1:], "ht:m:v",
                                       ["help", "tag", "message", "non-interactive", "verbose"])
    except getopt.GetoptError:
        # print help information and exit:
        usage()

    for o, a in opts:
        if o in ("-h", "--help"):
            usage()
        if o in ("-t", "--tag"):
            tag = a
        if o in ("-m", "--message"):
            message = a
        if o in ("--non-interactive"):
            interactive = 0
        if o in ("-v", "--verbose"):
            verbosew.verbose = 1

    if len(args) != 2:
        usage()

    new_release, path_or_url = args
    new_release = os.path.abspath(new_release)

    # Determine form. We cannot use urlparse, since c:\foo is a valid
    # URL. 
    using_url = re.match("\w+://", path_or_url) is not None

    if using_url:
        # Create a temp dir to hold our working copy
        wc_dir = tempfile.mkdtemp(prefix="svn-import")
        atexit.register(del_temp_tree, wc_dir)

        # Check out "current"
        print >>sys.stderr, "Checking out..."
        svn_call(["svn", "checkout", path_or_url, wc_dir])
    else:
        # We'll need an absolute URL, for various reasons. For
        # example, "svn copy . ../0.47" gives "Cannot copy path '.'
        # into its own child "
        path_or_url = os.path.abspath(path_or_url)
        wc_dir = path_or_url

    repo_root = get_repo_root(path_or_url, using_url)
    if repo_root == None:
        sys.exit("Error: %s is not a valid URL or working copy PATH" % path_or_url)

    # Verify tag directory
    if tag != None:
        tag_dest = url_join_dir(path_or_url, tag, using_url)
        if not tag_dest.startswith(repo_root):
            sys.exit("Error: %s is outside working copy %s" % (tag_dest, repo_root))

    os.chdir(wc_dir)

    # turn into new release
    print >>sys.stderr, "Adapting to %s..." % new_release
    delete_removed_files(new_release)
    add_new_files(new_release)

    if using_url and interactive:
        # Give the user a chance to fine-tune
        print >>sys.stderr, "If you want to fine-tune import, do so in working copy located at:", wc_dir
        print >>sys.stderr, "When done, press Enter to commit, or Ctrl-C to abort."
        try:
            sys.stdin.readline()
        except KeyboardInterrupt:
            sys.exit(0)

    if using_url:
        # Commit
        print >>sys.stderr, "Committing..."
        cmd = ["svn", "commit"]
        if message is not None:
            cmd.extend(["-m", message])
        svn_call(cmd)

    # If -t was specified, tag this release
    if tag != None:
        message = "Tagging %s as %s" % (removeprefix(path_or_url, repo_root), removeprefix(tag_dest, repo_root))
        print >>sys.stderr, message
        svn_call(["svn", "copy", "-m", message, path_or_url, tag_dest])
        

if __name__ == "__main__":
    DEVNULL = open(os.devnull, "w")
    main()