path: root/alot/helper.py

# -*- coding: utf-8 -*-
# Copyright (C) 2011-2012  Patrick Totzke <patricktotzke@gmail.com>
# Copyright © 2017-2018 Dylan Baker
# This file is released under the GNU GPL, version 3 or a later revision.
# For further details see the COPYING file
from collections import deque
import logging
import mimetypes
import os
import re
import shlex
import subprocess
import email
from email.mime.audio import MIMEAudio
from email.mime.base import MIMEBase
from email.mime.image import MIMEImage
from email.mime.text import MIMEText
import asyncio

import urwid
import magic


def split_commandline(s, comments=False, posix=True):
    """
    splits semi-colon separated commandlines
    """
    # shlex seems to remove unescaped quotes and backslashes
    s = s.replace('\\', '\\\\')
    s = s.replace('\'', '\\\'')
    s = s.replace('\"', '\\\"')
    lex = shlex.shlex(s, posix=posix)
    lex.whitespace_split = True
    lex.whitespace = ';'
    if not comments:
        lex.commenters = ''
    return list(lex)


def split_commandstring(cmdstring):
    """
    split command string into a list of strings to pass on to subprocess.Popen
    and the like. This simply calls shlex.split but works also with unicode
    bytestrings.
    """
    assert isinstance(cmdstring, str)
    return shlex.split(cmdstring)

def shorten(string, maxlen):
    """shortens string if longer than maxlen, appending ellipsis"""
    if 1 < maxlen < len(string):
        string = string[:maxlen - 1] + '…'
    return string[:maxlen]


def shorten_author_string(authors_string, maxlength):
    """
    Parse a list of authors concatenated as a text string (comma
    separated) and smartly adjust them to maxlength.

    1) If the complete list of sender names does not fit in maxlength, it
    tries to shorten names by using only the first part of each.

    2) If the list is still too long, hide authors according to the
    following priority:

      - First author is always shown (if too long is shorten with ellipsis)

      - If possible, last author is also shown (if too long, uses ellipsis)

      - If there are more than 2 authors in the thread, show the
        maximum of them. More recent senders have higher priority.

      - If it is finally necessary to hide any author, an ellipsis
        between first and next authors is added.
    """

    # I will create a list of authors by parsing author_string. I use
    # deque to do popleft without performance penalties
    authors = deque()

    # If author list is too long, it uses only the first part of each
    # name (gmail style)
    short_names = len(authors_string) > maxlength
    for au in authors_string.split(", "):
        if short_names:
            author_as_list = au.split()
            if len(author_as_list) > 0:
                authors.append(author_as_list[0])
        else:
            authors.append(au)

    # Author chain will contain the list of author strings to be
    # concatenated using commas for the final formatted author_string.
    authors_chain = deque()

    if len(authors) == 0:
        return ''

    # reserve space for first author
    first_au = shorten(authors.popleft(), maxlength)
    remaining_length = maxlength - len(first_au)

    # Tries to add an ellipsis if no space to show more than 1 author
    if authors and maxlength > 3 and remaining_length < 3:
        first_au = shorten(first_au, maxlength - 3)
        remaining_length += 3

    # Tries to add as more authors as possible. It takes into account
    # that if any author will be hidden, and ellipsis should be added
    while authors and remaining_length >= 3:
        au = authors.pop()
        if len(au) > 1 and (remaining_length == 3 or (authors and
                                                      remaining_length < 7)):
            authors_chain.appendleft('…')
            break
        else:
            if authors:
                # 5= ellipsis + 2 x comma and space used as separators
                au_string = shorten(au, remaining_length - 5)
            else:
                # 2 = comma and space used as separator
                au_string = shorten(au, remaining_length - 2)
            remaining_length -= len(au_string) + 2
            authors_chain.appendleft(au_string)

    # Add the first author to the list and concatenate list
    authors_chain.appendleft(first_au)
    authorsstring = ', '.join(authors_chain)
    return authorsstring


def call_cmd(cmdlist, stdin=None):
    """
    get a shell commands output, error message and return value and immediately
    return.

    .. warning::

        This returns with the first screen content for interactive commands.

    :param cmdlist: shellcommand to call, already splitted into a list accepted
                    by :meth:`subprocess.Popen`
    :type cmdlist: list of str
    :param stdin: string to pipe to the process
    :type stdin: str, bytes, or None
    :return: triple of stdout, stderr, return value of the shell command
    :rtype: str, str, int
    """
    termenc = urwid.util.detected_encoding
    try:

        logging.debug("Calling %s" % cmdlist)
        proc = subprocess.Popen(
            cmdlist,
            encoding = termenc, errors = 'backslashreplace',
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            stdin=subprocess.PIPE if stdin is not None else None)
    except OSError as e:
        out = ''
        err = e.strerror
        ret = e.errno
    else:
        out, err = proc.communicate(stdin)
        ret = proc.returncode

    return out, err, ret


async def call_cmd_async(cmdlist, stdin=None, env=None):
    """Given a command, call that command asynchronously and return the output.

    This function only handles `OSError` when creating the subprocess, any
    other exceptions raised either durring subprocess creation or while
    exchanging data with the subprocess are the caller's responsibility to
    handle.

    If such an `OSError` is caught, then returncode will be set to 1, and the
    error value will be set to the str() value of the exception.

    :type cmdlist: list of str
    :param stdin: string to pipe to the process
    :type stdin: str
    :return: Tuple of stdout, stderr, returncode
    :rtype: tuple[str, str, int]
    """
    termenc = urwid.util.detected_encoding
    cmdlist = [s.encode(termenc) for s in cmdlist]

    environment = os.environ.copy()
    if env is not None:
        environment.update(env)
    logging.debug('ENV = %s', environment)
    logging.debug('CMD = %s', cmdlist)
    try:
        proc = await asyncio.create_subprocess_exec(
            *cmdlist,
            env=environment,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
            stdin=asyncio.subprocess.PIPE if stdin else None)
    except OSError as e:
        return ('', str(e), 1)
    out, err = await proc.communicate(stdin.encode(termenc) if stdin else None)
    return (out.decode(termenc), err.decode(termenc), proc.returncode)


def guess_mimetype(blob):
    """
    uses file magic to determine the mime-type of the given data blob.

    :param blob: file content as read by file.read()
    :type blob: data
    :returns: mime-type, falls back to 'application/octet-stream'
    :rtype: str
    """
    mimetype = 'application/octet-stream'
    # this is a bit of a hack to support different versions of python magic.
    # Hopefully at some point this will no longer be necessary
    #
    # the version with open() is the bindings shipped with the file source from
    # http://darwinsys.com/file/ - this is what is used by the python-magic
    # package on Debian/Ubuntu. However, it is not available on pypi/via pip.
    #
    # the version with from_buffer() is available at
    # https://github.com/ahupp/python-magic and directly installable via pip.
    #
    # for more detail see https://github.com/pazz/alot/pull/588
    if hasattr(magic, 'open'):
        m = magic.open(magic.MAGIC_MIME_TYPE)
        m.load()
        magictype = m.buffer(blob)
    elif hasattr(magic, 'from_buffer'):
        # cf. issue #841
        magictype = magic.from_buffer(blob, mime=True) or magictype
    else:
        raise Exception('Unknown magic API')

    # libmagic does not always return proper mimetype strings, cf. issue #459
    if re.match(r'\w+\/\w+', magictype):
        mimetype = magictype
    return mimetype


def guess_encoding(blob):
    """
    uses file magic to determine the encoding of the given data blob.

    :param blob: file content as read by file.read()
    :type blob: data
    :returns: encoding
    :rtype: str
    """
    # this is a bit of a hack to support different versions of python magic.
    # Hopefully at some point this will no longer be necessary
    #
    # the version with open() is the bindings shipped with the file source from
    # http://darwinsys.com/file/ - this is what is used by the python-magic
    # package on Debian/Ubuntu.  However it is not available on pypi/via pip.
    #
    # the version with from_buffer() is available at
    # https://github.com/ahupp/python-magic and directly installable via pip.
    #
    # for more detail see https://github.com/pazz/alot/pull/588
    if hasattr(magic, 'open'):
        m = magic.open(magic.MAGIC_MIME_ENCODING)
        m.load()
        return m.buffer(blob)
    elif hasattr(magic, 'from_buffer'):
        m = magic.Magic(mime_encoding=True)
        return m.from_buffer(blob)
    else:
        raise Exception('Unknown magic API')


def try_decode(blob):
    """Guess the encoding of blob and try to decode it into a str.

    :param bytes blob: The bytes to decode
    :returns: the decoded blob
    :rtype: str
    """
    assert isinstance(blob, bytes), 'cannot decode a str or non-bytes object'
    return blob.decode(guess_encoding(blob))


def libmagic_version_at_least(version):
    """
    checks if the libmagic library installed is more recent than a given
    version.

    :param version: minimum version expected in the form XYY (i.e. 5.14 -> 514)
                    with XYY >= 513
    """
    if hasattr(magic, 'open'):
        magic_wrapper = magic._libraries['magic']
    elif hasattr(magic, 'from_buffer'):
        magic_wrapper = magic.libmagic
    else:
        raise Exception('Unknown magic API')

    if not hasattr(magic_wrapper, 'magic_version'):
        # The magic_version function has been introduced in libmagic 5.13,
        # if it's not present, we can't guess right, so let's assume False
        return False

    # Depending on the libmagic/ctypes version, magic_version is a function or
    # a callable:
    if callable(magic_wrapper.magic_version):
        return magic_wrapper.magic_version() >= version

    return magic_wrapper.magic_version >= version


# TODO: make this work on blobs, not paths
def mimewrap(path, filename=None, ctype=None):
    """Take the contents of the given path and wrap them into an email MIME
    part according to the content type.  The content type is auto detected from
    the actual file contents and the file name if it is not given.

    :param path: the path to the file contents
    :type path: str
    :param filename: the file name to use in the generated MIME part
    :type filename: str or None
    :param ctype: the content type of the file contents in path
    :type ctype: str or None
    :returns: the message MIME part storing the data from path
    :rtype: subclasses of email.mime.base.MIMEBase
    """

    with open(path, 'rb') as f:
        content = f.read()
    if not ctype:
        ctype = guess_mimetype(content)
        # libmagic < 5.12 incorrectly detects excel/powerpoint files as
        # 'application/msword' (see #179 and #186 in libmagic bugtracker)
        # This is a workaround, based on file extension, useful as long
        # as distributions still ship libmagic 5.11.
        if (ctype == 'application/msword' and
                not libmagic_version_at_least(513)):
            mimetype, _ = mimetypes.guess_type(path)
            if mimetype:
                ctype = mimetype

    maintype, subtype = ctype.split('/', 1)
    if maintype == 'text':
        part = MIMEText(content.decode(guess_encoding(content), 'replace'),
                        _subtype=subtype,
                        _charset='utf-8')
    elif maintype == 'image':
        part = MIMEImage(content, _subtype=subtype)
    elif maintype == 'audio':
        part = MIMEAudio(content, _subtype=subtype)
    else:
        part = MIMEBase(maintype, subtype)
        part.set_payload(content)
        # Encode the payload using Base64
        email.encoders.encode_base64(part)
    # Set the filename parameter
    if not filename:
        filename = os.path.basename(path)
    part.add_header('Content-Disposition', 'attachment',
                    filename=filename)
    return part


def shell_quote(text):
    """Escape the given text for passing it to the shell for interpretation.
    The resulting string will be parsed into one "word" (in the sense used in
    the shell documentation, see sh(1)) by the shell.

    :param text: the text to quote
    :type text: str
    :returns: the quoted text
    :rtype: str
    """
    return "'%s'" % text.replace("'", """'"'"'""")


def get_xdg_env(env_name, fallback):
    """ Used for XDG_* env variables to return fallback if unset *or* empty """
    env = os.environ.get(env_name)
    return env if env else fallback

def formataddr(pair):
    """ this is the inverse of email.utils.parseaddr:
    other than email.utils.formataddr, this
    - *will not* re-encode unicode strings, and
    - *will* re-introduce quotes around real names containing commas
    """
    name, address = pair
    if not name:
        return address
    elif ',' in name:
        name = "\"" + name + "\""
    return "{0} <{1}>".format(name, address)