1454 lines · 47.7 KB · Python
RawDownload
# patch.py - patch file parsing routines## Copyright 2006 Brendan Cully <brendan@kublai.com># Copyright 2007 Chris Mason <chris.mason@oracle.com>## This software may be used and distributed according to the terms of the# GNU General Public License version 2, incorporated herein by reference. from i18n import _from node import hex, nullid, shortimport base85, cmdutil, mdiff, util, diffhelpers, copiesimport cStringIO, email.Parser, os, reimport sys, tempfile, zlib gitre = re.compile('diff --git a/(.*) b/(.*)') class PatchError(Exception):    pass class NoHunks(PatchError):    pass # helper functions def copyfile(src, dst, basedir):    abssrc, absdst = [util.canonpath(basedir, basedir, x) for x in [src, dst]]    if os.path.exists(absdst):        raise util.Abort(_("cannot create %s: destination already exists") %                         dst)     dstdir = os.path.dirname(absdst)    if dstdir and not os.path.isdir(dstdir):        try:            os.makedirs(dstdir)        except IOError:            raise util.Abort(                _("cannot create %s: unable to create destination directory")                % dst)     util.copyfile(abssrc, absdst) # public functions def extract(ui, fileobj):    '''extract patch from data read from fileobj.     patch can be a normal patch or contained in an email message.     return tuple (filename, message, user, date, node, p1, p2).    Any item in the returned tuple can be None. If filename is None,    fileobj did not contain a patch. Caller must unlink filename when done.'''     # attempt to detect the start of a patch    # (this heuristic is borrowed from quilt)    diffre = re.compile(r'^(?:Index:[ \t]|diff[ \t]|RCS file: |'                        r'retrieving revision [0-9]+(\.[0-9]+)*$|'                        r'(---|\*\*\*)[ \t])', re.MULTILINE)     fd, tmpname = tempfile.mkstemp(prefix='hg-patch-')    tmpfp = os.fdopen(fd, 'w')    try:        msg = email.Parser.Parser().parse(fileobj)         subject = msg['Subject']        user = msg['From']        gitsendmail = 'git-send-email' in msg.get('X-Mailer', '')        # should try to parse msg['Date']        date = None        nodeid = None        branch = None        parents = []         if subject:            if subject.startswith('[PATCH'):                pend = subject.find(']')                if pend >= 0:                    subject = subject[pend+1:].lstrip()            subject = subject.replace('\n\t', ' ')            ui.debug('Subject: %s\n' % subject)        if user:            ui.debug('From: %s\n' % user)        diffs_seen = 0        ok_types = ('text/plain', 'text/x-diff', 'text/x-patch')        message = ''        for part in msg.walk():            content_type = part.get_content_type()            ui.debug('Content-Type: %s\n' % content_type)            if content_type not in ok_types:                continue            payload = part.get_payload(decode=True)            m = diffre.search(payload)            if m:                hgpatch = False                ignoretext = False                 ui.debug(_('found patch at byte %d\n') % m.start(0))                diffs_seen += 1                cfp = cStringIO.StringIO()                for line in payload[:m.start(0)].splitlines():                    if line.startswith('# HG changeset patch'):                        ui.debug(_('patch generated by hg export\n'))                        hgpatch = True                        # drop earlier commit message content                        cfp.seek(0)                        cfp.truncate()                        subject = None                    elif hgpatch:                        if line.startswith('# User '):                            user = line[7:]                            ui.debug('From: %s\n' % user)                        elif line.startswith("# Date "):                            date = line[7:]                        elif line.startswith("# Branch "):                            branch = line[9:]                        elif line.startswith("# Node ID "):                            nodeid = line[10:]                        elif line.startswith("# Parent "):                            parents.append(line[10:])                    elif line == '---' and gitsendmail:                        ignoretext = True                    if not line.startswith('# ') and not ignoretext:                        cfp.write(line)                        cfp.write('\n')                message = cfp.getvalue()                if tmpfp:                    tmpfp.write(payload)                    if not payload.endswith('\n'):                        tmpfp.write('\n')            elif not diffs_seen and message and content_type == 'text/plain':                message += '\n' + payload    except:        tmpfp.close()        os.unlink(tmpname)        raise     if subject and not message.startswith(subject):        message = '%s\n%s' % (subject, message)    tmpfp.close()    if not diffs_seen:        os.unlink(tmpname)        return None, message, user, date, branch, None, None, None    p1 = parents and parents.pop(0) or None    p2 = parents and parents.pop(0) or None    return tmpname, message, user, date, branch, nodeid, p1, p2 GP_PATCH  = 1 << 0  # we have to run patchGP_FILTER = 1 << 1  # there's some copy/rename operationGP_BINARY = 1 << 2  # there's a binary patch class patchmeta(object):    """Patched file metadata     'op' is the performed operation within ADD, DELETE, RENAME, MODIFY    or COPY.  'path' is patched file path. 'oldpath' is set to the    origin file when 'op' is either COPY or RENAME, None otherwise. If    file mode is changed, 'mode' is a tuple (islink, isexec) where    'islink' is True if the file is a symlink and 'isexec' is True if    the file is executable. Otherwise, 'mode' is None.    """    def __init__(self, path):        self.path = path        self.oldpath = None        self.mode = None        self.op = 'MODIFY'        self.lineno = 0        self.binary = False     def setmode(self, mode):        islink = mode & 020000        isexec = mode & 0100        self.mode = (islink, isexec) def readgitpatch(lr):    """extract git-style metadata about patches from <patchname>"""     # Filter patch for git information    gp = None    gitpatches = []    # Can have a git patch with only metadata, causing patch to complain    dopatch = 0     lineno = 0    for line in lr:        lineno += 1        line = line.rstrip(' \r\n')        if line.startswith('diff --git'):            m = gitre.match(line)            if m:                if gp:                    gitpatches.append(gp)                src, dst = m.group(1, 2)                gp = patchmeta(dst)                gp.lineno = lineno        elif gp:            if line.startswith('--- '):                if gp.op in ('COPY', 'RENAME'):                    dopatch |= GP_FILTER                gitpatches.append(gp)                gp = None                dopatch |= GP_PATCH                continue            if line.startswith('rename from '):                gp.op = 'RENAME'                gp.oldpath = line[12:]            elif line.startswith('rename to '):                gp.path = line[10:]            elif line.startswith('copy from '):                gp.op = 'COPY'                gp.oldpath = line[10:]            elif line.startswith('copy to '):                gp.path = line[8:]            elif line.startswith('deleted file'):                gp.op = 'DELETE'                # is the deleted file a symlink?                gp.setmode(int(line[-6:], 8))            elif line.startswith('new file mode '):                gp.op = 'ADD'                gp.setmode(int(line[-6:], 8))            elif line.startswith('new mode '):                gp.setmode(int(line[-6:], 8))            elif line.startswith('GIT binary patch'):                dopatch |= GP_BINARY                gp.binary = True    if gp:        gitpatches.append(gp)     if not gitpatches:        dopatch = GP_PATCH     return (dopatch, gitpatches) class linereader(object):    # simple class to allow pushing lines back into the input stream    def __init__(self, fp, textmode=False):        self.fp = fp        self.buf = []        self.textmode = textmode     def push(self, line):        if line is not None:            self.buf.append(line)     def readline(self):        if self.buf:            l = self.buf[0]            del self.buf[0]            return l        l = self.fp.readline()        if self.textmode and l.endswith('\r\n'):            l = l[:-2] + '\n'        return l     def __iter__(self):        while 1:            l = self.readline()            if not l:                break            yield l # @@ -start,len +start,len @@ or @@ -start +start @@ if len is 1unidesc = re.compile('@@ -(\d+)(,(\d+))? \+(\d+)(,(\d+))? @@')contextdesc = re.compile('(---|\*\*\*) (\d+)(,(\d+))? (---|\*\*\*)') class patchfile(object):    def __init__(self, ui, fname, opener, missing=False, eol=None):        self.fname = fname        self.eol = eol        self.opener = opener        self.ui = ui        self.lines = []        self.exists = False        self.missing = missing        if not missing:            try:                self.lines = self.readlines(fname)                self.exists = True            except IOError:                pass        else:            self.ui.warn(_("unable to find '%s' for patching\n") % self.fname)         self.hash = {}        self.dirty = 0        self.offset = 0        self.rej = []        self.fileprinted = False        self.printfile(False)        self.hunks = 0     def readlines(self, fname):        fp = self.opener(fname, 'r')        try:            return list(linereader(fp, self.eol is not None))        finally:            fp.close()     def writelines(self, fname, lines):        fp = self.opener(fname, 'w')        try:            if self.eol and self.eol != '\n':                for l in lines:                    if l and l[-1] == '\n':                        l = l[:-1] + self.eol                    fp.write(l)            else:                fp.writelines(lines)        finally:            fp.close()     def unlink(self, fname):        os.unlink(fname)     def printfile(self, warn):        if self.fileprinted:            return        if warn or self.ui.verbose:            self.fileprinted = True        s = _("patching file %s\n") % self.fname        if warn:            self.ui.warn(s)        else:            self.ui.note(s)      def findlines(self, l, linenum):        # looks through the hash and finds candidate lines.  The        # result is a list of line numbers sorted based on distance        # from linenum         try:            cand = self.hash[l]        except:            return []         if len(cand) > 1:            # resort our list of potentials forward then back.            cand.sort(key=lambda x: abs(x - linenum))        return cand     def hashlines(self):        self.hash = {}        for x, s in enumerate(self.lines):            self.hash.setdefault(s, []).append(x)     def write_rej(self):        # our rejects are a little different from patch(1).  This always        # creates rejects in the same form as the original patch.  A file        # header is inserted so that you can run the reject through patch again        # without having to type the filename.         if not self.rej:            return         fname = self.fname + ".rej"        self.ui.warn(            _("%d out of %d hunks FAILED -- saving rejects to file %s\n") %            (len(self.rej), self.hunks, fname))         def rejlines():            base = os.path.basename(self.fname)            yield "--- %s\n+++ %s\n" % (base, base)            for x in self.rej:                for l in x.hunk:                    yield l                    if l[-1] != '\n':                        yield "\n\ No newline at end of file\n"         self.writelines(fname, rejlines())     def write(self, dest=None):        if not self.dirty:            return        if not dest:            dest = self.fname        self.writelines(dest, self.lines)     def close(self):        self.write()        self.write_rej()     def apply(self, h, reverse):        if not h.complete():            raise PatchError(_("bad hunk #%d %s (%d %d %d %d)") %                            (h.number, h.desc, len(h.a), h.lena, len(h.b),                            h.lenb))         self.hunks += 1        if reverse:            h.reverse()         if self.missing:            self.rej.append(h)            return -1         if self.exists and h.createfile():            self.ui.warn(_("file %s already exists\n") % self.fname)            self.rej.append(h)            return -1         if isinstance(h, githunk):            if h.rmfile():                self.unlink(self.fname)            else:                self.lines[:] = h.new()                self.offset += len(h.new())                self.dirty = 1            return 0         # fast case first, no offsets, no fuzz        old = h.old()        # patch starts counting at 1 unless we are adding the file        if h.starta == 0:            start = 0        else:            start = h.starta + self.offset - 1        orig_start = start        if diffhelpers.testhunk(old, self.lines, start) == 0:            if h.rmfile():                self.unlink(self.fname)            else:                self.lines[start : start + h.lena] = h.new()                self.offset += h.lenb - h.lena                self.dirty = 1            return 0         # ok, we couldn't match the hunk.  Lets look for offsets and fuzz it        self.hashlines()        if h.hunk[-1][0] != ' ':            # if the hunk tried to put something at the bottom of the file            # override the start line and use eof here            search_start = len(self.lines)        else:            search_start = orig_start         for fuzzlen in xrange(3):            for toponly in [ True, False ]:                old = h.old(fuzzlen, toponly)                 cand = self.findlines(old[0][1:], search_start)                for l in cand:                    if diffhelpers.testhunk(old, self.lines, l) == 0:                        newlines = h.new(fuzzlen, toponly)                        self.lines[l : l + len(old)] = newlines                        self.offset += len(newlines) - len(old)                        self.dirty = 1                        if fuzzlen:                            fuzzstr = "with fuzz %d " % fuzzlen                            f = self.ui.warn                            self.printfile(True)                        else:                            fuzzstr = ""                            f = self.ui.note                        offset = l - orig_start - fuzzlen                        if offset == 1:                            msg = _("Hunk #%d succeeded at %d %s"                                    "(offset %d line).\n")                        else:                            msg = _("Hunk #%d succeeded at %d %s"                                    "(offset %d lines).\n")                        f(msg % (h.number, l+1, fuzzstr, offset))                        return fuzzlen        self.printfile(True)        self.ui.warn(_("Hunk #%d FAILED at %d\n") % (h.number, orig_start))        self.rej.append(h)        return -1 class hunk(object):    def __init__(self, desc, num, lr, context, create=False, remove=False):        self.number = num        self.desc = desc        self.hunk = [ desc ]        self.a = []        self.b = []        if context:            self.read_context_hunk(lr)        else:            self.read_unified_hunk(lr)        self.create = create        self.remove = remove and not create     def read_unified_hunk(self, lr):        m = unidesc.match(self.desc)        if not m:            raise PatchError(_("bad hunk #%d") % self.number)        self.starta, foo, self.lena, self.startb, foo2, self.lenb = m.groups()        if self.lena is None:            self.lena = 1        else:            self.lena = int(self.lena)        if self.lenb is None:            self.lenb = 1        else:            self.lenb = int(self.lenb)        self.starta = int(self.starta)        self.startb = int(self.startb)        diffhelpers.addlines(lr, self.hunk, self.lena, self.lenb, self.a, self.b)        # if we hit eof before finishing out the hunk, the last line will        # be zero length.  Lets try to fix it up.        while len(self.hunk[-1]) == 0:            del self.hunk[-1]

Lines 1–500 of 1454.