import sys
import argparse
import numpy as np
from pyspark import SparkContext
import csv

def toLowerCase(s):
    """ Convert a sting to lowercase. E.g., 'BaNaNa' becomes 'banana'
    """
    return s.lower()

def stripNonAlpha(s):
    """ Remove non alphabetic characters. E.g. 'B:a,n+a1n$a' becomes 'Banana' """
    return ''.join([c for c in s if c.isalpha()])


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description = 'Text Analysis through TFIDF computation',formatter_class=argparse.ArgumentDefaultsHelpFormatter)
    parser.add_argument('mode', help='Mode of operation',choices=['TF','IDF','TFIDF','SIM','TOP'])
    parser.add_argument('input', help='Input file or list of files.')
    parser.add_argument('output', help='File in which output is stored')
    parser.add_argument('--master',default="local[20]",help="Spark Master")
    parser.add_argument('--idfvalues',type=str,default="idf", help='File/directory containing IDF values. Used in TFIDF mode to compute TFIDF')
    parser.add_argument('--other',type=str,help = 'Score to which input score is to be compared. Used in SIM mode')
    args = parser.parse_args()

    sc = SparkContext(args.master, 'Text Analysis')
    print "Hello there"

    if args.mode=='TF':
        # Read text file at args.input, compute TF of each term,
        # and store result in file args.output. All terms are first converted to
        # lowercase, and have non alphabetic characters removed
        # (i.e., 'Ba,Na:Na.123' and 'banana' count as the same term). Empty strings, i.e., ""
        # are also removed
        # pass
        # rdd = sc.parallelize()
        rdd1 = sc.textFile(args.input)
        rdd1.flatMap(lambda s: s.split()).map(toLowerCase).map(stripNonAlpha)\
           .map(lambda word: (word, 1)).reduceByKey(lambda x, y: x + y)\
           .sortBy(lambda (x,y):y, ascending=False).saveAsTextFile(args.output)



    if args.mode=='TOP':
        # Read file at args.input, comprizing strings representing pairs of the form (TERM,VAL),
        # where TERM is a string and VAL is a numeric value. Find the pairs with the top 20 values,
        # and store result in args.output
        rdd2 = sc.textFile(args.input)

        rdd2.flatMap(lambda s: s.split()).map(toLowerCase).map(stripNonAlpha)\
           .map(lambda word: (word, 1)).reduceByKey(lambda x, y: x + y)\
           .takeOrdered(20,lambda (x,y):-y)#.saveAsTextFile(args.output)
           #.sortBy(lambda (x,y):y, ascending=False).take(20)
        #rdd2.saveAsTextFile(args.output)
        svFile = sc.parallelize(rdd2)
        svFile.collect().saveAsTextFile(args.output)





    if args.mode=='IDF':
        # Read list of files from args.input, compute IDF of each term,
        # and store result in file args.output.  All terms are first converted to
        # lowercase, and have non alphabetic characters removed
        # (i.e., 'Ba,Na:Na.123' and 'banana' count as the same term). Empty strings ""
        # are removed
        pass






    if args.mode=='TFIDF':
        # Read  TF scores from file args.input the IDF scores from file args.idfvalues,
        # compute TFIDF score, and store it in file args.output. Both input files contain
        # strings representing pairs of the form (TERM,VAL),
        # where TERM is a lowercase letter-only string and VAL is a numeric value.
        pass






    if args.mode=='SIM':
        # Read  scores from file args.input the scores from file args.other,
        # compute the cosine similarity between them, and store it in file args.output. Both input files contain
        # strings representing pairs of the form (TERM,VAL),
        # where TERM is a lowercase, letter-only string and VAL is a numeric value.
        pass