# create_DT.acc_v_anal.r

fileSelfDescription(
"creates  DT.acc_v_anal  from the combination of three tables:
  *  DT.totalpaidunits.acc   (raw pull from fact_sales)
  *  DT.totalpaidunits.anal  (raw pull from fact_analytics)
  *  DT.totalpaidunits.aErr  (raw pull from fact_analytics_error)
(the latter two are rbind then merged with the first)
")


create_DT.acc_v_anal <- function(
  assign.to="DT.acc_v_anal"
, envir.to.assign=globalenv()
, include_country   = TRUE
, include_trans     = TRUE
, include_label     = FALSE
, include_sc_group  = FALSE
, where = NULL
, where.acc  = where
, where.anal = where
, minDate = as.Date("2012-01-01")
, maxDate = monthFloor(today() - 60)
, storeid_to_ignore = c(Hulu=392, YouTube=453, "YouTube Movies"=463)
, transid_to_ignore = c(33:36)
, countryid_to_ignore = c()
, stores_only = NULL
, jesus = re_pull_data
, confirm = (!dont_crop_at_maxDate)
, stores_confirm = c(1, 7, 187, 348, 286)
, mnD_confirm = as.Date("2014-01-01")
, mxD_confirm = as.Date("2014-12-31")
, re_pull_data = TRUE
, dont_crop_at_maxDate = FALSE # make TRUE for predicting
, projName = "Acc_vs_Anal_2015"
, subProj = NULL
, cluster.using = 09
, verbose.for_qry = verbose
, verbose = TRUE
) {

  ##  repull : whether to repull the raw data or load from disk
  ##  jesus : whether or not to save the intermediate DTs

  ## set project if needed
  setScienceIfNot(projName=projName, subProj=subProj, load=FALSE)

  if (dont_crop_at_maxDate && confirm) {
    message("When dont_crop_at_maxDate is TRUE, the confirmation steps will fail, hence setting that to FALSE")
    confirm <- FALSE
  }

  if (isTRUE(include_sc_group)) {
    warning("Cannot include_sc_group since that requires bi.accounting (we are currently using fact_sales\nRequired step: Confirm numbers are constant between the two tables then switch over to the other table\n\n .")
    Sys.sleep(3)
    include_sc_group <- FALSE
  }

  ## 
  envir.working <- environment()

  ## Set cluster if necessary
  previous.cluster <- getCluster()
  if (!identical(previous.cluster, previous.cluster)) {
    if (!is.null(previous.cluster))
      warning (sprintf("cluster will be changed from %02i to %02i", as.numeric(previous.cluster), as.numeric(cluster.using) ))
    setDBall(cluster=cluster.using)
  }

  ## Column info
  ## ---------------------------------- ##
  dateCol.acc  <- "activityperiodid"
  dateCol.anal <- "download_activity_date"
  colsToPull   <- c("storeid")
  if (include_trans)     colsToPull <- c(colsToPull, transid="transactiontypeid")
  if (include_country)   colsToPull <- c(colsToPull, "countryid")
  if (include_label)     colsToPull <- c(colsToPull, "labelid")
  if (include_sc_group)  colsToPull <- c(colsToPull, "label_sc_group")

  colsToAgg.acc  <- c(revenue="gross", paidunits = "sales")
  colsToAgg.anal <- c(revenue="royaltydollar", "paidunits")

  kCols <- c("date", colNamesFromVector(colsToPull))

  ## the confirm dates cannot exceed the min/max dates
  mnD_confirm <- max(minDate, mnD_confirm)
  mxD_confirm <- min(maxDate, mxD_confirm)

  if (dont_crop_at_maxDate) {
    ## use the string "NULL" to not mess up sprintf
    mxD_confirm <- maxDate <- NULL
  }

  if (!is.null(stores_only)) {
    where.acc  <- c(where.acc,  storeid = stores_only)
    where.anal <- c(where.anal, storeid = stores_only)
    stores_confirm <- if (length(intersect(stores_only, stores_confirm))) intersect(stores_only, stores_confirm) else stores_only
  }

  ## meta info for saving objects
  if (is.null(subProj))
    subProj <- pasteC(c(if (include_country) "byCountry", if (include_trans) "byTrans", if (include_label) "byLabel", if (include_sc_group) "bySCgroup"), C="_") 
  jinfo <- sprintf("%sminDate%s_and_maxDate%s", ifelse(nchar(subProj), paste0(subProj, "_"), ""), format(minDate, "%Y%m%d"), if (is.null(maxDate)) "NULL" else format(maxDate, "%Y%m%d"))
  if (!is.null(stores_only))
    jinfo <- paste0(jinfo, "_storeidIN", pasteC(stores_only, C="."))

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

  browser(expr=inDebugMode(c("create_DT.acc_v_anal")), text="in create_DT.acc_v_anal() at top before pulls")

  # ------------------------------------------------------------------------------------------- #
  # --------------------------------------------------------------------------------- #
  #    DATA PULL                                                                      # 
  # --------------------------------------------------------------------------------- #
    rawPull.jinfo <- pasteC(c("rawPull", jinfo), C="_")
    if (re_pull_data) {
      Q.acc  <- makeQry(tbl="fact_sales",     schema="production", colsToPull=c(dateCol.acc,  colsToPull), colsToAgg=colsToAgg.acc,  where=where.acc,  minDate=minDate, maxDate=maxDate, dateCol=dateCol.acc)
      Q.anal <- makeQry(tbl="fact_analytics", schema="production", colsToPull=c(dateCol.anal, colsToPull), colsToAgg=colsToAgg.anal, where=where.anal, minDate=minDate, maxDate=maxDate, dateCol=dateCol.anal)
      Q.aErr <- convertQry.fa_to_faerrors(Q.anal)

      DT.totalpaidunits.acc  <- runQry(Q.acc,  cluster=cluster.using, dont.setkey=TRUE, verbose=verbose.for_qry)
      DT.totalpaidunits.anal <- runQry(Q.anal, cluster=cluster.using, dont.setkey=TRUE, verbose=verbose.for_qry)
      DT.totalpaidunits.aErr <- runQry(Q.aErr, cluster=cluster.using, dont.setkey=TRUE, verbose=verbose.for_qry)

      storeid_in_analytics <- sort(intersect(DT.totalpaidunits.acc$storeid, DT.totalpaidunits.anal$storeid))

      ## Dont Jesus if stores_only
      if (is.null(stores_only)) {
        jesusForData(DT.totalpaidunits.acc , info=rawPull.jinfo, envir=envir.working, doNothing=!isTRUE(jesus), git=FALSE)
        jesusForData(DT.totalpaidunits.anal, info=rawPull.jinfo, envir=envir.working, doNothing=!isTRUE(jesus), git=FALSE)
        jesusForData(DT.totalpaidunits.aErr, info=rawPull.jinfo, envir=envir.working, doNothing=!isTRUE(jesus), git=FALSE)
      }

    } else {
      loadIfNotExists(DT.totalpaidunits.acc,  colsNeeded=colNamesFromVector(colsToPull), envir=envir.working, info=rawPull.jinfo, verbose=TRUE)
      loadIfNotExists(DT.totalpaidunits.anal, colsNeeded=colNamesFromVector(colsToPull), envir=envir.working, info=rawPull.jinfo, verbose=TRUE)
      loadIfNotExists(DT.totalpaidunits.aErr, colsNeeded=colNamesFromVector(colsToPull), envir=envir.working, info=rawPull.jinfo, verbose=TRUE) 
    }  


    ## if older version existed, it might have additional columns which will mess up the joins and hence crash
    confirm_col_not_present <- function(DT, flags=c(colName="logical_flag_value")) {
        if (flags[[1]] == "logical_flag_value")  stop ("you forgot to set the flags in confirm_col_not_present()")
        as_expected <- sapply(names(flags), function(col) flags[col] || !(col %in% names(DT)) )
        if (any(!as_expected)) {
          DT.nm <- capture.output(substitute(DT))
          stop (DT.nm, " contains columns that it shold not. This is likely to a previous load with different flags set.\n  HINT: suggested fixes are  \n\t\t(1) set  re_pull_data=TRUE \n\t\t(2) try rm(", DT.nm, ") so that the DT is reloaded from jesus or \n\t\t(3) Aggregate out the column using aggregateDT()")
        }
        return(invisible(TRUE))
    }
    flags <- c(countryid=include_country, transid=include_trans, labelid=include_label, label_sc_group=include_sc_group)
    confirm_col_not_present(DT.totalpaidunits.acc,  flags=flags)
    confirm_col_not_present(DT.totalpaidunits.anal, flags=flags)
    confirm_col_not_present(DT.totalpaidunits.aErr, flags=flags)


    ## Confirm DT is properly sized.  NOTE: This should happen in the loadFromJesus portion
    if(!truelength(DT.totalpaidunits.acc))  setDT(DT.totalpaidunits.acc)
    if(!truelength(DT.totalpaidunits.anal)) setDT(DT.totalpaidunits.anal)
    if(!truelength(DT.totalpaidunits.aErr)) setDT(DT.totalpaidunits.aErr)

    # BackUpOrRestore("DT.totalpaidunits.acc",  forceBak=TRUE)
    # BackUpOrRestore("DT.totalpaidunits.anal", forceBak=TRUE)
    # BackUpOrRestore("DT.totalpaidunits.aErr", forceBak=TRUE)

    # BackUpOrRestore("DT.totalpaidunits.acc")
    # BackUpOrRestore("DT.totalpaidunits.anal")
    # BackUpOrRestore("DT.totalpaidunits.aErr")
  # ------------------------------------------------------------------------------------------- #
  # ------------------------------------------------------------------------------------------- #


  # ------------------------------------------------------------------------------------------- #
  # --------------------------------------------------------------------------------- #
  #    Clean The Data                                                                 # 
  # --------------------------------------------------------------------------------- #
    ## MODIFY ANALYTICS, CREATING   DT.anals
    ## Anals == anal + aErr
    DT.anals <- combineAnalWithAErr(DT.totalpaidunits.anal, DT.totalpaidunits.aErr, fromErr=FALSE)
    DT.anals <- changeAndAggregate(DT.anals, colsToAgg=colNamesFromVector(colsToAgg.anal), colsToDrop=c("download_activity_date"), changeFunc=changeFunc.anals544, browseOnFail=TRUE)
    # jesusForData(DT.anals, info=jinfo, envir=envir.working, doNothing=!isTRUE(jesus), git=FALSE)

    ## MODIFY ACCOUNTING
    if ("activityperiodid" %in% names(DT.totalpaidunits.acc))
          setnames(DT.totalpaidunits.acc, "activityperiodid", "periodid")
    addDateCols.periodid_(DT.totalpaidunits.acc, drop=TRUE, showWarnings=FALSE)

    ## suffixes and the main columns
    suffixes <- c(".acc", ".anal")
    revCols   <- paste0("revenue",   suffixes)
    unitsCols <- paste0("paidunits", suffixes)
    revAndUnitsCols <- c(revCols, unitsCols)


    ## Debugging, middle
    browser(expr=inDebugMode("create_DT.acc_v_anal", "create", "create_DT.acc_v_anal.middle"), text="in create_DT.acc_v_anal MIDDLE after aggregating DT.anals")

    # PREP FOR MERGING
    matchKey(DT.totalpaidunits.acc, DT.anals, key=kCols, organize=TRUE)

    ## RM PREVIOUS VERSION, JUST IN CASE
    if (exists(assign.to, envir=envir.to.assign, inherit=FALSE)) {
        message("removing previous version of ", assign.to, " in environment ", cleanEnvirString(envir.to.assign))
        rm(list=assign.to, envir=envir.to.assign)
    }

    ## MERGE, CREATING DT.acc_v_anal
    ## SIMULTANEOUSLY:  ASSIGN INTO WORKING ENVIRONMENT  and ASSIGN IT LOCALLY TO  DT.acc_v_anal
    DT.acc_v_anal <- 
      assignWithInfo(
            name = assign.to
          , value = merge(DT.totalpaidunits.acc, DT.anals, suffix=suffixes, all=TRUE, allow=TRUE, envir=envir.to.assign)
          , info  = padTo("Merge of DT.anals and DT.totalpaidunits.acc\nDate from acc is activityperiodid\nDate from anal is download_activity_date (544 Adjusted)", 55)
          , envir = envir.to.assign, warnOnEnvir=FALSE
      )
  # ------------------------------------------------------------------------------------------- #
  # ------------------------------------------------------------------------------------------- #


  # ------------------------------------------------------------------------------------------- #
  # --------------------------------------------------------------------------------- #
  #    Confirm cols are expected                                                      # 
  # --------------------------------------------------------------------------------- #
  if (isTRUE(confirm))
  {
    byCols.confirm <- intersect(c("storeid", "transid"), colNamesFromVector(colsToPull))
    cat("Confirming that the values from ", assign.to, " are the same as in the raw pulls ... ")
    ## DT.totalpaidunits.acc
    stopifnot(confirmSameValue(DT.totalpaidunits.acc, DT.acc_v_anal, colsToCompare="paidunits", suffix=suffixes, by=byCols.confirm, verbose=FALSE)$.equal)
    cat(".. ")
    stopifnot(confirmSameValue(DT.totalpaidunits.acc, DT.acc_v_anal, colsToCompare="revenue",   suffix=suffixes, by=byCols.confirm, verbose=FALSE)$.equal)
    ## DT.anals
    stopifnot(confirmSameValue(DT.anals, DT.acc_v_anal, colsToCompare="paidunits", suffix=".anal", by=byCols.confirm, verbose=FALSE)$.equal)
    cat(".. ")
    stopifnot(confirmSameValue(DT.anals, DT.acc_v_anal, colsToCompare="revenue",   suffix=".anal", by=byCols.confirm, verbose=FALSE)$.equal)
    cat("..   [ALL OK]\n")

    ## SECOND CONFIRMATION
    ## -------------------------------------------- ##
    {
      cat("Confirming against new pulls from the DB ")
      Q.fs.confirm <- makeQry(tbl="fact_sales",     colsToAgg=c(paidunits="sales",     revenue="gross"),         dateCol="activityperiodid",       colsToPull=c("storeid"), whereIn=list(storeid=stores_confirm), minDate=mnD_confirm, maxDate=mxD_confirm)
      Q.fa.confirm <- makeQry(tbl="fact_analytics", colsToAgg=c(paidunits="paidunits", revenue="royaltydollar"), dateCol="download_activity_date", colsToPull=c("storeid"), whereIn=list(storeid=stores_confirm), minDate=mnD_confirm, maxDate=mxD_confirm)
      Q.fR.confirm <- convertQry.fa_to_faerrors(Q.fa.confirm)

      DT.confirm.sales <- setkeyIfNot(runQry(Q.fs.confirm, verbose=verbose.for_qry), "storeid", verbose=FALSE)
      cat(".. ")
      DT.confirm.anal  <- setkeyIfNot(runQry(Q.fa.confirm, verbose=verbose.for_qry), "storeid", verbose=FALSE)
      cat(".. ")
      DT.confirm.aERr  <- setkeyIfNot(runQry(Q.fR.confirm, verbose=verbose.for_qry), "storeid", verbose=FALSE)
      cat(".. ")

      DT.confirm.anals <- aggregateDT(combineAnalWithAErr(DT.confirm.anal, DT.confirm.aERr, fromErr_col=FALSE), by="storeid")
      cat(".. \n")

      ## The values in DT.confirm.sales and DT.confirm.anals should be the same as in DT.acc_v_anal
      DT.confirm.What_I_have <- setcolorderpt(DT.acc_v_anal[date >= mnD_confirm & date <= mxD_confirm & storeid %in% stores_confirm, lapply(.SD, sumn), keyby=storeid, .SDcols=revAndUnitsCols], sort=TRUE)
      DT.confirm.What_DB_has <- setcolorderpt(merge(DT.confirm.sales, DT.confirm.anals, keyby="storeid", suffix=suffixes, all=TRUE, allow=TRUE), sort=TRUE)
      stopifnot(names(DT.confirm.What_I_have) == names(DT.confirm.What_DB_has))
      DT.confirm.mine_vs_DB <-  setkeyv(cbind(DT.confirm.What_DB_has[, list(STORE=storeid, "| " = "| ")], DT.confirm.What_DB_has - DT.confirm.What_I_have), c("STORE", "| "))
      DT.confirm.mine_vs_DB <- DT.confirm.mine_vs_DB[, lapply(.SD, function(x) ifelse(equals(0, x, tol=1e-3), 0, x)), keyby=key(DT.confirm.mine_vs_DB)]
      cat("\n    ----- These should all equal 0: ------  \n   (except for analytics-iTunes, which will be off by about 0.02% or 1/365 due to 544)\n   [    showing  'Correct' minus 'Mine'   ", as.character(mnD_confirm), " ~ ", as.character(mxD_confirm), "  ]\n\n"); print(DT.confirm.mine_vs_DB)
      Sys.sleep(3.5)
    }
    ## -------------------------------------------- ##
  } # // end isTRUE(Confrim)
  # ------------------------------------------------------------------------------------------- #
  # ------------------------------------------------------------------------------------------- #


  # ------------------------------------------------------------------------------------------- #
  # --------------------------------------------------------------------------------- #
  #    Organize and Calculate Percentages                                             # 
  # --------------------------------------------------------------------------------- #
    setcolorderpt(DT.acc_v_anal, startCols=kCols, sort.middle=TRUE)

    ## Calculate Percentage
    DT.acc_v_anal[, paidunits.perc.acc_of_anal := paidunits.acc / paidunits.anal]
    DT.acc_v_anal[, revenue.perc.acc_of_anal   := revenue.acc   / revenue.anal  ]
  # --------------------------------------------------------------------------------- #

  ## Backup
  if (is.null(stores_only))
    jesusForData(DT.acc_v_anal, info=jinfo, envir=envir.working, doNothing=!isTRUE(jesus), git=TRUE)

  ## Assign before returning
  if (!identical(DT.acc_v_anal, get(assign.to, envir=envir.to.assign, inherit=FALSE))) {
    cat("assigning results to ", assign.to, "\n")
    assign(assign.to, DT.acc_v_anal, envir=envir.to.assign)
  } else {
    cat(assign.to, " already has the correct value; no need to assign\n")
  }

  return(invisible(DT.acc_v_anal))
}