/brz/remove-bazaar

# Copyright (C) 2008 Canonical Ltd

#

# This program is free software; you can redistribute it and/or modify

# it under the terms of the GNU General Public License as published by

# the Free Software Foundation; either version 2 of the License, or

# (at your option) any later version.

#

# This program is distributed in the hope that it will be useful,

# but WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

# GNU General Public License for more details.

#

# You should have received a copy of the GNU General Public License

# along with this program; if not, write to the Free Software

# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA

"""Import processor that supports all Bazaar repository formats."""

import re

import time

from bzrlib import (

    builtins,

    bzrdir,

    delta,

    errors,

    generate_ids,

    inventory,

    lru_cache,

    osutils,

    progress,

    revision,

    revisiontree,

    transport,

    )

from bzrlib.trace import (

    note,

    warning,

    error,

    )

import bzrlib.util.configobj.configobj as configobj

from bzrlib.plugins.fastimport import (

    helpers,

    processor,

    revisionloader,

    )

# How many commits before automatically reporting progress

_DEFAULT_AUTO_PROGRESS = 1000

# How many commits before automatically checkpointing

_DEFAULT_AUTO_CHECKPOINT = 10000

class GenericProcessor(processor.ImportProcessor):

    """An import processor that handles basic imports.

    Current features supported:

    * blobs are cached in memory

    * files and symlinks commits are supported

    * checkpoints automatically happen at a configurable frequency

      over and above the stream requested checkpoints

    * timestamped progress reporting, both automatic and stream requested

    * LATER: reset support, tags for each branch

    * some basic statistics are dumped on completion.

    Here are the supported parameters:

    * info - name of a hints file holding the analysis generated

      by running the fast-import-info processor in verbose mode. When

      importing large repositories, this parameter is needed so

      that the importer knows what blobs to intelligently cache.

    * trees - update the working trees before completing.

      By default, the importer updates the repository

      and branches and the user needs to run 'bzr update' for the

      branches of interest afterwards.

    * checkpoint - automatically checkpoint every n commits over and

      above any checkpoints contained in the import stream.

      The default is 10000.

    * count - only import this many commits then exit. If not set,

      all commits are imported.

    """

    known_params = ['info', 'trees', 'checkpoint', 'count']

    def note(self, msg, *args):

        """Output a note but timestamp it."""

        msg = "%s %s" % (self._time_of_day(), msg)

        note(msg, *args)

    def warning(self, msg, *args):

        """Output a warning but timestamp it."""

        msg = "%s WARNING: %s" % (self._time_of_day(), msg)

        warning(msg, *args)

    def _time_of_day(self):

        """Time of day as a string."""

        # Note: this is a separate method so tests can patch in a fixed value

        return time.strftime("%H:%M:%S")

    def pre_process(self):

        self._start_time = time.time()

        self._load_info_and_params()

        self.cache_mgr = GenericCacheManager(self.info, verbose=self.verbose)

        self.init_stats()

        # mapping of tag name to revision_id

        self.tags = {}

        # Create a write group. This is committed at the end of the import.

        # Checkpointing closes the current one and starts a new one.

        self.repo.start_write_group()

    def _load_info_and_params(self):

        # Load the info file, if any

        info_path = self.params.get('info')

        if info_path is not None:

            self.info = configobj.ConfigObj(info_path)

        else:

            self.info = None

        # Decide how often to automatically report progress

        # (not a parameter yet)

        self.progress_every = _DEFAULT_AUTO_PROGRESS

        if self.verbose:

            self.progress_every = self.progress_every / 10

        # Decide how often to automatically checkpoint

        self.checkpoint_every = int(self.params.get('checkpoint',

            _DEFAULT_AUTO_CHECKPOINT))

        # Find the maximum number of commits to import (None means all)

        # and prepare progress reporting. Just in case the info file

        # has an outdated count of commits, we store the max counts

        # at which we need to terminate separately to the total used

        # for progress tracking.

        try:

            self.max_commits = int(self.params['count'])

            if self.max_commits < 0:

                self.max_commits = None

        except KeyError:

            self.max_commits = None

        if self.info is not None:

            self.total_commits = int(self.info['Command counts']['commit'])

            if (self.max_commits is not None and

                self.total_commits > self.max_commits):

                self.total_commits = self.max_commits

        else:

            self.total_commits = self.max_commits

    def _process(self, command_iter):

        # if anything goes wrong, abort the write group if any

        try:

            processor.ImportProcessor._process(self, command_iter)

        except:

            if self.repo is not None and self.repo.is_in_write_group():

                self.repo.abort_write_group()

            raise

    def post_process(self):

        # Commit the current write group.

        self.repo.commit_write_group()

        # Update the branches

        self.note("Updating branch information ...")

        updater = GenericBranchUpdater(self.repo, self.branch, self.cache_mgr,

            helpers.invert_dict(self.cache_mgr.heads),

            self.cache_mgr.last_ref)

        branches_updated, branches_lost = updater.update()

        self._branch_count = len(branches_updated)

        # Tell the user about branches that were not created

        if branches_lost:

            if not self.repo.is_shared():

                self.warning("Cannot import multiple branches into "

                    "an unshared repository")

            self.warning("Not creating branches for these head revisions:")

            for lost_info in branches_lost:

                head_revision = lost_info[1]

                branch_name = lost_info[0]

                note("\t %s = %s", head_revision, branch_name)

        # Update the working trees as requested and dump stats

        self._tree_count = 0

        remind_about_update = True

        if self.params.get('trees', False):

            trees = self._get_working_trees(branches_updated)

            if trees:

                self.note("Updating the working trees ...")

                if self.verbose:

                    report = delta._ChangeReporter()

                else:

                    reporter = None

                for wt in trees:

                    wt.update(reporter)

                    self._tree_count += 1

                remind_about_update = False

            else:

                self.warning("No working trees available to update")

        self.dump_stats()

        if remind_about_update:

            self.note("To refresh the working tree for a branch, "

                "use 'bzr update'")

    def _get_working_trees(self, branches):

        """Get the working trees for branches in the repository."""

        result = []

        wt_expected = self.repo.make_working_trees()

        for br in branches:

            if br == self.branch and br is not None:

                wt = self.working_tree

            elif wt_expected:

                try:

                    wt = br.bzrdir.open_workingtree()

                except errors.NoWorkingTree:

                    self.warning("No working tree for branch %s", br)

                    continue

            else:

                continue

            result.append(wt)

        return result

    def init_stats(self):

        self._revision_count = 0

    def dump_stats(self):

        time_required = progress.str_tdelta(time.time() - self._start_time)

        rc = self._revision_count

        bc = self._branch_count

        wtc = self._tree_count

        self.note("Imported %d %s, updating %d %s and %d %s in %s",

            rc, helpers.single_plural(rc, "revision", "revisions"),

            bc, helpers.single_plural(bc, "branch", "branches"),

            wtc, helpers.single_plural(wtc, "tree", "trees"),

            time_required)

    def blob_handler(self, cmd):

        """Process a BlobCommand."""

        if cmd.mark is not None:

            dataref = cmd.id

        else:

            dataref = osutils.sha_strings(cmd.data)

        self.cache_mgr.store_blob(dataref, cmd.data)

    def checkpoint_handler(self, cmd):

        """Process a CheckpointCommand."""

        # Commit the current write group and start a new one

        self.repo.commit_write_group()

        self.repo.start_write_group()

    def commit_handler(self, cmd):

        """Process a CommitCommand."""

        # 'Commit' the revision

        handler = GenericCommitHandler(cmd, self.repo, self.cache_mgr,

            self.verbose)

        handler.process()

        # Update caches

        self.cache_mgr.revision_ids[cmd.id] = handler.revision_id

        self.cache_mgr.last_ids[cmd.ref] = cmd.id

        self.cache_mgr.last_ref = cmd.ref

        # Report progress

        self._revision_count += 1

        self.report_progress("(%s)" % cmd.id)

        # Check if we should finish up or automatically checkpoint

        if (self.max_commits is not None and

            self._revision_count >= self.max_commits):

            self.note("stopping after reaching requested count of commits")

            self.finished = True

        elif self._revision_count % self.checkpoint_every == 0:

            self.note("%d commits - automatic checkpoint triggered",

                self._revision_count)

            self.checkpoint_handler(None)

    def report_progress(self, details=''):

        # TODO: use a progress bar with ETA enabled

        if self._revision_count % self.progress_every == 0:

            if self.total_commits is not None:

                counts = "%d/%d" % (self._revision_count, self.total_commits)

                eta = progress.get_eta(self._start_time, self._revision_count,

                    self.total_commits)

                eta_str = progress.str_tdelta(eta)

                if eta_str.endswith('--'):

                    eta_str = ''

                else:

                    eta_str = '[%s] ' % eta_str

            else:

                counts = "%d" % (self._revision_count,)

                eta_str = ''

            self.note("%s commits processed %s%s" % (counts, eta_str, details))

    def progress_handler(self, cmd):

        """Process a ProgressCommand."""

        # We could use a progress bar here instead

        self.note("progress %s" % (cmd.message,))

    def reset_handler(self, cmd):

        """Process a ResetCommand."""

        if cmd.ref.startswith('refs/tags/'):

            self._set_tag(cmd.ref[len('refs/tags/'):], cmd.from_)

        else:

            self.warning("resets are not supported yet"

                " - ignoring reset of '%s'", cmd.ref)

    def tag_handler(self, cmd):

        """Process a TagCommand."""

        self._set_tag(cmd.id, cmd.from_)

    def _set_tag(self, name, from_):

        """Define a tag given a name an import 'from' reference."""

        bzr_tag_name = name.decode('utf-8', 'replace')

        bzr_rev_id = self.cache_mgr.revision_ids[from_]

        self.tags[bzr_tag_name] = bzr_rev_id

class GenericCacheManager(object):

    """A manager of caches for the GenericProcessor."""

    def __init__(self, info, verbose=False, inventory_cache_size=10):

        """Create a manager of caches.

        :param info: a ConfigObj holding the output from

            the --info processor, or None if no hints are available

        """

        self.verbose = verbose

        # dataref -> data. datref is either :mark or the sha-1.

        # Sticky blobs aren't removed after being referenced.

        self._blobs = {}

        self._sticky_blobs = {}

        # revision-id -> Inventory cache

        # these are large and we probably don't need too many as

        # most parents are recent in history

        self.inventories = lru_cache.LRUCache(inventory_cache_size)

        # import-ref -> revision-id lookup table

        # we need to keep all of these but they are small

        self.revision_ids = {}

        # path -> file-ids - as generated

        self.file_ids = {}

        # Head tracking: last ref, last id per ref & map of commit mark to ref

        self.last_ref = None

        self.last_ids = {}

        self.heads = {}

        # Work out the blobs to make sticky - None means all

        #print "%r" % (info,)

        self._blobs_to_keep = None

        if info is not None:

            try:

                self._blobs_to_keep = info['Blob usage tracking']['multi']

            except KeyError:

                # info not in file - possible when no blobs used

                pass

    def store_blob(self, id, data):

        """Store a blob of data."""

        if (self._blobs_to_keep is None or data == '' or

            id in self._blobs_to_keep):

            self._sticky_blobs[id] = data

            if self.verbose:

                print "making blob %s sticky" % (id,)

        else:

            self._blobs[id] = data

    def fetch_blob(self, id):

        """Fetch a blob of data."""

        try:

            return self._sticky_blobs[id]

        except KeyError:

            return self._blobs.pop(id)

    def _delete_path(self, path):

        """Remove a path from caches."""

        # we actually want to remember what file-id we gave a path,

        # even when that file is deleted, so doing nothing is correct

        pass

    def _rename_path(self, old_path, new_path):

        """Rename a path in the caches."""

        # we actually want to remember what file-id we gave a path,

        # even when that file is renamed, so both paths should have

        # the same value and we don't delete any information

        self.file_ids[new_path] = self.file_ids[old_path]

class GenericCommitHandler(processor.CommitHandler):

    def __init__(self, command, repo, cache_mgr, verbose=False):

        processor.CommitHandler.__init__(self, command)

        self.repo = repo

        self.cache_mgr = cache_mgr

        self.verbose = verbose

        # smart loader that uses these caches

        self.loader = revisionloader.RevisionLoader(repo,

            lambda revision_ids: self._get_inventories(revision_ids))

    def pre_process_files(self):

        """Prepare for committing."""

        self.revision_id = self.gen_revision_id()

        self.inv_delta = []

        # cache of texts for this commit, indexed by file-id

        self.lines_for_commit = {}

        # Work out the true set of parents

        cmd = self.command

        if cmd.mark is None:

            last_id = self.cache_mgr.last_ids.get(cmd.ref)

            if last_id is not None:

                parents = [last_id]

            else:

                parents = []

        else:

            parents = cmd.parents

        # Track the heads

        for parent in parents:

            try:

                del self.cache_mgr.heads[parent]

            except KeyError:

                warning("didn't find parent %s while tracking heads",

                    parent)

        self.cache_mgr.heads[cmd.id] = cmd.ref

        # Get the parent inventories

        if parents:

            self.parents = [self.cache_mgr.revision_ids[p]

                for p in parents]

        else:

            self.parents = []

        # Seed the inventory from the previous one

        if len(self.parents) == 0:

            self.inventory = self.gen_initial_inventory()

        else:

            # use the bzr_revision_id to lookup the inv cache

            self.inventory = self.get_inventory(self.parents[0]).copy()

        if not self.repo.supports_rich_root():

            # In this repository, root entries have no knit or weave. When

            # serializing out to disk and back in, root.revision is always

            # the new revision_id.

            self.inventory.root.revision = self.revision_id

        # directory-path -> inventory-entry for current inventory

        self.directory_entries = dict(self.inventory.directories())

    def post_process_files(self):

        """Save the revision."""

        if self.verbose:

            note("applying inventory delta ...")

            for entry in self.inv_delta:

                note("  %r" % (entry,))

        self.inventory.apply_delta(self.inv_delta)

        self.cache_mgr.inventories[self.revision_id] = self.inventory

        if self.verbose:

            note("created inventory ...")

            for entry in self.inventory:

                note("  %r" % (entry,))

        # Load the revision into the repository

        rev_props = {}

        committer = self.command.committer

        who = "%s <%s>" % (committer[0],committer[1])

        author = self.command.author

        if author is not None:

            author_id = "%s <%s>" % (author[0],author[1])

            if author_id != who:

                rev_props['author'] = author_id

        rev = revision.Revision(

           timestamp=committer[2],

           timezone=committer[3],

           committer=who,

           message=self._escape_commit_message(self.command.message),

           revision_id=self.revision_id,

           properties=rev_props,

           parent_ids=self.parents)

        self.loader.load(rev, self.inventory, None,

            lambda file_id: self._get_lines(file_id))

    def _escape_commit_message(self, message):

        """Replace xml-incompatible control characters."""

        # It's crap that we need to do this at this level (but we do)

        # Code copied from bzrlib.commit.

        # Python strings can include characters that can't be

        # represented in well-formed XML; escape characters that

        # aren't listed in the XML specification

        # (http://www.w3.org/TR/REC-xml/#NT-Char).

        message, _ = re.subn(

            u'[^\x09\x0A\x0D\u0020-\uD7FF\uE000-\uFFFD]+',

            lambda match: match.group(0).encode('unicode_escape'),

            message)

        return message

    def modify_handler(self, filecmd):

        if filecmd.dataref is not None:

            data = self.cache_mgr.fetch_blob(filecmd.dataref)

        else:

            data = filecmd.data

        self._modify_inventory(filecmd.path, filecmd.kind,

            filecmd.is_executable, data)

    def delete_handler(self, filecmd):

        path = filecmd.path

        try:

            del self.inventory[self.bzr_file_id(path)]

        except errors.NoSuchId:

            warning("ignoring delete of %s - not in inventory" % (path,))

        finally:

            try:

                self.cache_mgr._delete_path(path)

            except KeyError:

                pass

    def copy_handler(self, filecmd):

        raise NotImplementedError(self.copy_handler)

    def rename_handler(self, filecmd):

        old_path = filecmd.old_path

        new_path = filecmd.new_path

        file_id = self.bzr_file_id(old_path)

        ie = self.inventory[file_id]

        self.inv_delta.append((old_path, new_path, file_id, ie))

        self.cache_mgr._rename_path(old_path, new_path)

    def deleteall_handler(self, filecmd):

        raise NotImplementedError(self.deleteall_handler)

    def bzr_file_id_and_new(self, path):

        """Get a Bazaar file identifier and new flag for a path.

        :return: file_id, is_new where

          is_new = True if the file_id is newly created

        """

        try:

            return self.cache_mgr.file_ids[path], False

        except KeyError:

            id = generate_ids.gen_file_id(path)

            self.cache_mgr.file_ids[path] = id

            return id, True

    def bzr_file_id(self, path):

        """Get a Bazaar file identifier for a path."""

        return self.bzr_file_id_and_new(path)[0]

    def gen_initial_inventory(self):

        """Generate an inventory for a parentless revision."""

        inv = inventory.Inventory(revision_id=self.revision_id)

        return inv

    def gen_revision_id(self):

        """Generate a revision id.

        Subclasses may override this to produce deterministic ids say.

        """

        committer = self.command.committer

        # Perhaps 'who' being the person running the import is ok? If so,

        # it might be a bit quicker and give slightly better compression?

        who = "%s <%s>" % (committer[0],committer[1])

        timestamp = committer[2]

        return generate_ids.gen_revision_id(who, timestamp)

    def get_inventory(self, revision_id):

        """Get the inventory for a revision id."""

        try:

            inv = self.cache_mgr.inventories[revision_id]

        except KeyError:

            print "Hmm - get_inventory cache miss for %s" % revision_id

            # Not cached so reconstruct from repository

            inv = self.repo.revision_tree(revision_id).inventory

            self.cache_mgr.inventories[revision_id] = inv

        return inv

    def _get_inventories(self, revision_ids):

        """Get the inventories for revision-ids.

        This is a callback used by the RepositoryLoader to

        speed up inventory reconstruction."""

        present = []

        inventories = []

        # If an inventory is in the cache, we assume it was

        # successfully loaded into the repsoitory

        for revision_id in revision_ids:

            try:

                inv = self.cache_mgr.inventories[revision_id]

                present.append(revision_id)

            except KeyError:

                print "Hmm - get_inventories cache miss for %s" % revision_id

                # Not cached so reconstruct from repository

                if self.repo.has_revision(revision_id):

                    rev_tree = self.repo.revision_tree(revision_id)

                    present.append(revision_id)

                else:

                    rev_tree = self.repo.revision_tree(None)

                inv = rev_tree.inventory

                self.cache_mgr.inventories[revision_id] = inv

            inventories.append(inv)

        return present, inventories

    def _get_lines(self, file_id):

        """Get the lines for a file-id."""

        return self.lines_for_commit[file_id]

    def _modify_inventory(self, path, kind, is_executable, data):

        """Add to or change an item in the inventory."""

        # Create the new InventoryEntry

        basename, parent_ie = self._ensure_directory(path)

        file_id = self.bzr_file_id(path)

        ie = inventory.make_entry(kind, basename, parent_ie.file_id, file_id)

        ie.revision = self.revision_id

        if isinstance(ie, inventory.InventoryFile):

            ie.executable = is_executable

            lines = osutils.split_lines(data)

            ie.text_sha1 = osutils.sha_strings(lines)

            ie.text_size = sum(map(len, lines))

            self.lines_for_commit[file_id] = lines

        elif isinstance(ie, inventory.InventoryLnk):

            ie.symlink_target = data

        else:

            raise errors.BzrError("Cannot import items of kind '%s' yet" %

                (kind,))

        # Record this new inventory entry

        if file_id in self.inventory:

            # HACK: no API for this (del+add does more than it needs to)

            self.inventory._byid[file_id] = ie

        else:

            self.inventory.add(ie)

    def _ensure_directory(self, path):

        """Ensure that the containing directory exists for 'path'"""

        dirname, basename = osutils.split(path)

        if dirname == '':

            # the root node doesn't get updated

            return basename, self.inventory.root

        try:

            ie = self.directory_entries[dirname]

        except KeyError:

            # We will create this entry, since it doesn't exist

            pass

        else:

            return basename, ie