summaryrefslogtreecommitdiff
path: root/alot/addressbook/external.py
blob: c01a525e70c6f9036a3e9723cffb40bdd55615e7 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
# Copyright (C) 2011-2015  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

import logging
import re
import subprocess

from urwid.util import detected_encoding

from . import AddressBook, AddressbookError

class ExternalAddressbook(AddressBook):
    """:class:`AddressBook` that parses a shell command's output"""

    def __init__(self, commandline, regex, reflags=0,
                 external_filtering=True,
                 **kwargs):
        """
        :param commandline: commandline
        :type commandline: str
        :param regex: regular expression used to match contacts in `commands`
                      output to stdout. Must define subparts named "email" and
                      "name".
        :type regex: str
        :param reflags: flags to use with regular expression.
                        Use the constants defined in :mod:`re` here
                        (`re.IGNORECASE` etc.)
        :type reflags: str
        :param external_filtering: if True the command is fired
                        with the given search string as parameter
                        and the result is not filtered further.
                        If set to False, the command is fired without
                        additional parameters and the result list is filtered
                        according to the search string.
        :type external_filtering: bool
        """
        super().__init__(**kwargs)
        self.commandline = commandline
        self.regex = regex
        self.reflags = reflags
        self.external_filtering = external_filtering

    def get_contacts(self):
        return self._call_and_parse(self.commandline)

    def lookup(self, prefix):  # pragma: no cover
        if self.external_filtering:
            return self._call_and_parse(self.commandline + " " + prefix)
        else:
            return AddressBook.lookup(self, prefix)

    def _call_and_parse(self, cmd):
        try:
            result = subprocess.run(cmd, shell = True, check = True,
                                    capture_output = True,
                                    stdin = subprocess.DEVNULL)
        except subprocess.CalledProcessError as e:
            msg = 'abook command "%s" failed with code %s' % (e.cmd, e.returncode)
            if e.stderr:
                msg += '; stderr:\n%s' % \
                       e.stderr.decode(detected_encoding, errors = 'backslashreplace')
            raise AddressbookError(msg)

        output = result.stdout.decode(detected_encoding, errors = 'backslashreplace')
        if not output:
            logging.debug("No contacts in address book (empty string)")
            return []

        lines = output.splitlines()
        res = []
        logging.debug("Apply %s on %d results" % (self.regex, len(lines)))
        for l in lines:
            m = re.match(self.regex, l, self.reflags)
            if m:
                info = m.groupdict()
                if 'email' in info and 'name' in info:
                    email = info['email'].strip()
                    name = info['name']
                    res.append((name, email))
                    logging.debug("New match name=%s mail=%s" % (name, email))
        return res