def split()

in eden/scm/edenscm/mercurial/patch.py [0:0]

• 95 lines of code
• 37 McCabe index (conditional complexity)

def split(stream):
    """return an iterator of individual patches from a stream"""

    def isheader(line, inheader):
        if inheader and line[:1] in (b" ", b"\t"):
            # continuation
            return True
        if line[:1] in (b" ", b"-", b"+"):
            # diff line - don't check for header pattern in there
            return False
        l = line.split(b": ", 1)
        return len(l) == 2 and b" " not in l[0]

    def chunk(lines):
        return stringio(b"".join(lines))

    def hgsplit(stream, cur):
        inheader = True

        for line in stream:
            if not line.strip():
                inheader = False
            if not inheader and line.startswith(b"# HG changeset patch"):
                yield chunk(cur)
                cur = []
                inheader = True

            cur.append(line)

        if cur:
            yield chunk(cur)

    def mboxsplit(stream, cur):
        for line in stream:
            if line.startswith(b"From "):
                for c in split(chunk(cur[1:])):
                    yield c
                cur = []

            cur.append(line)

        if cur:
            for c in split(chunk(cur[1:])):
                yield c

    def mimesplit(stream, cur):
        def msgfp(m):
            fp = stringio()
            g = email.Generator.Generator(fp, mangle_from_=False)
            g.flatten(m)
            fp.seek(0)
            return fp

        for line in stream:
            cur.append(line)
        c = chunk(cur)

        m = pycompat.parse_email(c)
        if not m.is_multipart():
            yield msgfp(m)
        else:
            ok_types = ("text/plain", "text/x-diff", "text/x-patch")
            for part in m.walk():
                ct = part.get_content_type()
                if ct not in ok_types:
                    continue
                yield msgfp(part)

    def headersplit(stream, cur):
        inheader = False

        for line in stream:
            if not inheader and isheader(line, inheader):
                yield chunk(cur)
                cur = []
                inheader = True
            if inheader and not isheader(line, inheader):
                inheader = False

            cur.append(line)

        if cur:
            yield chunk(cur)

    def remainder(cur):
        yield chunk(cur)

    class fiter(object):
        def __init__(self, fp):
            self.fp = fp

        def __iter__(self):
            return self

        def next(self):
            l = self.fp.readline()
            if not l:
                raise StopIteration
            return l

        __next__ = next

    inheader = False
    cur = []

    mimeheaders = ["content-type"]

    if not util.safehasattr(stream, "next"):
        # http responses, for example, have readline but not next
        stream = fiter(stream)

    for line in stream:
        cur.append(line)
        if line.startswith(b"# HG changeset patch"):
            return hgsplit(stream, cur)
        elif line.startswith(b"From "):
            return mboxsplit(stream, cur)
        elif isheader(line, inheader):
            inheader = True
            if line.split(b":", 1)[0].lower() in mimeheaders:
                # let email parser handle this
                return mimesplit(stream, cur)
        elif line.startswith(b"--- ") and inheader:
            # No evil headers seen by diff start, split by hand
            return headersplit(stream, cur)
        # Not enough info, keep reading

    # if we are here, we have a very plain patch
    return remainder(cur)