Script to perform monthly exports of iMessage to a remote SSH server using rsync.

imessage-export.py

import subprocess
import shutil
from datetime import datetime, timedelta
import netifaces
import os

SUCCESS_FILE = os.path.expanduser('~/.imessage_export/last_success')
REMOTE_BASE_PATH = "/path/to/messages"
SSH_HOST = "user@192.168.1.2"

def run_imessage_exporter(start_date, end_date, export_path):
    command = [
        "/opt/homebrew/bin/imessage-exporter",
        "--copy-method", "efficient",
        "--format", "html",
        "--export-path", export_path,
        "--start-date", start_date.strftime("%Y-%m-%d"),
        "--end-date", end_date.strftime("%Y-%m-%d")
    ]
    subprocess.run(command, check=True)

def is_on_local_network():
    for interface in netifaces.interfaces():
        addrs = netifaces.ifaddresses(interface)
        if netifaces.AF_INET in addrs:
            for addr in addrs[netifaces.AF_INET]:
                ip = addr['addr']
                if ip.startswith('192.168.1.'):
                    return True
    return False

def create_remote_directory(remote_path):
    command = [
        "ssh",
        SSH_HOST,
        f"mkdir -p {remote_path}"
    ]
    subprocess.run(command, check=True)

def rsync_to_server(source_path, destination):
    command = [
        "rsync",
        "-avz",
        "--delete",
        source_path,
        destination
    ]
    subprocess.run(command, check=True)

def send_notification(title, message):
    os.system(f"""osascript -e 'display notification "{message}" with title "{title}"'""")

def get_last_exported_month():
    command = [
        "ssh",
        SSH_HOST,
        f"ls -1 {REMOTE_BASE_PATH} | sort -n | tail -n 1"
    ]
    result = subprocess.run(command, capture_output=True, text=True, check=True)
    if result.returncode == 0 and result.stdout.strip():
        last_year = result.stdout.strip()
        command = [
            "ssh",
            SSH_HOST,
            f"ls -1 {REMOTE_BASE_PATH}/{last_year} | sort -n | tail -n 1"
        ]
        result = subprocess.run(command, capture_output=True, text=True, check=True)
        if result.returncode == 0 and result.stdout.strip():
            last_month = result.stdout.strip()
            return datetime.strptime(f"{last_year}-{last_month}-01", "%Y-%m-%d")
    return datetime(2014, 1, 1)  # Default start date if no exports found

def export_and_sync_month(year, month):
    start_date = datetime(year, month, 1)
    if month == 12:
        end_date = datetime(year + 1, 1, 1) - timedelta(days=1)
    else:
        end_date = datetime(year, month + 1, 1) - timedelta(days=1)
    
    base_export_path = "/tmp/iMessageBackup"
    final_destination = f"{REMOTE_BASE_PATH}/{year}/{month:02d}"

    try:
        print(f"Exporting messages for {year}-{month:02d}")
        shutil.rmtree(base_export_path, ignore_errors=True)
        os.mkdir(base_export_path)
        run_imessage_exporter(start_date, end_date, base_export_path)
        
        create_remote_directory(final_destination)
        rsync_destination = f"{SSH_HOST}:{final_destination}"
        print(rsync_destination)
        rsync_to_server(f"{base_export_path}/", rsync_destination)
        
        shutil.rmtree(base_export_path)
        print(f"Export and sync completed for {year}-{month:02d}")
        return True
    except Exception as e:
        print(f"Error occurred while processing {year}-{month:02d}: {str(e)}")
        return False

def is_sync_past_due():
    if not os.path.exists(SUCCESS_FILE):
        return True

    with open(SUCCESS_FILE, 'r') as f:
        last_sync_str = f.read().strip()

    last_sync_date = datetime.strptime(last_sync_str, "%Y-%m-%d")
    latest_date = get_latest_date()

    if last_sync_date >= latest_date:
        return False

    if last_sync_date < latest_date:
        if datetime.now() > last_sync_date + timedelta(days=45):
            return True
        else:
            return False

def get_latest_date():
    return (datetime.now().replace(day=1).replace(hour=0, minute=0, second=0, microsecond=0) - timedelta(days=1)).replace(day=1)

def main():
    if not is_on_local_network():
        print("Not on local network. Skipping export and sync.")
        if is_sync_past_due():
            send_notification("iMessage Export Failed", "Failed to export or sync messages for the past 3 weeks. Reason: Not on local network.")
        return

    export_date = get_last_exported_month()
        
    while export_date < get_latest_date():
        if export_date.month == 12:
            export_date = export_date.replace(year=export_date.year + 1, month=1)
        else:
            export_date = export_date.replace(month=export_date.month + 1)

        if export_and_sync_month(export_date.year, export_date.month):
            with open(SUCCESS_FILE, "w") as f:
                f.write(export_date.strftime("%Y-%m-%d"))
        else:
            send_notification("iMessage Export Failed", f"Failed to export or sync messages for {export_date.year}-{export_date.month:02d}")
            break

if __name__ == "__main__":
    main()