diff options
author | 2020-03-13 23:04:58 +0100 | |
---|---|---|
committer | 2020-03-13 23:39:03 +0100 | |
commit | d62c042be70ad19c82e80116145c883ebe66b599 (patch) | |
tree | 9094aba914fbd97fd6fb392b673b7be12e93fa47 | |
download | pimtools-d62c042be70ad19c82e80116145c883ebe66b599.tar.gz pimtools-d62c042be70ad19c82e80116145c883ebe66b599.tar.xz pimtools-d62c042be70ad19c82e80116145c883ebe66b599.zip |
Initial import
-rw-r--r-- | .gitignore | 6 | ||||
-rw-r--r-- | pimtools/__init__.py | 1 | ||||
-rw-r--r-- | pimtools/buddies.py | 203 | ||||
-rw-r--r-- | pimtools/parsers/__init__.py | 0 | ||||
-rw-r--r-- | pimtools/parsers/abstract.py | 187 | ||||
-rw-r--r-- | pimtools/parsers/vcard.py | 87 | ||||
-rw-r--r-- | pimtools/parsers/vtodo.py | 45 | ||||
-rw-r--r-- | pimtools/todos.py | 122 |
8 files changed, 651 insertions, 0 deletions
diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..14840db --- /dev/null +++ b/.gitignore @@ -0,0 +1,6 @@ +build +dist +venv +.eggs +pimtools.egg-info +__pycache__ diff --git a/pimtools/__init__.py b/pimtools/__init__.py new file mode 100644 index 0000000..3dc1f76 --- /dev/null +++ b/pimtools/__init__.py @@ -0,0 +1 @@ +__version__ = "0.1.0" diff --git a/pimtools/buddies.py b/pimtools/buddies.py new file mode 100644 index 0000000..5131efa --- /dev/null +++ b/pimtools/buddies.py @@ -0,0 +1,203 @@ +import os +import re +import sys +import glob +import base64 +import subprocess +from argparse import ArgumentParser +from pimtools import __version__ +from pimtools.parsers.vcard import VCardParser + + +ADDRESSBOOKS = ["~/.vdirsyncer/addressbooks"] + + +class Buddies: + def __init__(self): + parser = ArgumentParser() + parser.add_argument("-v", "--version", action="store_true", + help="Display version information and exit.") + parser.add_argument("-f", "--file", action="store_true", + dest="filepath", help="Expect given query " + "to be vcard file paths.") + parser.add_argument("-o", "--only", dest="property_type", + help="When a property have different " + "possible types (e.g. home or work email), " + "display only this type. (Only relevant " + "for --show and --clip mode)") + parser.add_argument("-u", "--uid", action="store_true", + help="Add vcard uid in displayed fields. " + "(Only relevant for --show mode)") + mutu = parser.add_mutually_exclusive_group() + mutu.add_argument("-c", "--clip", choices=["email", "tel"], + help="Do not print anything, but instead " + "copy the given property value in the " + "clipboard using xclip.") + mutu.add_argument("-e", "--edit", action="store_true", + help="Open corresponding vcf files in edition.") + mutu.add_argument("-s", "--show", default="fancy", + choices=["email", "fancy", "mu", "mutt", + "org-contact", "photo", "pine", + "vcard"], + help="Use this format to display information. " + "The default value is `fancy'.") + parser.add_argument("query_or_file", nargs="+", + help="Query or file to parse.") + + args = parser.parse_args() + + if args.version: + print("Buddies - v{}".format(__version__)) + sys.exit(0) + + self.modifiers = { + "format": "{name} <{email}>", + "prop_type": args.property_type, + "show_uid": args.uid + } + + if args.filepath: + vcards = [ + VCardParser(os.path.expanduser(f), self.modifiers) + for f in args.query_or_file + ] + else: + vcards = self.search(args.query_or_file) + + if args.edit: + # vCards will be edited one after the other + operation = self.edit + elif args.clip == "email": + operation = self.clip_email + elif args.clip == "tel": + operation = self.clip_tel + elif args.show == "vcard": + operation = self.show + elif args.show == "photo": + operation = self.open_photo + elif args.show == "mutt": + # Mutt imply email, thus it's checked before email + self.modifiers["format"] = "{email}\t{name}" + print() # Mutt needs a first empty line + operation = self.print_email + elif args.show == "mu": + # Mu imply email, thus it's checked before email too + magenta_name = "\x1b[35m{}\x1b[0m".format("{name}") + green_email = "\x1b[32m{}\x1b[0m".format("{email}") + self.modifiers["format"] = magenta_name + " " + green_email + operation = self.print_email + elif args.show == "email": + operation = self.print_email + elif args.show == "org-contact": + operation = self.convert_org + elif args.show == "pine": + operation = self.convert_pine + else: + operation = print + for card in vcards: + operation(card) + + def search(self, query): + addressbooks = ADDRESSBOOKS + vcf_files = [] + for book in addressbooks: + vcf_files += glob.glob( + os.path.join(os.path.expanduser(book), "**/*.vcf") + ) + if "all" in query: + return [VCardParser(m, self.modifiers) for m in vcf_files] + vcards = [] + # Compile regexp once for all + queryx = [re.compile(q, re.I) for q in query] + for mate in vcf_files: + m = VCardParser(mate, self.modifiers) + email = m.get("email") + if isinstance(email, list): + email = " ".join([e[0] for e in email]) + data = " ".join( + [m.get("fn"), email, m.get("nickname")] + ).strip() + for qx in queryx: + if qx.search(data) is None: + continue + vcards.append(m) + # One match is sufficient + break + return vcards + + def edit(self, card): + p = subprocess.run( + [os.getenv("EDITOR", "nano"), card.filepath] + ) + if p.returncode == 0: + return + print("Error while editing {}".format(card.filepath), + file=sys.stderr) + sys.exit(p.returncode) + + def show(self, card): + with open(card.filepath, "r") as f: + print(f.read()) + + def clip_data(self, value): + if value == "": + return + if isinstance(value, list): + value = ", ".join([v[0] for v in value]) + cmd = subprocess.Popen(["xclip", "-selection", "primary"], + stdin=subprocess.PIPE) + cmd.communicate(value.encode()) + + def clip_email(self, card): + self.clip_data(card.get("email")) + + def clip_tel(self, card): + self.clip_data(card.get("tel")) + + def print_email(self, card): + val = card.get("email") + if val == "": + return + template = self.modifiers.get("format") + name = card.get("fn") + if isinstance(val, list): + print("\n".join([ + template.format(name=name, email=v[0]) + for v in val + ])) + else: + print(template.format(name=name, email=val)) + + def convert_pine(self, card): + val = card.get("email") + if val == "": + return + line = [card.get("nickname"), card.get("fn")] + if isinstance(val, list): + line.append("({})".format(",".join([v[0] for v in val]))) + else: + line.append(val) + print("\t".join(line)) + + def convert_org(self, card): + print(card.print_org_contact()) + print() + + def open_photo(self, card): + photo = card.get("photo") + if photo == "": + print("No photo for {}".format(card.get("fn")), file=sys.stderr) + return + tmp_pic_path = "/tmp/picture_" + card.get("uid") + with open(tmp_pic_path, "wb") as f: + f.write(base64.b64decode(photo)) + p = subprocess.run(["gio", "open", tmp_pic_path]) + if p.returncode == 0: + return + print("Error while opening photo from {}".format(card.filepath), + file=sys.stderr) + sys.exit(p.returncode) + + +if __name__ == "__main__": + Buddies() diff --git a/pimtools/parsers/__init__.py b/pimtools/parsers/__init__.py new file mode 100644 index 0000000..e69de29 --- /dev/null +++ b/pimtools/parsers/__init__.py diff --git a/pimtools/parsers/abstract.py b/pimtools/parsers/abstract.py new file mode 100644 index 0000000..400fc36 --- /dev/null +++ b/pimtools/parsers/abstract.py @@ -0,0 +1,187 @@ +import sys + + +class AbstractParser: + def __init__(self, filepath, context={}): + self.filepath = filepath + self.context = context + self.atoms = {} + self.max_len = 8 # len of longest possible vcard term (birthday) + self.skip_terms = [] + self.displayed_terms = [] + self.other_allowed_terms = [] + self.terms_aliases = {} + + def parse(self): + with open(self.filepath, "r") as f: + lines = f.readlines() + self._last_atom = None + self._last_data = None + for line in lines: + stripped = line.strip() + shield_terms = [ + "", "BEGIN:VCARD", "END:VCARD", "BEGIN:VTODO", + "END:VTODO", "BEGIN:VCALENDAR", "END:VCALENDAR" + ] + if stripped in shield_terms: + # Ignore some useless values + continue + if line.startswith(" "): + if self._last_atom is None: + raise SyntaxError( + "Continuation line found without open atom in {}" + .format(self.filepath) + ) + # This is a continuation line + self._last_data["value"] += line.strip() + else: + # This is a new atom + if self._last_atom is not None: + # Append last opened atom to self.atoms + self._write_atom() + # Now process the current line + self._extract_atom(line) + # Do it for the last loop step + self._write_atom() + + def _extract_atom(self, line): + data = line.split(":", 1) + headers = data[0].split(";") + last_atom = headers.pop(0) + # Warning: Provide empty metadata to the extract headers method + # to avoid pollution through methods calls and variable stored + # by references + meta_data = self._extract_headers(headers, {}) + content = {"headers": meta_data, "value": data[1].strip()} + self._last_atom = last_atom + self._last_data = content + + def _extract_headers(self, headers, meta_data): + for meta in headers: + if meta == "": + # Ignore empty cells + continue + name = meta.lower().split("=") + if len(name) == 1: + meta_data.setdefault("others", []) + meta_data["others"].append(name[0]) + else: + cur_val = meta_data.get(name[0]) + if cur_val is None: + meta_data[name[0]] = name[1] + elif isinstance(cur_val, list): + meta_data[name[0]].append(name[1]) + else: + meta_data[name[0]] = [cur_val, name[1]] + return meta_data + + def _write_atom(self): + if self._last_atom in self.skip_terms: + self._last_atom = None + self._last_data = None + return + all_allowed_terms = self.other_allowed_terms + self.displayed_terms + custom_handler = "_handle_{}_field".format(self._last_atom.lower()) + if hasattr(self, custom_handler): + getattr(self, custom_handler)() + elif self._last_atom not in all_allowed_terms: + print("WARNING: This card use a non standard term {}!\n> {}" + .format(self._last_atom, self.filepath), file=sys.stderr) + else: + new_val = self._last_data["value"].split(";") + if len(new_val) > 1: + self._last_data["value"] = [v for v in new_val if v != ""] + cur_val = self.atoms.get(self._last_atom) + if self._last_atom == "UID" and cur_val is not None: + print("WARNING: This card has 2 UID!\n> {}" + .format(self.filepath), file=sys.stderr) + if cur_val is None: + self.atoms[self._last_atom] = self._last_data + elif isinstance(cur_val, list): + self.atoms[self._last_atom].append(self._last_data) + else: + self.atoms[self._last_atom] = [cur_val, self._last_data] + self._last_atom = None + self._last_data = None + + def _match_property_type(self, prop_type, cur_type, headers): + if prop_type is None: + return True + if prop_type == "pref" and headers.get("pref") is not None: + return True + if isinstance(cur_type, list): + return prop_type in cur_type + return prop_type == cur_type + + def get(self, name): + content = self.atoms.get(name.upper(), {}) + if isinstance(content, list): + prop_type = self.context.get("prop_type") + values = [] + for v in content: + cur_type = v["headers"].get("type", "") + if self._match_property_type(prop_type, + cur_type, + v["headers"]): + values.append((v["value"], cur_type)) + return values + return content.get("value", "") + + def _format_line(self, title, value, default=None): + title = title.ljust(self.max_len) + if isinstance(value, list): + first = value.pop(0) + + def linefmt(title, value): + if isinstance(value, str): + return "{title} {value}".format( + title=title, value=value + ) + if isinstance(value[1], list): + val = ", ".join(value[1]) + else: + val = value[1] + return "{title} {value} ({info})".format( + title=title, value=value[0], info=val + ) + + results = [linefmt(title + " :", first)] + results += [linefmt(" " * (self.max_len + 2), v) for v in value] + return "\n".join(results) + value = value.replace(r"\n", "\n") + if value == "": + if default is None: + return None + value = default + return "{title} : {val}".format(title=title, val=value) + + def _format_title(self): + """This method must be overidden by subclasses""" + return [] + + def _format_output(self): + lines = [] + terms_iter = self.displayed_terms + if self.context.get("show_uid"): + terms_iter.insert(0, "UID") + lines += [ + self._format_line( + self.terms_aliases.get(term, term.capitalize()), + self.get(term) + ) for term in terms_iter + ] + return lines + + def __str__(self): + lines = self._format_title() + self._format_output() + lines.append("") + return "\n".join([l for l in lines if l is not None]) + + def __repr__(self): + return "[" + ", ".join(self.atoms.keys()) + "]" + + def __getattr__(self, name): + uname = name.upper() + if uname in self.atoms.keys(): + return self.atoms[uname] + raise AttributeError("No such attribute: {}".format(uname)) diff --git a/pimtools/parsers/vcard.py b/pimtools/parsers/vcard.py new file mode 100644 index 0000000..29be765 --- /dev/null +++ b/pimtools/parsers/vcard.py @@ -0,0 +1,87 @@ +from .abstract import AbstractParser + + +class VCardParser(AbstractParser): + def __init__(self, filepath, context={}): + super().__init__(filepath, context) + self.skip_terms = ["PRODID", "VERSION", "REV", "N"] + self.displayed_terms = [ + "NICKNAME", "TEL", "EMAIL", "IMPP", "ADR", "BDAY", "URL", + "TITLE", "ORG", "NOTE", "RELATED" + ] + self.other_allowed_terms = [ + "UID", "FN", "PHOTO", "X-DEPARTMENT", "X-GENDER" + ] + self.terms_aliases = { + "NICKNAME": "Nick", + "BDAY": "Birthday", + "IMPP": "IM", + "ADR": "Address" + } + self.parse() + + def _handle_photo_field(self): + new_val = self._last_data["value"].split(",", 1) + if len(new_val) <= 1: + return + self._last_data["value"] = new_val[1] + new_meta = self._extract_headers( + new_val[0].split(";"), + self._last_data["headers"] + ) + self._last_data["headers"] = new_meta + + def print_org_contact(self): + lines = [] + name = self.get("fn") + if name == "": + name = "Anonymous" + lines.append("* " + name) + lines.append(":PROPERTIES:") + + def linefmt(title, value): + if isinstance(value, str): + val = value + elif isinstance(value, list): + new_list = [v[0] if isinstance(v, tuple) else v + for v in value] + val = ", ".join(new_list) + if val == "": + return "" + return ":{title}: {value}".format( + title=title.upper(), value=val + ) + + email = linefmt("email", self.get("email")) + if email != "": + lines.append(email) + phone = linefmt("phone", self.get("tel")) + if phone != "": + lines.append(phone) + nick = linefmt("nickname", self.get("nickname")) + if nick != "": + lines.append(nick) + adr = linefmt("address", self.get("adr")) + if adr != "": + lines.append(adr) + bday = linefmt("birthday", self.get("bday")) + if bday != "": + lines.append(bday) + + lines.append(":END:") + return "\n".join(lines) + + def _format_title(self): + name = self.get("fn") + if name == "": + lines = ["Anonymous"] + else: + lines = [name] + lines.append("-" * len(lines[0])) + return lines + + def _format_output(self): + lines = super()._format_output() + if self.get("photo") != "": + lines.append(self._format_line("Photo", "yes")) + return lines diff --git a/pimtools/parsers/vtodo.py b/pimtools/parsers/vtodo.py new file mode 100644 index 0000000..c0eb826 --- /dev/null +++ b/pimtools/parsers/vtodo.py @@ -0,0 +1,45 @@ +from datetime import datetime +from .abstract import AbstractParser + + +class VTodoParser(AbstractParser): + def __init__(self, filepath, context={}): + super().__init__(filepath, context) + self.max_len = 11 # len of longest possible vtodo term (description) + self.skip_terms = [ + "PRODID", "VERSION", "CREATED", "DTSTAMP", "LAST-MODIFIED", + "BEGIN", "END", "RDATE", "TZNAME", "TZOFFSETTO", + "TZOFFSETFROM", "TZURL", "TZID", "DTSTART", "RRULE", + "X-LIC-LOCATION" + ] + self.displayed_terms = [ + "DUE", "PRIORITY", "STATUS", "COMPLETED", + "PERCENT-COMPLETE", "LOCATION", "DESCRIPTION", "CLASS" + ] + self.other_allowed_terms = [ + "SUMMARY", "UID", "X-MOZ-GENERATION", "SEQUENCE" + ] + self.terms_aliases = { + 'PERCENT-COMPLETE': 'Completion' + } + self.parse() + + def get(self, name): + value = super().get(name) + if value == "" or isinstance(value, list): + return value + if name == "DUE": + _raw_date = datetime.strptime(value, "%Y%m%dT%H%M%S") + value = _raw_date.strftime("%d/%m/%Y") + elif name == "PERCENT-COMPLETE": + value += "%" + return value + + def _format_title(self): + summary = self.get("summary") + if summary == "": + lines = ["N/A"] + else: + lines = [summary] + lines.append("-" * len(lines[0])) + return lines diff --git a/pimtools/todos.py b/pimtools/todos.py new file mode 100644 index 0000000..f81f5bb --- /dev/null +++ b/pimtools/todos.py @@ -0,0 +1,122 @@ +import os +import re +import sys +import glob +import subprocess +from argparse import ArgumentParser +from pimtools import __version__ +from pimtools.parsers.vtodo import VTodoParser + + +CALENDARS = ["~/.vdirsyncer/calendars"] + + +class Todos: + def __init__(self): + parser = ArgumentParser() + parser.add_argument("-v", "--version", action="store_true", + help="Display version information and exit.") + parser.add_argument("-f", "--file", action="store_true", + dest="filepath", help="Expect given query " + "to be vtodo file paths.") + parser.add_argument("-u", "--uid", action="store_true", + help="Add todo uid in displayed fields. " + "(Only relevant for --show mode)") + parser.add_argument("-a", "--all", action="store_true", + help="Show all, even completed tasks. " + "(Only relevant for --show mode)") + mutu = parser.add_mutually_exclusive_group() + mutu.add_argument("-e", "--edit", action="store_true", + help="Open corresponding ics files in edition.") + mutu.add_argument("-s", "--show", default="fancy", + choices=["fancy", "org", "vtodo"], + help="Use this format to display information. " + "The default value is `fancy'.") + parser.add_argument("query_or_file", nargs="+", + help="Query or file to parse.") + + args = parser.parse_args() + + if args.version: + print("Todos - v{}".format(__version__)) + sys.exit(0) + + self.modifiers = { + "show_uid": args.uid, + "show_all": args.all + } + + if args.filepath: + vtodos = self.filter_vtodo_files(args.query_or_file) + else: + vtodos = self.search(args.query_or_file) + + if args.edit: + # vtodos will be edited one after the other + operation = self.edit + elif args.show == "vtodo": + operation = self.show + elif args.show == "org": + operation = self.convert_org + else: + operation = print + for card in vtodos: + operation(card) + + def filter_vtodo_files(self, ics_files): + vtodos = [] + for _path in ics_files: + path = os.path.expanduser(_path) + with open(path, "r") as f: + file_lines = f.readlines() + for line in file_lines: + if line.strip() == "BEGIN:VTODO": + todo = VTodoParser(path, self.modifiers) + if self.modifiers.get("show_all") \ + or todo.get("status") != "COMPLETED": + vtodos.append(todo) + break + return vtodos + + def search(self, query): + calendars = CALENDARS + _ics_files = [] + for book in calendars: + _ics_files += glob.glob( + os.path.join(os.path.expanduser(book), "**/**/*.ics") + ) + ics_files = self.filter_vtodo_files(_ics_files) + if "all" in query: + return ics_files + vtodos = [] + # Compile regexp once for all + queryx = [re.compile(q, re.I) for q in query] + for todo in ics_files: + for qx in queryx: + if qx.search(todo.get("summary")) is None: + continue + vtodos.append(todo) + break + return vtodos + + def edit(self, card): + p = subprocess.run( + [os.getenv("EDITOR", "nano"), card.filepath] + ) + if p.returncode == 0: + return + print("Error while editing {}".format(card.filepath), + file=sys.stderr) + sys.exit(p.returncode) + + def show(self, card): + with open(card.filepath, "r") as f: + print(f.read()) + + def convert_org(self, card): + print(card.print_org_contact()) + print() + + +if __name__ == "__main__": + Todos() |