from typing import List
from typing import Union

from pyspark.sql import DataFrame
from pyspark.sql import functions as F
from pyspark.sql.window import Window


def filter_all_not_null(df: DataFrame, reference_columns: List[str]) -> DataFrame:
    """
    Filter rows which have NULL values in all the specified columns.
    From households_aggregate_processes.xlsx, filter number 2.
    Parameters
    ----------
    df
    reference_columns
        Columns to check for missing values in, all
        must be missing for the record to be dropped.
    """
    return df.na.drop(how="all", subset=reference_columns)


def filter_duplicates_by_time_and_threshold(
    df: DataFrame,
    first_reference_column: str,
    second_reference_column: str,
    third_reference_column: str,
    fourth_reference_column: str,
    time_threshold: float = 1.5,
    float_threshold: float = 0.00001,
) -> DataFrame:
    """
    Drop duplicates based on two identitical column values if third and fourth column and not both within
    a threshold difference from the first duplicate record.
    From households_aggregate_processes.xlsx, filter number 4.
    Parameters
    ----------
    df
    first_reference_column
        First column with duplicate value
    second_reference_column
        Second column with duplicate value
    third_reference_column
        Column used for time based threshold difference, timestamp
    fourth_reference_column
        Column used for numeric based threshold difference, float
    """

    window = Window.partitionBy(first_reference_column, second_reference_column).orderBy(third_reference_column)

    df = df.withColumn("duplicate_id", F.row_number().over(window))

    df = df.withColumn(
        "within_time_threshold",
        (
            F.abs(
                F.first(third_reference_column).over(window).cast("long") - F.col(third_reference_column).cast("long")
            )
            / (60 * 60)
        )
        < time_threshold,
    )

    df = df.withColumn(
        "within_float_threshold",
        F.abs(F.first(fourth_reference_column).over(window) - F.col(fourth_reference_column)) < float_threshold,
    )

    df = df.filter((F.col("duplicate_id") == 1) | ~(F.col("within_time_threshold") & (F.col("within_float_threshold"))))

    return df.drop("duplicate_id", "within_time_threshold", "within_float_threshold")


def filter_by_cq_diff(
    df: DataFrame, comparing_column: str, ordering_column: str, tolerance: float = 0.00001
) -> DataFrame:
    """
    This function works out what columns have a float value difference less than 10-^5 or 0.00001
        (or any other tolerance value inputed) given all the other columns are the same and
        considers it to be the same dropping or deleting the repeated values and only keeping one entry.
    Parameters
    ----------
    df
    comparing_column
    ordering_column
    tolerance
    """
    column_list = df.columns
    column_list.remove(comparing_column)

    windowSpec = Window.partitionBy(column_list).orderBy(ordering_column)
    df = df.withColumn("first_value_in_duplicates", F.first(comparing_column).over(windowSpec))
    df = df.withColumn(
        "duplicates_first_record", F.abs(F.col("first_value_in_duplicates") - F.col(comparing_column)) < tolerance
    )

    difference_window = Window.partitionBy(column_list + ["duplicates_first_record"]).orderBy(ordering_column)
    df = df.withColumn("duplicate_number", F.row_number().over(difference_window))

    df = df.filter(~(F.col("duplicates_first_record") & (F.col("duplicate_number") != 1)))
    df = df.drop("first_value_in_duplicates", "duplicates_first_record", "duplicate_number")

    return df


def assign_date_interval_and_flag(
    df: DataFrame,
    column_name_inside_interval: str,
    column_name_time_interval: str,
    start_datetime_reference_column: str,
    end_datetime_reference_column: str,
    lower_interval: Union[int, float],
    upper_interval: Union[int, float],
    interval_format: str = "hours",
) -> DataFrame:
    """
    This function gives the time interval in either hours (by default) or days
    in a column by given two date columns and says whether it is inside and
    upper and lower interval. If the difference of dates is within the upper and
    lower time intervals, the function will output None and an integer 1 if the
    difference in dates are outside of those intervals.
    Parameters
    ----------
    df
    column_name_inside_interval
        Name of the column that returns whether the difference in dates are
        within the upper/lower limits if within, it will return None, if outside
        will return an integer 1.
    column_name_time_interval
        Name of the column that returns the difference between start and end
        date and adds at the end of the column name whether it is in hours or
        days
    start_datetime_reference_column
        Earliest date in string format yyyy-mm-dd hh:mm:ss.
    end_datetime_reference_column
        Latest date in string format yyyy-mm-dd hh:mm:ss.
    lower_interval
        Marks how much NEGATIVE time difference can have between
        end_datetime_reference_column and start_datetime_reference_column.
        Meaning how the end_datetime_reference_column can be earlier than
        start_datetime_reference_column
    upper_interval
        Marks how much POSITIVE time difference can have between
        end_datetime_reference_column and start_datetime_reference_column
    interval_format
        By default will be a string called 'hours' if upper and lower
        intervals are input as days, define interval_format to 'days'.
        These are the only two possible formats.
    Notes
    -----
    Lower_interval should be a negative value if start_datetime_reference_column
    is after end_datetime_reference_column."""

    # by default, Hours but if days, apply change factor
    if interval_format == "hours":  # to convert hours to seconds
        conversion_factor = 3600  # 1h has 60s*60min seconds = 3600 seconds
    elif interval_format == "days":
        conversion_factor = 86400  # 1 day has 60s*60min*24h seconds = 86400 seconds

    column_name_time_interval = column_name_time_interval + "_" + interval_format

    # FORMULA: (end_datetime_reference_column - start_datetime_reference_column) in
    # seconds/conversion_factor in seconds
    df = df.withColumn(
        column_name_time_interval,
        (
            F.to_timestamp(F.col(end_datetime_reference_column)).cast("long")
            - F.to_timestamp(F.col(start_datetime_reference_column)).cast("long")
        )
        / conversion_factor,  # 1 day has 60s*60min*24h seconds = 86400 seconds
    )

    return df.withColumn(
        column_name_inside_interval,
        F.when(~F.col(column_name_time_interval).between(lower_interval, upper_interval), 1).otherwise(None),
    )


def file_exclude(df: DataFrame, source_file_col: str, files_to_exclude: list):
    """
    Function to exclude specific files from pipeline processing

    Parameters
    --------
    df
    source_file_column = Column in input dataframe which contains the source file
    files_to_exclude = List of files to exclude (feed in from config)
    """
    for item in files_to_exclude:
        df = df.filter(~F.col(source_file_col).isin(item))

    return df

[((49, 13, 49, 80), 'pyspark.sql.window.Window.partitionBy', 'Window.partitionBy', ({(49, 32, 49, 54): 'first_reference_column', (49, 56, 49, 79): 'second_reference_column'}, {}), '(first_reference_column, second_reference_column)', False, 'from pyspark.sql.window import Window\n'), ((91, 17, 91, 48), 'pyspark.sql.window.Window.partitionBy', 'Window.partitionBy', ({(91, 36, 91, 47): 'column_list'}, {}), '(column_list)', False, 'from pyspark.sql.window import Window\n'), ((97, 24, 97, 85), 'pyspark.sql.window.Window.partitionBy', 'Window.partitionBy', ({(97, 43, 97, 84): "column_list + ['duplicates_first_record']"}, {}), "(column_list + ['duplicates_first_record'])", False, 'from pyspark.sql.window import Window\n'), ((51, 39, 51, 53), 'pyspark.sql.functions.row_number', 'F.row_number', ({}, {}), '()', True, 'from pyspark.sql import functions as F\n'), ((69, 20, 69, 41), 'pyspark.sql.functions.col', 'F.col', ({(69, 26, 69, 40): '"""duplicate_id"""'}, {}), "('duplicate_id')", True, 'from pyspark.sql import functions as F\n'), ((92, 52, 92, 77), 'pyspark.sql.functions.first', 'F.first', ({(92, 60, 92, 76): 'comparing_column'}, {}), '(comparing_column)', True, 'from pyspark.sql import functions as F\n'), ((98, 43, 98, 57), 'pyspark.sql.functions.row_number', 'F.row_number', ({}, {}), '()', True, 'from pyspark.sql import functions as F\n'), ((100, 21, 100, 53), 'pyspark.sql.functions.col', 'F.col', ({(100, 27, 100, 52): '"""duplicates_first_record"""'}, {}), "('duplicates_first_record')", True, 'from pyspark.sql import functions as F\n'), ((66, 62, 66, 92), 'pyspark.sql.functions.col', 'F.col', ({(66, 68, 66, 91): 'fourth_reference_column'}, {}), '(fourth_reference_column)', True, 'from pyspark.sql import functions as F\n'), ((69, 52, 69, 82), 'pyspark.sql.functions.col', 'F.col', ({(69, 58, 69, 81): '"""within_time_threshold"""'}, {}), "('within_time_threshold')", True, 'from pyspark.sql import functions as F\n'), ((69, 86, 69, 117), 'pyspark.sql.functions.col', 'F.col', ({(69, 92, 69, 116): '"""within_float_threshold"""'}, {}), "('within_float_threshold')", True, 'from pyspark.sql import functions as F\n'), ((94, 41, 94, 75), 'pyspark.sql.functions.col', 'F.col', ({(94, 47, 94, 74): '"""first_value_in_duplicates"""'}, {}), "('first_value_in_duplicates')", True, 'from pyspark.sql import functions as F\n'), ((94, 78, 94, 101), 'pyspark.sql.functions.col', 'F.col', ({(94, 84, 94, 100): 'comparing_column'}, {}), '(comparing_column)', True, 'from pyspark.sql import functions as F\n'), ((100, 57, 100, 82), 'pyspark.sql.functions.col', 'F.col', ({(100, 63, 100, 81): '"""duplicate_number"""'}, {}), "('duplicate_number')", True, 'from pyspark.sql import functions as F\n'), ((190, 24, 190, 46), 'pyspark.sql.functions.col', 'F.col', ({(190, 30, 190, 45): 'source_file_col'}, {}), '(source_file_col)', True, 'from pyspark.sql import functions as F\n'), ((66, 14, 66, 46), 'pyspark.sql.functions.first', 'F.first', ({(66, 22, 66, 45): 'fourth_reference_column'}, {}), '(fourth_reference_column)', True, 'from pyspark.sql import functions as F\n'), ((167, 27, 167, 63), 'pyspark.sql.functions.col', 'F.col', ({(167, 33, 167, 62): 'end_datetime_reference_column'}, {}), '(end_datetime_reference_column)', True, 'from pyspark.sql import functions as F\n'), ((168, 29, 168, 67), 'pyspark.sql.functions.col', 'F.col', ({(168, 35, 168, 66): 'start_datetime_reference_column'}, {}), '(start_datetime_reference_column)', True, 'from pyspark.sql import functions as F\n'), ((57, 76, 57, 105), 'pyspark.sql.functions.col', 'F.col', ({(57, 82, 57, 104): 'third_reference_column'}, {}), '(third_reference_column)', True, 'from pyspark.sql import functions as F\n'), ((175, 16, 175, 48), 'pyspark.sql.functions.col', 'F.col', ({(175, 22, 175, 47): 'column_name_time_interval'}, {}), '(column_name_time_interval)', True, 'from pyspark.sql import functions as F\n'), ((57, 16, 57, 47), 'pyspark.sql.functions.first', 'F.first', ({(57, 24, 57, 46): 'third_reference_column'}, {}), '(third_reference_column)', True, 'from pyspark.sql import functions as F\n')]