git.oblomov.eu Git - git/blob - git_remote_helpers/hg/hg.py

   1 import urllib
   2 import re
   3
   4 import difflib
   5
   6 from itertools import imap;
   7
   8 class ApplyDeltaError(Exception):
   9     """Indicates that applying a delta failed."""
  10
  11     def __init__(self, *args, **kwargs):
  12         Exception.__init__(self, *args, **kwargs)
  13
  14 def chunks_length(chunks):
  15     return sum(imap(len, chunks))
  16
  17 def apply_delta(src_buf, delta):
  18     """Based on the similar function in git's patch-delta.c.
  19
  20     :param src_buf: Source buffer
  21     :param delta: Delta instructions
  22     """
  23     if type(src_buf) != str:
  24         src_buf = ''.join(src_buf)
  25     if type(delta) != str:
  26         delta = ''.join(delta)
  27     out = []
  28     index = 0
  29     delta_length = len(delta)
  30     def get_delta_header_size(delta, index):
  31         size = 0
  32         i = 0
  33         while delta:
  34             cmd = ord(delta[index])
  35             index += 1
  36             size |= (cmd & ~0x80) << i
  37             i += 7
  38             if not cmd & 0x80:
  39                 break
  40         return size, index
  41     src_size, index = get_delta_header_size(delta, index)
  42     dest_size, index = get_delta_header_size(delta, index)
  43     assert src_size == len(src_buf), '%d vs %d' % (src_size, len(src_buf))
  44     while index < delta_length:
  45         cmd = ord(delta[index])
  46         index += 1
  47         if cmd & 0x80:
  48             cp_off = 0
  49             for i in range(4):
  50                 if cmd & (1 << i):
  51                     x = ord(delta[index])
  52                     index += 1
  53                     cp_off |= x << (i * 8)
  54             cp_size = 0
  55             for i in range(3):
  56                 if cmd & (1 << (4+i)):
  57                     x = ord(delta[index])
  58                     index += 1
  59                     cp_size |= x << (i * 8)
  60             if cp_size == 0:
  61                 cp_size = 0x10000
  62             if (cp_off + cp_size < cp_size or
  63                 cp_off + cp_size > src_size or
  64                 cp_size > dest_size):
  65                 break
  66             out.append(src_buf[cp_off:cp_off+cp_size])
  67         elif cmd != 0:
  68             out.append(delta[index:index+cmd])
  69             index += cmd
  70         else:
  71             raise ApplyDeltaError('Invalid opcode 0')
  72
  73     if index != delta_length:
  74         raise ApplyDeltaError('delta not empty: %r' % delta[index:])
  75
  76     if dest_size != chunks_length(out):
  77         raise ApplyDeltaError('dest size incorrect')
  78
  79     return ''.join(out)
  80
  81 class GitHg(object):
  82     """Class that handles various aspects of converting a hg commit to git.
  83     """
  84
  85     def __init__(self, warn):
  86         """Initializes a new GitHg object with the specified warner.
  87         """
  88
  89         self.warn = warn
  90
  91     def format_timezone(self, offset):
  92         if offset % 60 != 0:
  93             raise ValueError("Unable to handle non-minute offset.")
  94         sign = (offset < 0) and '-' or '+'
  95         offset = abs(offset)
  96         return '%c%02d%02d' % (sign, offset / 3600, (offset / 60) % 60)
  97
  98     def get_committer(self, ctx):
  99         extra = ctx.extra()
 100
 101         if 'committer' in extra:
 102             # fixup timezone
 103             (name_timestamp, timezone) = extra['committer'].rsplit(' ', 1)
 104             try:
 105                 timezone = self.format_timezone(-int(timezone))
 106                 return '%s %s' % (name_timestamp, timezone)
 107             except ValueError:
 108                 self.warn("Ignoring committer in extra, invalid timezone in r%s: '%s'.\n" % (ctx.rev(), timezone))
 109
 110         return None
 111
 112     def get_message(self, ctx):
 113         extra = ctx.extra()
 114
 115         message = ctx.description() + "\n"
 116         if 'message' in extra:
 117             message = apply_delta(message, extra['message'])
 118
 119         # HG EXTRA INFORMATION
 120         # TODO FIXME these should go into notes
 121         add_extras = False
 122         extra_message = ''
 123         if not ctx.branch() == 'default':
 124             add_extras = True
 125             extra_message += "branch : " + ctx.branch() + "\n"
 126
 127         renames = []
 128         for f in ctx.files():
 129             if f not in ctx.manifest():
 130                 continue
 131             rename = ctx.filectx(f).renamed()
 132             if rename:
 133                 renames.append((rename[0], f))
 134
 135         if renames:
 136             add_extras = True
 137             for oldfile, newfile in renames:
 138                 extra_message += "rename : " + oldfile + " => " + newfile + "\n"
 139
 140         for key, value in extra.iteritems():
 141             if key in ('author', 'committer', 'encoding', 'message', 'branch', 'hg-git'):
 142                 continue
 143             else:
 144                 add_extras = True
 145                 extra_message += "extra : " + key + " : " +  urllib.quote(value) + "\n"
 146
 147         if add_extras:
 148             message += "\n--HG--\n" + extra_message
 149
 150         return message
 151
 152     def get_author(self, ctx):
 153         # hg authors might not have emails
 154         author = ctx.user()
 155
 156         # check for git author pattern compliance
 157         regex = re.compile('^(.*?) \<(.*?)\>(.*)$')
 158         a = regex.match(author)
 159
 160         if a:
 161             name = a.group(1)
 162             email = a.group(2)
 163             if len(a.group(3)) > 0:
 164                 name += ' ext:(' + urllib.quote(a.group(3)) + ')'
 165             author = name + ' <' + email + '>'
 166         else:
 167             author = author + ' <none@none>'
 168
 169         if 'author' in ctx.extra():
 170             author = apply_delta(author, ctx.extra()['author'])
 171
 172         (time, timezone) = ctx.date()
 173         date = str(int(time)) + ' ' + self.format_timezone(-timezone)
 174
 175         return author + ' ' + date
 176
 177     def get_parents(self, ctx):
 178         def is_octopus_part(ctx):
 179             return ctx.extra().get('hg-git', None) in ('octopus', 'octopus-done')
 180
 181         parents = []
 182         if ctx.extra().get('hg-git', None) == 'octopus-done':
 183             # implode octopus parents
 184             part = ctx
 185             while is_octopus_part(part):
 186                 (p1, p2) = part.parents()
 187                 assert not is_octopus_part(p1)
 188                 parents.append(p1)
 189                 part = p2
 190             parents.append(p2)
 191         else:
 192             parents = ctx.parents()
 193
 194         return parents