# Ingest UPC Candidates.r



.us()
setScience("EU_PricingTest_2015", create=TRUE, subl=FALSE, load=FALSE)
setDBall(cluster = 09)

lib(reshape2, quiet=TRUE)
lib(Hmisc,    quiet=TRUE)

countries_using <- c(GB=3, DE=4, FR=6, ES=15, IE=90)
# runQry(paste0("SELECT * FROM dim_country where Country_code in ", pasteQ(names(countries_using), C=", "), " ORDER BY countryid"))
# runQry(paste0("SELECT * FROM dim_country where countryid in ", pasteQ(countries_using, C=", "), " ORDER BY countryid"))



### ------------------------------------------------------------------------- ###
###                    INGEST
### ------------------------------------------------------------------------- ###

f.apprvd_titles <- "FINAL EU iTUNES PRICE ANALYSIS TITLE LIST - APPROVED TITLES 20150205_1201.csv"
f.owned_and_tvt <- "FINAL EU iTUNES PRICE ANALYSIS TITLE LIST - Owned Catalog +1 Sale in 2014  20150205_1217.csv"

DT.apprvd_titles <- fread(ingest.p(f.apprvd_titles), colClasses = "character")[!(UPC == "")]
DT.owned_and_tvt <- fread(ingest.p(f.owned_and_tvt), colClasses = "character")[!(UPC == "")]
setInfo(DT.apprvd_titles, sprintf("fread from\n'%s'", f.apprvd_titles))
setInfo(DT.owned_and_tvt ,  sprintf("fread from\n'%s'", f.owned_and_tvt))

## Add row index number to preserve the order
DT.apprvd_titles[, .rowindex := seq(.N)]
DT.owned_and_tvt[, .rowindex := seq(.N)]

desc(DT.apprvd_titles)
desc(DT.owned_and_tvt)

UPCs.apprvd_titles <- unique(DT.apprvd_titles$UPC)
UPCs.owned_and_tvt <- unique(DT.owned_and_tvt$UPC)

if (L.err <- length(intersect(UPCs.apprvd_titles, UPCs.owned_and_tvt)))
  stop ("There are ", L.err, " common titles between 'owned_and_tvt' and 'apprvd_titles'")

## Victory Records will get special treatment. Spearate their UPCs
UPCs.victory <- DT.apprvd_titles[grepl("Victory", Label, ignore.case=TRUE), unique(UPC)]

### ------------------------------------------------------------------------- ###


### ------------------------------------------------------------------------- ###
###                     ADD RELEASE DATE
### ------------------------------------------------------------------------- ###

DT.rel_dates.apprvd_titles <- runQry(makeQry(tbl="analytics", schema="bi", distinct=TRUE, colsToPull=c(UPC="releaseid", releasedate="release_releasedate"), colsToAgg=NULL, whereIn=list(releaseid = UPCs.apprvd_titles), limit=NULL))
DT.rel_dates.owned_and_tvt <- runQry(makeQry(tbl="analytics", schema="bi", distinct=TRUE, colsToPull=c(UPC="releaseid", releasedate="release_releasedate"), colsToAgg=NULL, whereIn=list(releaseid = UPCs.owned_and_tvt), limit=NULL))

## Clean up names
setnames(DT.rel_dates.apprvd_titles, "upc", "UPC")
setnames(DT.rel_dates.owned_and_tvt, "upc", "UPC")

## Convert UPC to character, for joining with CSV data
DT.rel_dates.apprvd_titles[, UPC := as.character(UPC)]
DT.rel_dates.owned_and_tvt[, UPC := as.character(UPC)]

## Confirm same number of rows
stopifnot(nrow(DT.rel_dates.apprvd_titles) == nrow(DT.apprvd_titles))
stopifnot(nrow(DT.rel_dates.owned_and_tvt) == nrow(DT.owned_and_tvt))

## add in a count for max weeks since release.  Rounding down
DT.rel_dates.apprvd_titles[, weeks_released_by20150206 := as.numeric(today() - releasedate) %/% 7, by=releasedate]
DT.rel_dates.owned_and_tvt[, weeks_released_by20150206 := as.numeric(today() - releasedate) %/% 7, by=releasedate]

## Define as Frontline or Catalog
## Frontline is upto 18 months or 78 weeks.   Cutoff is Frontline := 0 < w <= 78;    Catalog := 78 < w <= Inf
DT.rel_dates.apprvd_titles[, category := ifelse(weeks_released_by20150206 > 78, "Catalog", "Frontline")]
DT.rel_dates.owned_and_tvt[, category := ifelse(weeks_released_by20150206 > 78, "Catalog", "Frontline")]

## Add the info back into the original DT that stemmed from the CSV
addColsFrom_(DT.apprvd_titles, DT.rel_dates.apprvd_titles, joinCols="UPC")
addColsFrom_(DT.owned_and_tvt, DT.rel_dates.owned_and_tvt, joinCols="UPC")
### ------------------------------------------------------------------------- ###


### ------------------------------------------------------------------------- ###
###                     GET COUNTS  
### ------------------------------------------------------------------------- ###


dateCol    <- "download_activity_date"
colsToPull <- c(date=dateCol, "releaseid", "storeid", transac_typeid = "transactiontypeid")
colsToAgg  <- c("paidunits", "royaltydollar")
minDate    <- as.Date("2014-09-02") 

kCols <- c("date", "releaseid")
kCols.rst <- c("releaseid", "storeid", "transac_type_abbr")
kCols.tr <- c("transac_type_abbr", "releaseid")

## Create Queries
Q.apprvd_titles <- makeQry(tbl="fact_analytics", schema="production", colsToPull=colsToPull, colsToAgg=colsToAgg, dateCol=dateCol[[1]], minDate=minDate, whereIn=list(releaseid = UPCs.apprvd_titles, countryid=countries_using), key=kCols)
Q.owned_and_tvt <- makeQry(tbl="fact_analytics", schema="production", colsToPull=colsToPull, colsToAgg=colsToAgg, dateCol=dateCol[[1]], minDate=minDate, whereIn=list(releaseid = UPCs.owned_and_tvt, countryid=countries_using), key=kCols)

## Excute query
BackUpOrRestore("DT.counts.apprvd_titles", clear=TRUE)
BackUpOrRestore("DT.counts.owned_and_tvt", clear=TRUE)
(DT.counts.apprvd_titles <- runQry(Q.apprvd_titles, cluster=9))
(DT.counts.owned_and_tvt <- runQry(Q.owned_and_tvt, cluster=9))

## Take backup
BackUpOrRestore("DT.counts.apprvd_titles")
BackUpOrRestore("DT.counts.owned_and_tvt")

## Convert UPCs to character
DT.counts.apprvd_titles[, releaseid := as.character(releaseid)]
DT.counts.owned_and_tvt[, releaseid := as.character(releaseid)]

## add in a week_number
DT.counts.apprvd_titles [, week_number := weekday_decimal(date)]
DT.counts.owned_and_tvt [, week_number := weekday_decimal(date)]

## count the days in the week  (This will only be != 7 for the weeks at the boundaries of the queries and the year)
DT.counts.apprvd_titles [, days_in_week := as.numeric(lunique(date)), by=week_number]
DT.counts.owned_and_tvt [, days_in_week := as.numeric(lunique(date)), by=week_number]

## Add transaction details
addTransacInfo_ (DT.counts.apprvd_titles, description=FALSE)
addTransacInfo_ (DT.counts.owned_and_tvt, description=FALSE)
### ------------------------------------------------------------------------- ###


### ------------------------------------------------------------------------- ###
###        Aggregate by Week and Sort into groups
### ------------------------------------------------------------------------- ###

## Aggregate by week, ignoring any incomplete week
func_weeklyAvg <- function(x) sum(x, na.rm=TRUE) / 7
DT.weekly_avg.apprvd_titles <- aggregateDT(DT.counts.apprvd_titles[days_in_week >= 7], colsToAgg=colsToAgg, exclude=c("date"), by=NULL, aggFunc=func_weeklyAvg)
DT.weekly_avg.owned_and_tvt <- aggregateDT(DT.counts.owned_and_tvt[days_in_week >= 7], colsToAgg=colsToAgg, exclude=c("date"), by=NULL, aggFunc=func_weeklyAvg)

## Clean up any new years weeks
DT.weekly_avg.apprvd_titles[days_in_week > 7, c(colsToAgg) := lapply(.SD, function(x) x * (7 / days_in_week) ), .SDcols=colsToAgg, by=days_in_week]
DT.weekly_avg.owned_and_tvt[days_in_week > 7, c(colsToAgg) := lapply(.SD, function(x) x * (7 / days_in_week) ), .SDcols=colsToAgg, by=days_in_week]
### ------------------------------------------------------------------------- ###