reportxml-chats.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pandas as pd\n",
    "from bs4 import BeautifulSoup as bs\n",
    "import re\n",
    "import numpy as np\n",
    "from datetime import datetime\n",
    "from pathlib import Path"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "def getme(base, subfield):\n",
    "    \n",
    "    '''\n",
    "    returns a value from within a tag\n",
    "    '''\n",
    "    \n",
    "    try:\n",
    "        return base.find(\"field\", {\"name\": subfield}).value.text\n",
    "    except AttributeError:\n",
    "        return np.nan\n",
    "    \n",
    "def getme_date(base, subfield, t):\n",
    "    \n",
    "    '''\n",
    "    returns date or time (based on the 't' parameter)\n",
    "    '''\n",
    "    \n",
    "    try:\n",
    "        if t == \"date\":\n",
    "            return datetime.fromisoformat(base.find(\"field\", {\"name\": subfield}).value.text).date().isoformat()\n",
    "        else:\n",
    "            return datetime.fromisoformat(base.find(\"field\", {\"name\": subfield}).value.text).time().isoformat()                              \n",
    "    except AttributeError:\n",
    "        return np.nan\n",
    "    \n",
    "def getme_attribute(base, attr):\n",
    "    \n",
    "    '''\n",
    "    returns attribute of a tag\n",
    "    '''\n",
    "    \n",
    "    try:\n",
    "        return base.get(attr)\n",
    "    \n",
    "    except AttributeError:\n",
    "        return np.nan\n",
    "    \n",
    "def getme_attachment(base, subfield):\n",
    "    \n",
    "    '''\n",
    "    returns attachment name and type\n",
    "    '''\n",
    "    \n",
    "    try:\n",
    "        a = base.find(\"model\", {\"type\" : subfield})\n",
    "        return a.find(\"field\", {\"name\" : \"Filename\"}).value.text, a.find(\"field\", {\"name\" : \"ContentType\"}).value.text\n",
    "        \n",
    "    except AttributeError:\n",
    "        return np.nan,np.nan\n",
    "\n",
    "def getme_from(base):\n",
    "    \n",
    "    '''\n",
    "    returns the sender of a message\n",
    "    '''\n",
    "\n",
    "    try:\n",
    "        return base.find(\"modelField\", {\"name\": \"From\"}).find(\"field\", {\"name\" : \"Identifier\"}).value.text\n",
    "    except AttributeError:\n",
    "        #print(base)\n",
    "        return np.nan\n",
    "    \n",
    "def getme_id(base):\n",
    "    \n",
    "    '''\n",
    "    returns the Chat id\n",
    "    '''\n",
    "    \n",
    "    return base.parent.parent.get(\"id\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "def make_export(soup,TYPE):\n",
    "    \n",
    "    parent_model =  TYPE[0][0]\n",
    "    child_model = TYPE[0][1]\n",
    "    values = TYPE[1].items()\n",
    "\n",
    "    print(parent_model, child_model, values)\n",
    "    \n",
    "    reader = soup.find_all(\"modelType\", {\"type\" : parent_model})[0]\n",
    "   \n",
    "    data = []\n",
    "    \n",
    "    for el in reader.find_all(\"model\", {\"type\": child_model}):        \n",
    "        row = {}\n",
    "        for v in values:\n",
    "            \n",
    "            if v[1] == \"TimeStamp\":\n",
    "                row[v[0]] = getme_date(el, v[1], v[0])\n",
    "            elif v[0] == \"erased\":\n",
    "                row[v[0]] = getme_attribute(el, v[1])\n",
    "            elif v[0] == \"attachment\":\n",
    "                row[v[0]], row[\"type\"] = getme_attachment(el, v[1])\n",
    "            elif v[0] == \"od\":\n",
    "                row[v[0]] = getme_from(el)\n",
    "            elif v[0] == \"chat_id\":\n",
    "                row[v[0]] = getme_id(el)\n",
    "            else: \n",
    "                row[v[0]] = getme(el, v[1])\n",
    "        \n",
    "        data.append(row)\n",
    "      \n",
    "    return data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "data_types = {\n",
    "    (\"Chat\", \"InstantMessage\") :    {\n",
    "    \"content\" : \"Body\",\n",
    "    \"name\" : \"Name\",\n",
    "    \"app\" : \"SourceApplication\",\n",
    "    \"date\" : \"TimeStamp\",\n",
    "    \"time\" : \"TimeStamp\",\n",
    "    \"erased\" : \"deleted_state\",\n",
    "    \"from\" : (\"From\", \"Identifier\"),\n",
    "    \"attachment\" : \"Attachment\",\n",
    "    \"chat_id\" : \"\",\n",
    "    }}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "please be aware that parsing very big xmls (about 1 GB) this way will require at least 32 GB of RAM. lxml might be a better way to go"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Chat InstantMessage dict_items([('content', 'Body'), ('name', 'Name'), ('app', 'SourceApplication'), ('date', 'TimeStamp'), ('time', 'TimeStamp'), ('erased', 'deleted_state'), ('from', ('From', 'Identifier')), ('attachment', 'Attachment'), ('chat_id', '')])\n"
     ]
    }
   ],
   "source": [
    "xml = \"path/to/report.xml\"\n",
    "xml = Path(xml)\n",
    "target_dir = xml.parent # the csv will be saved to the same folder as the xml\n",
    "\n",
    "with open(xml) as infile:\n",
    "    soup = bs(infile.read(), \"xml\").find(\"decodedData\")\n",
    "\n",
    "DT = [d for d in data_types.items()][0]\n",
    "data = make_export(soup,DT)\n",
    "\n",
    "df = pd.DataFrame(data)\n",
    "\n",
    "df[\"app\"] = df[\"app\"].str.replace(\"\\+?[0-9]+\", \"\", regex=True).str.strip().str.replace(\":\",\"\")\n",
    "df[\"from\"] = df[\"from\"].str.replace(\"@s.whatsapp.net\",\"\").str.replace(\"+\",\"\")\n",
    "df[\"erased\"] = df[\"erased\"].str.replace(\"Intact\", \"0\").str.replace(\"Deleted\", \"1\")\n",
    "df[\"from\"] = df[\"from\"].fillna(\"owner\") # if message was sent through viber, the 'from' part is empty. This is a hack and might backfire in unforeseen ways.\n",
    "\n",
    "df.to_csv(Path(target_dir, \"all_chats.csv\"),  index=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.7.1"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}