twilio_message_export.py

#!/usr/bin/env python3
"""
Twilio Message Bulk Export Script

Exports the last 30 days of message records from your Twilio account
and saves them to a single JSONL (JSON Lines) file.

Usage:
    1. Set environment variables:
       export TWILIO_ACCOUNT_SID="your_account_sid"
       export TWILIO_AUTH_TOKEN="your_auth_token"
    
    2. Run the script:
       python twilio_message_export.py

Output:
    - messages_export_YYYY-MM-DD_to_YYYY-MM-DD.jsonl
"""

import os
import sys
import json
import gzip
import time
from datetime import datetime, timedelta, timezone
from io import BytesIO

# Check for required packages before importing
try:
    from twilio.rest import Client
    from twilio.base.exceptions import TwilioRestException
except ImportError:
    print("Error: twilio package not installed.")
    print("Please run: pip install twilio")
    sys.exit(1)

try:
    import requests
except ImportError:
    print("Error: requests package not installed.")
    print("Please run: pip install requests")
    sys.exit(1)


# Configuration
POLL_INTERVAL_SECONDS = 30
MAX_POLL_ATTEMPTS = 240  # 2 hours max wait time
DOWNLOAD_RETRIES = 3
DOWNLOAD_RETRY_DELAY = 5


def get_twilio_client():
    """Initialize and return Twilio client."""
    account_sid = os.environ.get("TWILIO_ACCOUNT_SID")
    auth_token = os.environ.get("TWILIO_AUTH_TOKEN")
    
    if not account_sid or not auth_token:
        print("=" * 60)
        print("ERROR: Missing Twilio credentials")
        print("=" * 60)
        print()
        print("Please set the following environment variables:")
        print()
        print("  export TWILIO_ACCOUNT_SID='ACxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx'")
        print("  export TWILIO_AUTH_TOKEN='your_auth_token'")
        print()
        print("You can find these at: https://console.twilio.com/")
        print()
        sys.exit(1)
    
    # Validate SID format
    if not account_sid.startswith("AC") or len(account_sid) != 34:
        print("ERROR: TWILIO_ACCOUNT_SID appears invalid.")
        print(f"  Got: {account_sid[:10]}...")
        print("  Expected format: ACxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx (34 characters)")
        sys.exit(1)
    
    return Client(account_sid, auth_token), account_sid, auth_token


def create_export_job(client, start_day, end_day):
    """Create a bulk export job for the specified date range."""
    print(f"Creating export job for {start_day} to {end_day}...")
    
    try:
        job = client.bulkexports.v1.exports("Messages").export_custom_jobs.create(
            start_day=start_day,
            end_day=end_day,
            friendly_name=f"Message Export {start_day} to {end_day}"
        )
        print(f"  Job created successfully: {job.job_sid}")
        return job.job_sid
    
    except TwilioRestException as e:
        print(f"ERROR: Failed to create export job")
        print(f"  Status: {e.status}")
        print(f"  Message: {e.msg}")
        if e.status == 401:
            print("  -> Check your TWILIO_ACCOUNT_SID and TWILIO_AUTH_TOKEN")
        elif e.status == 429:
            print("  -> Rate limited. You may have exceeded the 366 days/day limit.")
            print("  -> Wait until the next UTC day and try again.")
        sys.exit(1)


def get_job_status(client, job_sid):
    """Fetch and parse job status. Returns (is_complete, is_failed, status_summary)."""
    try:
        job = client.bulkexports.v1.exports.jobs(job_sid).fetch()
    except TwilioRestException as e:
        print(f"  Warning: Failed to fetch job status: {e.msg}")
        return False, False, "Unknown"
    
    details = job.details or []
    
    # Build status summary
    status_counts = {}
    for detail in details:
        if isinstance(detail, dict):
            status = detail.get("status", "Unknown")
            count = detail.get("count", 0)
            status_counts[status] = status_counts.get(status, 0) + count
    
    # Determine state
    is_failed = "Failed" in status_counts
    
    total_days = sum(status_counts.values())
    completed = status_counts.get("Completed", 0) + status_counts.get("CompletedEmptyRecords", 0)
    is_complete = total_days > 0 and completed == total_days
    
    # Format summary
    summary_parts = [f"{status}: {count}" for status, count in status_counts.items()]
    summary = ", ".join(summary_parts) if summary_parts else "Pending"
    
    # Add queue info if available
    if job.job_queue_position and job.job_queue_position != "0":
        summary += f" | Queue position: {job.job_queue_position}"
    
    return is_complete, is_failed, summary


def wait_for_job_completion(client, job_sid):
    """Poll the job status until it completes or fails."""
    print("Waiting for export job to complete...")
    print("  (This may take several minutes for large date ranges)")
    print()
    
    for attempt in range(MAX_POLL_ATTEMPTS):
        is_complete, is_failed, summary = get_job_status(client, job_sid)
        
        timestamp = datetime.now().strftime("%H:%M:%S")
        print(f"  [{timestamp}] {summary}")
        
        if is_failed:
            print()
            print("ERROR: Export job failed.")
            print("  This can happen due to temporary Twilio issues.")
            print("  Please try running the script again.")
            sys.exit(1)
        
        if is_complete:
            print()
            print("  Export job completed successfully!")
            return
        
        time.sleep(POLL_INTERVAL_SECONDS)
    
    print()
    print("ERROR: Timed out waiting for export job to complete.")
    print(f"  Job SID: {job_sid}")
    print("  The job may still be processing. You can check status at:")
    print("  https://console.twilio.com/")
    sys.exit(1)


def download_day_file(account_sid, auth_token, day_date):
    """Download and decompress a single day's export file with retries."""
    url = f"https://bulkexports.twilio.com/v1/Exports/Messages/Days/{day_date}"
    
    for attempt in range(DOWNLOAD_RETRIES):
        try:
            # Fetch the day resource to get redirect URL
            response = requests.get(url, auth=(account_sid, auth_token), timeout=30)
            
            if response.status_code == 404:
                # No data for this day
                return [], None
            
            response.raise_for_status()
            data = response.json()
            
            redirect_url = data.get("redirect_to")
            if not redirect_url:
                # No data file available
                return [], None
            
            # Download the gzipped file
            file_response = requests.get(redirect_url, timeout=120)
            file_response.raise_for_status()
            
            # Decompress and parse
            with gzip.GzipFile(fileobj=BytesIO(file_response.content)) as f:
                content = f.read().decode("utf-8")
            
            # Parse JSON lines
            records = []
            for line in content.strip().split("\n"):
                if line.strip():
                    try:
                        records.append(json.loads(line))
                    except json.JSONDecodeError:
                        continue  # Skip malformed lines
            
            return records, None
            
        except requests.exceptions.RequestException as e:
            error = str(e)
            if attempt < DOWNLOAD_RETRIES - 1:
                time.sleep(DOWNLOAD_RETRY_DELAY)
                continue
            return [], error
        
        except gzip.BadGzipFile as e:
            return [], f"Invalid gzip file: {e}"
        
        except Exception as e:
            return [], str(e)
    
    return [], "Max retries exceeded"


def download_all_days(account_sid, auth_token, start_day, end_day, output_file):
    """Download all day files and stream directly to output file."""
    print("Downloading and saving message records...")
    print()
    
    # Generate list of dates
    start = datetime.strptime(start_day, "%Y-%m-%d")
    end = datetime.strptime(end_day, "%Y-%m-%d")
    
    total_records = 0
    failed_days = []
    
    with open(output_file, "w", encoding="utf-8") as f:
        current = start
        while current <= end:
            day_str = current.strftime("%Y-%m-%d")
            
            records, error = download_day_file(account_sid, auth_token, day_str)
            
            if error:
                print(f"  {day_str}: ERROR - {error}")
                failed_days.append(day_str)
            elif records:
                print(f"  {day_str}: {len(records):,} messages")
                for record in records:
                    f.write(json.dumps(record) + "\n")
                total_records += len(records)
            else:
                print(f"  {day_str}: 0 messages")
            
            current += timedelta(days=1)
    
    return total_records, failed_days


def main():
    print()
    print("=" * 60)
    print("  Twilio Message Bulk Export")
    print("=" * 60)
    print()
    
    # Calculate date range
    # End day must be at least 2 days before current UTC day (Twilio requirement)
    today = datetime.now(timezone.utc).date()
    end_date = today - timedelta(days=2)
    start_date = end_date - timedelta(days=29)  # 30 days inclusive
    
    end_day = end_date.strftime("%Y-%m-%d")
    start_day = start_date.strftime("%Y-%m-%d")
    
    print(f"  Date range: {start_day} to {end_day} (30 days)")
    print(f"  Current UTC date: {today}")
    print()
    
    # Initialize client
    client, account_sid, auth_token = get_twilio_client()
    
    # Test authentication
    print("Verifying Twilio credentials...")
    try:
        client.api.accounts(account_sid).fetch()
        print("  Credentials verified successfully!")
        print()
    except TwilioRestException as e:
        print(f"ERROR: Authentication failed")
        print(f"  {e.msg}")
        print()
        print("Please verify your TWILIO_ACCOUNT_SID and TWILIO_AUTH_TOKEN")
        sys.exit(1)
    
    # Create export job
    job_sid = create_export_job(client, start_day, end_day)
    print()
    
    # Wait for completion
    wait_for_job_completion(client, job_sid)
    print()
    
    # Prepare output file
    output_file = f"messages_export_{start_day}_to_{end_day}.jsonl"
    
    # Download all day files (streaming to disk to avoid memory issues)
    total_records, failed_days = download_all_days(
        account_sid, auth_token, start_day, end_day, output_file
    )
    
    # Summary
    print()
    print("=" * 60)
    if failed_days:
        print("  COMPLETED WITH WARNINGS")
        print("=" * 60)
        print()
        print(f"  Total messages exported: {total_records:,}")
        print(f"  Output file: {output_file}")
        print()
        print(f"  WARNING: Failed to download {len(failed_days)} day(s):")
        for day in failed_days:
            print(f"    - {day}")
        print()
        print("  You may want to re-run the script to retry failed days.")
    elif total_records == 0:
        print("  COMPLETED - NO MESSAGES FOUND")
        print("=" * 60)
        print()
        print(f"  No messages found in the date range {start_day} to {end_day}")
        print()
        # Remove empty file
        if os.path.exists(output_file):
            os.remove(output_file)
    else:
        print("  SUCCESS")
        print("=" * 60)
        print()
        print(f"  Total messages exported: {total_records:,}")
        print(f"  Output file: {output_file}")
        print()
        
        # Show file size
        file_size = os.path.getsize(output_file)
        if file_size > 1_000_000_000:
            size_str = f"{file_size / 1_000_000_000:.2f} GB"
        elif file_size > 1_000_000:
            size_str = f"{file_size / 1_000_000:.2f} MB"
        elif file_size > 1_000:
            size_str = f"{file_size / 1_000:.2f} KB"
        else:
            size_str = f"{file_size} bytes"
        print(f"  File size: {size_str}")
    
    print()


if __name__ == "__main__":
    main()