## EXAMPLE
if (FALSE)
  getISRCMeta("BRDEP0900129", "CAAA10430105", "SEVFZ1105714", "BRPUI0900690", drop_parent = TRUE)[]

get_streams_for_ISRC <- function(isrc) {
    DT <- makeQry(showWarnings=FALSE, tbl=tbl_new, schema=schema_new, colsToPull="*", limit=12000, where=list(isrc = isrc)) %>% sfQry %>% setkey(track_id, source, unknown_uri, weekof)
    DT[, source2 := ifelse(source=="others_playlist" & unknown_uri == "unknown", "unknown_others_playlist", source)]
    DT[source2 %in% c("album", "artist", "search"), source2 := "album+artist+search"]
    return(DT[])
}


graph_by_stream <- function(DT) {
  title <- DT[1, sprintf("%s (%s)", track_artists, isrc)]
  ggLinegraph(DT[, list(streams=sum(streams)), by=list(source2, weekof, track_id)], x="weekof", y="streams", color="source2", facet_y="track_id", title=title)

  ## New Option for graphing multiple
  ggLinegraph(DT.12_ISRCS, x="weekof", y="streams", facet_y=rev(colsToBring), title=title, y_facet_relative_text=.55, facet_scales="free_y")

}

get_some_top_performers_OLD <- function(n=25, minDate=as.Date("2015-12-01"), min_streams=1000, min_diff=500, min_prev_streams=80, non_using_isrcs = c("GBPS81523905", "US2H51200724"), verbose=TRUE) {
### This was the version of the function before I added the metadata directly to the table weekly_counts_by_track_and_source
### The new version filters directly during the original query and does not need to pull directly from the meta

  tmp_DT <- makeQry(
            showWarnings=FALSE
          , tbl=tbl_new
          , schema=schema_new
          , colsToPull=c("weekof", "upc", "track_id", "streams", "perc_increase")
          , limit=min(max(10000, n * 4), 5000000)
          , where=list(
                "perc_increase > 8"
              , sprintf("weekof > '%s'", minDate)
              , sprintf("streams > %s", min_streams)
              , sprintf("diff_in_streams > %s", min_diff)
              , sprintf("lag_streams > %s", min_prev_streams)
              , unknown_uri="unknown"
              , sprintf("NOT isrc IN %s", pasteQ(non_using_isrcs))
              , source="others_playlist"
              )
          ) %>% sfQry(verbose=FALSE) %>% setIDCols(verbose=FALSE)

  ## Add Metadata, such as sc_group and is_dthree
  ## ------------------------------------------------ 
  verboseMsg(verbose, "Grabbing metadata for UPCs")
  tmp_DT.meta <- 
    makeQry(
        schema="bi", tbl="RELEASE_WITH_METADATA_VIEW"
      , colsToPull=c("upc"="releaseid", "label_sc_group", "label_is_dthree", "label_is_o_and_o", "labelid")
      , colsToAgg=NULL
      , key="upc"
      , releaseid=unique(tmp_DT$upc)
      , limit=NULL
      ) %>% 
    sfQry(verbose=FALSE) %>% setIDCols(verbose=FALSE)

  addColsFrom_(tmp_DT, tmp_DT.meta, joinCols="upc")
  tmp_DT[, label_is_dthree  := as.logical(label_is_dthree)]
  tmp_DT[, label_is_o_and_o := as.logical(label_is_o_and_o)]
  ## ------------------------------------------------ 


  ## Grab the track_id's using, by grabbing the top 50% of streams, then sorting on perc_increase
  track_ids <- tmp_DT[label_sc_group=="Orchard" & (!label_is_dthree) & (!label_is_o_and_o)][streams > quantile(streams, .5)] [order(perc_increase, decreasing=TRUE)] [, unique(track_id) %>% head(n=n)]

  ## pull the details based on track_id's selected
  ret_DT <- 
  makeQry(
        showWarnings=FALSE
      , tbl=tbl_new
      , schema=schema_new
      , colsToPull="*"
      , limit=5000000
      , where=list(
            track_id = track_ids
          , sprintf("NOT isrc IN %s", pasteQ(non_using_isrcs))
        )
     , orderby=c("upc", "track_id", "isrc", "weekof")) %>% 
  sfQry(verbose=verbose) %>%
  setIDCols(verbose=FALSE) %>%
  setkeyIfNot(upc, track_id, track_uri, isrc, weekof, unknown_uri, source, organize=TRUE, verbose=FALSE)

  ## Add the metadata from perviously pulled
  addColsFrom_(ret_DT, tmp_DT.meta, joinCols="upc")
  ret_DT[, label_is_dthree  := as.logical(label_is_dthree)]
  ret_DT[, label_is_o_and_o := as.logical(label_is_o_and_o)]


  ## This should expand by date.  However, not sure if it is needed, as why would a date be missing, except at the tails, unless it is missing from the whole DT which is a bigger problem
  if (FALSE) {
    dates <- ret_DT[, seq.Date(from=min(weekof), to=max(weekof), by="7 days")]
    DT.dates <- ret_DT[, list(weekof=dates), keyby=c(setdiff(key(ret_DT), "weekof"))] %>% setkeyIfNot(key(ret_DT), organize=TRUE, verbose=FALSE)
    ret_DT <- ret_DT[DT.dates]
  }

  ret_DT[, unknown_uri := toFactorWithExpectedLevels(unknown_uri, levels=c("unknown", "known"))]

  return(ret_DT[])
}


get_some_top_performers_OLD_2 <- function(n=25, minDate=as.Date("2015-12-01"), min_streams=1000, min_diff=500, min_prev_streams=80, label_sc_group="Orchard", label_is_dthree=0, non_using_isrcs = c("GBPS81523905", "US2H51200724"), verbose=TRUE) {
### This was the version of the function before I added the metadata directly to the table weekly_counts_by_track_and_source
### The new version filters directly during the original query and does not need to pull directly from the meta

  tmp_DT <- makeQry(
            showWarnings=FALSE
          , tbl=tbl_new
          , schema=schema_new
          , colsToPull=c("weekof", "upc", "track_id", "streams", "perc_increase")
          , limit=min(max(10000, n * 10), 5000000)
          , where=list(
                "perc_increase > 8"
              , sprintf("weekof > '%s'", minDate)
              , sprintf("streams > %s", min_streams)
              , sprintf("diff_in_streams > %s", min_diff)
              , sprintf("lag_streams > %s", min_prev_streams)
              , unknown_uri="unknown"
              , sprintf("NOT isrc IN %s", pasteQ(non_using_isrcs))
              , source="others_playlist"
              , label_sc_group = label_sc_group
              , label_is_dthree = label_is_dthree
              )
          ) %>% sfQry(verbose=FALSE)

  ## Grab the track_id's using, by grabbing the top 50% of streams, then sorting on perc_increase
  track_ids <- tmp_DT[streams > quantile(streams, .6)] [order(perc_increase, decreasing=TRUE)] [, unique(track_id) %>% head(n=n)]
  ## If we did not select enough track_ids, then simply sort on percent, without filtering by streams. Afterall, there is already a filter on streams in the function call
  if (length(track_ids) < n)
    track_ids <- tmp_DT[order(perc_increase, decreasing=TRUE)] [, unique(track_id) %>% head(n=n)]

  ## pull the details based on track_id's selected
  ret_DT <- 
  makeQry(
        showWarnings=FALSE
      , tbl=tbl_new
      , schema=schema_new
      , colsToPull="*"
      , limit=5000000
      , where=list(
            track_id = track_ids
          , sprintf("NOT isrc IN %s", pasteQ(non_using_isrcs))
        )
     , orderby=c("upc", "track_id", "isrc", "weekof")) %>% 
  sfQry(verbose=verbose) %>%
  setIDCols(verbose=FALSE) %>%
  setkeyIfNot(upc, track_id, track_uri, isrc, weekof, unknown_uri, source, organize=TRUE, verbose=FALSE)

  ## This should expand by date.  However, not sure if it is needed, as why would a date be missing, except at the tails, unless it is missing from the whole DT which is a bigger problem
  ## ------------------------------
  dates <- ret_DT[, seq.Date(from=min(weekof), to=max(weekof), by="7 days")]
  DT.dates <- ret_DT[, list(weekof=dates), keyby=c(setdiff(key(ret_DT), "weekof"))] %>% setkeyIfNot(key(ret_DT), organize=TRUE, verbose=FALSE)
  if (nrow(DT.dates) > nrow(ret_DT))
    warning("Not all dates are present for all values. Might want to CJ() or modify the get_some_top_performers() function")
  if (FALSE) {
    ret_DT <- ret_DT[DT.dates]
  }
  ## ------------------------------

  ## Set levels for ret_DT
  ret_DT[, unknown_uri := toFactorWithExpectedLevels(unknown_uri, levels=c("unknown", "known"))]

  return(ret_DT[])
}


~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
## TODO: 
USE This
NTILE( constant_value ) OVER ( [ window_partition_by_clause ] window_order_by_clause )
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

get_some_top_performers <- function(n=25
  , minDate=as.Date("2015-12-01")
  , min_streams=1000
  , min_perc_increase=1.2
  , min_diff=500
  , min_prev_streams=80
  , min_accel_scaled=0.89
  
  , label_sc_group="Orchard"
  , label_is_dthree=0
  , source_using = "others_playlist"
  , unknown_uri_using = "unknown"
  , non_using_isrcs = c("GBPS81523905", "US2H51200724")
  , verbose=TRUE
) {
### This was the version of the function before I added the metadata directly to the table weekly_counts_by_track_and_source
### The new version filters directly during the original query and does not need to pull directly from the meta

  qry.inner <- makeQry(
            showWarnings=FALSE
          , tbl=tbl_new
          , schema=schema_new
          , colsToPull=c(accel_rank="rank() OVER (order by accel_score_scaled DESC)", "track_id", "accel_score_scaled")
          , distinct = TRUE
          , limit=min(max(10000, n * 10), 5000000)
          # , limit=n
          , where=list(
                sprintf("weekof > '%s'", minDate)
              , sprintf("streams > %s", min_streams)
              , sprintf("perc_increase > %s", min_perc_increase)
              , sprintf("diff_in_streams > %s", min_diff)
              , sprintf("lag_streams > %s", min_prev_streams)
              # , sprintf("accel_score_scaled > %s", min_accel_scaled)
              , unknown_uri=unknown_uri_using
              , sprintf("NOT isrc IN %s", pasteQ(non_using_isrcs))
              , source=source_using
              , label_sc_group = label_sc_group
              , label_is_dthree = label_is_dthree
              )
          , order=c("accel_score_scaled DESC")
          )

  qry.middle <- makeQry(tbl=qry.inner, schema=NULL, colsToPull=c("track_id", "accel_rank"), distinct=TRUE, where=sprintf("accel_rank <= %s", n * 1.5), limit=n, order="accel_rank") %>% sprintf("SELECT track_id FROM (%s)", .)

  qry.outter <- makeQry(
        showWarnings=FALSE
      , tbl=tbl_new
      , schema=schema_new
      , colsToPull="*"
      , limit=5000000
      , expandStar = FALSE
      , where=list(
            sprintf("track_id in (%s)", qry.middle)
          , sprintf("NOT isrc IN %s", pasteQ(non_using_isrcs))
        )
     , orderby=c("upc", "isrc", "track_id", "weekof")
     )

  ## pull the details based on track_id's selected
  ret_DT <- sfQry(qry.outter, verbose=verbose) %>%
              setIDCols(verbose=FALSE) %>%
              setkeyIfNot(upc, track_id, track_uri, isrc, weekof, unknown_uri, source, organize=TRUE, verbose=FALSE)

  ## This should expand by date.  However, not sure if it is needed, as why would a date be missing, except at the tails, unless it is missing from the whole DT which is a bigger problem
  ## ------------------------------
  dates <- ret_DT[, seq.Date(from=min(weekof), to=max(weekof), by="7 days")]
  DT.dates <- ret_DT[, list(weekof=dates), keyby=c(setdiff(key(ret_DT), "weekof"))] %>% setkeyIfNot(key(ret_DT), organize=TRUE, verbose=FALSE)
  if (nrow(DT.dates) > nrow(ret_DT))
    warning("Not all dates are present for all values. Might want to CJ() or modify the get_some_top_performers() function")
  if (FALSE) {
    ret_DT <- ret_DT[DT.dates]
  }
  ## ------------------------------

  ## Set levels for ret_DT
  ret_DT[, unknown_uri := toFactorWithExpectedLevels(unknown_uri, levels=c("unknown", "known"))]

  return(ret_DT[])
}


&&&&  LEFT OFF HERE .... ADD Metdata to 
paste_info_string_top_performer(DT.top_perf_examples)[, catn(V1, sep="\n\n")]
paste_info_string_top_performer <- function(DT, label_included=TRUE) {
  DT[, j=tail(.SD, 5)[, {
     scores <- sprintf("%2.1f", 100*accel_score_scaled) %>% pasteC(C=" | ")
     minDate <- format(minn(weekof), format="%Y-%b-%d")
     maxDate <- ifelse(minn(weekof) != maxn(weekof), format(maxn(weekof), format="%Y-%b-%d"), "")

     line1 <- sprintf("%s | %s | %s", artist_name[[1]], shorten(release_name[[1]], 20, thresh=5), shorten(track_name[[1]], 25, thresh=5) )
     line2 <- sprintf("%s (%s %s)", label_name[[1]], labelid[[1]], label_client_manager[[1]])
     line3 <- sprintf("%s  %s  %s", minDate, scores, maxDate)

     lines <- c(line1, if (label_included) line2, line3)
     pasteC(lines, C="\n")
   }], by=list(upc, isrc, source, unknown_uri), .SDcols=c("artist_name", "release_name", "track_name", "label_name", "labelid", "label_client_manager", "accel_score_scaled", "weekof")]
}

plot_top_performers <- function(DT) {
&&&&  LEFT OFF HERE .... ADD Metdata to 
&&&&             crop on the left
&&&&             Add to looker
&&&&             Create a report

  
  Title = "Streams by Source"
  DT <- copy(DT)

  DT[1, ] %>% t
  ggLinegraph(DT, x="weekof", y="streams", color="source", facet_y=c("track_name_from_spotify", "artist_name", "label_name"), facet_scales="free_y", thickness=.65, title=Title, linetype="unknown_uri") + scale_linetype_discrete(c("solid", "longdash"))
}

if (FALSE) {
  DT.top_perf_examples <- get_some_top_performers(8)
  jesusForData(DT.top_perf_examples)
  # if (.Pfm == "Darwin")
  #   "~rsaporta/git/orch/data/Discover_Weekly/DT.top_perf_examples-20160117_1504-1366x15.RDS" %>%
  #   loadFromJesus(over=TRUE)
  P <- plot_top_performers(DT.top_perf_examples)
  f.out <- ggsave.out(plot=P, nm="DW Candidates", height=15)
  .o(f.out)
}