SMS_XML_to_TXT_CSV_JSON.ipynb

{
  "nbformat": 4,
  "nbformat_minor": 0,
  "metadata": {
    "colab": {
      "provenance": [],
      "authorship_tag": "ABX9TyP0urp02ZkAGzg/mAi71LwZ",
      "include_colab_link": true
    },
    "kernelspec": {
      "name": "python3",
      "display_name": "Python 3"
    },
    "language_info": {
      "name": "python"
    }
  },
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "view-in-github",
        "colab_type": "text"
      },
      "source": [
        "<a href=\"https://colab.research.google.com/github/isaac-mackey/mind-uploading/blob/main/Template_SMS_XML_to_TXT_CSV_JSON.ipynb\" target=\"_parent\"><img src=\"https://colab.research.google.com/assets/colab-badge.svg\" alt=\"Open In Colab\"/></a>"
      ]
    },
    {
      "cell_type": "markdown",
      "source": [
        "**How To Use**\n",
        "\n",
        "1) Get export xml file from \"SMS Backup & Restore\" Android app\n",
        "\n",
        "2) Put the xml file in Google Drive\n",
        "\n",
        "3) Run this notebook using that xml file\n",
        "\n",
        "4) Export to TXT, CSV, or JSON file"
      ],
      "metadata": {
        "id": "KnjoVJxDM7AT"
      }
    },
    {
      "cell_type": "markdown",
      "source": [
        "## XML helper library"
      ],
      "metadata": {
        "id": "fvKyICuQtduU"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "!pip install lxml"
      ],
      "metadata": {
        "id": "5ONT-NGCfdAr"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "source": [
        "## XML helper functions"
      ],
      "metadata": {
        "id": "mPZut2CutioW"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "from lxml import etree\n",
        "\n",
        "def raw_extract_sms_from_xml(file_name):\n",
        "    sms_messages = []\n",
        "    sent_messages = []\n",
        "    contacts = {}\n",
        "\n",
        "    # Create an iterative parser\n",
        "    context = etree.iterparse(file_name, events=(\"end\",), tag=\"sms\")\n",
        "\n",
        "    for event, elem in context:\n",
        "        # Extract SMS data\n",
        "        sms_data = {\n",
        "            'address': elem.get('address'),\n",
        "            'date': elem.get('date'),\n",
        "            'type': elem.get('type'),\n",
        "            'body': elem.get('body'),\n",
        "            'contact_name': elem.get('contact_name')\n",
        "        }\n",
        "        sms_messages.append(sms_data)\n",
        "\n",
        "        # Only sent messages\n",
        "        # if sms_data['type'] == '2':\n",
        "        #     sent_messages.append(sms_data)\n",
        "\n",
        "        # Update contacts dictionary\n",
        "        contact_name = sms_data['contact_name']\n",
        "        if contact_name not in contacts:\n",
        "            contacts[contact_name] = {\n",
        "                'address': sms_data['address'],\n",
        "                'messages': [],\n",
        "                'contact_name': sms_data['contact_name']\n",
        "            }\n",
        "        contacts[contact_name]['messages'].append(sms_data)\n",
        "\n",
        "        # Clear the element to free up memory\n",
        "        elem.clear()\n",
        "\n",
        "        # Also eliminate now-empty references from the root node to <sms>\n",
        "        while elem.getprevious() is not None:\n",
        "            del elem.getparent()[0]\n",
        "\n",
        "    # Close the context to free resources\n",
        "    del context\n",
        "\n",
        "    return sms_messages, sent_messages, contacts"
      ],
      "metadata": {
        "id": "hjxxtjEWAXO6"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "from datetime import datetime\n",
        "\n",
        "def UNIX_timestamp_to_formatted_datetime(date):\n",
        "    unix_timestamp = int(date) / 1000  # Convert to integer and then to seconds\n",
        "    date_time_obj = datetime.utcfromtimestamp(unix_timestamp)\n",
        "    # Format the datetime object as a string\n",
        "    formatted_date = date_time_obj.strftime('%Y-%m-%d %H:%M:%S')\n",
        "    return formatted_date\n",
        "\n",
        "def write_output_to_text_file(output_file_path, sms_messages, contacts):\n",
        "    # Writing the output to a text file\n",
        "    with open(output_file_path, 'w') as file:\n",
        "        print(\"output_file_path\" + output_file_path)\n",
        "        file.write('Total SMS messages: ' + str(len(sms_messages)) + '\\n')\n",
        "        file.write('Contacts found: ' + str(len(contacts)) + '\\n')\n",
        "\n",
        "        earliest = sms_messages[0]['date']\n",
        "        latest = sms_messages[0]['date']\n",
        "\n",
        "        for message in combined_sms_messages:\n",
        "            message_date = message['date']\n",
        "            if message['date'] < earliest:\n",
        "                earliest = message_date\n",
        "            if message['date'] > latest:\n",
        "                latest = message_date\n",
        "\n",
        "        file.write('Earliest message: ' + UNIX_timestamp_to_formatted_datetime(earliest) + '\\n')\n",
        "        file.write('Latest message: ' + UNIX_timestamp_to_formatted_datetime(latest) + '\\n')\n",
        "\n",
        "        file.write('\\n')\n",
        "\n",
        "        for contact in contacts.values():\n",
        "            if contact['contact_name'] == '(Unknown)':\n",
        "                continue\n",
        "            file.write('Conversation with '+contact['contact_name'] + ' at ' + contact['address'] + '\\n')\n",
        "            contact_name = contact['contact_name']\n",
        "            other_name = \"Isaac Mackey\"\n",
        "            # Determine the length of the longer string\n",
        "            max_length = max(len(contact_name), len(other_name))\n",
        "            padded_contact_name = contact_name.ljust(max_length)\n",
        "            padded_other_name = other_name.ljust(max_length)\n",
        "\n",
        "            file.write(str(len(contact['messages'])) + \" messages\" + '\\n')\n",
        "            date_sorted_messages = sorted(contact['messages'], key=lambda x: x['date'])\n",
        "\n",
        "            earliest = date_sorted_messages[0]['date']\n",
        "            latest = date_sorted_messages[-1]['date']\n",
        "\n",
        "            file.write('Earliest message: ' + UNIX_timestamp_to_formatted_datetime(earliest) + '\\n')\n",
        "            file.write('Latest message: ' + UNIX_timestamp_to_formatted_datetime(latest) + '\\n')\n",
        "\n",
        "            for message in date_sorted_messages:\n",
        "\n",
        "                # Convert UNIX timestamp to datetime object\n",
        "                unix_timestamp = int(message['date']) / 1000  # Convert to integer and then to seconds\n",
        "                date_time_obj = datetime.utcfromtimestamp(unix_timestamp)\n",
        "                # Format the datetime object as a string\n",
        "                formatted_date = date_time_obj.strftime('%Y-%m-%d %H:%M:%S')\n",
        "\n",
        "                if message['type'] == '1':\n",
        "                    file.write(formatted_date + \": \" + padded_contact_name + \": \"+message['body'] + '\\n')\n",
        "                if message['type'] == '2':\n",
        "                    file.write(formatted_date + \": \" + padded_other_name + \": \" + message['body'] + '\\n')\n",
        "            file.write('\\n')\n",
        "\n",
        "    print('File closed')"
      ],
      "metadata": {
        "id": "ohKYboxq3qK9"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "source": [
        "## write_prompt_completion_pairs_to_csv"
      ],
      "metadata": {
        "id": "Xrj7KkGflMTM"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "from datetime import datetime\n",
        "import csv\n",
        "\n",
        "# Function to check if a message is a question\n",
        "def is_question(message):\n",
        "    question_indicators = [\"who\", \"what\", \"where\", \"when\", \"why\", \"how\", \"is\", \"are\", \"was\", \"were\", \"do\", \"does\", \"did\", \"can\", \"could\", \"should\", \"would\", \"will\", \"whose\", \"which\"]\n",
        "\n",
        "    message_lower = str(message.lower())\n",
        "    if any(message_lower.startswith(sub) for sub in question_indicators):\n",
        "        return True\n",
        "    return message.endswith('?')\n",
        "\n",
        "def write_prompt_completion_pairs_to_csv(output_file_path, sms_messages, contacts):\n",
        "    if not 'csv' in output_file_path:\n",
        "        print(\"csv not found in\",output_file_path)\n",
        "        return\n",
        "\n",
        "    # Writing the output to a text file\n",
        "    with open(output_file_path, 'w', newline='') as csvfile:\n",
        "        print(\"writing to output_file_path:\" + output_file_path)\n",
        "        writer = csv.writer(csvfile)\n",
        "\n",
        "        # Write the header\n",
        "        writer.writerow(['prompt','completion'])\n",
        "\n",
        "        for contact in contacts.values():\n",
        "            if contact['contact_name'] == '(Unknown)':\n",
        "                continue\n",
        "            contact_name = contact['contact_name']\n",
        "            other_name = \"Isaac Mackey\"\n",
        "            # Determine the length of the longer string\n",
        "            max_length = max(len(contact_name), len(other_name))\n",
        "            padded_contact_name = contact_name.ljust(max_length)\n",
        "            padded_other_name = other_name.ljust(max_length)\n",
        "\n",
        "            date_sorted_messages = sorted(contact['messages'], key=lambda x: x['date'])\n",
        "\n",
        "            # Iterate over messages\n",
        "            for i in range(len(date_sorted_messages) - 1):\n",
        "                current_message = date_sorted_messages[i]\n",
        "                next_message = date_sorted_messages[i + 1]\n",
        "\n",
        "                # Check if the current message is a question\n",
        "                if is_question(current_message['body']):\n",
        "                    if current_message['type'] == '1' and next_message['type'] == '2':\n",
        "                        # Write the current message (prompt) and the next message (completion)\n",
        "                        writer.writerow([current_message['body'], next_message['body']])\n",
        "\n",
        "    print('File closed')"
      ],
      "metadata": {
        "id": "FLURPeDa3vSB"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "source": [
        "## write_messages_to_role_system_user_format_json"
      ],
      "metadata": {
        "id": "yNuzsk4IlKFz"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "from datetime import datetime\n",
        "import json\n",
        "\n",
        "def write_messages_to_role_system_user_format_json(output_file_path, sms_messages, contacts):\n",
        "    if not 'json' in output_file_path:\n",
        "        print(\"json not in\",output_file_path)\n",
        "        return\n",
        "\n",
        "    # Writing the output to a text file\n",
        "    with open(output_file_path, 'w') as file:\n",
        "        print(\"writing to output_file_path:\", output_file_path)\n",
        "\n",
        "        for contact in contacts.values():\n",
        "            if contact['contact_name'] == '(Unknown)':\n",
        "                continue\n",
        "            contact_name = contact['contact_name']\n",
        "            other_name = \"Isaac Mackey\"\n",
        "\n",
        "            date_sorted_messages = sorted(contact['messages'], key=lambda x: x['date'])\n",
        "\n",
        "            conversation = []\n",
        "\n",
        "            system_message = (\"You are a computer science PhD graduate in the Marine Corps\"\n",
        "                              \" who wants to achieve immortality through mind-uploading.\"\n",
        "                              \" Be polite and formal. Do not apologize. Use correct grammar and avoid logic fallacies.\")\n",
        "            conversation.append({\"role\": \"system\", \"content\": system_message})\n",
        "\n",
        "            assistant_present = False\n",
        "            # Iterate over messages\n",
        "            for message in date_sorted_messages:\n",
        "                if message['type'] == '1':\n",
        "                    role = \"user\"\n",
        "                    content = message['body']\n",
        "                    conversation.append({\"role\": role, \"content\":content})\n",
        "\n",
        "                if message['type'] == '2':\n",
        "                    role = \"assistant\"\n",
        "                    content = message['body']\n",
        "                    conversation.append({\"role\": role, \"content\":content})\n",
        "                    assistant_present = True\n",
        "\n",
        "            # conversation.append({\"role\": \"separator\", \"content\": \"<END_OF_CONVERSATION>\"})\n",
        "            if assistant_present:\n",
        "                json_record = json.dumps({'messages':conversation})\n",
        "                file.write(json_record + '\\n')\n",
        "\n",
        "    print('File closed')"
      ],
      "metadata": {
        "id": "VlGPvhK1gjLa"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "source": [
        "## Mount Google Drive"
      ],
      "metadata": {
        "id": "PxmkLkTnBU1X"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "from google.colab import drive\n",
        "\n",
        "# It would be really great to automate this step... selenium?\n",
        "drive.mount('/content/drive')\n",
        "\n",
        "import os\n",
        "for x in os.listdir('/content/drive/MyDrive'):\n",
        "  if 'sms-' in x:\n",
        "    print(x)\n",
        "    # pass"
      ],
      "metadata": {
        "id": "WHhq-dxwKuNS"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "source": [
        "## Load SMS XML files"
      ],
      "metadata": {
        "id": "jqv2h40iXrTk"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "from datetime import datetime\n",
        "# Format the current date and time in a human-readable format\n",
        "print(datetime.now().strftime(\"%Y-%m-%d %H:%M:%S\"))\n",
        "\n",
        "# Replace with the latest filename\n",
        "input_file_name = 'sms-20210904143609.xml'\n",
        "input_file_path = '/content/drive/My Drive/'+input_file_name\n",
        "\n",
        "print('input_file_name:',input_file_name)\n",
        "print('input_file_path:',input_file_path)\n",
        "\n",
        "import os\n",
        "if not input_file_name in os.listdir('/content/drive/MyDrive'):\n",
        "    print(\"Input file doesn't exist\")"
      ],
      "metadata": {
        "id": "W_jC3WiS35tE"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "source": [
        "## Parse SMS XML into Python dictionary"
      ],
      "metadata": {
        "id": "dJnkUvuev7cg"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "from datetime import datetime\n",
        "# Format the current date and time in a human-readable format\n",
        "print(datetime.now().strftime(\"%Y-%m-%d %H:%M:%S\"))\n",
        "\n",
        "# Extract contents of input file\n",
        "sms_messages, sent_messages, contacts = raw_extract_sms_from_xml(input_file_path)\n",
        "\n",
        "print('input_file_name:',input_file_name)\n",
        "print('input_file_path:',input_file_path)\n",
        "# Print the lengths of the lists\n",
        "print('Total SMS messages:', len(sms_messages))\n",
        "# print('Sent SMS messages:', len(sent_messages))\n",
        "print('Contacts found:', len(contacts))"
      ],
      "metadata": {
        "id": "1XbnW5CLK8WL"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "combined_sms_messages = sms_messages"
      ],
      "metadata": {
        "id": "rI0NWAVfSD4a"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "len(combined_sms_messages)"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "_yY9t1zuBwNX",
        "outputId": "9a8e23e0-b53f-4719-ec63-ce9f87691cfe"
      },
      "execution_count": null,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "text/plain": [
              "31016"
            ]
          },
          "metadata": {},
          "execution_count": 8
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "for c in contacts.keys():\n",
        "    print(c)\n",
        "    print(contacts[c]['contact_name'])"
      ],
      "metadata": {
        "id": "bZcmBhpAPJIr"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "for c in sorted(contacts.keys()):\n",
        "    print(c)"
      ],
      "metadata": {
        "id": "4h2lmA-Z14nA"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        " # Examine contacts extracted from input file\n",
        "pairs = []\n",
        "for contact in contacts.values():\n",
        "      pairs.append((contact['contact_name'],contact['address']))\n",
        "pairs = sorted(pairs, key=lambda x: x[0])\n",
        "for pair in pairs:\n",
        "      # print(pair)\n",
        "      pass"
      ],
      "metadata": {
        "id": "7UhaCrLvGCn3"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "source": [
        "## OUTPUT CONTACT-SORTED CHATLOGS TO TEXT FILE"
      ],
      "metadata": {
        "id": "k-yJPEa6R2UL"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "from datetime import datetime\n",
        "# Format the current date and time in a human-readable format\n",
        "print(datetime.now().strftime(\"%Y-%m-%d %H:%M:%S\"))\n",
        "\n",
        "print('input_file_name:',input_file_name)\n",
        "print('input_file_path:',input_file_path)\n",
        "# Print the lengths of the lists\n",
        "print('Total SMS messages:', len(combined_sms_messages))\n",
        "# print('Sent SMS messages:', len(sent_messages))\n",
        "print('Contacts found:', len(combined_contacts))\n",
        "\n",
        "# Path for the output text file\n",
        "input_file_name = 'sms-combined.xml'\n",
        "output_file_name = input_file_name+'-dated-messages.txt'\n",
        "\n",
        "output_file_path = '/content/drive/My Drive/'+output_file_name\n",
        "\n",
        "print('output_file_name:',output_file_name)\n",
        "print('output_file_path:',output_file_path)\n",
        "\n",
        "if output_file_name in os.listdir('/content/drive/MyDrive'):\n",
        "    print(\"Output file already exists\")\n",
        "else:\n",
        "  # Write contents into output file\n",
        "  write_output_to_text_file(output_file_path, combined_sms_messages, combined_contacts)"
      ],
      "metadata": {
        "id": "0pc1ptLwCv5J"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "source": [
        "## OUTPUT PROMPT/COMPLETIONS TO CSV FILE"
      ],
      "metadata": {
        "id": "aH4vV8wQR6g6"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "from datetime import datetime\n",
        "# Format the current date and time in a human-readable format\n",
        "print(datetime.now().strftime(\"%Y-%m-%d %H:%M:%S\"))\n",
        "\n",
        "# Path for the output text file\n",
        "output_file_name = input_file_name+'-prompt-completion-1.csv'\n",
        "output_file_path = '/content/drive/My Drive/'+output_file_name\n",
        "\n",
        "print('output_file_name:',output_file_name)\n",
        "print('output_file_path:',output_file_path)\n",
        "\n",
        "if output_file_name in os.listdir('/content/drive/MyDrive'):\n",
        "    print(\"Output file already exists\")\n",
        "\n",
        "# Write contents into output file\n",
        "write_prompt_completion_pairs_to_csv(output_file_path, sms_messages, contacts)"
      ],
      "metadata": {
        "id": "c1uiJfJRRv0U"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "source": [
        "## OUTPUT ROLE/SYSTEM/USER TO JSON FILE"
      ],
      "metadata": {
        "id": "GX7xGAwqUCOQ"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "from datetime import datetime\n",
        "# Format the current date and time in a human-readable format\n",
        "print(datetime.now().strftime(\"%Y-%m-%d %H:%M:%S\"))\n",
        "\n",
        "# Path for the output text file\n",
        "input_file_name = 'sms-combined.xml'\n",
        "output_file_name = input_file_name+'-role-system-user-7.json'\n",
        "output_file_path = '/content/drive/My Drive/'+output_file_name\n",
        "\n",
        "print('output_file_name:',output_file_name)\n",
        "print('output_file_path:',output_file_path)\n",
        "\n",
        "if output_file_name in os.listdir('/content/drive/MyDrive'):\n",
        "    print(\"Output file already exists\")\n",
        "\n",
        "# Write contents into output file\n",
        "write_messages_to_role_system_user_format_json(output_file_path, sms_messages, contacts)"
      ],
      "metadata": {
        "id": "VJj1Lc9la-i7"
      },
      "execution_count": null,
      "outputs": []
    }
  ]
}