Convert JSON exported contacts from Android/Telegram to vCard (.vcf)

json2vcf_converter.py

import enum
import json
import sys
from typing import Any, Dict, Iterable, List, TextIO, Union


DOUBLE_UNDERSCORE = '__'
SPACE = ' '


class TransformerType(enum.Enum):
    AFTER_LAST_SPACE = 'after_last_space'
    BEFORE_LAST_SPACE = 'before_last_space'


class InputContactType(enum.Enum):
    ANDROID = 'android_contact'
    TELEGRAM = 'telegram_contact'


# https://stackoverflow.com/a/67292548
# InputContactValue = typing.Literal['android_contact', 'telegram_contact', ]
# assert (
#     set(typing.get_args(InputContactValue))
#     == {item.value for item in InputContactType}
# )


class InternalField:
    FIRST_NAME: str = 'first_name'
    LAST_NAME: str = 'last_name'
    PHONE: str = 'phone_number'
    ORGANIZATION: str = 'organization'


class TelegramField:
    FIRST_NAME: str = 'first_name'
    LAST_NAME: str = 'last_name'
    PHONE: str = 'phone_number'


class AndroidField:
    FIRST_NAME: str = (
        'display_name' + DOUBLE_UNDERSCORE
        + TransformerType.BEFORE_LAST_SPACE.value
    )
    LAST_NAME: str = (
        'display_name' + DOUBLE_UNDERSCORE
        + TransformerType.AFTER_LAST_SPACE.value
    )
    PHONE: str = 'data4'
    ORGANIZATION: str = 'company'


def to_vcf_entity(data: dict) -> str:
    first_name = data[InternalField.FIRST_NAME]
    last_name = data[InternalField.LAST_NAME]

    vcf_lines = []
    vcf_lines.append('BEGIN:VCARD')
    vcf_lines.append('VERSION:4.0')
    vcf_lines.append('N:%s' % (last_name + ';' + first_name))
    vcf_lines.append('FN:%s' % (
        first_name + ('' if len(last_name) == 0 else ' ') + last_name
    ))
    vcf_lines.append('TEL:%s' % data[InternalField.PHONE])
    try:
        vcf_lines.append('ORG:%s' % data[InternalField.ORGANIZATION])
    except KeyError:
        # The contact has no organization info
        pass
    vcf_lines.append('END:VCARD')
    vcf_string = '\n'.join(vcf_lines) + '\n'
    return vcf_string


def extract_date_from_dict_using_path_part_list(target_dict: dict, path: List[str]):
    result = target_dict
    for path_part in path:
        result = result.get(path_part)
        if result is None:
            break
    return result


class JSONContactImporter:
    def __init__(self):
        self.field_constant_class = InternalField

    @staticmethod
    def get_contact_list(raw_content: Iterable) -> List[Dict[str, Any]]:
        return list(raw_content)

    def from_contact_data(
            self, contact: Dict[str, Any], skip_no_name_contact: bool = False,
    ) -> Dict[str, Any]:
        if (
                skip_no_name_contact
                and len(contact[self.field_constant_class.FIRST_NAME]) == 0
                and len(contact[self.field_constant_class.LAST_NAME]) == 0
        ):
            return {}

        result = {}
        for field_id, field_name in self.field_constant_class.__dict__.items():
            if (
                    field_id.startswith(DOUBLE_UNDERSCORE)
                    and field_id.endswith(DOUBLE_UNDERSCORE)
            ):
                continue

            if not(hasattr(InternalField, field_id)):
                raise ValueError(
                    'The provided field is not known to the system:' + field_id
                )

            if DOUBLE_UNDERSCORE not in field_name:
                field_value = contact.get(field_name)
            else:
                field_name_parts = field_name.split(DOUBLE_UNDERSCORE)
                if field_name_parts[-1] == TransformerType.AFTER_LAST_SPACE.value:
                    field_value = extract_date_from_dict_using_path_part_list(
                        contact, field_name_parts[:-1],
                    )
                    last_space_index = field_value.rfind(SPACE)
                    field_value = field_value[last_space_index + 1:]
                elif field_name_parts[-1] == TransformerType.BEFORE_LAST_SPACE.value:
                    field_value = extract_date_from_dict_using_path_part_list(
                        contact, field_name_parts[:-1],
                    )
                    last_space_index = field_value.rfind(SPACE)
                    field_value = field_value[:last_space_index]
                else:
                    field_value = extract_date_from_dict_using_path_part_list(
                        contact, field_name_parts,
                    )

            if field_value is not None:
                result[getattr(InternalField, field_id)] = field_value

        return result

    def import_contacts(self, json_file: TextIO) -> List[Dict[str, str]]:
        data = json.load(json_file)
        contacts_list = self.get_contact_list(data)

        result = [self.from_contact_data(contact) for contact in contacts_list]
        return result

    def to_vcf_entity_list(self, json_file: TextIO) -> List[str]:
        imported_contact_list = self.import_contacts(json_file)

        vcf_entity_list = [
            to_vcf_entity(imported_contact)
            for imported_contact in imported_contact_list
        ]
        return vcf_entity_list


class TelegramContactImporter(JSONContactImporter):
    def __init__(self):
        super().__init__()
        self.field_constant_class = TelegramField

    @staticmethod
    def get_contact_list(
            raw_content: Dict[str, Dict[str, List[Dict[str, Any]]]]
    ) -> List[Dict[str, Any]]:
        return raw_content['contacts']['list']


class AndroidContactImporter(JSONContactImporter):
    def __init__(self):
        super().__init__()
        self.field_constant_class= AndroidField

    @staticmethod
    def get_contact_list(
            raw_content: List[Dict[
                str, Union[str, List[Dict[str, Union[str, List[Dict[str, str]]]]]]
            ]]
    ) -> List[Dict[str, str]]:
        result = []
        for contact_root in raw_content:
            for raw_contact in contact_root['raw_contacts']:
                for contacts_data_item in raw_contact['contacts_data']:
                    if (
                            contacts_data_item['mimetype']
                            == 'vnd.android.cursor.item/phone_v2'
                    ):
                        result.append(contacts_data_item)
        return result


def main(input_contact_type: InputContactType):
    input_file = sys.argv[1]
    output_file = sys.argv[2]

    contact_importer: JSONContactImporter
    if input_contact_type == InputContactType.TELEGRAM:
        contact_importer = TelegramContactImporter()
    elif input_contact_type == InputContactType.ANDROID:
        contact_importer = AndroidContactImporter()
    else:
        raise ValueError('Unknown input contact type: ' + str(input_contact_type))

    with open(input_file) as json_file:
        vcf_entity_list = contact_importer.to_vcf_entity_list(json_file)

    with open(output_file, 'w') as output_file:
        for vcf_entity in vcf_entity_list:
            output_file.write(vcf_entity)


if __name__ == '__main__':
    # input_type = InputContactType.TELEGRAM
    input_type = InputContactType.ANDROID
    main(input_type)