Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
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
#!/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()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.