# Dejan Belic

import datetime
from pyspark import SparkConf, SparkContext

conf = SparkConf().setMaster("local").setAppName("Festivali")
sc = SparkContext(conf=conf)

def transform_muzicari_input(x):
    splits = x.split(',')
    return (splits[0], splits[1:])

def transform_sastav(x):
    splits = x.split(',')
    date = [int(x) for x in splits[2].split('-')]
    datetime1 = datetime.datetime(date[0], date[1], date[2])
    if splits[3] == "NULL" :
        datetime2 = datetime.datetime.now()
    else:
        date = [int(x) for x in splits[3].split('-')]
        datetime2 = datetime.datetime(date[0], date[1], date[2])
    
    return ((splits[0], splits[1]), datetime2 - datetime1)

def transform_bendovi(x):
    splits = x.split(',')
    return (splits[0], splits[1])

muzicari = sc.textFile("../festivali/muzicari.csv").map(lambda x: transform_muzicari_input(x))
sastav = sc.textFile("../festivali/sastav_benda.csv").map(lambda x: transform_sastav(x)).reduceByKey(lambda x, y: x + y)
bendovi = sc.textFile("../festivali/bendovi.csv").map(lambda x: transform_bendovi(x))

def longer_stay(x, y):
    if (x[1] > y[1]):
        return x
    return y

najvise_ostao = sastav.map(lambda x: (x[0][0], (x[0][1], x[1]))).reduceByKey(lambda x, y: longer_stay(x, y))

j1 = muzicari.join(najvise_ostao).map(lambda x: (x[1][1][0], (x[1][0], x[1][1][1])))
j2 = j1.join(bendovi).mapValues(lambda x: [x[0][0][0], x[0][0][1], x[1], f'{x[0][1].days} dana']) # Transformacija u odgovarajuci oblik

for x in j2.collect():
    print(x)
    
# Upis fajlova u csv
import csv
from io import StringIO
def writeRecords(records):
    output = StringIO()
    writer = csv.writer(output)
    for record in records:
        writer.writerow(record)
    return [output.getvalue()]
j2.map(lambda x: x[1]).mapPartitions(writeRecords).saveAsTextFile("out")