path: root/alot/helper.py

# -*- coding: utf-8 -*-
# Copyright (C) 2011-2012  Patrick Totzke <patricktotzke@gmail.com>
# This file is released under the GNU GPL, version 3 or a later revision.
# For further details see the COPYING file
from __future__ import absolute_import

from datetime import timedelta
from datetime import datetime
from collections import deque
from cStringIO import StringIO
import logging
import mimetypes
import os
import re
import shlex
import subprocess
import email
from email.generator import Generator
from email.mime.audio import MIMEAudio
from email.mime.base import MIMEBase
from email.mime.image import MIMEImage
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart

import urwid
import magic
from twisted.internet import reactor
from twisted.internet.protocol import ProcessProtocol
from twisted.internet.defer import Deferred


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('\"', '\\\"')
    # encode s to utf-8 for shlex
    if isinstance(s, unicode):
        s = s.encode('utf-8')
    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.
    """
    if isinstance(cmdstring, unicode):
        cmdstring = cmdstring.encode('utf-8', errors='ignore')
    return shlex.split(cmdstring)


def string_sanitize(string, tab_width=8):
    r"""
    strips, and replaces non-printable characters

    :param tab_width: number of spaces to replace tabs with. Read from
                      `globals.tabwidth` setting if `None`
    :type tab_width: int or `None`

    >>> string_sanitize(' foo\rbar ', 8)
    ' foobar '
    >>> string_sanitize('foo\tbar', 8)
    'foo     bar'
    >>> string_sanitize('foo\t\tbar', 8)
    'foo             bar'
    """

    string = string.replace('\r', '')

    lines = list()
    for line in string.split('\n'):
        tab_count = line.count('\t')

        if tab_count > 0:
            line_length = 0
            new_line = list()
            for i, chunk in enumerate(line.split('\t')):
                line_length += len(chunk)
                new_line.append(chunk)

                if i < tab_count:
                    next_tab_stop_in = tab_width - (line_length % tab_width)
                    new_line.append(' ' * next_tab_stop_in)
                    line_length += next_tab_stop_in
            lines.append(''.join(new_line))
        else:
            lines.append(line)

    return '\n'.join(lines)


def string_decode(string, enc='ascii'):
    """
    safely decodes string to unicode bytestring, respecting `enc` as a hint.
    """

    if enc is None:
        enc = 'ascii'
    try:
        string = unicode(string, enc, errors='replace')
    except LookupError:  # malformed enc string
        string = string.decode('ascii', errors='replace')
    except TypeError:  # already unicode
        pass
    return string


def shorten(string, maxlen):
    """shortens string if longer than maxlen, appending ellipsis"""
    if 1 < maxlen < len(string):
        string = string[:maxlen - 1] + u'\u2026'
    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 u''

    # 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(u'\u2026')
            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 pretty_datetime(d):
    """
    translates :class:`datetime` `d` to a "sup-style" human readable string.

    >>> now = datetime.now()
    >>> now.strftime('%c')
    'Sat 31 Mar 2012 14:47:26 '
    >>> pretty_datetime(now)
    u'just now'
    >>> pretty_datetime(now - timedelta(minutes=1))
    u'1min ago'
    >>> pretty_datetime(now - timedelta(hours=5))
    u'5h ago'
    >>> pretty_datetime(now - timedelta(hours=12))
    u'02:54am'
    >>> pretty_datetime(now - timedelta(days=1))
    u'yest 02pm'
    >>> pretty_datetime(now - timedelta(days=2))
    u'Thu 02pm'
    >>> pretty_datetime(now - timedelta(days=7))
    u'Mar 24'
    >>> pretty_datetime(now - timedelta(days=356))
    u'Apr 2011'
    """
    ampm = d.strftime('%p').lower()
    if len(ampm):
        hourfmt = '%I' + ampm
        hourminfmt = '%I:%M' + ampm
    else:
        hourfmt = '%Hh'
        hourminfmt = '%H:%M'

    now = datetime.now()
    today = now.date()
    if d.date() == today or d > now - timedelta(hours=6):
        delta = datetime.now() - d
        if delta.seconds < 60:
            string = 'just now'
        elif delta.seconds < 3600:
            string = '%dmin ago' % (delta.seconds / 60)
        elif delta.seconds < 6 * 3600:
            string = '%dh ago' % (delta.seconds / 3600)
        else:
            string = d.strftime(hourminfmt)
    elif d.date() == today - timedelta(1):
        string = d.strftime('yest ' + hourfmt)
    elif d.date() > today - timedelta(7):
        string = d.strftime('%a ' + hourfmt)
    elif d.year != today.year:
        string = d.strftime('%b %Y')
    else:
        string = d.strftime('%b %d')
    return string_decode(string, 'UTF-8')


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
    :return: triple of stdout, stderr, return value of the shell command
    :rtype: str, str, int
    """

    out, err, ret = '', '', 0
    try:
        if stdin:
            proc = subprocess.Popen(cmdlist, stdin=subprocess.PIPE,
                                    stdout=subprocess.PIPE,
                                    stderr=subprocess.PIPE)
            out, err = proc.communicate(stdin)
            ret = proc.poll()
        else:
            try:
                out = subprocess.check_output(cmdlist)
            except subprocess.CalledProcessError as e:
                err = e.output
                ret = e.returncode
    except OSError as e:
        err = e.strerror
        ret = e.errno

    out = string_decode(out, urwid.util.detected_encoding)
    err = string_decode(err, urwid.util.detected_encoding)
    return out, err, ret


def call_cmd_async(cmdlist, stdin=None, env=None):
    """
    get a shell commands output, error message and return value as a deferred.

    :type cmdlist: list of str
    :param stdin: string to pipe to the process
    :type stdin: str
    :return: deferred that calls back with triple of stdout, stderr and
             return value of the shell command
    :rtype: `twisted.internet.defer.Deferred`
    """

    class _EverythingGetter(ProcessProtocol):
        def __init__(self, deferred):
            self.deferred = deferred
            self.outBuf = StringIO()
            self.errBuf = StringIO()
            self.outReceived = self.outBuf.write
            self.errReceived = self.errBuf.write

        def processEnded(self, status):
            termenc = urwid.util.detected_encoding
            out = string_decode(self.outBuf.getvalue(), termenc)
            err = string_decode(self.errBuf.getvalue(), termenc)
            if status.value.exitCode == 0:
                self.deferred.callback(out)
            else:
                terminated_obj = status.value
                terminated_obj.stderr = err
                self.deferred.errback(terminated_obj)

    d = Deferred()
    environment = os.environ
    if env is not None:
        environment.update(env)
    logging.debug('ENV = %s', environment)
    logging.debug('CMD = %s', cmdlist)
    proc = reactor.spawnProcess(_EverythingGetter(d), executable=cmdlist[0],
                                env=environment,
                                args=cmdlist)
    if stdin:
        logging.debug('writing to stdin')
        proc.write(stdin)
        proc.closeStdin()
    return d


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 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

    return magic_wrapper.magic_version >= version


# TODO: make this work on blobs, not paths
def mimewrap(path, filename=None, ctype=None):
    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):
    return "'%s'" % text.replace("'", """'"'"'""")


def tag_cmp(a, b):
    r'''
    Sorting tags using this function puts all tags of length 1 at the
    beginning. This groups all tags mapped to unicode characters.
    '''
    if min(len(a), len(b)) == 1 and max(len(a), len(b)) > 1:
        return cmp(len(a), len(b))
    else:
        return cmp(a.lower(), b.lower())


def humanize_size(size):
    for factor, format_string in ((1, '%i'),
                                  (1024, '%iK'),
                                  (1024 * 1024, '%.1fM')):
        if size / factor < 1024:
            return format_string % (float(size) / factor)
    return format_string % (size / factor)


def parse_mailcap_nametemplate(tmplate='%s'):
    """this returns a prefix and suffix to be used
    in the tempfile module for a given mailcap nametemplate string"""
    nt_list = tmplate.split('%s')
    template_prefix = ''
    template_suffix = ''
    if len(nt_list) == 2:
        template_suffix = nt_list[1]
        template_prefix = nt_list[0]
    else:
        template_suffix = tmplate
    return (template_prefix, template_suffix)


def parse_mailto(mailto_str):
    """
    Interpret mailto-string

    :param mailto_str: the string to interpret. Must conform to :rfc:2368.
    :type mailto_str: str
    :return: the header fields and the body found in the mailto link as a tuple
        of length two
    :rtype: tuple(dict(str->list(str)), str)
    """
    if mailto_str.startswith('mailto:'):
        import urllib
        to_str, parms_str = mailto_str[7:].partition('?')[::2]
        headers = {}
        body = u''

        to = urllib.unquote(to_str)
        if to:
            headers['To'] = [to]

        for s in parms_str.split('&'):
            key, value = s.partition('=')[::2]
            key = key.capitalize()
            if key == 'Body':
                body = urllib.unquote(value)
            elif value:
                headers[key] = [urllib.unquote(value)]
        return (headers, body)
    else:
        return (None, None)


def mailto_to_envelope(mailto_str):
    """
    Interpret mailto-string into a :class:`alot.db.envelope.Envelope`
    """
    from alot.db.envelope import Envelope
    headers, body = parse_mailto(mailto_str)
    return Envelope(bodytext=body, headers=headers)


def RFC3156_canonicalize(text):
    """
    Canonicalizes plain text (MIME-encoded usually) according to RFC3156.

    This function works as follows (in that order):

    1. Convert all line endings to \\\\r\\\\n (DOS line endings).
    2. Ensure the text ends with a newline (\\\\r\\\\n).
    3. Encode all occurences of "From " at the beginning of a line
       to "From=20" in order to prevent other mail programs to replace
       this with "> From" (to avoid MBox conflicts) and thus invalidate
       the signature.

    :param text: text to canonicalize (already encoded as quoted-printable)
    :rtype: str
    """
    text = re.sub("\r?\n", "\r\n", text)
    if not text.endswith("\r\n"):
        text += "\r\n"
    text = re.sub("^From ", "From=20", text, flags=re.MULTILINE)
    return text


def email_as_string(mail):
    """
    Converts the given message to a string, without mangling "From" lines
    (like as_string() does).

    :param mail: email to convert to string
    :rtype: str
    """
    fp = StringIO()
    g = Generator(fp, mangle_from_=False, maxheaderlen=78)
    g.flatten(mail)
    as_string = RFC3156_canonicalize(fp.getvalue())

    if isinstance(mail, MIMEMultipart):
        # Get the boundary for later
        boundary = mail.get_boundary()

        # Workaround for http://bugs.python.org/issue14983:
        # Insert a newline before the outer mail boundary so that other mail
        # clients can verify the signature when sending an email which contains
        # attachments.
        as_string = re.sub(r'--(\r\n)--' + boundary,
                           r'--\g<1>\g<1>--' + boundary,
                           as_string, flags=re.MULTILINE)

    return as_string