// taking a look at types and columns
rawDf.schema()

rawDf.size() // rowsCount x columnsCount

rawDf.head() // return the first five rows

// Getting general statistics and info for each columns
rawDf.describe()

val df = rawDf.dropNulls { date_added } // remove rows where `date_added` is not specified
    .convert { date_added }.toLocalDate("MMMM d, yyyy", java.util.Locale.ENGLISH) // convert date_added to LocalDate using date pattern
    .sortBy { date_added } // and let's also sort by date for easy operation later
df

// let's look at what type of column it turned out
df.date_added.type()

rawDf
    .valueCounts(sort = false) { type }
    .plot {
        bars {
            x(type)
            y("count")
            fillColor(type) {
                scale = categorical(range = listOf(Color.hex("#00BCD4"), Color.hex("#009688")))
            }
        }

        layout {
            title = "Count of TV Shows and Movies"
            size = 900 to 550
        }
    }

val df_date_count = df
    .convert { date_added }.with { it.year } // converting `date_added` to extract `year`
    .groupBy { date_added } // grouping by `year` stored in `date_added`
    .aggregate {
        count { type == "TV Show" } into "tvshows" // counting TV Shows into column `tvshows`
        count { type == "Movie" } into "movies" // counting Movies into column `movies`
    }
df_date_count

val df_date_count = df
    .groupBy { date_added.map { it.year } } // grouping by year added extracted from `date_added`
    .aggregate {
        count { type == "TV Show" } into "tvshows" // counting TV Shows into column `tvshows`
        count { type == "Movie" } into "movies" // counting Movies into column `movies`
    }
df_date_count

df.groupBy { date_added.map { it.year } }
    .pivot { type }

df.groupBy { date_added.map { it.year } }
    .pivot { type }.aggregate { count() }

df.groupBy { date_added.map { it.year } }
    .pivot { type }.count()

val df_date_count = df
    .groupBy { date_added.map { it.year } }.pivotCounts { type }
df_date_count

df_date_count.plot {
    x(date_added) { axis.name = "year" }

    area {
        y(type.`TV Show`) { axis.name = "count" }
        fillColor = Color.hex("#BF360C")
        borderLine.color = Color.hex("#BF360C")
        alpha = .5
    }

    area {
        y(type.Movie)
        fillColor = Color.hex("#01579B")
        borderLine.color = Color.hex("#01579B")
        alpha = .5
    }

    layout {
        title = "Number of titles by year"
        size = 800 to 500
        style {
            panel {
                background {
                    fillColor = Color.hex("#ECEFF1")
                    borderLineColor = Color.hex("#ECEFF1")
                }
                grid.lineGlobal { blank = true }
            }
        }
    }
}

val df_cumsum_titles = df_date_count
    .sortBy { date_added } // sorting by date_added
    .cumSum { type.colsOf<Int>() } // count cumulative sum for columns `TV Show` and `Movie` that are nested under column `type`
df_cumsum_titles

df_cumsum_titles.plot {
    x(date_added) { axis.name = "year" }

    area {
        y(type.`TV Show`) { axis.name = "cumulative count" }
        fillColor = Color.hex("#BF360C")
        borderLine.color = Color.hex("#BF360C")
        alpha = .5
    }

    area {
        y(type.Movie)
        fillColor = Color.hex("#01579B")
        borderLine.color = Color.hex("#01579B")
        alpha = .5
    }

    layout {
        title = "Cumulative count of titles by year"
        size = 800 to 500

        style {
            panel {
                background {
                    fillColor = Color.hex("#ECEFF1")
                    borderLineColor = Color.hex("#ECEFF1")
                }
                grid.lineGlobal { blank = true }
            }
        }
    }
}

val p1 = df_days.select { type and days_on_platform }.plot {
    histogram(days_on_platform, binsOption = BinsOption.byNumber(30)) {
        y(Stat.density)
        fillColor = Color.hex("#ef0b0b")
        borderLine.color = Color.hex("#ECEFF1")
    }

    statBin(days_on_platform, binsOption = BinsOption.byNumber(30)) {
        area {
            x(Stat.x)
            y(Stat.density)
            alpha = .5
            fillColor = Color.hex("#0befef")
        }
    }

    layout {
        xAxisLabel = "days"
        title = "Age distribution (in days) on Netflix"
    }
}

val p2 = df_days.select { type and days_on_platform }.plot {
    boxplot(x = type, y = days_on_platform) {
        boxes {
            fillColor(Stat.x) {
                scale = categorical(range = listOf(Color.hex("#792020"), Color.hex("#207979")))
            }
        }
    }
    layout {
        yAxisLabel = "days"
        title = "Boxplot for age (in days) by type"
    }
}

plotBunch {
    add(p1, 0, 0, 500, 450)
    add(p2, 500, 0, 500, 450)
}

df_days.valueCounts(sort = false) { type and years_on_platform }.plot {
    bars {
        x(years_on_platform) { axis.name = "years" }
        y("count")
        fillColor(type) {
            scale = categorical(range = listOf(Color.hex("#bc3076"), Color.hex("#30bc76")))
        }
        position = Position.dodge()
    }
    layout {
        title = "Years of Movies and TV Shows on Netflix"
        size = 900 to 500
    }
}

val df_years = df
    // adding a new column of the difference between the year of release and the year of addition
    .add("years_off_platform") {
        date_added.year - release_year
    }
    // dropping negative values and equal to zero
    .filter { "years_off_platform"<Int>() > 0 }
df_years

df_years.valueCounts(false) { years_off_platform }.plot {
    x(years_off_platform) { axis.name = "years" }
    points {
        y("count")
        size = 7.5
        color(years_off_platform) {
            scale = continuous(range = Color.hex("#97a6d9")..Color.hex("#00256e"))
        }
    }
    layout {
        title = "How long does it take for a title to be added to Netflix?"
        size = 1000 to 500
    }
}

// Top 5 oldest movies
df_days
    .filter { type == "Movie" } // filtering by type
    .sortByDesc { days_on_platform } // sorting by number of days on Netflix
    .select { cols(type, title, country, date_added, release_year, duration) } // selecting required columns
    .head() // taking first five rows

// Top 5 newest movies
df_days
    .filter { type == "Movie" }
    .sortBy { days_on_platform }
    .select { cols(type, title, country, date_added, release_year, duration) }
    .head()

// Top 5 oldest shows
df_days
    .filter { type == "TV Show" }
    .sortByDesc { days_on_platform }
    .select { cols(type, title, country, date_added, release_year, duration) }
    .head()

// Top 5 newest shows
df_days
    .filter { type == "TV Show" }
    .sortBy { days_on_platform }
    .select { cols(type, title, country, date_added, release_year, duration) }
    .head()

val df_split_date = df
    // splitting dates into four columns
    .split { date_added }.by { listOf(it, it.dayOfWeek, it.month, it.year) }
    .into("date", "day", "month", "year")
    .sortBy("month") // sorting by month
df_split_date