summaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
authorGravatar √Čtienne Deparis 2020-03-13 23:04:58 +0100
committerGravatar √Čtienne Deparis 2020-03-13 23:39:03 +0100
commitd62c042be70ad19c82e80116145c883ebe66b599 (patch)
tree9094aba914fbd97fd6fb392b673b7be12e93fa47
downloadpimtools-d62c042be70ad19c82e80116145c883ebe66b599.tar.gz
pimtools-d62c042be70ad19c82e80116145c883ebe66b599.tar.xz
pimtools-d62c042be70ad19c82e80116145c883ebe66b599.zip
Initial import
-rw-r--r--.gitignore6
-rw-r--r--pimtools/__init__.py1
-rw-r--r--pimtools/buddies.py203
-rw-r--r--pimtools/parsers/__init__.py0
-rw-r--r--pimtools/parsers/abstract.py187
-rw-r--r--pimtools/parsers/vcard.py87
-rw-r--r--pimtools/parsers/vtodo.py45
-rw-r--r--pimtools/todos.py122
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()