parse-fauci-emails.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "# pdftotext -layout -r 300 leopold-nih-foia-anthony-fauci-emails.pdf \n",
    "import copy\n",
    "import importlib\n",
    "import json\n",
    "import numpy as np\n",
    "import re\n",
    "from datetime import datetime\n",
    "from string import ascii_lowercase\n",
    "from collections import Counter"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "def line_metadata_re(name): return re.compile(\"[\\s]*\" + \"[\\s]*\".join(list(name)) + \"[\\s]*:.*\")\n",
    "from_re = line_metadata_re(\"from\")\n",
    "time_re = line_metadata_re(\"sent\")\n",
    "to_re   = line_metadata_re(\"to\")\n",
    "cc_re   = line_metadata_re(\"cc\")\n",
    "subj_re = line_metadata_re(\"subject\")\n",
    "\n",
    "def is_from(line): return from_re.match(line.lower())\n",
    "def is_time(line): return time_re.match(line.lower())\n",
    "def is_to(line):   return to_re.match(line.lower())\n",
    "def is_cc(line):   return cc_re.match(line.lower())\n",
    "def is_subj(line): return subj_re.match(line.lower())\n",
    "\n",
    "fauci_re = re.compile(\".*\" + \"[\\s]*\".join(list(\"fauci\")) + \".*\")\n",
    "def contains_fauci(line): return fauci_re.match(line.lower())\n",
    "\n",
    "def has_redacted_sender(line):\n",
    "    ll = line.lower()\n",
    "    ind = ll.find(':')\n",
    "    if ind == -1: return False\n",
    "    ll = ll[ind + 1:]\n",
    "    if not ('b' in ll and '6' in ll): return False\n",
    "    if not ('(' in ll and ')' in ll): return False\n",
    "    return sum([c in ascii_lowercase for c in ll]) <= 5"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "class Email:\n",
    "    def __init__(self):\n",
    "        self.sender = \"\"\n",
    "        self.time = \"\"\n",
    "        self.recipients = \"\"\n",
    "        self.cc = \"\"\n",
    "        self.subject = \"\"\n",
    "        self.text = \"\"\n",
    "        self.nih = \"\"\n",
    "        self.redacted_sender = False\n",
    "        \n",
    "    def __str__(self):\n",
    "        return f'from: {self.sender}\\nsent: {self.time}\\nto: {self.recipients}\\ncc: {self.cc}\\nsubject: {self.subject}\\n\\n{self.text}'\n",
    "\n",
    "    def has_sender(self): return self.sender != \"\"\n",
    "    def has_recipient(self): return self.recipients != \"\" and self.recipients != []\n",
    "    def has_cc(self): return self.cc != \"\" and self.cc != []\n",
    "    def has_timestamp(self): return self.time != \"\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "def clean_metadata_lines(lines): \n",
    "    if len(lines) == 0:  return ''\n",
    "    first = lines[0][(lines[0].find(':') + 1):].strip(' \\\"\\'')\n",
    "    if len(lines) == 1: return first\n",
    "    return first + '\\n'.join([line.strip(' \\\"\\'') for line in lines[1:]])\n",
    "\n",
    "def find_next_from(chunk, start_ind):    \n",
    "    \"\"\" \n",
    "    Find index of first line after start_ind that is a from line. \n",
    "    Returns length of chunk if there is no such line.\n",
    "    \"\"\"\n",
    "    for ind in range(start_ind + 1, len(chunk)):\n",
    "        if is_from(chunk[ind]): return ind\n",
    "    return len(chunk)\n",
    "\n",
    "def handle_special_case_1(chunk):\n",
    "    \"\"\"\n",
    "    Manage the following two lines \n",
    "    # From: \n",
    "    # From: Thomas Quinn\n",
    "    \"\"\"\n",
    "    line_to_remove = -1\n",
    "    for j in range(len(chunk) - 1):\n",
    "        if is_from(chunk[j]) and is_from(chunk[j + 1]):\n",
    "            if chunk[j + 1].lower().find(\"quinn\") != -1:\n",
    "                line_to_remove = j\n",
    "                break\n",
    "    if line_to_remove != -1:\n",
    "        chunk.pop(line_to_remove)\n",
    "        \n",
    "def find_line(lines, start_ind, check_func):\n",
    "    for i in range(start_ind, len(lines)):\n",
    "        if check_func(lines[i]): return i\n",
    "    return -1\n",
    "\n",
    "def process_chunk(chunk):\n",
    "    \"\"\" \n",
    "    Extract structured email data from single thread. The first email \n",
    "    is from Fauci. Subsequent ones are extraced from the thread.\n",
    "    \"\"\"\n",
    "    if not is_from(chunk[0]):\n",
    "        print(chunk[0])\n",
    "    \n",
    "    assert(is_from(chunk[0]))\n",
    "    \n",
    "    handle_special_case_1(chunk)\n",
    "    \n",
    "    emails = []\n",
    "    start_ind = 0\n",
    "    end_ind = find_next_from(chunk, start_ind)\n",
    "    \n",
    "    while start_ind < len(chunk):\n",
    "        c_chunk = copy.deepcopy(chunk[start_ind:end_ind])\n",
    "        email = Email()\n",
    "        \n",
    "        from_ind = find_line(c_chunk, 0, is_from)\n",
    "        time_ind = find_line(c_chunk, 0, is_time)\n",
    "        to_ind   = find_line(c_chunk, 0, is_to)\n",
    "        cc_ind   = find_line(c_chunk, 0, is_cc)\n",
    "        subj_ind = find_line(c_chunk, 0, is_subj)\n",
    "        \n",
    "        email.sender     = clean_metadata_lines(c_chunk[from_ind:max(from_ind+1, time_ind)])\n",
    "        email.time       = clean_metadata_lines(c_chunk[time_ind:max(time_ind+1, to_ind)])\n",
    "        email.recipients = clean_metadata_lines(c_chunk[to_ind:max(to_ind+1, cc_ind)])\n",
    "        email.cc         = clean_metadata_lines(c_chunk[cc_ind:max(cc_ind+1, subj_ind)])\n",
    "        email.subject    = clean_metadata_lines(c_chunk[subj_ind:subj_ind+1])\n",
    "\n",
    "        # Text concsits of remaining lines\n",
    "        email.text = '\\n'.join(c_chunk[subj_ind+1:end_ind]).strip()\n",
    "        \n",
    "        # Update positions\n",
    "        start_ind = end_ind\n",
    "        end_ind = find_next_from(chunk, start_ind)        \n",
    "        \n",
    "        emails.append(email)\n",
    "        \n",
    "    if has_redacted_sender(chunk[0]):\n",
    "        emails[0].redacted_sender = True        \n",
    "\n",
    "    return emails"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Extract text in chunks corresponding to each email thread\n",
    "all_chunks = []\n",
    "\n",
    "with open(\"leopold-nih-foia-anthony-fauci-emails.txt\", \"r\") as f:\n",
    "    last_line_new_page = True  # first line is the start of new page\n",
    "    curr_chunk = []\n",
    "    last_nih_num = 0\n",
    "\n",
    "    for (j, line) in enumerate(f):\n",
    "        # Check for start of new page\n",
    "        possible_fauci = contains_fauci(line) or has_redacted_sender(line)\n",
    "        if j == 0 or (line[0] == '\\f' and is_from(line) and possible_fauci):\n",
    "            if len(curr_chunk) > 0:\n",
    "                all_chunks.append(curr_chunk)\n",
    "                curr_chunk = []\n",
    "                \n",
    "        line = line.strip()\n",
    "        if len(line) > 0:\n",
    "            curr_chunk.append(line)\n",
    "            \n",
    "    if len(curr_chunk) > 0:\n",
    "        all_chunks.append(curr_chunk)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Extract emails from chunks\n",
    "emails = []\n",
    "for (i, chunk) in enumerate(all_chunks):\n",
    "    chain = [e for e in process_chunk(chunk) if e.has_recipient()]\n",
    "    if len(chain) > 0:\n",
    "        emails.append(chain)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "import parse_timestamps\n",
    "importlib.reload(parse_timestamps)\n",
    "from parse_timestamps import *"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [],
   "source": [
    "import parse_names\n",
    "importlib.reload(parse_names)\n",
    "from parse_names import *"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [],
   "source": [
    "import parse_text\n",
    "importlib.reload(parse_text)\n",
    "from parse_text import *"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Map timestamps and names\n",
    "nih_names = Counter()\n",
    "hhs_names = Counter()\n",
    "cdc_names = Counter()\n",
    "fda_names = Counter()\n",
    "os_names = Counter()\n",
    "eop_names = Counter()\n",
    "\n",
    "def check_membership(name, p_name):\n",
    "    if name == '': return\n",
    "    l_name = name.lower()\n",
    "    def is_match(s): return l_name.find(s) != -1\n",
    "    \n",
    "    if p_name == \"lenihan, keagan\": \n",
    "        fda_names[p_name] += 1  # don't match nih\n",
    "        return\n",
    "    if p_name == \"aspa deputies\":\n",
    "        hhs_names[p_name] += 1\n",
    "        return\n",
    "        \n",
    "    if p_name in [\"shoc\", \"farrar, jeremey\", \"schwartlander, bernhard\",\n",
    "                  \"banks, lynn\", \"kanarek, morgan\", \"lapook, jon\",\n",
    "                  \"elias, chris\", \"pfeifer, hazel\", \"smith, steven\",\n",
    "                  \"hatchett, richard\", \"lancman, christine\",\n",
    "                  \"morrison, stephen\", \"oneail, shawn\", \"callahan, michael\",\n",
    "                  \"mcnamara, tracey\", \"caneva, duane\", \"verma, seema\"]: return\n",
    "    \n",
    "    if is_match(\"(nih\") or p_name.find(\"niaid\") != -1 or p_name.find(\"nih\") != -1: nih_names[p_name] += 1\n",
    "    if is_match(\"(cdc\") or p_name.find(\"cdc\") != -1: cdc_names[p_name] += 1        \n",
    "    if is_match(\"(fda\") or p_name.find(\"fda\") != -1: fda_names[p_name] += 1           \n",
    "    if is_match(\"(hhs\") or p_name.find(\"hhs\") != -1: hhs_names[p_name] += 1\n",
    "    if is_match(\"(os\"):  os_names[p_name] += 1\n",
    "    if is_match(\"eop/\"): eop_names[p_name] += 1\n",
    "        \n",
    "parsed_emails = []\n",
    "for (i, chain) in enumerate(emails):    \n",
    "    parsed_chain = []\n",
    "    for email in chain:\n",
    "        p_email = copy.copy(email)\n",
    "        p_email.time = parse_timestamp(email.time)\n",
    "        \n",
    "        s_name = parse_name(str(email.sender))\n",
    "        if email.redacted_sender:\n",
    "            s_name = \"fauci, anthony\"\n",
    "        check_membership(str(email.sender), s_name)\n",
    "        p_email.sender = s_name\n",
    "            \n",
    "        r_names = []\n",
    "        for name in email.recipients.split(';'): \n",
    "            r_name = parse_name(str(name))\n",
    "            if type(r_name) == type([]):\n",
    "                r_names += r_name\n",
    "            else:\n",
    "                r_names.append(r_name)\n",
    "                check_membership(str(name), r_name)\n",
    "        p_email.recipients = list(filter(lambda name: name != '', r_names))\n",
    "                \n",
    "        cc_names = []\n",
    "        if email.has_cc():\n",
    "            for name in email.cc.split(';'):\n",
    "                cc_name = parse_name(str(name))\n",
    "                if type(cc_name) == type([]):\n",
    "                    cc_names += cc_name\n",
    "                else:\n",
    "                    cc_names.append(cc_name)\n",
    "                    check_membership(str(name), cc_name)\n",
    "        p_email.cc = list(filter(lambda name: name != '', cc_names))\n",
    "        \n",
    "        # Special case of duplicate Ashley Parkers\n",
    "        if p_email.sender == \"abutaleb, yasmeen\":\n",
    "            if p_email.recipients[0] == \"conrad, patricia\":\n",
    "                if p_email.subject.find(\"Washington Post\") != -1:\n",
    "                    p_email.cc = [\"parker, ashley (wapo)\"]\n",
    "\n",
    "        parsed_chain.append(p_email)\n",
    "        \n",
    "    parsed_emails.append(parsed_chain)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Map all names to integers \n",
    "name_map = dict()\n",
    "def name_id(name):\n",
    "    if name not in name_map:\n",
    "        n = len(name_map)\n",
    "        name_map[name] = n\n",
    "    return name_map[name]\n",
    "\n",
    "mapped_emails = []\n",
    "for (i, chain) in enumerate(parsed_emails):    \n",
    "    mapped_chain = []\n",
    "    for email in chain:\n",
    "        # Skip if we can't parse timestamp\n",
    "        if not email.has_timestamp(): continue\n",
    "        # Skip if we didn't find a sender\n",
    "        if not email.has_sender(): continue\n",
    "        # Skip if we didn't find a recipient\n",
    "        if not email.has_recipient(): continue\n",
    "            \n",
    "        mapped_email = dict()\n",
    "        mapped_email[\"sender\"] = name_id(email.sender)\n",
    "        mapped_email[\"recipients\"] = list(set([name_id(recip) for recip in email.recipients]))\n",
    "        mapped_email[\"cc\"] = list(set([name_id(cc_name) for cc_name in email.cc]))\n",
    "        mapped_email[\"time\"] = email.time.isoformat()  # read with date.fromisoformat()\n",
    "        mapped_email[\"subject\"] = parse_text(email.subject)\n",
    "        mapped_email[\"body\"] = parse_text(email.text)\n",
    "        mapped_chain.append(mapped_email)\n",
    "        \n",
    "        if len(mapped_email[\"recipients\"]) == 0:\n",
    "            print(list(email.recipients))\n",
    "        \n",
    "    if len(mapped_chain) > 0:\n",
    "        mapped_emails.append(mapped_chain)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Write out the digest\n",
    "names = []\n",
    "clusters = []\n",
    "for node_id, name in sorted([(name_map[k], k) for k in name_map]):\n",
    "    names.append(name)\n",
    "    counts = [nih_names[name], hhs_names[name], cdc_names[name],\n",
    "              fda_names[name], os_names[name], eop_names[name]]\n",
    "    if max(counts) == 0:\n",
    "        # \"other\" cluster\n",
    "        clusters.append(len(counts) + 1)\n",
    "    else:\n",
    "        clusters.append(int(np.argmax(counts)) + 1)\n",
    "    \n",
    "cluster_names = [\"NIH\", \"HHS\", \"CDC\", \"FDA\", \"OS\", \"EOP\", \"other\"]\n",
    "        \n",
    "data = {\"names\": names,\n",
    "        \"clusters\": clusters,\n",
    "        \"cluster_names\": cluster_names,\n",
    "        \"emails\": mapped_emails\n",
    "       }\n",
    "with open('manual-fix/fauci-email-initial.json', 'w') as f:\n",
    "    json.dump(data, f)"
   ]
  },
  {
   "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.9.5"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}