""" Python 3 bytes encoder. Used to print filename in

# [ ] figure out how to print non-utf-8 filenames without

# information loss

""" Copied from Python 2.7 to avoid getting

def handle(self, record):

pass

def emit(self, record):

pass

def createLock(self):

global debugmode, streamhandler

debugmode = True

loglevel = logging.DEBUG

logformat = "%(levelname)8s %(message)s"

logger.setLevel(loglevel)

if streamhandler not in logger.handlers:

# when used as a library, streamhandler is not added

# by default

logger.addHandler(streamhandler)

""" Cross-platform version of `os.path.isabs()`

if filename.startswith(b'/'): # Linux/Unix

return True

elif filename.startswith(b'\\'): # Windows

return True

elif re.match(b'\\w:[\\\\/]', filename): # Windows

return True

""" Cross-platform version of os.path.normpath """

# replace escapes and Windows slashes

normalized = posixpath.normpath(path).replace(b'\\', b'/')

# fold the result

""" Make relative path out of absolute by stripping

while xisabs(filename):

# strip windows drive with all slashes

if re.match(b'\\w:[\\\\/]', filename):

filename = re.sub(b'^\\w+:[\\\\/]+', b'', filename)

# strip all slashes

elif re.match(b'[\\\\/]', filename):

filename = re.sub(b'^[\\\\/]+', b'', filename)

""" Parse patch file. If successful, returns

patchset = PatchSet()

debug("reading %s" % filename)

fp = open(filename, "rb")

res = patchset.parse(fp)

fp.close()

if res is True:

return patchset

""" Parse text string and return PatchSet()

ps = PatchSet(StringIO(s))

if ps.errors == 0:

return ps

""" Parse patch from an URL, return False

ps = PatchSet(urllib_request.urlopen(url))

if ps.errors == 0:

return ps

""" Strip n leading components from the given path """

pathlist = [path]

while os.path.dirname(pathlist[0]) != b'':

pathlist[0:1] = os.path.split(pathlist[0])

encodings = {codecs.BOM_UTF8: "utf_8_sig",

codecs.BOM_UTF16_BE: "utf_16_be",

codecs.BOM_UTF16_LE: "utf_16_le",

codecs.BOM_UTF32_BE: "utf_32_be",

codecs.BOM_UTF32_LE: "utf_32_le",

b'\x2b\x2f\x76\x38': "utf_7",

b'\x2b\x2f\x76\x39': "utf_7",

b'\x2b\x2f\x76\x2b': "utf_7",

b'\x2b\x2f\x76\x2f': "utf_7",

b'\x2b\x2f\x76\x38\x2d': "utf_7"}

for bom in sorted(encodings, key=len, reverse=True):

if text.startswith(bom):

try:

return text[len(bom):].decode(encodings[bom])

except UnicodeDecodeError:

continue

decoders = ["utf-8", "Windows-1252"]

for decoder in decoders:

try:

return text.decode(decoder)

except UnicodeDecodeError:

continue

logger.warning("can't decode %s" % str(text))

""" Loads a file content """

with open(path, 'rb') as handle:

tmp = handle.read()

"""

try:

os.makedirs(os.path.dirname(path))

except Exception:

pass

new_content = to_file_bytes(content)

if only_if_modified and os.path.exists(path):

old_content = load(path, binary=True)

if old_content == new_content:

return

with open(path, "wb") as handle:

""" Parsed hunk data container (hunk starts with @@ -R +R @@) """

def __init__(self):

self.startsrc = None # : line count starts with 1

self.linessrc = None

self.starttgt = None

self.linestgt = None

self.invalid = False

self.desc = ''

""" Patch for a single file.

def __init__(self):

self.source = None

self.target = None

self.hunks = []

self.hunkends = []

self.header = []

self.type = None

def __iter__(self):

for h in self.hunks:

""" PatchSet is a patch parser and container.

def __init__(self, stream=None):

# --- API accessible fields ---

# name of the PatchSet (filename or ...)

self.name = None

# patch set type - one of constants

self.type = None

# list of Patch objects

self.items = []

self.errors = 0 # fatal parsing errors

self.warnings = 0 # non-critical warnings

# --- /API ---

if stream:

self.parse(stream)

def __len__(self):

return len(self.items)

def __iter__(self):

for i in self.items:

yield i

def parse(self, stream):

""" parse unified diff

lineends = dict(lf=0, crlf=0, cr=0)

nexthunkno = 0 #: even if index starts with 0 user messages number hunks from 1

p = None

hunk = None

# hunkactual variable is used to calculate hunk lines for comparison

hunkactual = dict(linessrc=None, linestgt=None)

class wrapumerate(enumerate):

"""Enumerate wrapper that uses boolean end of stream status instead of

def __init__(self, *args, **kwargs):

# we don't call parent, it is magically created by __new__ method

self._exhausted = False

self._lineno = False # after end of stream equal to the num of lines

self._line = False # will be reset to False after end of stream

def next(self):

"""Try to read the next line and return True if it is available,

if self._exhausted:

return False

try:

self._lineno, self._line = compat_next(super(wrapumerate, self))

except StopIteration:

self._exhausted = True

self._line = False

return False

return True

@property

def is_empty(self):

return self._exhausted

@property

def line(self):

return self._line

@property

def lineno(self):

return self._lineno

# define states (possible file regions) that direct parse flow

headscan = True # start with scanning header

filenames = False # lines starting with --- and +++

hunkhead = False # @@ -R +R @@ sequence

hunkbody = False #

hunkskip = False # skipping invalid hunk mode

hunkparsed = False # state after successfully parsed hunk

# regexp to match start of hunk, used groups - 1,3,4,6

re_hunk_start = re.compile(br"^@@ -(\d+)(,(\d+))? \+(\d+)(,(\d+))? @@")

self.errors = 0

# temp buffers for header and filenames info

header = []

srcname = None

tgtname = None

# start of main cycle

# each parsing block already has line available in fe.line

fe = wrapumerate(stream)

while fe.next():

# -- deciders: these only switch state to decide who should process

# -- line fetched at the start of this cycle

if hunkparsed:

hunkparsed = False

if re_hunk_start.match(fe.line):

hunkhead = True

elif fe.line.startswith(b"--- "):

filenames = True

else:

headscan = True

# -- ------------------------------------

# read out header

if headscan:

while not fe.is_empty and not fe.line.startswith(b"--- "):

header.append(fe.line)

fe.next()

if fe.is_empty:

if p is None:

debug("no patch data found") # error is shown later

self.errors += 1

else:

info("%d unparsed bytes left at the end of stream" % len(b''.join(header)))

self.warnings += 1

# TODO check for \No new line at the end..

# TODO test for unparsed bytes

# otherwise error += 1

# this is actually a loop exit

continue

headscan = False

# switch to filenames state

filenames = True

line = fe.line

lineno = fe.lineno

# hunkskip and hunkbody code skipped until definition of hunkhead is parsed

if hunkbody:

# [x] treat empty lines inside hunks as containing single space

# (this happens when diff is saved by copy/pasting to editor

# that strips trailing whitespace)

if line.strip(b"\r\n") == b"":

debug("expanding empty line in a middle of hunk body")

self.warnings += 1

line = b' ' + line

# process line first

if re.match(b"^[- \\+\\\\]", line):

# gather stats about line endings

if line.endswith(b"\r\n"):

p.hunkends["crlf"] += 1

elif line.endswith(b"\n"):

p.hunkends["lf"] += 1

elif line.endswith(b"\r"):

p.hunkends["cr"] += 1

if line.startswith(b"-"):

hunkactual["linessrc"] += 1

elif line.startswith(b"+"):

hunkactual["linestgt"] += 1

elif not line.startswith(b"\\"):

hunkactual["linessrc"] += 1

hunkactual["linestgt"] += 1

hunk.text.append(line)

# todo: handle \ No newline cases

else:

warning("invalid hunk no.%d at %d for target file %s" % (nexthunkno, lineno + 1, p.target))

# add hunk status node

hunk.invalid = True

p.hunks.append(hunk)

self.errors += 1

# switch to hunkskip state

hunkbody = False

hunkskip = True

# check exit conditions

if hunkactual["linessrc"] > hunk.linessrc or hunkactual["linestgt"] > hunk.linestgt:

warning("extra lines for hunk no.%d at %d for target %s" % (nexthunkno, lineno + 1, p.target))

# add hunk status node

hunk.invalid = True

p.hunks.append(hunk)

self.errors += 1

# switch to hunkskip state

hunkbody = False

hunkskip = True

elif hunk.linessrc == hunkactual["linessrc"] and hunk.linestgt == hunkactual["linestgt"]:

# hunk parsed successfully

p.hunks.append(hunk)

# switch to hunkparsed state

hunkbody = False

hunkparsed = True

# detect mixed window/unix line ends

ends = p.hunkends

if ((ends["cr"] != 0) + (ends["crlf"] != 0) + (ends["lf"] != 0)) > 1:

warning("inconsistent line ends in patch hunks for %s" % p.source)

self.warnings += 1

if debugmode:

debuglines = dict(ends)

debuglines.update(file=p.target, hunk=nexthunkno)

debug("crlf: %(crlf)d lf: %(lf)d cr: %(cr)d\t - file: %(file)s hunk: %(hunk)d" % debuglines)

# fetch next line

continue

if hunkskip:

if re_hunk_start.match(line):

# switch to hunkhead state

hunkskip = False

hunkhead = True

elif line.startswith(b"--- "):

# switch to filenames state

hunkskip = False

filenames = True

if debugmode and len(self.items) > 0:

debug("- %2d hunks for %s" % (len(p.hunks), p.source))

if filenames:

if line.startswith(b"--- "):

if srcname is not None:

# XXX testcase

warning("skipping false patch for %s" % srcname)

srcname = None

# XXX header += srcname

# double source filename line is encountered

# attempt to restart from this second line

# Files dated at Unix epoch don't exist, e.g.:

# '1970-01-01 01:00:00.000000000 +0100'

# They include timezone offsets.

# .. which can be parsed (if we remove the nanoseconds)

# .. by strptime() with:

# '%Y-%m-%d %H:%M:%S %z'

# .. but unfortunately this relies on the OSes libc

# strptime function and %z support is patchy, so we drop

# everything from the . onwards and group the year and time

# separately.

re_filename_date_time = br"^--- ([^\t]+)(?:\s([0-9-]+)\s([0-9:]+)|.*)"

match = re.match(re_filename_date_time, line)

# todo: support spaces in filenames

if match:

srcname = match.group(1).strip()

date = match.group(2)

time = match.group(3)

if (date == b'1970-01-01' or date == b'1969-12-31') and time.split(b':', 1)[1] == b'00:00':

srcname = b'/dev/null'

else:

warning("skipping invalid filename at line %d" % (lineno + 1))

self.errors += 1

# XXX p.header += line

# switch back to headscan state

filenames = False

headscan = True

elif not line.startswith(b"+++ "):

if srcname is not None:

warning("skipping invalid patch with no target for %s" % srcname)

self.errors += 1

srcname = None

# XXX header += srcname

# XXX header += line

else:

# this should be unreachable

warning("skipping invalid target patch")

filenames = False

headscan = True

else:

if tgtname is not None:

# XXX seems to be a dead branch

warning("skipping invalid patch - double target at line %d" % (lineno + 1))

self.errors += 1

srcname = None

tgtname = None

# XXX header += srcname

# XXX header += tgtname

# XXX header += line

# double target filename line is encountered

# switch back to headscan state

filenames = False

headscan = True

else:

re_filename_date_time = br"^\+\+\+ ([^\t]+)(?:\s([0-9-]+)\s([0-9:]+)|.*)"

match = re.match(re_filename_date_time, line)

if not match:

warning("skipping invalid patch - no target filename at line %d" % (lineno + 1))

self.errors += 1

srcname = None

# switch back to headscan state

filenames = False

headscan = True

else:

tgtname = match.group(1).strip()

date = match.group(2)

time = match.group(3)

if (date == b'1970-01-01' or date == b'1969-12-31') and time.split(b':', 1)[1] == b'00:00':

tgtname = b'/dev/null'

if p: # for the first run p is None

self.items.append(p)

p = Patch()

p.source = srcname

srcname = None

p.target = tgtname

tgtname = None

p.header = header

header = []

# switch to hunkhead state

filenames = False

hunkhead = True

nexthunkno = 0

p.hunkends = lineends.copy()

continue

if hunkhead:

match = re.match(

br"^@@ -(\d+)(,(\d+))? \+(\d+)(,(\d+))? @@(.*)", line)

if not match:

if not p.hunks:

warning("skipping invalid patch with no hunks for file %s" % p.source)

self.errors += 1

# XXX review switch

# switch to headscan state

hunkhead = False

headscan = True

continue

else:

# TODO review condition case

# switch to headscan state

hunkhead = False

headscan = True

else:

hunk = Hunk()

hunk.startsrc = int(match.group(1))

hunk.linessrc = 1

if match.group(3):

hunk.linessrc = int(match.group(3))

hunk.starttgt = int(match.group(4))

hunk.linestgt = 1

if match.group(6):

hunk.linestgt = int(match.group(6))

hunk.invalid = False

hunk.desc = match.group(7)[1:].rstrip()

hunk.text = []

hunkactual["linessrc"] = hunkactual["linestgt"] = 0

# switch to hunkbody state

hunkhead = False

hunkbody = True

nexthunkno += 1

continue

# /while fe.next()

if p:

self.items.append(p)

if not hunkparsed:

if hunkskip:

warning("warning: finished with errors, some hunks may be invalid")

elif headscan:

if len(self.items) == 0:

warning("error: no patch data found!")

return False

else: # extra data at the end of file

pass

else:

warning("error: patch stream is incomplete!")

self.errors += 1

if len(self.items) == 0:

return False

if debugmode and len(self.items) > 0:

debug("- %2d hunks for %s" % (len(p.hunks), p.source))

# XXX fix total hunks calculation

debug("total files: %d total hunks: %d" % (len(self.items),

sum(len(p.hunks) for p in self.items)))

# ---- detect patch and patchset types ----

for idx, p in enumerate(self.items):

self.items[idx].type = self._detect_type(p)

types = set([p.type for p in self.items])

if len(types) > 1:

self.type = MIXED

else:

self.type = types.pop()

# --------

self._normalize_filenames()

return (self.errors == 0)

def _detect_type(self, p):

""" detect and return type for the specified Patch object

# check for SVN

# - header starts with Index:

# - next line is ===... delimiter

# - filename is followed by revision number

# TODO add SVN revision

if (len(p.header) > 1 and p.header[-2].startswith(b"Index: ") and p.header[-1].startswith(b"=" * 67)):

return SVN

# common checks for both HG and GIT

DVCS = ((p.source.startswith(b'a/') or p.source == b'/dev/null') and (p.target.startswith(b'b/') or p.target == b'/dev/null'))

# GIT type check

# - header[-2] is like "diff --git a/oldname b/newname"

# - header[-1] is like "index <hash>..<hash> <mode>"

# TODO add git rename diffs and add/remove diffs

# add git diff with spaced filename

# TODO http://www.kernel.org/pub/software/scm/git/docs/git-diff.html

# Git patch header len is 2 min

if len(p.header) > 1:

# detect the start of diff header - there might be some comments before

for idx in reversed(range(len(p.header))):

if p.header[idx].startswith(b"diff --git"):

break

if p.header[idx].startswith(b'diff --git a/'):

if (idx + 1 < len(p.header) and re.match(b'(?:index \\w{7}..\\w{7} \\d{6}|new file mode \\d*)', p.header[idx + 1])):

if DVCS:

return GIT

# HG check

#

# - for plain HG format header is like "diff -r b2d9961ff1f5 filename"

# - for Git-style HG patches it is "diff --git a/oldname b/newname"

# - filename starts with a/, b/ or is equal to /dev/null

# - exported changesets also contain the header

# # HG changeset patch

# # User [email protected]

# ...

# TODO add MQ

# TODO add revision info

if len(p.header) > 0:

if DVCS and re.match(b'diff -r \\w{12} .*', p.header[-1]):

return HG

if DVCS and p.header[-1].startswith(b'diff --git a/'):

if len(p.header) == 1: # native Git patch header len is 2

return HG

elif p.header[0].startswith(b'# HG changeset patch'):

return HG

return PLAIN

def _normalize_filenames(self):

""" sanitize filenames, normalizing paths, i.e.:

if debugmode:

debug("normalize filenames")

for i, p in enumerate(self.items):

if debugmode:

debug(" patch type = %s" % p.type)

debug(" source = %s" % p.source)

debug(" target = %s" % p.target)

if p.type in (HG, GIT):

debug("stripping a/ and b/ prefixes")

if p.source != b'/dev/null':

if not p.source.startswith(b"a/"):

warning("invalid source filename")

else:

p.source = p.source[2:]

if p.target != b'/dev/null':

if not p.target.startswith(b"b/"):

warning("invalid target filename")

else:

p.target = p.target[2:]

p.source = xnormpath(p.source)

p.target = xnormpath(p.target)

sep = b'/' # sep value can be hardcoded, but it looks nice this way

# references to parent are not allowed

if p.source.startswith(b".." + sep):

warning("error: stripping parent path for source file patch no.%d" % (i + 1))

self.warnings += 1

while p.source.startswith(b".." + sep):

p.source = p.source.partition(sep)[2]

if p.target.startswith(b".." + sep):

warning("error: stripping parent path for target file patch no.%d" % (i + 1))

self.warnings += 1

while p.target.startswith(b".." + sep):

p.target = p.target.partition(sep)[2]

# absolute paths are not allowed

if (xisabs(p.source) and p.source != b'/dev/null') or \

(xisabs(p.target) and p.target != b'/dev/null'):

warning("error: absolute paths are not allowed - file no.%d" % (i + 1))

self.warnings += 1

if xisabs(p.source) and p.source != b'/dev/null':

warning("stripping absolute path from source name '%s'" % p.source)

p.source = xstrip(p.source)

if xisabs(p.target) and p.target != b'/dev/null':

warning("stripping absolute path from target name '%s'" % p.target)

p.target = xstrip(p.target)

self.items[i].source = p.source

self.items[i].target = p.target

def diffstat(self):

""" calculate diffstat and return as a string

names = []

insert = []

delete = []

delta = 0 # size change in bytes

namelen = 0

maxdiff = 0 # max number of changes for single file

# (for histogram width calculation)

for patch in self.items:

i, d = 0, 0

for hunk in patch.hunks:

for line in hunk.text:

if line.startswith(b'+'):

i += 1

delta += len(line) - 1

elif line.startswith(b'-'):

d += 1

delta -= len(line) - 1

names.append(patch.target)

insert.append(i)

delete.append(d)

namelen = max(namelen, len(patch.target))

maxdiff = max(maxdiff, i + d)

output = ''

statlen = len(str(maxdiff)) # stats column width

for i, n in enumerate(names):

# %-19s | %-4d %s

format = " %-" + str(namelen) + "s | %" + str(statlen) + "s %s\n"

hist = ''

# -- calculating histogram --

width = len(format % ('', '', ''))

histwidth = max(2, 80 - width)

if maxdiff < histwidth:

hist = "+" * insert[i] + "-" * delete[i]

else:

iratio = (float(insert[i]) / maxdiff) * histwidth

dratio = (float(delete[i]) / maxdiff) * histwidth

# make sure every entry gets at least one + or -

iwidth = 1 if 0 < iratio < 1 else int(iratio)

dwidth = 1 if 0 < dratio < 1 else int(dratio)

# print(iratio, dratio, iwidth, dwidth, histwidth)

hist = "+" * int(iwidth) + "-" * int(dwidth)

# -- /calculating +- histogram --

output += (format % (tostr(names[i]), str(insert[i] + delete[i]), hist))

output += (" %d files changed, %d insertions(+), %d deletions(-), %+d bytes"

% (len(names), sum(insert), sum(delete), delta))

return output

def findfiles(self, old, new):

""" return tuple of source file, target file """

if old == b'/dev/null':

handle, abspath = tempfile.mkstemp(suffix='pypatch')

abspath = abspath.encode()

# The source file must contain a line for the hunk matching to succeed.

os.write(handle, b' ')

os.close(handle)

if not exists(new):

handle = open(new, 'wb')

handle.close()

return abspath, new

elif exists(old):

return old, old

elif exists(new):

return new, new

elif new == b'/dev/null':

return None, None

else:

# [w] Google Code generates broken patches with its online editor

debug("broken patch from Google Code, stripping prefixes..")

if old.startswith(b'a/') and new.startswith(b'b/'):

old, new = old[2:], new[2:]

debug(" %s" % old)

debug(" %s" % new)

if exists(old):

return old, old

elif exists(new):

return new, new

return None, None

def _strip_prefix(self, filename):

if filename.startswith(b'a/') or filename.startswith(b'b/'):

return filename[2:]

return filename

def decode_clean(self, path, prefix):

path = path.decode("utf-8").replace("\\", "/")

if path.startswith(prefix):

path = path[2:]

return path

def strip_path(self, path, base_path, strip=0):

tokens = path.split("/")

if len(tokens) > 1:

tokens = tokens[strip:]

path = "/".join(tokens)

if base_path:

path = os.path.join(base_path, path)

return path

# account for new and deleted files, upstream dep won't fix them

def apply(self, strip=0, root=None, fuzz=False):

""" Apply parsed patch, optionally stripping leading components

items = []

for item in self.items:

source = self.decode_clean(item.source, "a/")

target = self.decode_clean(item.target, "b/")

if "dev/null" in source:

target = self.strip_path(target, root, strip)

hunks = [s.decode("utf-8") for s in item.hunks[0].text]

new_file = "".join(hunk[1:] for hunk in hunks)

save(target, new_file)

elif "dev/null" in target:

source = self.strip_path(source, root, strip)

safe_unlink(source)

else:

items.append(item)

self.items = items

if root:

prevdir = os.getcwd()

os.chdir(root)

total = len(self.items)

errors = 0

if strip:

# [ ] test strip level exceeds nesting level

# [ ] test the same only for selected files

# [ ] test if files end up being on the same level

try:

strip = int(strip)

except ValueError:

errors += 1

warning("error: strip parameter '%s' must be an integer" % strip)

strip = 0

# for fileno, filename in enumerate(self.source):

for i, p in enumerate(self.items):

if strip:

debug("stripping %s leading component(s) from:" % strip)

debug(" %s" % p.source)

debug(" %s" % p.target)

old = p.source if p.source == b'/dev/null' else pathstrip(p.source, strip)

new = p.target if p.target == b'/dev/null' else pathstrip(p.target, strip)