def extract()

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

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

def extract(ui, fileobj):
    """extract patch from data read from fileobj.

    patch can be a normal patch or contained in an email message.

    return a dictionary. Standard keys are:
      - filename,
      - message,
      - user,
      - date,
      - branch,
      - node,
      - p1,
      - p2.
    Any item can be missing from the dictionary. If filename is missing,
    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(
        br"^(?:Index:[ \t]|diff[ \t]-|RCS file: |"
        br"retrieving revision [0-9]+(\.[0-9]+)*$|"
        br"---[ \t].*?^\+\+\+[ \t]|"
        br"\*\*\*[ \t].*?^---[ \t])",
        re.MULTILINE | re.DOTALL,
    )

    data = {}
    fd, tmpname = tempfile.mkstemp(prefix="hg-patch-")
    tmpfp = util.fdopen(fd, "wb")
    try:
        msg = pycompat.parse_email(fileobj)

        subject = msg["Subject"] and mail.headdecode(msg["Subject"])
        data["user"] = msg["From"] and mail.headdecode(msg["From"])
        if not subject and not data["user"]:
            # Not an email, restore parsed headers if any
            subject = "\n".join(": ".join(h) for h in msg.items()) + "\n"

        # should try to parse msg['Date']
        parents = []

        if subject:
            if subject.startswith("[PATCH"):
                pend = subject.find("]")
                if pend >= 0:
                    subject = subject[pend + 1 :].lstrip()
            subject = re.sub(r"\n[ \t]+", " ", subject)
            ui.debug("Subject: %s\n" % subject)
        if data["user"]:
            ui.debug("From: %s\n" % data["user"])
        diffs_seen = 0
        ok_types = ("text/plain", "text/x-diff", "text/x-patch")
        message = b""
        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
                hgpatchheader = False
                ignoretext = False

                ui.debug("found patch at byte %d\n" % m.start(0))
                diffs_seen += 1
                cfp = stringio()
                for line in payload[: m.start(0)].splitlines():
                    if line.startswith(b"# HG changeset patch") and not hgpatch:
                        ui.debug("patch generated by hg export\n")
                        hgpatch = True
                        hgpatchheader = True
                        # drop earlier commit message content
                        cfp.seek(0)
                        cfp.truncate()
                        subject = None
                    elif hgpatchheader:
                        if line.startswith(b"# User "):
                            data["user"] = pycompat.decodeutf8(line[7:])
                            ui.debug("From: %s\n" % data["user"])
                        elif line.startswith(b"# Parent "):
                            parents.append(pycompat.decodeutf8(line[9:].lstrip()))
                        elif line.startswith(b"# "):
                            for header, key in patchheadermap:
                                prefix = b"# %s " % header
                                if line.startswith(prefix):
                                    data[key] = pycompat.decodeutf8(line[len(prefix) :])
                        else:
                            hgpatchheader = False
                    elif line == b"---":
                        ignoretext = True
                    if not hgpatchheader and not ignoretext:
                        cfp.write(line)
                        cfp.write(b"\n")
                message = cfp.getvalue()
                if tmpfp:
                    tmpfp.write(payload)
                    if not payload.endswith(b"\n"):
                        tmpfp.write(b"\n")
            elif not diffs_seen and message and content_type == "text/plain":
                message += b"\n" + payload
    except:  # re-raises
        tmpfp.close()
        os.unlink(tmpname)
        raise

    message = pycompat.decodeutf8(message)
    if subject and not message.startswith(subject):
        message = "%s\n%s" % (subject, message)
    data["message"] = message
    tmpfp.close()
    if parents:
        data["p1"] = parents.pop(0)
        if parents:
            data["p2"] = parents.pop(0)

    if diffs_seen:
        data["filename"] = tmpname
    else:
        os.unlink(tmpname)
    return data