""" Wrapper for the `mu` command line. """ import email.message import email.policy from email.parser import BytesParser import subprocess from subprocess import PIPE import xml.dom.minidom import xml.dom from typing import ( Optional, ) parser = BytesParser(policy=email.policy.default) def find_file(id: str) -> Optional[str]: cmd = subprocess.run(['mu', 'find', '-u', f'i:{id}', '--fields', 'l'], stdout=PIPE) filename = cmd.stdout.decode('UTF-8').strip() if cmd.returncode == 4: return None if cmd.returncode != 0: raise MuError(cmd.returncode) return filename def get_mail(id: str) -> email.message.Message: """ Lookup email by Message-ID. [Raises] MuError """ with open(find_file(id), "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 __str__(self): return repr(self) def search(query: str, sortfield: Optional[str] = 'subject', reverse: bool = False) -> list[dict[str, str]]: """ [Parameters] query - Search query as per mu-find(1). sortfield - Field to sort the values by reverse - If the sort should be reversed [Returns] >>> {'from': 'Hugo Hörnquist ', 'date': '1585678375', 'size': '377', 'msgid': 'SAMPLE-ID@localhost', 'path': '/home/hugo/.local/var/mail/INBOX/cur/filename', 'maildir': '/INBOX' } """ if not query: raise ValueError('Query required for mu_search') cmdline = ['mu', 'find', '--format=xml', query] if sortfield: cmdline.extend(['--sortfield', sortfield]) if reverse: cmdline.append('--reverse') cmd = subprocess.run(cmdline, capture_output=True) if cmd.returncode == 1: raise MuError(cmd.returncode) if cmd.returncode == 4: # no matches return [] if cmd.returncode != 0: raise MuError(cmd.returncode) dom = xml.dom.minidom.parseString(cmd.stdout.decode('UTF-8')) 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 def info(): cmd = subprocess.Popen('mu info --nocolor'.split(' '), stdout=subprocess.PIPE, text=True) out = {} for line in cmd.stdout: if not line: continue if line[0] == '+': continue key, *value = [s.strip() for s in line.split('|') if s and not s.isspace()] out[key] = '|'.join(value) return out