# streams_per_playlist.r


"This queries staging_raw_spotify_v2 to see how many streams are due to each playlist"

run_the_query <- TRUE

setScience("Spotify Playlist Webscrape", subl=FALSE, load=FALSE, create=FALSE)
loadIfNotExists("DT.pl")
loadIfNotExists("DT.streams_per_source")

options('db.all.pfm' = TRUE)
# cluster <- getOption("db.defaultcluster")
tbl <- "staging_raw_spotify_v2"
schema <- "production"

cluster <- 7
qShowCols(tbl, schema=schema)

minDate <- monthCeiling(today() - 125)

colsToPull.all <- c("download_date", "upc", artist="track_album_artist", "album_name", age = "2015 - birth_year"
                            , "source_uri", "product", "user_region", "user_country", "gender")
colsToPull <- c("source_uri")

if (run_the_query && !exists("DT.streams_per_source")) {
  DT.streams_per_source <- {
        makeQry(tbl=tbl
              , schema=schema
              , colsToAgg=c(streams="*")
              , colsToPull=colsToPull
              , whereIn=c("(NOT source_uri is NULL) AND (NOT source_uri = '')")
              , minDate=minDate
              , dateCol="download_date"
              , limit = NULL
            ) %>% 
        runQry(cluster=cluster, verbose=TRUE, allow.large.groupby=TRUE)
  }
  jesusForData(DT.streams_per_source)
}


DT.streams_per_source[, rank := rank(-streams)]
DT.top_playlists <- DT.streams_per_source[rank >= 50]
setnames(DT.top_playlists, "source_uri", "uri")

addColsFrom_(DT.pl, DT.top_playlists, joinCols="uri")


DT.top_with_metadata <- DT.pl[uri %in% DT.top_playlists$uri][!grepl("spotify", owner.id)]


DT.our_tops <- {
    DT.top_with_metadata[, list(our_tracks = sum(track.is_orchard_dist)
                  , all_tracks=.N
                  , perc=percTrue(track.is_orchard_dist)
                  , numb_of_albums = lunique(track.album.name)
                  , followers=total_followers[[1]]
                  , streams=streams[[1]]
                  # , rank=rank[[1]]
                )
          , by=list(uri, owner.id, id, playlist_name=name)]
}

DT.our_tops