#  ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ ##
#  ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ CREATE AN AGGREGATED DB IN SNOWFLAKE ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ ##
#
#  ## NOTES
#  we will create a table 
#  weekof | unknown_uri | source | streams | lag_streams | diff in streams | perc_increase
#  
#  Do not include in the table:
#    * year = "DATE_PART(year, TMSTAMP)"
#    * week = "DATE_PART(week, TMSTAMP)"
#    * days_of_data_in_week = "count(distinct TMSTAMP::date)"
#    * artistid :  Why not?
#  
#  `week`` is similar to weekof, but cycles at 53. Therefore we cannot order by it across years
#  adding `year` almost helps, except for the week which goes across years. We get half of the week in one year the other half in the other year.  And then ordering still is a problem for week 53, in that the first few days of Jan 2016 are put at the end of Dec 2016.
#  Combining either `week` or `year` with `weekof` introduces similar problem.
#  Instead, using ONLY `weekof` which we can order by and does not cut up the year. (instead early Jan 2016 is included with end of Dec 2015)
#  
#  days_of_data_in_week :: We would like to track how many days of data we have in each week (to confirm that it is in fact 7)
#  However, the idea behind this is that there not be data errors. Yet if there were that should be caught elsewhere. As for a particular track with no streams on a given day, there is no need for this value, since we should still expect it to be 7, and we consider the days without data to be 0 streams. We still avg daily streams for the week by 7 days.
#  
#  
#  Use `track_album_artist` instead of `track_artists` since `track_artists` seems to have minor data-entry differences amongst the same ISRC
#  
#  
#  album_code VS upc:
#  ------------------
#     We checked the two columns, and for the data in production.staging_raw_spotify_v2, they are always the same
#  
#  ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ ##
#  ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ ##


# screen -xRR DiscoverWeekly
setScience("Discover_Weekly", create=TRUE)
setGitBranchToSystem(); .g()

wh <- "ETL_JOBS_Standard"
wh <- "SPOTIFYAGGREGATES"
dbname <- "prod"
# if (interactive())
#   wh <- getWH_by_interactive(default=getWH_already_on(default=wh))

setSnowflake(wh=wh, dbname=dbname, start=TRUE)
sfWaitForWarehouse(wh=wh, N.seconds=20, max_iterations=500, verbose=TRUE)

tbl <- "sos_from_raw_view"
schema <- "spotify"

tbl_new <- "weekly_counts_by_track_and_source"
schema_new <- "spotify"

smaxDate_in_tbl <- qMaxDate(tbl=tbl, schema=schema, wh=wh, snowflake=TRUE)
maxDate <- maxDate_in_tbl + 7 - which(getWdays()[wday(maxDate_in_tbl - (6:0))] == "Sun")

## Discover Weekly was press releasesed in July 2015
## May 4th 2015 is a monday
minDate <- as.Date("2015-05-04")   

## the source of interest
## THIS VAR IS NOT USED ANYWHERE, YET
source_op <- "others_playlist"

## DON'T REMEMBER WHAT THESE ARE FOR? 
candidates <- c("BRRGE1502999", "USCL81402807", "GBARL1200767", "CAU2O1010217")
non_using_isrcs <- c("GBPS81523905", "US2H51200724")

# ---------------------------------------------------------------------------------------------------
# ~~~~~~~~~~ BEGIN ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~


## NOT IMPLEMENTING THIS RIGHT NOW, BUT WE SHOULD LOOK INTO THESE
## .... or alternatively, decide that we dont give a shit about 35 trackids, and move on with life
if (FALSE) {
    ## ~~~~~~~~~~~~~~~~~~~~~~~~~~~    | DATE HYGIENE |   ~~~~~~~~~~~~~~~~~~~~~~~~~~~~
    ## There are about 35+ trackid's that have more than one UPC+ISRC to them
    ## These should be flagged
    qry.track_ids_with_more_than_one_upc_isrc_combo <- setQry("
    SELECT track_id, 1 AS has_multiple_upc_isrc_combo_per_track_id
    FROM (
      SELECT   
          upc
        , isrc
        , track_id
      FROM  production.staging_raw_spotify_v2
      WHERE TMSTAMP::date >= '2015-05-04'
      GROUP BY 1, 2, 3
    )
    GROUP BY 1
    HAVING count(*) > 1")
    ## ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
}


dateCol <- "TMSTAMP::date"

colsToPull.inner <- {
c(
    weekof = sprintf("friday_weekof(%s)", dateCol)
  # , week = "DATE_PART(week, TMSTAMP)"
  # , year = "DATE_PART(year, TMSTAMP)"
  , "source"
  , unknown_uri="CASE WHEN playlist_is_unknown = 1 THEN 'unknown' ELSE 'known'"
  , UPC = "album_code"  ## There is also a UPC column, however, I am not certain that this column is provided by Spotify. It might be added by us during the ETL.
  , "ISRC"
  ## All other metadata to come from the spotify.meta_from_raw_v2 table
)}

## some upc+isrc combos have more than one track_id, track_artists etc
colsWithaggFunc.inner <- 
    c( 
        track_name_from_spotify  = "max(track_name)"
      , no_of_track_names        = "count(distinct track_name)"
      )
colsToAgg.inner = c(streams="*")


stbl.release_meta <- "bi.release_with_metadata_view"
upcCol.release_meta <- "RELEASEID"

colsToPull.release_meta <- c("labelid", "label_client_manager", "label_is_dthree", "label_is_ioda", "label_is_o_and_o", "label_name", "label_owner", "label_ownerid", "label_priority", "label_sc_group", "label_subaccountid", "label_subaccount_name", "release_date", "release_genre", "release_is_compilation", "release_marketing_priority", "release_name", "release_number_of_tracks", "release_number_of_cds", "artistid", "artist_name")
colsToPull.spotify_meta <- c("track_id", "track_artists", "track_name", "track_uri", "track_album_artist", "album_name", "has_sloppy_metadata", "last_tmstamp")


LG <- "(LAG(streams, 1, NULL) OVER (PARTITION BY track_id, unknown_uri, source ORDER BY weekof ASC))"
diffCol <- c(diff_in_streams = sprintf("(streams - %1$s)", LG))
percCol <- c(  perc_increase = sprintf("(streams - %1$s)/%1$s", LG))
accelCol <- c( accel_score        = sprintf("score_accel_before_after(%s, streams)", LG)
             , accel_score_scaled = sprintf("scale_minus_one_to_plus_one(score_accel_before_after(%s, streams))", LG)
             )

colsToPull.outer <- c("S.*", lag_streams=LG, diffCol, percCol, accelCol) %>% 
                      c(., paste0("R.", colsToPull.release_meta)) %>% 
                      c(., paste0("M.", colsToPull.spotify_meta))

## BUILD QUERRIES

stbl.spotify_meta <- "spotify.meta_from_raw_v2"
join.outer_to_release <- sprintf("S LEFT JOIN %s R ON R.%s = S.upc\nLEFT JOIN %s M on M.album_code = S.UPC and M.isrc = S.isrc", stbl.release_meta, upcCol.release_meta, stbl.spotify_meta)

qry_inner <- makeQry(colsToPull=colsToPull.inner, tbl=tbl, schema=schema, aggFunc="count", colsWithaggFunc=colsWithaggFunc.inner, colsToAgg=colsToAgg.inner, minDate=minDate, maxDate=maxDate, dateCol=dateCol, join=NULL)

qry_count <- makeQry(colsToPull=colsToPull.outer, tbl=qry_inner, schema=NULL, colsToAgg=NULL, join=join.outer_to_release, expandStar=FALSE, limit=NULL, order=c("UPC", "weekof"))


### THE QUICK WAY 
### No metadata
{
  s.t({
    sfPopulateTable(tbl=tbl_new, schema=schema_new, qry=qry_count, overwrite=TRUE, transient=TRUE, just_qry=FALSE)
    print(headDB(tbl=tbl_new, schema=schema_new, snowflake=TRUE, n=12)[])
  })

  ## CONFIRM -----------------------------------------
  {
    tmp_DT.confirm <- makeQry(tbl=tbl_new, schema=schema_new, colsToPull=c("unknown_uri", "weekof"), colsToAgg="streams", limit=NULL, order=TRUE) %>% sfQry()
    setkeyIfNot(tmp_DT.confirm, unknown_uri, weekof, organize=TRUE, verbose=FALSE)
    ## All weeks should have a "known" & "unknown" set of uris
    if (any(tmp_DT.confirm[, .N, keyby=weekof][, N != 2]))
      warning ("Some weeks do not have two alues for unknown_uri")
    ## Every set should be exactly one week apart. Nothing in between
    if (!all(tmp_DT.confirm[, diff(weekof), by=unknown_uri]$V1 == 7))
      warning("The date-diff between each week (by unknown_uri) is NOT all 7.\n\nMeaning something went wrong, they are not exactly a week apart")

    ## Confirm that the number of streams is correct
    tmp.streams_from_new <- makeQry(tbl=tbl_new, schema=schema_new, colsToAgg="streams", minDate=minDate, maxDate=minDate+41, dateCol="weekof", order=TRUE) %>% sfQry()
    tmp.streams_from_orig <- makeQry(tbl=tbl, schema=schema, colsToAgg=c(streams="*"), minDate=minDate, maxDate=minDate+41, dateCol=dateCol, order=TRUE) %>% sfQry()
    stopifnot(unlist(tmp.streams_from_new) == unlist(tmp.streams_from_orig))
  }
  ## CONFIRM -----------------------------------------

# ~~~~~~~~~~~~~~~~~~~~~~ ENDS HERE? ~~~~~~~~~~~~~~~~~~~~
# ~~~~~~~~~~~~~~~~~~~~~~ ENDS HERE? ~~~~~~~~~~~~~~~~~~~~
# ~~~~~~~~~~~~~~~~~~~~~~ ENDS HERE? ~~~~~~~~~~~~~~~~~~~~

  if (FALSE)  ## This takes forever to run
    DT.DW <- sfQry(makeQry(tbl=tbl_new, schema=schema_new, minDate="2015-12-21", dateCol="weekof", colsToPul="*"))

  # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ MANUALLY SEARCH FOR SOME DW EXAMPLES ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

  ## EXAMPLE OF A FEW RANDOM ISRCS
  # getISRCMeta("BRDEP0900129", "CAAA10430105", "SEVFZ1105714", "BRPUI0900690")[]

  ## Manually run this
  DT.top_perf_examples <- get_some_top_performers(n=57, minDate=today()-20, min_accel_scaled=0.9, non_using_isrcs=non_using_isrcs)
  print(DT.top_perf_examples[unknown_uri == "unknown" & source == "others_playlist"], nrow=322)

  ## TEMP
  {
    DT2 <- DT.top_perf_examples[unknown_uri == "unknown" & source == "others_playlist"]
    writeDT(DT2, to=getRS())
  }
}

jesusForData(DT.top_perf_examples)

## Then pick an ISRC and pull the metadata and streams for it
isrc_using <- "CLGG29700005"
getISRCMeta(isrc_using)

## PULL A TOP LIST
DT.misc <- get_streams_for_ISRC(isrc_using)

## Save it, then bring it to local machine for plotting
if (FALSE) {
  jesusForData(DT.misc)
   "~rsaporta/git/orch/data/Discover_Weekly/DT.misc-20150923_0254-140x13.RDS" %>%
   loadFromJesus(over=TRUE)
}

## GRAPH IT 
graph_by_stream(DT.misc)

## Get metadata
getISRCMeta("CAU2O1010217")