import pandas as pd
import re

"""
Explanation: WordCount

Record Reader
Input Split
Map
Combiner
Shuffle and Sort
Reduce
Output Format
End of explanation
"""


def recordReader(path_file):
    text = None
    try:
        text = open(path_file, encoding='utf-8').read()
        return text 
    except (TypeError, ZeroDivisionError, ValueError):
        print("Error: "+e)
        return text

text = recordReader("./data/mercado_ti.txt")
text2 = recordReader("./data/data_science.txt")

"""
Explanation: 1. Record Reader
Descreva uma função que realize a leitura de dados "mercado_ti.txt" e "data_science.txt"
End of explanation
"""


def input_split(text):
    text = re.sub(u'[^a-zA-ZáéíóúÁÉÍÓÚâêîôÂÊÎÔãõÃÕçÇ: ]', '', text)
    return pd.DataFrame(text.split(" "))

df_file1 = input_split(text)
df_file2 = input_split(text2)

"""
Explanation: 2. Input split
defina uma função split que recebe os dados e quebre em partes menores para usar como entrada na função mapper.
Exemplo: string.split()
End of explanation
"""


def mapper(df):
    df.rename(columns={0:"key"}, inplace=True)
    df.at[:,'value'] = [1]
    return df

df1 = mapper(df_file1)
df2 = mapper(df_file2)

df1.head()

df2.head()

"""
Explanation: 3. Map
O usuário define uma função intermediária que recebe os dados e mapeia  pares de chave e valor (definido pelo usuário)
End of explanation
"""


def combiner(df_list):
    return pd.concat(df_list)

df_combiner = combiner([df_file1, df_file2])

df_combiner.head()

"""
Explanation: 4. Combiner
Faz a combinação de "n" dataframes para obter um único dataframe que será usado na função Shufle
End of explanation
"""


def shufle(df):
    df.sort_values(by=['key'], inplace=True)
    df = df[df.key != ""]
    return df

df_combiner = shufle(df_combiner)

df_combiner.head()

"""
Explanation: 5. Shuffle
Descreva uma função Shufle para ordenar, filtrar e agrupar dados como pares de chave e valor que serão a entrada na função reducer
End of explanation
"""


def reducer(df):
    return df_combiner.groupby('key').count()

df_reducer = reducer(df_combiner)
df_reducer.head(10)

"""
Explanation: 6. Reduce
Use uma função que agregue dados pelo valor (v) de acordo com a chave (k) para enviar pares de chave-valor para saída
End of explanation
"""


df_reducer.sort_values(by=['value'], ascending=False, inplace=True)

df_reducer.head(20)

"""
Explanation: 7.Output Format
Forneça os pares de chave-valor finais ordenados por maior valor
End of explanation
"""


