# ---
# jupyter:
#   jupytext:
#     text_representation:
#       extension: .py
#       format_name: light
#       format_version: '1.5'
#       jupytext_version: 1.15.2
#   kernelspec:
#     display_name: Python 2
#     language: python
#     name: python2
# ---

# # Coursework 2: Data Processing
#
# ## Task 1
# This coursework will assess your understanding of using NoSQL to store and retrieve data.  You will perform operations on data from the Enron email dataset in a MongoDB database, and write a report detailing the suitability of different types of databases for data science applications.  You will be required to run code to answer the given questions in the Jupyter notebook provided, and write a report describing alternative approaches to using MongoDB.
#
# Download the JSON version of the Enron data (using the “Download as zip” to download the data file from http://edshare.soton.ac.uk/19548/, the file is about 380MB) and import into a collection called messages in a database called enron.  You do not need to set up any authentication.  In the Jupyter notebook provided, perform the following tasks, using the Python PyMongo library.
#
# Answers should be efficient in terms of speed.  Answers which are less efficient will not get full marks.

import pymongo
from pymongo import MongoClient
from datetime import datetime
from pprint import pprint


# ### 1)
# Write a function which returns a MongoDB connection object to the "messages" collection. [4 points] 

# + nbgrader={"grade": false, "grade_id": "get_collection", "locked": false, "schema_version": 1, "solution": true}
def get_collection():
    """
    Connects to the server, and returns a collection object
    of the `messages` collection in the `enron` database
    """
    # YOUR CODE HERE

    return None


# -

# ### 2)
#
# Write a function which returns the amount of emails in the messages collection in total. [4 points] 

# + nbgrader={"grade": false, "grade_id": "get_amount_of_messages", "locked": false, "schema_version": 1, "solution": true}
def get_amount_of_messages(collection):
    """
    :param collection A PyMongo collection object
    :return the amount of documents in the collection
    """    
    # YOUR CODE HERE
    
    pass
    


# -

# ### 3) 
#
# Write a function which returns each person who was BCCed on an email.  Include each person only once, and display only their name according to the X-To header. [4 points] 
#
#

# + nbgrader={"grade": false, "grade_id": "get_bcced_people", "locked": false, "schema_version": 1, "solution": true}
def get_bcced_people(collection):
    """
    :param collection A PyMongo collection object
    :return the names of the people who have received an email by BCC
    """    
    # YOUR CODE HERE

    pass
    


# -

# ### 4)
#
# Write a function with parameter subject, which gets all emails in a thread with that parameter, and orders them by date (ascending). “An email thread is an email message that includes a running list of all the succeeding replies starting with the original email.”, check for detail descriptions at https://www.techopedia.com/definition/1503/email-thread [4 points]

# + nbgrader={"grade": false, "grade_id": "get_emails_in_thread", "locked": false, "schema_version": 1, "solution": true}
def get_emails_in_thread(collection, subject):
    """
    :param collection A PyMongo collection object
    :return All emails in the thread with that subject
    """    
    # YOUR CODE HERE    
    
    pass


# -

# ### 5)
#
# Write a function which returns the percentage of emails sent on a weekend (i.e., Saturday and Sunday) as a `float` between 0 and 1. [6 points]

# + nbgrader={"grade": false, "grade_id": "get_percentage_sent_on_weekend", "locked": false, "schema_version": 1, "solution": true}
def get_percentage_sent_on_weekend(collection):
    """
    :param collection A PyMongo collection object
    :return A float between 0 and 1
    """    
    # YOUR CODE HERE
    
    pass


# -

# ### 6)
#
# Write a function with parameter limit. The function should return for each email account: the number of emails sent, the number of emails received, and the total number of emails (sent and received). Use the following format: [{"contact": "michael.simmons@enron.com", "from": 42, "to": 92, "total": 134}] and the information contained in the To, From, and Cc headers. Sort the output in descending order by the total number of emails. Use the parameter limit to specify the number of results to be returned. If limit is null, the function should return all results. If limit is higher than null, the function should return the number of results specified as limit. limit cannot take negative values. [10 points]

# + nbgrader={"grade": false, "grade_id": "get_emails_between_contacts", "locked": false, "schema_version": 1, "solution": true}
def get_emails_between_contacts(collection, limit):
    """
    Shows the communications between contacts
    Sort by the descending order of total emails using the To, From, and Cc headers.
    :param `collection` A PyMongo collection object    
    :param `limit` An integer specifying the amount to display, or
    if null will display all outputs
    :return A list of objects of the form:
    [{
        'contact': <<Another email address>>
        'from': 
        'to': 
        'total': 
    },{.....}]
    """    
    # YOUR CODE HERE
    
    pass


# -

# ### 7)
# Write a function to find out the number of senders who were also direct receivers. Direct receiver means the email is sent to the person directly, not via cc or bcc. [4 points]

def get_from_to_people(collection):
    """
    :param collection A PyMongo collection object
    :return the NUMBER of the people who have sent emails and received emails as direct receivers.
    """    
    # YOUR CODE HERE

    pass


# ### 8)
# Write a function with parameters start_date and end_date, which returns the number of email messages that have been sent between those specified dates, including start_date and end_date [4 points] 

def get_emails_between_dates(collection, start_date, end_date):
    """
    :param collection A PyMongo collection object
    :return All emails between the specified start_date and end_date
    """    
    # YOUR CODE HERE    
    
    pass


# ## Task 2
# This task will assess your ability to use the Hadoop Streaming API and MapReduce to process data. For each of the questions below, you are expected to write two python scripts, one for the Map phase and one for the Reduce phase. You are also expected to provide the correct parameters to the `hadoop` command to run the MapReduce process. Write down your answers in the specified cells below.
#
# To get started, you need to download and unzip the YouTube dataset (available at http://edshare.soton.ac.uk/19547/) onto the machine where you have Hadoop installed (this should be the virtual machine provided).
#
# To help you, `%%writefile` has been added to the top of the cells, automatically writing them to "mapper.py" and "reducer.py" respectively when the cells are run.

# ### 1) 
# Using Youtube01-Psy.csv, find the hourly interval in which most spam was sent. The output should be in the form of a single key-value pair, where the value is a datetime at the start of the hour with the highest number of spam comments. [9 points]

from datetime import datetime
import csv
import sys

# +
# DEBUGGING SCRIPT FOR MAPPER

dates = [
'2013-11-07T06:20:48',
'2013-11-07T12:37:15',
'2014-01-19T04:27:18',
'2014-01-19T08:55:53',
'2014-01-19T20:31:10'
]

spam_class = [1,1,0,0,1]

for x in range(len(dates)):
    if spam_class[x] == 1:
        date = dates[x].strip()
        date_as_date = datetime.strptime(date, '%Y-%m-%dT%H:%M:%S')
        day = date_as_date.date().day
        month = date_as_date.date().month
        year = date_as_date.date().year
        hour = date_as_date.hour

        print (str(day) + '|' + str(month) + '|' + str(year) + '|' + str(hour) + '\t' + '1')


# +
test = [1,2,3]

test = test[1:]

# +
# %%writefile mapper.py
# #!/usr/bin/env python
# MAPPER

import csv
import sys
from datetime import datetime


lines = sys.stdin.readlines()


csvreader = csv.reader(lines)

dates = []

spam_class = []

input_for_reducer = []

counter = 0
for row in csvreader:
    if counter > 0:
        dates.append(row[2])
        spam_class.append(row[4])
    counter += 1
    

if (len(dates) != len(spam_class)):
    print ('Unequal number of entries in Date and Class columns... Aborting...')
    sys.exit()


for x in range(len(dates)):
    if spam_class[x] == '1':
        date = dates[x].strip()
        date_as_date = datetime.strptime(date, '%Y-%m-%dT%H:%M:%S')
        day = date_as_date.date().day
        month = date_as_date.date().month
        year = date_as_date.date().year
        hour = date_as_date.hour
    
        print (str(day) + '|' + str(month) + '|' + str(year) + '|' + str(hour) + '\t' + '1')

        
# -

# If the dates in our input file are arranged such that the dates (at an hourly interval) occur in groups, we can perform the Reduce operation in linear time.
#
# It is observed in the data that the column 'Date' is indeed sorted in ascending order
#
# So the dates (at an hourly interval) are in groups
#
#
#

# +
# DEBUGGING SCRIPT FOR REDUCER

input_pairs = [
    '7|11|2013|6	1',
    '7|11|2013|6	1',
    '7|11|2013|12	1',
    '7|11|2013|12	1',
    '7|11|2013|12	1',
    '19|1|2014|20	1'
]

dates_list = []

date_count_dict = dict()

final_dict = {
    'hour_with_most_spam': None,
    'value_of_max_spam_count': 0
}


for input_pair in input_pairs:
    input_list = input_pair.split('\t', 1)
    if (len(input_list) != 2):
        continue
    
    dates_list.append(input_list[0])
        
dates_list

for date in dates_list:
    if date in date_count_dict.keys():
        date_count_dict[date] += 1
    else:
        date_count_dict[date] = 1
        
date_count_dict_sorted = sorted(date_count_dict.items(), key=lambda date_count_value: date_count_value[1], 
                                reverse=True)

final_dict['hour_with_most_spam'] = date_count_dict_sorted[0][0]
final_dict['value_of_max_spam_count'] = date_count_dict_sorted[0][1]

final_dict

# +
# %%writefile reducer.py
# #!/usr/bin/env python
# REDUCER

import sys
from datetime import datetime

input_pairs = sys.stdin.readlines()

dates_list = []

date_count_dict = dict()

final_dict = {
    'hour_with_most_spam': None,
    'value_of_max_spam_count': 0
}


for input_pair in input_pairs:
    input_list = input_pair.split('\t', 1)
    if (len(input_list) != 2):
        continue
    
    dates_list.append(input_list[0])
        
dates_list

for date in dates_list:
    if date in date_count_dict.keys():
        date_count_dict[date] += 1
    else:
        date_count_dict[date] = 1
        
date_count_dict_sorted = sorted(date_count_dict.items(), key=lambda date_count_value: date_count_value[1], 
                                reverse=True)

final_dict['hour_with_most_spam'] = date_count_dict_sorted[0][0]
final_dict['value_of_max_spam_count'] = date_count_dict_sorted[0][1]


for key, value in final_dict.items():
    print (key + "\t" + str(value))

# +
myList = [1,1,1,2,2,2,2,3,3]

max_count = 1
max_elem = myList[0]
curr_count = 1

for x in range(1, len(myList)):
    if (myList[x] == myList[x-1]):
        # same elem, inc counter 
        curr_count += 1
    else:
        # diff elem
        if curr_count > max_count:
            max_count = curr_count
            max_elem = myList[x - 1]
        curr_count = 1
# last element check
if curr_count > max_count:
    max_count = curr_count
    max_elem = myList[x - 1]

    
print (max_elem)

# + language="bash"
# cat ./Youtube01-Psy.csv | ./mapper.py  | ./reducer.py

# + language="bash"
#
# # Clear output
# rm -rf output1
#
# # Make sure hadoop is in standalone mode
# hadoop-standalone-mode.sh
#
# # Main pipeline command
# hadoop jar $HADOOP_HOME/share/hadoop/tools/lib/hadoop-streaming-*.jar \
# -files mapper.py,reducer.py \
# -input Youtube01-Psy.csv \
# -mapper ./mapper.py \
# -reducer ./reducer.py \
# -output output1

# + language="bash"
# #Hadoop command to run the map reduce.
#
# hadoop jar $HADOOP_HOME/share/hadoop/tools/lib/hadoop-streaming-*.jar \
# -files mapper.py,reducer.py   \
# -input Youtube01-Psy.csv   \
# -mapper  ./mapper.py \
# -reducer ./reducer.py  \
# -output output

# +
#Expected key-value output format:
#hour_with_most_spam	"2013-11-10T10:00:00"

#Additional key-value pairs are acceptable, as long as the hour_with_most_spam pair is correct.
# -

# ### 2) 
# Find all comments associated with a username (the AUTHOR field). Return a JSON array of all comments associated with that username. (This should use the data from all 5 data files: Psy, KatyPerry, LMFAO, Eminem, Shakira) [11 points]

# +
# %%writefile mapper1.py
# #!/usr/bin/env python
#Answer for mapper.py

# importing the libraries
import csv
import sys


def mapper_function(required_username):

    # function that accepts an username as input 

    # counter keeps track of number of rows left, so that we can skip the first row (headers)
    counter = 0
    for row in csvreader:
        if counter > 0:
            usernames.append(row[1])
            comments.append(row[3])
        counter += 1

    if (len(usernames) != len(comments)):
        print ('Unequal number of entries in Author and Content... Aborting...')
        sys.exit()

    # pass the required username and the comments for that username to reducer stage
    for x in range(len(usernames)):
        if required_username == usernames[x]:
            print (str(usernames[x]) + '\t' + str(comments[x]))
            


lines = sys.stdin.readlines()

# read from csv

csvreader = csv.reader(lines)

usernames = []

comments = []


# get username from command line argument

required_username = str(sys.argv[1])

mapper_function(required_username)





# +
# %%writefile reducer1.py
# #!/usr/bin/env python
#Answer for reducer.py

import sys


final_dict = {
    'username': None,
    'comments': []
}

# get input from mapper job

input_pairs = sys.stdin.readlines()



for input_pair in input_pairs:
    # split the tab separated input (username\tcomment)
    input_list = input_pair.split('\t', 1)
    if (len(input_list) != 2):
        continue
        
    # append each comment
    final_dict['comments'].append(input_list[1])
    # set the username if it is not set
    if final_dict['username'] is None:
        final_dict['username'] = input_list[0]
    

# print out the output in desired form: username\t[..comments..]
print (final_dict.values()[0] + '\t' + str(final_dict.values()[1]))
    

# + language="bash"
# cat ./test_files/Youtube02-KatyPerry.csv ./test_files/Youtube01-Psy.csv \
# ./test_files/Youtube03-LMFAO.csv ./test_files/Youtube04-Eminem.csv ./test_files/Youtube05-Shakira.csv | ./mapper1.py 'Mini' | ./reducer1.py

# + language="bash"
#
# # Clear output
# rm -rf output2
#
# # Make sure hadoop is in standalone mode
# hadoop-standalone-mode.sh
#
# # Main pipeline command
# hadoop jar $HADOOP_HOME/share/hadoop/tools/lib/hadoop-streaming-*.jar \
# -files mapper1.py,reducer1.py \
# -input ./test_files/Youtube01-Psy.csv ./test_files/Youtube02-KatyPerry.csv ./test_files/Youtube03-LMFAO.csv \
# -mapper 'mapper1.py Mini' -file ./mapper1.py  \
# -reducer ./reducer1.py \
# -output output2

# +
#Expected key-value output format:
#John Smith	["Comment 1", "Comment 2", "Comment 3", "etc."]
#Jane Doe	["Comment 1", "Comment 2", "Comment 3", "etc."]
# -


cel(writer,"audience_by_group_both",index=False)
df_output_audience_tireonly.to_excel(writer,"audience_by_group_tireonly",index=False)

df_both_by_trans.to_excel(writer,"trans_detail",index=False)
df_output_by_store_period.to_excel(writer,"store_by_each_period",index=False)

writer.save()
# -

df_both_by_trans.shape


