#!/bin/usr/env python3
# (c) 2017 - 2020 L.Spiegelberg
# following is a simple baseline experiment
import time
import sys
import json
import argparse
import os
import glob

from pyspark.sql import *
from pyspark.sql.types import *
from pyspark import SparkContext, SparkConf
from pyspark.sql import *
from pyspark.sql.functions import udf
from pyspark.sql.types import *
import csv
import math

def runRDDTuple(path, output_path):

    """
    runs spark using RDDs and tuples as internal data format
    :param path:
    :param output_path:
    :return:
    """
    jstart = time.time()

    def extractZipcode(x):
        x = tuple(x)
        try:
            return x + ('{:05}'.format(int(float(x[4]))),)
        except:
            return x + (None,)

    def cleanCity(x):
        x = tuple(x)
        try:
            return x[:2] + (x[2][0].upper() + x[2][1:].lower(),) + x[3:]
        except:
            return x[:2] + (None,) + x[3:]

    def extractBd(x):
        x = tuple(x)
        try:
            val = x[6]
            max_idx = val.find(' bd')
            if max_idx < 0:
                max_idx = len(val)
            s = val[:max_idx]

            # find comma before
            split_idx = s.rfind(',')
            if split_idx < 0:
                split_idx = 0
            else:
                split_idx += 2
            r = s[split_idx:]
            return x + (int(r),)
        except:
            return x + (None,)

    def extractBa(x):
        x = tuple(x)
        try:
            val = x[6]
            max_idx = val.find(' ba')
            if max_idx < 0:
                max_idx = len(val)
            s = val[:max_idx]

            # find comma before
            split_idx = s.rfind(',')
            if split_idx < 0:
                split_idx = 0
            else:
                split_idx += 2
            r = s[split_idx:]
            ba = math.ceil(2.0 * float(r)) / 2.0
            return x + (ba,)
        except:
            return x + (None,)

    def extractSqft(x):
        x = tuple(x)
        try:
            val = x[6]
            max_idx = val.find(' sqft')
            if max_idx < 0:
                max_idx = len(val)
            s = val[:max_idx]

            split_idx = s.rfind('ba ,')
            if split_idx < 0:
                split_idx = 0
            else:
                split_idx += 5
            r = s[split_idx:]
            r = r.replace(',', '')
            return x + (int(r),)
        except:
            return x + (None,)

    def extractOffer(x):
        x = tuple(x)
        try:
            offer = x[0].lower()
            if 'sale' in offer:
                return x + ('sale',)
            if 'rent' in offer:
                return x + ('rent',)
            if 'sold' in offer:
                return x + ('sold',)
            if 'foreclose' in offer.lower():
                return x + ('foreclosed',)
            return x + (offer,)
        except:
            return x + (None,)

    def extractType(x):
        x = tuple(x)
        try:
            t = x[0].lower()
            type = 'unknown'
            if 'condo' in t or 'apartment' in t:
                type = 'condo'
            if 'house' in t:
                type = 'house'
            return x + (type,)
        except:
            return x + (None,)

    def extractPrice(x):
        x = tuple(x)
        try:
            price = x[5]
            if x[15] == 'sold':
                # price is to be calculated using price/sqft * sqft
                val = x[6]
                s = val[val.find('Price/sqft:') + len('Price/sqft:') + 1:]
                r = s[s.find('$')+1:s.find(', ') - 1]
                price_per_sqft = int(r)
                price = price_per_sqft * x['sqft']
            elif x[15] == 'rent':
                max_idx = price.rfind('/')
                price = int(price[1:max_idx].replace(',', ''))
            else:
                # take price from price column
                price = int(price[1:].replace(',', ''))

            return x[:5] + (price,) + x[6:]
        except:
            return x[:5] + (None,) + x[6:]

    def selectCols(x):
        x = tuple(x)
        return (x[8],x[12],x[1],x[2],x[3],x[10],x[13],x[14],x[15],x[11],x[5],)

    def filterPriceAndOffer(x):
        x = tuple(x)
        try:
            # in original 6th column is price
            return 100000 < x[5] <= 2e7 and x[15] == 'sale'
        except:
            return False

    def filterType(x):
        x = tuple(x)
        try:
            return x[-1] == 'condo'
        except:
            return False

    def filterBd(x):
        x = tuple(x)
        try:
            return x[-1] < 10
        except:
            return False

    tstart = time.time()

    spark = SparkSession.builder \
        .master("local[16]") \
        .config("spark.hadoop.validateOutputSpecs", "false") \
        .appName("read csv") \
        .getOrCreate()

    sc = spark.sparkContext

    tparse_start = time.time()

    # Spark pipeline
    # note the ignore header to have the contents ready
    rdd1 = sc.textFile(path)
    header = rdd1.first()

    # tuple based approach
    rdd2 = rdd1.zipWithIndex().filter(lambda x: x[1] > 0) \
        .map(lambda x: x[0]) \
        .map(lambda t: tuple(csv.reader([t]))[0])

    # pipeline will be done here
    rddFinal = rdd2.map(extractBd) \
        .filter(filterBd) \
        .map(extractType) \
        .filter(filterType) \
        .map(extractZipcode) \
        .map(cleanCity) \
        .map(extractBa) \
        .map(extractSqft) \
        .map(extractOffer) \
        .map(extractPrice) \
        .filter(filterPriceAndOffer) \
        .map(selectCols)

    # write to text file
    columns = ['url', 'zipcode', 'address', 'city', 'state', 'bedrooms', 'bathrooms', 'sqft', 'offer', 'type', 'price']

    def tocsvstr(a):
        # escape strings if necessary
        def escapecsv(s):
            s = s.replace('"', '""')  # quote quote
            if ',' in s or '\n' in s:
                s = '"' + s + '"'  # quote if contains , or newline
            return s

        a = [escapecsv(str(x)) for x in a]
        return ','.join(a)

    rddHeader = sc.parallelize([tuple(columns)])

    # experiment is to save to ONE file
    rddHeader.union(rddFinal).map(tocsvstr).saveAsTextFile(output_path)

    tend = time.time()

    sc = spark.sparkContext
    framework = 'Spark {} {}'.format(sc.version, sc.master)

    stats = {'type': 'RDD tuple',
             'framework': framework,
             'run_time': tend - tstart,
             'pipeline_time': tend - tparse_start,
             'startup_time': tparse_start - tstart,
             'job_time': tend - jstart}

    print(stats)


def runRDDDict(path, output_path):

    """
    runs spark using RDDs and dicts as internal data format
    :param path:
    :param output_path:
    :return:
    """
    jstart = time.time()

    def extractZipcode(x):
        try:
            return {'zipcode': '%05d' % int(float(x['postal_code'])), **x}
        except:
            return {'zipcode': None, **x}

    def cleanCity(x):
        try:
            return {**x, 'city': x['city'][0].upper() + x['city'][1:].lower()}
        except:
            return {**x, 'city': None}

    def extractBd(x):
        try:
            val = x['facts and features']
            max_idx = val.find(' bd')
            if max_idx < 0:
                max_idx = len(val)
            s = val[:max_idx]

            # find comma before
            split_idx = s.rfind(',')
            if split_idx < 0:
                split_idx = 0
            else:
                split_idx += 2
            r = s[split_idx:]
            return {**x, 'bedrooms': int(r)}
        except:
            return {**x, 'bedrooms': None}

    def extractBa(x):
        try:
            val = x['facts and features']
            max_idx = val.find(' ba')
            if max_idx < 0:
                max_idx = len(val)
            s = val[:max_idx]

            # find comma before
            split_idx = s.rfind(',')
            if split_idx < 0:
                split_idx = 0
            else:
                split_idx += 2
            r = s[split_idx:]
            ba = math.ceil(2.0 * float(r)) / 2.0
            return {**x, 'bathrooms': ba}
        except:
            return {**x, 'bathrooms': None}

    def extractSqft(x):
        try:
            val = x['facts and features']
            max_idx = val.find(' sqft')
            if max_idx < 0:
                max_idx = len(val)
            s = val[:max_idx]

            split_idx = s.rfind('ba ,')
            if split_idx < 0:
                split_idx = 0
            else:
                split_idx += 5
            r = s[split_idx:]
            r = r.replace(',', '')
            return {**x, 'sqft': int(r)}
        except:
            return {**x, 'sqft': None}

    def extractOffer(x):
        try:
            offer = x['title'].lower()

            if 'sale' in offer:
                offer = 'sale'
            if 'rent' in offer:
                offer = 'rent'
            if 'sold' in offer:
                offer = 'sold'
            if 'foreclose' in offer:
                offer = 'foreclosed'

            return {**x, 'offer': offer}
        except:
            return {**x, 'offer': None}

    def extractType(x):
        try:
            t = x['title'].lower()
            type = 'unknown'
            if 'condo' in t or 'apartment' in t:
                type = 'condo'
            if 'house' in t:
                type = 'house'
            return {**x, 'type': type}
        except:
            return {**x, 'type': None}

    def extractPrice(x):
        try:
            price = x['price']

            if x['offer'] == 'sold':
                # price is to be calculated using price/sqft * sqft
                val = x['facts and features']
                s = val[val.find('Price/sqft:') + len('Price/sqft:') + 1:]
                r = s[s.find('$') + 1:s.find(', ') - 1]
                price_per_sqft = int(r)
                price = price_per_sqft * x['sqft']
            elif x['offer'] == 'rent':
                max_idx = price.rfind('/')
                price = int(price[1:max_idx].replace(',', ''))
            else:
                # take price from price column
                price = int(price[1:].replace(',', ''))

            return {**x, 'price': price}
        except:
            return {**x, 'price': None}

    def selectCols(x):
        columns = ['url', 'zipcode', 'address', 'city', 'state', 'bedrooms', 'bathrooms', 'sqft', 'offer', 'type',
                   'price']
        return {key: x[key] for key in x.keys() if key in columns}

    def filterPriceAndOffer(x):
        try:
            return 100000 < x['price'] <= 2e7 and x['offer'] == 'sale'
        except:
            return False

    def filterType(x):
        try:
            return x['type'] == 'condo'
        except:
            return False

    def filterBd(x):
        try:
            return x['bedrooms'] < 10
        except:
            return False

    tstart = time.time()

    spark = SparkSession.builder \
        .master("local[16]") \
        .config("spark.hadoop.validateOutputSpecs", "false") \
        .appName("read csv") \
        .getOrCreate()

    sc = spark.sparkContext

    tparse_start = time.time()

    # Spark pipeline
    # note the ignore header to have the contents ready
    rdd1 = sc.textFile(path)
    header = rdd1.take(1)[0]
    initialCols = header.split(',')

    # spark has some trouble making correct closure, hence here again columns explicitly
    initialCols = 'title,address,city,state,postal_code,price,facts and features,real estate provider,url,sales_date'.split(
        ',')
    # dict based approach
    rdd2 = rdd1.zipWithIndex().filter(lambda x: x[1] > 0) \
        .map(lambda x: x[0]) \
        .map(lambda t: tuple(csv.reader([t]))[0]) \
        .map(lambda t: dict(zip(initialCols, t)))

    # pipeline will be done here
    rddFinal = rdd2.map(extractBd) \
        .filter(filterBd) \
        .map(extractType) \
        .filter(filterType) \
        .map(extractZipcode) \
        .map(cleanCity) \
        .map(extractBa) \
        .map(extractSqft) \
        .map(extractOffer) \
        .map(extractPrice) \
        .filter(filterPriceAndOffer) \
        .map(selectCols)

    # write to text file
    columns = ['url', 'zipcode', 'address', 'city', 'state', 'bedrooms', 'bathrooms', 'sqft', 'offer', 'type', 'price']

    def tocsvstr(d):
        # escape strings if necessary
        def escapecsv(s):
            s = s.replace('"', '""')  # quote quote
            if ',' in s or '\n' in s:
                s = '"' + s + '"'  # quote if contains , or newline
            return s

        a = []
        for c in columns:
            a.append(escapecsv(str(d[c])))
        return ','.join(a)

    rddHeader = sc.parallelize([','.join(columns)])

    # experiment is to save to ONE file
    rddHeader.union(rddFinal.map(tocsvstr)).saveAsTextFile(output_path)

    tend = time.time()

    sc = spark.sparkContext
    framework = 'Spark {} {}'.format(sc.version, sc.master)

    stats = {'type': 'RDD dict',
             'framework': framework,
             'run_time': tend - tstart,
             'pipeline_time': tend - tparse_start,
             'startup_time': tparse_start - tstart,
             'job_time': tend - jstart}

    print(stats)


def runDF(path, output_path):

    """
    runs spark using Spark 2.x Dataframe API
    :param path:
    :param output_path:
    :return:
    """

    jstart = time.time()

    def extractZipcode(x):
        try:
            return '%05d' % int(float(x))
        except:
            return None

    def cleanCity(city):
        try:
            return city[0].upper() + city[1:].lower()
        except:
            return None

    def extractBd(val):
        try:
            max_idx = val.find(' bd')
            if max_idx < 0:
                max_idx = len(val)
            s = val[:max_idx]

            # find comma before
            split_idx = s.rfind(',')
            if split_idx < 0:
                split_idx = 0
            else:
                split_idx += 2
            r = s[split_idx:]
            return int(r)
        except:
            return None

    def extractBa(val):
        try:
            max_idx = val.find(' ba')
            if max_idx < 0:
                max_idx = len(val)
            s = val[:max_idx]

            # find comma before
            split_idx = s.rfind(',')
            if split_idx < 0:
                split_idx = 0
            else:
                split_idx += 2
            r = s[split_idx:]
            ba = math.ceil(2.0 * float(r)) / 2.0
            return ba
        except:
            return None

    def extractSqft(val):
        try:
            max_idx = val.find(' sqft')
            if max_idx < 0:
                max_idx = len(val)
            s = val[:max_idx]

            split_idx = s.rfind('ba ,')
            if split_idx < 0:
                split_idx = 0
            else:
                split_idx += 5
            r = s[split_idx:]
            r = r.replace(',', '')
            return int(r)
        except:
            return None

    def extractOffer(title):
        try:
            offer = title.lower()

            if 'sale' in offer:
                return 'sale'
            if 'rent' in offer:
                return 'rent'
            if 'sold' in offer:
                return 'sold'
            if 'foreclose' in offer:
                return 'foreclosed'
            return offer
        except:
            return None

    def extractType(title):
        try:
            t = title.lower()
            type = 'unknown'
            if 'condo' in t or 'apartment' in t:
                type = 'condo'
            if 'house' in t:
                type = 'house'
            return type
        except:
            return None

    def extractPrice(price, offer, facts, sqft):
        try:

            if offer == 'sold':
                # price is to be calculated using price/sqft * sqft
                val = facts
                s = val[val.find('Price/sqft:') + len('Price/sqft:') + 1:]
                r = s[s.find('$') + 1:s.find(', ') - 1]
                price_per_sqft = int(r)
                price = price_per_sqft * sqft
            elif offer == 'rent':
                max_idx = price.rfind('/')
                price = int(price[1:max_idx].replace(',', ''))
            else:
                # take price from price column
                price = int(price[1:].replace(',', ''))

            return price
        except:
            return None

    # spark DataFrame udfs
    udfBd = udf(extractBd, IntegerType())
    udfType = udf(extractType, StringType())
    udfZipcode = udf(extractZipcode, StringType())
    udfCity = udf(cleanCity, StringType())
    udfBa = udf(extractBa, DoubleType())
    udfSqft = udf(extractSqft, IntegerType())
    udfOffer = udf(extractOffer, StringType())
    udfPrice = udf(extractPrice, IntegerType())

    def init_spark(app_name, master_config):
        """
        :params app_name: Name of the app
        :params master_config: eg. local[4]
        :returns SparkContext, SQLContext, SparkSession:
        """
        conf = (SparkConf().setAppName(app_name).setMaster(master_config))

        sc = SparkContext(conf=conf)
        #sc.setLogLevel("ERROR")
        sql_ctx = SQLContext(sc)
        spark = SparkSession(sc)

        return (sc, sql_ctx, spark)

    tstart = time.time()
    sc, sql_ctx, spark = init_spark("read csv", "local[16]")

    tparse_start = time.time()

    #df = spark.read.options(header=True, mode='PERMISSIVE', multiLine=True, escape='"') \
    df = spark.read.options(header=True, mode='PERMISSIVE', escape='"') \
        .csv(path) \
        .withColumn('bedrooms', udfBd("facts and features")) \
        .filter("bedrooms  < 10") \
        .withColumn("type", udfType("title")) \
        .filter("type == 'condo'") \
        .withColumn("zipcode", udfZipcode("postal_code")) \
        .withColumn("city", udfCity("city")) \
        .withColumn('bathrooms', udfBa("facts and features")) \
        .withColumn('sqft', udfSqft("facts and features")) \
        .withColumn('offer', udfOffer("title")) \
        .withColumn('price', udfPrice("price", "offer", "facts and features", "sqft")) \
        .filter("100000 < price") \
        .filter("price <= 2e7 and offer == 'sale'") \
        .select('url', 'zipcode', 'address', 'city', 'state', 'bedrooms', 'bathrooms', 'sqft', 'offer', 'type', 'price')

    # write to file (single partition)
    df.write.csv(output_path, mode='overwrite', header=True)

    tend = time.time()

    framework = 'Spark {} {}'.format(sc.version, sc.master)

    stats = {'type': 'DataFrame',
             'framework': framework,
             'run_time': tend - tstart,
             'pipeline_time': tend - tparse_start,
             'startup_time': tparse_start - tstart,
             'job_time' : tend - jstart}

    print(stats)

if __name__ == '__main__':
    parser = argparse.ArgumentParser(description='Zillow cleaning')
    parser.add_argument('--path', type=str, dest='data_path', default='data/large100MB.csv',
                        help='path or pattern to zillow data')
    parser.add_argument('--mode', type=str, dest='mode', default='tuple',
                        help='specify tuple or dict for internal data representation')
    parser.add_argument('--output-path', type=str, dest='output_path', default='pyspark_output',
                        help='specify path where to save output data files')
    args = parser.parse_args()

    assert args.data_path, 'need to set data path!'

    # config vars
    paths = [args.data_path]
    output_path = args.output_path

    # explicit globbing because dask can't handle patterns well...
    if not os.path.isfile(args.data_path):
        paths = sorted(glob.glob(os.path.join(args.data_path, '*.csv')))
    else:
        paths = [args.data_path]

    if not paths:
        print('found no zillow data to process, abort.')
        sys.exit(1)

    print('>>> running pyspark on {}'.format(paths))
    print('>>> data mode: {}'.format(args.mode))

    # run specific pipeline (either tuples or dicts)
    if 'tuple' in args.mode :
        runRDDTuple(','.join(paths), output_path)
    elif 'dict' in args.mode:
        runRDDDict(','.join(paths), output_path)
    elif args.mode == 'sql' or args.mode == 'df' or args.mode == 'dataframe':
        runDF(','.join(paths), output_path)
    else:
        print('invalid mode {}, abort.'.format(args.mode))
        sys.exit(1)
