raw to text data

import mailbox
import os
import re
import datetime
import pandas as pd
from pathlib import Path
from dotenv import load_dotenv
import sys

load_dotenv("../../.env")
sys.path.append("../..")
from src import utils  # noqa

# collect paths of either the entire dataset, or only the most recent month if running in automation

BASE_PATH = os.getenv("LOCAL_DATA_PATH", "../../data")
path = Path(BASE_PATH).joinpath("raw/fedora-devel-list")
mboxes = [x.name for x in list(path.glob("*.mbox"))]

if os.getenv("RUN_IN_AUTOMATION"):
    LAST_MONTH_DATE = datetime.datetime.now().replace(
        day=1
    ) - datetime.timedelta(days=1)
    y = LAST_MONTH_DATE.year
    m = LAST_MONTH_DATE.month
    mboxes = [x.name for x in list(path.glob(f"*{y}-{m}.mbox"))]

# Define a function to convert mbox data into row,column format for analysis
# using pandas


def mbox_to_text(mbox):

    csv = []
    for msg in mbox:
        msg_id = msg["Message-ID"]
        date = msg["Date"]
        body = []
        for m in msg.get_payload():
            body.append(m.get_payload())

        csv.append((msg_id, date, body))
    df = pd.DataFrame(csv, columns=["Message-ID", "Date", "Body"])
    return df

# Clean the data before storing it
def strip_thread(text):
    text = "".join(text)
    text = text.replace("\r", "")
    lines = text.split("\n")
    lines = [line for line in lines if len(line) > 0]
    lines = [line for line in lines if line[0] != ">"]
    lines = [line for line in lines if line[:3] != "Re:"]
    lines = [line for line in lines if line[:7] != "Subject"]
    lines = [line for line in lines if line[:5] != "From:"]
    lines = [line for line in lines if line[:5] != "Date:"]
    lines = [line for line in lines if "BEGIN PGP SIGNED MESSAGE" not in line]
    lines = [line for line in lines if line[:5] != "Hash:"]
    lines = [line for line in lines if line[:10] != "Version: G"]
    lines = [line for line in lines if "wrote:" not in line]
    lines = [line for line in lines if "wrote :" not in line]
    lines = [line for line in lines if "writes:" not in line]
    lines = [line for line in lines if line[:7] != "Am Mit,"]
    lines = [line for line in lines if line[:7] != "Am Don,"]
    lines = [line for line in lines if line[:7] != "Am Mon,"]
    lines = [line for line in lines if line[:7] != "Quoting"]
    lines = [line for line in lines if line[:10] != "Em Quinta,"]
    lines = [line for line in lines if "said:" not in line]
    lines = [
        line
        for line in lines
        # matching lines that are "On Weekday, Day Month, Year" or "On Weekday, Month Day, Year"
        # ie, Monday, 6 March, 2020 or Monday, March 6, 2020
        if re.match(
            ".*n ((Sun|Mon|Tue(s)?|Wed(nes)?|Thu(rs)?|Fri|Sat(ur)?)(day)?)(,)? (.*)?"
            "(Jan(uary)?|Feb(ruary)?|Mar(ch)?|Apr(il)?|May|Jun(e)?|Jul(y)?|Aug(ust)?|Sep((t)?ember)?|Oct(ober)?|Nov(ember)?|Dec(ember)?)( .*)? 20..*"  # noqa
            "(at [\d]{2}:[\d]{2} AM|PM)?",  # noqa
            line,
        )
        is None
    ]
    lines = [
        line
        for line in lines
        # matching lines that ar "On Day Month Year"
        # ie, 6 March 2020
        if re.match(
            (
                ".*n .(.)? (Jan(uary)?|Feb(ruary)?|Mar(ch)?|Apr(il)?|May|Jun(e)?|Jul(y)?|Aug(ust)?|Sep((t)?ember)?|Oct(ober)?|Nov(ember)?|Dec(ember)?) 20..*"  # noqa
            ),
            line,
        )
        is None
    ]
    lines = [
        line
        for line in lines
        # matching lines that are "On Weekday, YYYY-MM-DD at HH:MM"
        # ie, Monday, 2020-03-06 at 12:01
        if re.match(
            ".*n ((Sun|Mon|Tue(s)?|Wed(nes)?|Thu(rs)?|Fri|Sat(ur)?)(day)?), 20[\d]{2}-[\d]{2}-[\d]{2} at.*",  # noqa
            line,
        )
        is None
    ]
    lines = [line for line in lines if line[-6:] != "said: "]
    lines = [line for line in lines if line[-8:] != "babbled:"]
    lines = [line for line in lines if line[-7:] != "wrot=e:"]
    lines = [line for line in lines if line[-8:] != "A9crit :"]
    lines = [line for line in lines if line[0] != "|"]
    return "\n".join(lines)


# format for CSV, clean special characters, and remove extranous emails
def pandas_clean(emails):
    emails["Body"].replace(
        to_replace=[
            r"\n",
            "\n",
        ],
        value=" ",
        regex=True,
        inplace=True,
    )
    emails["Body"].replace(
        to_replace=[r"\'", "'", ">", "<", "= ", "-", r"http\S+"],
        value="",
        regex=True,
        inplace=True,
    )
    emails["Body"].replace(
        to_replace=[r"\\\s+", r"\\s+", "="], value="", regex=True, inplace=True
    )
    emails["Body"].replace(
        to_replace=["   ", "  "], value=" ", regex=True, inplace=True
    )
    emails["Body"].replace(
        to_replace=["_", "3D"], value="", regex=True, inplace=True
    )
    emails["Body"].replace(
        to_replace=["   ", "  "], value=" ", regex=True, inplace=True
    )
    emails["Body"].replace(
        to_replace=["   ", "  "], value=" ", regex=True, inplace=True
    )
    emails["Body"] = emails["Body"].apply(
        lambda x: x.strip().replace(r"\n", "")
    )

    emails.drop(emails.index[emails["Body"] == ""], inplace=True)
    emails.drop(emails.index[emails["Body"] == " "], inplace=True)
    emails.dropna(subset=["Body"], inplace=True)

    emails = emails.reset_index()
    emails.drop("index", axis=1, inplace=True)
    return emails

# Ensure datset location exists
dataset_base_path = Path(f"{BASE_PATH}/interim/text")
dataset_base_path.mkdir(parents=True, exist_ok=True)

# [WIP]avoid edge cases
subset_mboxes = mboxes[:5]

# Register all created dataset slices for later upload
new_files = []

# Save each dataset into its own monthly csv
for mbox in subset_mboxes:
    output_path = dataset_base_path.joinpath(f"{mbox}.csv")
    monthly_mbox = mailbox.mbox(path.joinpath(mbox))
    df = mbox_to_text(monthly_mbox)
    df["Body"] = df["Body"].apply(strip_thread)
    df = pandas_clean(df)
    df.to_csv(output_path)
    new_files.append(output_path)
    print(f"{output_path} saved")

sample_mbox = mailbox.mbox(path.joinpath("fedora-devel-2018-6.mbox"))
sample_mbox_df = mbox_to_text(sample_mbox)
sample_mbox_df.head()

sample_cleaned_mbox = pd.read_csv(
    "../../data/interim/text/fedora-devel-2018-6.mbox.csv"
)
sample_cleaned_mbox.head()

# Push all the files to ceph

if os.getenv("RUN_IN_AUTOMATION"):
    utils.upload_files(
        (f.as_posix(), f"interim/text/{Path(f).name}") for f in new_files
    )