import spark.implicits._
import io.delta.tables._
import com.aamend.spark.gdelt._
import org.apache.spark.sql.{Dataset,DataFrame,SaveMode}
import org.apache.spark.sql.functions._
import org.apache.spark.sql.expressions._
import org.lamastex.spark.trendcalculus._
import java.sql.{Date,Timestamp}
import java.text.SimpleDateFormat

import spark.implicits._
import io.delta.tables._
import com.aamend.spark.gdelt._
import org.apache.spark.sql.{Dataset, DataFrame, SaveMode}
import org.apache.spark.sql.functions._
import org.apache.spark.sql.expressions._
import org.lamastex.spark.trendcalculus._
import java.sql.{Date, Timestamp}
import java.text.SimpleDateFormat

"./000a_finance_utils"

"./000b_gdelt_utils"

val rootPath = GdeltUtils.getGdeltV1Path
val rootCheckpointPath = GdeltUtils.getEOICheckpointPath
val gkgPath = rootPath+"gkg"
val eventPath = rootPath+"events"

val gkg_v1 = spark.read.format("delta").load(gkgPath).as[GKGEventV1]
val eve_v1 = spark.read.format("delta").load(eventPath).as[EventV1]

val gkg_v1_filt = gkg_v1.filter($"publishDate">"2013-04-01 00:00:00" && $"publishDate"<"2019-12-31 00:00:00")
val oil_gas_themeGKG = gkg_v1_filt.filter(c =>c.themes.contains("ENV_GAS") || c.themes.contains("ENV_OIL"))
                              .select(explode($"eventIds"))
                              .toDF("eventId")
                              .groupBy($"eventId")
                              .agg(count($"eventId"))
                              .toDF("eventId","count") 
val oil_gas_eventDF = eve_v1.toDF()
                            .join( oil_gas_themeGKG, "eventId")

oil_gas_eventDF.write.parquet(rootCheckpointPath + "oil_gas_event_v1")

val  oil_gas_eventDF = spark.read.parquet(rootCheckpointPath + "oil_gas_event_v1")

// Counting number of articles for each country, each day and applying a moving average on each country's coverage
def movingAverage(df:DataFrame,size:Int,avgOn:String):DataFrame = {
  val windowSpec = Window.partitionBy($"country").orderBy($"date").rowsBetween(-size/2, size/2)
  return df.withColumn("coverage",avg(avgOn).over(windowSpec))
}
val (mean_articles, std_articles) = oilEventTemp.select(mean("articles"), stddev("articles"))
  .as[(Double, Double)]
  .first() 
val oilEventTemp = oil_gas_eventDF
  .filter(length(col("eventGeo.countryCode")) > 0)
  .groupBy(
    col("eventGeo.countryCode").as("country"),
    col("eventDay").as("date")
  )
  .agg(
    sum(col("numArticles")).as("articles"),
  )
                                        
//Applying moving averge over weeks and of normalized number of articles.
val oilEventWeeklyCoverage = movingAverage(oilEventTemp.withColumn("normArticles",
                                                                   ($"articles"-mean_articles) / std_articles),
                                                                    7,
                                                                    "normArticles")
  .drop("normArticles")      

movingAverage: (df: org.apache.spark.sql.DataFrame, size: Int, avgOn: String)org.apache.spark.sql.DataFrame
oilEventTemp: org.apache.spark.sql.DataFrame = [country: string, date: date ... 1 more field]
mean_articles: Double = 1593.3213115310818
std_articles: Double = 10276.10388376719
oilEventWeeklyCoverage: org.apache.spark.sql.DataFrame = [country: string, date: date ... 2 more fields]

oilEventWeeklyCoverage.write.parquet(rootCheckpointPath +"oil_gas_cov_norm")

val oil_gas_cov_norm = spark.read.parquet(rootCheckpointPath +"oil_gas_cov_norm")

val oilEventWeeklyCoverageC = oil_gas_cov_norm.drop($"articles").toDF("country","tempDate","coverage")
val oilEventCoverageDF = oilEventWeeklyCoverageC.join(oil_gas_eventDF,oil_gas_eventDF("eventDay") === oilEventWeeklyCoverageC("tempDate") && oil_gas_eventDF("eventGeo.countryCode")
                       === oilEventWeeklyCoverageC("country"))

oilEventCoverageDF.write.parquet(rootCheckpointPath+"oil_gas_eve_cov")

val oilEventCoverageDF = spark.read.parquet(rootCheckpointPath+"oil_gas_eve_cov")

val tot_cov_2018 = oil_gas_cov_norm.filter($"date" >"2018-01-01" && $"date"<"2018-12-31").groupBy($"date").agg(sum($"coverage").as("coverage")).orderBy(desc("coverage"))

display(tot_cov_2018)

display(oil_gas_cov_norm.filter($"date" >"2018-01-01" && $"date"<"2018-12-31").orderBy(desc("coverage")).limit(1000))

display(oil_gas_cov_norm.filter($"date" >"2018-01-01" && $"date"<"2018-12-31" && $"country" =!="US").orderBy(desc("coverage")).limit(1000))