From be69ac901db8c494087e95b3a8511a45bcc9e2a3 Mon Sep 17 00:00:00 2001 From: Eric Wheeler Date: Wed, 13 May 2026 15:10:15 -0700 Subject: [PATCH] scripts: add contact-to-vcf converter with deduplication libpff exports one Contact.txt per item, leading to duplicate records when processing large mailbox archives. This script reads a list of Contact.txt paths from stdin, merges duplicates keyed on primary email or display name, and emits a single vCard 3.0 .vcf file to stdout. - parse Contact.txt fields into a structured dict per file - filter Exchange internal addresses (/o=) from email fields - key contacts on first real email, falling back to display name - accumulate phones and emails when merging duplicate keys - map phone label substrings to vCard TEL type tokens - emit vCard 3.0 blocks with proper value escaping - report input/error/skip/merge/unique tallies to stderr Signed-off-by: Eric Wheeler --- scripts/contact-to-vcf.py | 304 ++++++++++++++++++++++++++++++++++++++ 1 file changed, 304 insertions(+) create mode 100644 scripts/contact-to-vcf.py diff --git a/scripts/contact-to-vcf.py b/scripts/contact-to-vcf.py new file mode 100644 index 00000000..23f8846a --- /dev/null +++ b/scripts/contact-to-vcf.py @@ -0,0 +1,304 @@ +#!/usr/bin/env python3 +"""Convert libpff Contact.txt exports to a merged vCard 3.0 .vcf file. + +Usage: + find ... -name Contact.txt | python3 debug/contact_to_vcf.py > contacts.vcf + python3 debug/contact_to_vcf.py < file_list.txt > contacts.vcf +""" + +import sys +import re + +# Maps source phone label substrings to vCard TEL type tokens. +_PHONE_TYPE_MAP = { + 'mobile': 'CELL', + 'business': 'WORK', + 'home': 'HOME', + 'other': 'VOICE', + 'fax': 'FAX', +} + + +def _phone_vcf_type(label): + """Return vCard TEL type token for a source phone field label.""" + label_lower = label.lower() + for keyword, vcf_type in _PHONE_TYPE_MAP.items(): + if keyword in label_lower: + return vcf_type + return 'VOICE' + + +def is_real_email(value): + """Return True if value looks like a real email address.""" + if not value: + return False + if value.startswith('/o=') or value.startswith('/O='): + return False + return bool(re.match(r'^[^@\s]+@[^@\s]+\.[^@\s]+$', value)) + + +def _display_name(fields): + """Derive a display name from available name fields.""" + if fields['display']: + return fields['display'] + parts = [] + if fields['given']: + parts.append(fields['given']) + if fields['surname']: + parts.append(fields['surname']) + return ' '.join(parts) + + +def parse_contact(path): + """Parse a single Contact.txt file into a field dict. + + Returns dict with keys: given, surname, display, file_under, initials, + name_title, generational, company, job_title, office, department, + locality, country, postal (list of lines), phones (list of (vcf_type, number)), + emails (list of str). + """ + try: + with open(path, encoding='utf-8', errors='replace') as fh: + lines = fh.readlines() + except OSError: + return None + + fields = { + 'given': '', + 'surname': '', + 'display': '', + 'file_under': '', + 'initials': '', + 'name_title': '', + 'generational': '', + 'company': '', + 'job_title': '', + 'office': '', + 'department': '', + 'locality': '', + 'country': '', + 'postal': [], + 'phones': [], # list of (vcf_type, number) + 'emails': [], + } + + in_contact = False + in_postal = False + postal_lines = [] + + for line in lines: + stripped = line.rstrip('\n') + + if stripped.strip() == 'Contact:': + in_contact = True + continue + + if stripped.strip() == 'Body:': + if in_postal and postal_lines: + fields['postal'] = postal_lines + in_contact = False + in_postal = False + continue + + if not in_contact: + continue + + # Postal address continuation: no label prefix + if in_postal and not re.match(r'^[A-Za-z][\w\s]*:\s', stripped): + if stripped.strip(): + postal_lines.append(stripped.strip()) + continue + + if in_postal and postal_lines: + fields['postal'] = postal_lines + in_postal = False + + m = re.match(r'^([A-Za-z][\w\s]*):\s+(.*)', stripped) + if not m: + continue + + label = m.group(1).strip() + label_lower = label.lower() + value = m.group(2).strip() + + if label_lower == 'given name': + fields['given'] = fields['given'] or value + elif label_lower == 'surname': + fields['surname'] = fields['surname'] or value + elif label_lower == 'display name': + fields['display'] = fields['display'] or value + elif label_lower == 'file under': + fields['file_under'] = fields['file_under'] or value + elif label_lower == 'initials': + fields['initials'] = fields['initials'] or value + elif label_lower == 'title': + fields['name_title'] = fields['name_title'] or value + elif label_lower == 'generational abbreviation': + fields['generational'] = fields['generational'] or value + elif label_lower == 'company name': + fields['company'] = fields['company'] or value + elif label_lower == 'job title': + fields['job_title'] = fields['job_title'] or value + elif label_lower == 'office location': + fields['office'] = fields['office'] or value + elif label_lower == 'department name': + fields['department'] = fields['department'] or value + elif label_lower == 'locality': + fields['locality'] = fields['locality'] or value + elif label_lower == 'country': + fields['country'] = fields['country'] or value + elif label_lower == 'postal address': + postal_lines = [value] if value else [] + in_postal = True + elif re.search(r'phone|fax', label_lower): + if value: + vcf_type = _phone_vcf_type(label) + if (vcf_type, value) not in fields['phones']: + fields['phones'].append((vcf_type, value)) + elif re.match(r'email address \d', label_lower): + if is_real_email(value) and value.lower() not in [e.lower() for e in fields['emails']]: + fields['emails'].append(value) + + if in_postal and postal_lines: + fields['postal'] = postal_lines + + return fields + + +def contact_key(fields): + """Return the merge key for a contact record. + + Keyed on first real email (lowercased) when present, else display name. + """ + if fields['emails']: + return ('email', fields['emails'][0].lower()) + display = _display_name(fields) + return ('display', display.lower()) if display else None + + +def merge_into(base, incoming): + """Merge incoming fields into base, accumulating phones and emails.""" + scalar_fields = ( + 'given', 'surname', 'display', 'file_under', 'initials', + 'name_title', 'generational', 'company', 'job_title', + 'office', 'department', 'locality', 'country', + ) + for f in scalar_fields: + if not base[f] and incoming[f]: + base[f] = incoming[f] + + if not base['postal'] and incoming['postal']: + base['postal'] = incoming['postal'] + + for phone_entry in incoming['phones']: + if phone_entry not in base['phones']: + base['phones'].append(phone_entry) + + for email in incoming['emails']: + if email.lower() not in [e.lower() for e in base['emails']]: + base['emails'].append(email) + + +def vcard_escape(value): + """Escape special characters per vCard 3.0 value encoding rules.""" + value = value.replace('\\', '\\\\') + value = value.replace(',', '\\,') + value = value.replace(';', '\\;') + value = value.replace('\n', '\\n') + return value + + +def format_vcard(fields): + """Render a merged contact dict as a vCard 3.0 string.""" + output_lines = ['BEGIN:VCARD', 'VERSION:3.0'] + + given = vcard_escape(fields['given']) + surname = vcard_escape(fields['surname']) + prefix = vcard_escape(fields['name_title']) + suffix = vcard_escape(fields['generational']) + initials = vcard_escape(fields['initials']) + + display = _display_name(fields) or 'Unknown' + output_lines.append('FN:' + vcard_escape(display)) + + # N: surname;given;additional;prefix;suffix + output_lines.append(f'N:{surname};{given};{initials};{prefix};{suffix}') + + if fields['company']: + org_parts = [vcard_escape(fields['company'])] + if fields['department']: + org_parts.append(vcard_escape(fields['department'])) + output_lines.append('ORG:' + ';'.join(org_parts)) + + if fields['job_title']: + output_lines.append('TITLE:' + vcard_escape(fields['job_title'])) + + for vcf_type, number in fields['phones']: + output_lines.append(f'TEL;TYPE={vcf_type}:{vcard_escape(number)}') + + for email in fields['emails']: + output_lines.append('EMAIL;TYPE=INTERNET:' + vcard_escape(email)) + + if fields['postal'] or fields['locality'] or fields['country']: + street = '\\n'.join(vcard_escape(ln) for ln in fields['postal']) + locality = vcard_escape(fields['locality']) + country = vcard_escape(fields['country']) + # ADR: pobox;ext;street;locality;region;postalcode;country + output_lines.append(f'ADR;TYPE=WORK:;;{street};{locality};;;{country}') + + output_lines.append('END:VCARD') + return '\r\n'.join(output_lines) + '\r\n' + + +def _log(msg): + """Write an informational line to stderr.""" + print(msg, file=sys.stderr) + + +def main(): + contacts = {} # key -> fields dict + n_inputs = 0 + n_parse_errors = 0 + n_no_key = 0 + n_merged = 0 + + for raw_path in sys.stdin: + path = raw_path.strip() + if not path: + continue + + n_inputs += 1 + fields = parse_contact(path) + if fields is None: + n_parse_errors += 1 + _log(f'error: could not read {path}') + continue + + key = contact_key(fields) + if key is None: + n_no_key += 1 + _log(f'skip: no key (no email, no name) in {path}') + continue + + if key in contacts: + n_merged += 1 + merge_into(contacts[key], fields) + else: + contacts[key] = fields + + for fields in contacts.values(): + sys.stdout.write(format_vcard(fields)) + + n_outputs = len(contacts) + _log('') + _log(f'inputs read: {n_inputs}') + _log(f'parse errors: {n_parse_errors}') + _log(f'skipped (no key): {n_no_key}') + _log(f'merged duplicates: {n_merged}') + _log(f'unique contacts: {n_outputs}') + _log(f'accounted: {n_parse_errors + n_no_key + n_merged + n_outputs}' + f' (expect {n_inputs})') + + +if __name__ == '__main__': + main()