Share your Python scripts

contacts tool

this script can take 1 or multiple vcf files and remove duplicates. Also works on db files.

Features:

  • remove duplicate contacts.
  • merge 2 vcf files into 1 vcf file. (useful if your contacts are not the same between different phones.)
  • extract contacts from .db file (if you can’t get a vcf)
  • change all American phone numbers to start with +1 (useful for americans in Israel going crazy every time you gotta call some1 from your contacts… Iykyk​:wink:)

example usage:
python contacts.py -vc -o output.vcf in1.vcf in2.vcf contacts2.db (happens to be the exact use case I used it for :rofl:)
that’s basically using all options to show full capability.

-v is for verbose which will display all the duplicates it found.
-c is for adding +1 in the output vcf
-o is self explanatory for specifying an output file.

To clarify the .db is if your phone can’t export vcf files (some have limits -wink TCL-) but you’re rooted, the actual contacts are usually stored in Android at /data/data/com.android.providers.contacts/databases/contacts2.db just copy that file and the script will do the rest.

Code:

#!/usr/bin/env python3
import sqlite3
import os
import sys
import argparse
import re
import quopri
import base64
import binascii

# --- 1. The Schema (Internal Representation) ---
class StandardContact:
    def __init__(self):
        # Name components
        self.name = {
            'family': '', 'given': '', 'middle': '', 'prefix': '', 'suffix': '', 'fn': ''
        }
        # Data fields (Using sets to auto-deduplicate within a single contact)
        self.phones = set() # (number, type)
        self.emails = set() # (email, type)
        self.addresses = set() # (street, city, region, zip, country, type)
        self.urls = set()
        self.org = {'company': '', 'title': ''}
        self.note = ""
        self.photo = None # {'data': bytes, 'type': str}

    def add_phone(self, number, type_label='VOICE'):
        # Normalize number for storage: remove surrounding whitespace
        clean_num = number.strip()
        if clean_num:
            self.phones.add((clean_num, type_label.upper()))

    def fingerprint(self):
        """
        Generates a strict deduplication signature.
        Two contacts are duplicates ONLY if all fields match EXACTLY.
        """
        sig = []

        # 1. Name (Normalized)
        n_parts = [self.name[k].strip() for k in sorted(self.name.keys()) if k != 'fn']
        sig.append(f"N:{'|'.join(n_parts)}")

        # 2. Numbers
        # We strip non-digits for the comparison signature to catch (555) vs 555
        p_sigs = []
        for num, type_ in self.phones:
            digits = re.sub(r'\D', '', num)
            # US Rule: If 11 digits and starts with 1, strip it for comparison
            if len(digits) == 11 and digits.startswith('1'):
                digits = digits[1:]
            p_sigs.append(f"{digits}")
        sig.append(f"TEL:{','.join(sorted(p_sigs))}")

        # 3. Emails
        e_sigs = [e[0].strip().lower() for e in self.emails]
        sig.append(f"EMAIL:{','.join(sorted(e_sigs))}")

        # 4. Org
        sig.append(f"ORG:{self.name.get('company','').strip()}|{self.name.get('title','').strip()}")

        return "||".join(sig)

    def has_data(self):
        # Check if contact is not empty
        return any(self.name.values()) or self.phones or self.emails or self.note or self.addresses

# --- 2. The Parsers (Extract & Transform) ---
class ContactParser:
    def parse_db(self, db_path):
        print(f"[*] Parsing DB: {db_path}")
        contacts = {} # Map ID -> StandardContact

        try:
            conn = sqlite3.connect(db_path)
            cur = conn.cursor()

            # Select relevant columns.
            # Note: The meaning of data1-data15 depends on mimetype!
            q = """
            SELECT data.raw_contact_id, mimetypes.mimetype, 
                   data.data1, data.data2, data.data3, data.data4, data.data5, 
                   data.data6, data.data7, data.data8, data.data9, data.data10, data.data15
            FROM data 
            JOIN mimetypes ON data.mimetype_id = mimetypes._id
            JOIN raw_contacts ON data.raw_contact_id = raw_contacts._id
            WHERE data.raw_contact_id IS NOT NULL
            ORDER BY data.raw_contact_id
            """
            cur.execute(q)

            for row in cur.fetchall():
                cid, mime, d1, d2, d3, d4, d5, d6, d7, d8, d9, d10, d15 = row
                if cid not in contacts:
                    contacts[cid] = StandardContact()

                c = contacts[cid]

                if mime == 'vnd.android.cursor.item/name':
                    # d2=Given, d3=Family, d4=Prefix, d5=Middle, d6=Suffix
                    c.name['given'] = d2 or ""
                    c.name['family'] = d3 or ""
                    c.name['prefix'] = d4 or ""
                    c.name['middle'] = d5 or ""
                    c.name['suffix'] = d6 or ""
                    c.name['fn'] = d1 or "" # data1 is usually formatted name

                elif mime == 'vnd.android.cursor.item/phone_v2':
                    # d1=Number, d2=Type (int)
                    if d1:
                        # Map Android type int to string
                        t_map = {1:'HOME', 2:'CELL', 3:'WORK', 12:'MAIN'}
                        t_str = t_map.get(d2, 'VOICE')
                        c.add_phone(d1, t_str)

                elif mime == 'vnd.android.cursor.item/email_v2':
                    if d1: c.emails.add((d1, 'INTERNET'))

                elif mime == 'vnd.android.cursor.item/postal-address_v2':
                    # d4=Street, d7=City, d8=Region, d9=Postcode, d10=Country
                    if d1: c.addresses.add((d1, 'HOME')) # Simplified address

                elif mime == 'vnd.android.cursor.item/postal-address_v2':
                    # data1=Formatted, data4=Street, data7=City, data8=Region, data9=Postcode, data10=Country
                    street = d4 or ""
                    city = d7 or ""
                    region = d8 or ""
                    postcode = d9 or ""
                    country = d10 or ""
                    # Fallback: If structured fields are empty but data1 exists, put data1 in street
                    if not (street or city or region or postcode or country) and d1:
                        street = d1

                    # Type mapping (1=Home, 2=Work, 3=Other)
                    addr_type = {1:'HOME', 2:'WORK', 3:'OTHER'}.get(d2, 'HOME')

                    # Store as tuple: (street, city, region, postcode, country, type)
                    c.addresses.add((street, city, region, postcode, country, addr_type))

                elif mime == 'vnd.android.cursor.item/organization':
                    c.org['company'] = d1 or ""
                    c.org['title'] = d4 or ""

                elif mime == 'vnd.android.cursor.item/note':
                    if d1: c.note = d1

                elif mime == 'vnd.android.cursor.item/website':
                    if d1: c.urls.add(d1)

                elif mime == 'vnd.android.cursor.item/photo':
                    # d15 is the BLOB
                    if d15:
                        c.photo = {'data': d15, 'type': 'JPEG'}

            conn.close()
            return list(contacts.values())

        except Exception as e:
            print(f"Error reading DB: {e}")
            return []

    def parse_vcf(self, vcf_path):
        print(f"[*] Parsing VCF: {vcf_path}")
        contacts = []
        try:
            # Handle encoding
            try:
                with open(vcf_path, 'r', encoding='utf-8') as f: lines = f.readlines()
            except:
                with open(vcf_path, 'r', encoding='latin-1') as f: lines = f.readlines()

            # Unfold lines
            unfolded = []
            for line in lines:
                if line.startswith(' '):
                    if unfolded: unfolded[-1] = unfolded[-1].strip() + line[1:]
                else:
                    unfolded.append(line.strip())

            current = None

            for line in unfolded:
                if line.startswith("BEGIN:VCARD"):
                    current = StandardContact()
                elif line.startswith("END:VCARD"):
                    if current and current.has_data():
                        contacts.append(current)
                    current = None
                elif current:
                    # Parse Line: KEY;PARAM=VAL:VALUE
                    if ':' not in line: continue
                    key_part, value = line.split(':', 1)

                    # Handle Quoted-Printable
                    if "ENCODING=QUOTED-PRINTABLE" in key_part.upper():
                        try:
                            value = quopri.decodestring(value).decode('utf-8', errors='replace')
                        except: pass

                    # Split Key and Params
                    key_split = key_part.split(';')
                    tag = key_split[0].upper()
                    params = key_split[1:]

                    if tag == 'N':
                        # Family;Given;Middle;Prefix;Suffix
                        parts = value.split(';')
                        if len(parts) >= 1: current.name['family'] = parts[0]
                        if len(parts) >= 2: current.name['given'] = parts[1]
                        if len(parts) >= 3: current.name['middle'] = parts[2]
                        if len(parts) >= 4: current.name['prefix'] = parts[3]
                        if len(parts) >= 5: current.name['suffix'] = parts[4]

                    elif tag == 'FN':
                        current.name['fn'] = value

                    elif tag == 'TEL':
                        # Try to find TYPE
                        t_type = 'VOICE'
                        for p in params:
                            if p.startswith('TYPE='): t_type = p.split('=')[1]
                            elif p in ['CELL', 'HOME', 'WORK']: t_type = p
                        current.add_phone(value, t_type)

                    elif tag == 'EMAIL':
                        current.emails.add((value, 'INTERNET'))

                    elif tag == 'ORG':
                        parts = value.split(';')
                        current.org['company'] = parts[0]
                        if len(parts) > 1: current.org['title'] = parts[1]

                    elif tag == 'NOTE':
                        current.note = value

                    elif tag == 'URL':
                        current.urls.add(value)

                    elif tag == 'ADR':
                        # ADR Format: ;;Street;City;Region;Zip;Country
                        parts = value.split(';')
                        # Pad with empty strings to avoid index errors
                        parts += [''] * (7 - len(parts))

                        street = parts[2].strip()
                        city = parts[3].strip()
                        region = parts[4].strip()
                        zip_code = parts[5].strip()
                        country = parts[6].strip()

                        # Find TYPE
                        a_type = 'HOME'
                        for p in params:
                            if p.startswith('TYPE='): a_type = p.split('=')[1].upper()
                            elif p in ['WORK', 'HOME', 'DOM', 'INTL', 'POSTAL', 'PARCEL']: a_type = p

                        current.addresses.add((street, city, region, zip_code, country, a_type))

                    elif tag == 'PHOTO':
                        # Value is base64 string
                        try:
                            # Strip whitespace
                            b64 = "".join(value.split())
                            raw = base64.b64decode(b64)
                            current.photo = {'data': raw, 'type': 'JPEG'}
                        except: pass

            return contacts
        except Exception as e:
            print(f"Error reading VCF: {e}")
            return []

# --- 3. The Writer (Load) ---
class VcfWriter:
    def write(self, contacts, out_path, add_us_code=False):
        print(f"[*] Writing {len(contacts)} contacts to {out_path}")
        try:
            with open(out_path, 'w', encoding='utf-8') as f:
                for c in contacts:
                    f.write("BEGIN:VCARD\n")
                    f.write("VERSION:3.0\n")

                    # Name (Reconstruct N and FN if missing)
                    n_str = f"{c.name['family']};{c.name['given']};{c.name['middle']};{c.name['prefix']};{c.name['suffix']}"
                    f.write(f"N:{n_str}\n")

                    fn = c.name['fn']
                    if not fn:
                        # Build FN from parts
                        parts = [c.name['prefix'], c.name['given'], c.name['middle'], c.name['family'], c.name['suffix']]
                        fn = " ".join([p for p in parts if p]).strip()
                    f.write(f"FN:{fn}\n")

                    # Phones (Apply Country Code Logic Here)
                    for num, type_ in c.phones:
                        final_num = num
                        if add_us_code:
                            digits = re.sub(r'\D', '', num)
                            # 10 digits, no leading 0, original didn't start with +
                            if len(digits) == 10 and not digits.startswith('0') and not num.strip().startswith('+'):
                                final_num = f"+1{digits}"
                        f.write(f"TEL;TYPE={type_}:{final_num}\n")

                    # Emails
                    for em, type_ in c.emails:
                        f.write(f"EMAIL;TYPE={type_}:{em}\n")

                    # Addresses
                    for street, city, region, zip_code, country, a_type in c.addresses:
                        # We escape semicolons in the values just in case
                        def esc(s): return s.replace(';', '\\;')

                        adr_val = f";;{esc(street)};{esc(city)};{esc(region)};{esc(zip_code)};{esc(country)}"
                        f.write(f"ADR;TYPE={a_type}:{adr_val}\n")

                    # Org
                    if c.org['company'] or c.org['title']:
                        f.write(f"ORG:{c.org['company']};{c.org['title']}\n")

                    # Note
                    if c.note:
                        # Escape newlines for VCF
                        clean_note = c.note.replace('\n', '\\n')
                        f.write(f"NOTE:{clean_note}\n")

                    # Urls
                    for u in c.urls:
                        f.write(f"URL:{u}\n")

                    # Photo
                    if c.photo:
                        b64 = base64.b64encode(c.photo['data']).decode('utf-8')
                        f.write(f"PHOTO;ENCODING=b;TYPE={c.photo['type']}:{b64}\n")

                    f.write("END:VCARD\n")
            return True
        except Exception as e:
            print(f"Error writing: {e}")
            return False

# --- 4. Main Controller ---
def main():
    parser = argparse.ArgumentParser(description="Contact Tool")
    parser.add_argument('inputs', nargs='+', help="Input files (.vcf or .db)")
    parser.add_argument('-o', '--output', help="Custom output filename")
    parser.add_argument('-c', '--add-us-code', action='store_true', help="Add +1 to 10-digit US numbers")
    parser.add_argument('-v', '--verbose', action='store_true', help="Print list of removed duplicates")
    args = parser.parse_args()

    # Verify inputs first
    for f in args.inputs:
        if not os.path.exists(f):
            print(f"Error: File not found: {f}")
            sys.exit(1)

    all_contacts = []
    contact_parser = ContactParser()

    # Extract
    for f in args.inputs:
        if f.lower().endswith('.db'):
            all_contacts.extend(contact_parser.parse_db(f))
        else:
            all_contacts.extend(contact_parser.parse_vcf(f))

    # Deduplicate
    unique_contacts = []
    seen_hashes = set()
    duplicates_count = 0
    removed_list = []

    print(f"[*] Processing {len(all_contacts)} raw contacts...")

    for c in all_contacts:
        h = c.fingerprint()
        if h in seen_hashes:
            duplicates_count += 1
            # Try to construct a readable name for the log
            parts = [c.name['prefix'], c.name['given'], c.name['middle'], c.name['family'], c.name['suffix']]
            full_name = " ".join([p for p in parts if p]).strip() or c.name['fn'] or "Unnamed"
            removed_list.append(full_name)
        else:
            seen_hashes.add(h)
            unique_contacts.append(c)

    print(f"[-] Removed {duplicates_count} exact duplicates.")

    # Load (Write)
    if args.output:
        out_name = args.output
    else:
        base, _ = os.path.splitext(args.inputs[0])
        out_name = f"{base}_clean.vcf"

    writer = VcfWriter()
    if writer.write(unique_contacts, out_name, args.add_us_code):
        print(f"[+] Successfully saved {len(unique_contacts)} contacts to '{out_name}'")

        # Verbose Logic: Only print if -v is passed AND there are duplicates
        if args.verbose and removed_list:
            print("\n--- Removed Duplicates ---")
            for name in removed_list:
                print(f"  x {name}")
    else:
        print("[!] Write failed.")

if __name__ == "__main__":
    main()

took me longer then i thought. (a nice few hours.) hope someone enjoys it!
(any questions just ask)

was bored so put it on github too.