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
|
import email.message
import email.policy
from email.parser import BytesParser
import subprocess
from subprocess import PIPE
import xml.dom.minidom
import xml.dom
parser = BytesParser(policy=email.policy.default)
def get_mail(id: str) -> email.message.Message:
cmd = subprocess.run(['mu', 'find', '-u', f'i:{id}',
'--fields', 'l'],
stdout=PIPE)
filename = cmd.stdout.decode('UTF-8').strip()
with open(filename, "rb") as f:
mail = parser.parse(f)
return mail
class MuError(Exception):
codes = {
1: 'General Error',
2: 'No Matches',
4: 'Database is corrupted'
}
def __init__(self, returncode: int):
self.returncode: int = returncode
self.msg: str = MuError.codes.get(returncode, 'Unknown Error')
def __repr__(self):
return f'MuError({self.returncode}, "{self.msg}")'
def mu_search(query, sortfield='subject', reverse=False):
if not query:
raise ValueError('Query required for mu_search')
cmdline = ['mu', 'find',
'--format=xml',
query,
'--sortfield', sortfield]
if reverse:
cmdline.append('--reverse')
cmd = subprocess.Popen(cmdline, stdout=subprocess.PIPE)
dom = xml.dom.minidom.parse(cmd.stdout)
if returncode := cmd.wait():
raise MuError(returncode)
message_list = []
messages = dom.childNodes[0]
assert messages.localName == 'messages'
for message in messages.childNodes:
msg_dict = {}
if message.nodeType != xml.dom.Node.ELEMENT_NODE:
continue
for kv in message.childNodes:
if kv.nodeType != xml.dom.Node.ELEMENT_NODE:
continue
msg_dict[kv.localName] = kv.childNodes[0].data
message_list.append(msg_dict)
return message_list
|
