2. Extract Conversations.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Import the required libraries"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import psycopg2 as psy\n",
    "import pandas as pd\n",
    "import json\n",
    "import os\n",
    "import time"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Establish connection with the PostgreSQL database"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "database_host = \"localhost\"\n",
    "database_name = \"dbl_data_challenge\"\n",
    "database_user = \"admin\"\n",
    "database_pass = \"vZtbqKNXGz27cQCH\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "con = psy.connect(\n",
    "    host = database_host,\n",
    "    database = database_name,\n",
    "    user = database_user,\n",
    "    password = database_pass\n",
    ")\n",
    "cur = con.cursor()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### This cell gets all the tweet ids and the tweet ids of the tweets they're replying to, creates a pair dictionary, which is subsequently used to retrieve the start of a conversation. The conversation id is the id of the tweet which forms the root of the conversation (the starting tweet). When iterating over the pairs, the tweets are added to the converstaion id key in the dictionary which they belong to"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "cur.execute(\"\"\"\n",
    "    SELECT id_str, in_reply_to_status_id_str FROM tweets\n",
    "\"\"\")\n",
    "\n",
    "data = cur.fetchall()\n",
    "\n",
    "convo_pairs = {}\n",
    "counter = 0\n",
    "\n",
    "for row in data: \n",
    "    convo_pairs[row[0]] = row[1]\n",
    "\n",
    "total_count = len(convo_pairs)\n",
    "\n",
    "conversations = []\n",
    "\n",
    "start = time.time()\n",
    "\n",
    "for twt_id in convo_pairs: \n",
    "    counter += 1\n",
    "    found_last = False\n",
    "    convo_id = None\n",
    "    current = twt_id\n",
    "    current_fallback = None\n",
    "    is_snippet = False\n",
    "    while not found_last: \n",
    "        try: \n",
    "            if convo_pairs[current] != None: \n",
    "                current_fallback = current\n",
    "                current = convo_pairs[current]\n",
    "            else: \n",
    "                found_last = True\n",
    "        except KeyError: \n",
    "            convo_pairs[current_fallback] = None\n",
    "            convo_id = current_fallback\n",
    "            is_snippet = True\n",
    "            found_last = True\n",
    "    convo_id = current\n",
    "    conversations.append((twt_id, convo_id, is_snippet))\n",
    "    if counter % 100000 == 0: \n",
    "        print(\"[Pairing] Currently at tweet: \"+twt_id+\" (\"+str(counter)+\"/\"+str(total_count)+\") (\"+str(round((time.time()-start), 2))+\"s)\")\n",
    "\n",
    "counter = 0\n",
    "\n",
    "start = time.time()\n",
    "\n",
    "cur.execute(\"\"\"\n",
    "    DROP TABLE IF EXISTS conversations\n",
    "\"\"\")\n",
    "\n",
    "cur.execute(\"\"\"\n",
    "    CREATE TABLE conversations (tweet_id VARCHAR(50), conversation_id VARCHAR(50), is_snippet BOOLEAN)\n",
    "\"\"\")\n",
    "\n",
    "con.commit()\n",
    "\n",
    "for twt_tuple in conversations: \n",
    "    counter += 1\n",
    "    cur.execute(\"\"\"\n",
    "        INSERT INTO conversations (tweet_id, conversation_id, is_snippet) VALUES (%s, %s, %s)\n",
    "    \"\"\", (twt_tuple[0], twt_tuple[1], twt_tuple[2]))\n",
    "    if counter % 100000 == 0: \n",
    "        print(\"[SQL Insert] Currently at tweet: \"+twt_tuple[0]+\" (\"+str(counter)+\"/\"+str(total_count)+\") (\"+str(round((time.time()-start), 2))+\"s)\")\n",
    "        con.commit()\n",
    "\n",
    "print(\"[SQL Insert] Completed update of data in \"+str(round((time.time()-start), 2))+\"s\")\n",
    "\n",
    "con.commit()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### A function which is a able to return all the tweets from a certain conversation id, presented in a dataframe"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def GetConversation(convo_id): \n",
    "    columns = [\"created_at\",\"timestamp_ms\",\"userid_str\",\"id_str\",\"full_text\",\"in_reply_to_status_id_str\",\"in_reply_to_user_id_str\",\"quoted_status_id_str\",\"quote_count\",\"reply_count\",\"retweet_count\",\"favorite_count\",\"hashtags\",\"user_mentions\",\"conversation_id\",\"thread_id\"]\n",
    "    cur.execute(\"\"\"\n",
    "        SELECT tweet.* FROM tweet, conversations\n",
    "        WHERE tweet.id_str = conversations.tweet_id AND conversations.conversation_id = %(convo_id)s\n",
    "    \"\"\", {\"convo_id\": convo_id})\n",
    "    df_tweets = pd.DataFrame(data=cur.fetchall(), columns=columns)\n",
    "    return df_tweets\n",
    "\n",
    "GetConversation(\"1149272217349427203\")"
   ]
  },
  {
   "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.8.5"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}