# RV AIV - Ingest and Ship To Snowflake 20150716.r


### TODO:  
setScience("RetailVid_AIV", create=TRUE)

## DATE FORMAT
"2015/07/14 13:52:58"
frmt_time <- "%Y/%m/%d %H:%M:%S"
tz.guessed <- "UTC"

wh      <- getSnowflakeWH(); # "SCIENCE"
dbname  <- "prod"
schema  <- "Amazon"
tbl <- "RetailVid_AIV"
comment_tbl <- "AIV daily_paid file, generally sent via email to Retail Video team"

stage_name <- "ETLd_CSVs_for_snowflake_loading"
format_name <- "AIV_daily_paid"
delimiter <- ","
comment_format <- "For ingesting data from AIV daily_paid file, generally sent via email to Retail Video team"
comment_stage <- "For ingesting data from RAW CSV. Generally for quick feeding into Looker via Snowflake"

datetimestamp <- timeStamp(seconds=TRUE)


### -- CREATE FILE FORMAT & STAGES
"One Time"
if (FALSE)
{
  # setSnowflake(wh=wh, dbname=dbname, schema=schema, showWarnings=FALSE)
  sfCreateFormatAndStage(stage_name=stage_name, comment_stage=comment_stage, format_name=format_name, comment_format=comment_format, overwrite=FALSE, header=TRUE, delimiter=delimiter, gzip=FALSE, quote_string='"', wh=wh, dbname=dbname, schema=schema)

  sfQry(sprintf("DESC FILE FORMAT %s", dbschematbl(db=dbname, schema=schema, tbl=format_name)))
  sfQry(sprintf("DESC STAGE %s", dbschematbl(db=dbname, schema=schema, tbl=stage_name)))
}


# fbase <- dir(ingestDir)
file.zip <- ingest.p("Orchard_Digital_Music_Group_2015-07-14_new_detailed.zip")
stopifnot(file.exists(file.zip))

# ## FOR TESTING
# if (FALSE)
#   file.zip %<>% c("~/git/orch/ingest/RetailVid_AIV/Orchard_Digital_Music_Group_2015-07-14_new_detailed_02.zip")

## NOTE: This may not get the filename exactly
output_from_unzip <- file.zip %>% path.expand %>% shellClean %>% sprintf("unzip -o %s -d %s", ., shellClean(ingestDir)) %>% lapply(system, intern=TRUE)
file.unzip <- output_from_unzip %>% sapply(function(x) gsub("^.+?: ", "", x[[2]]) %>% trim)

## file.unzip will be path.unexpand'ed

if (any(!file.exists(file.unzip))) {
  stop("some files don't exist")
}

## MOVE THE ZIP FILES TO ARCHIVE
zArchive(file.zip, createIfNotExists.folder=TRUE)


### -- CREATE TABLE
"One Time"
if (FALSE)
{
  DT.raw <- fread(file.unzip) %>% setnamesSpaceToUnderscore_ %>% setnamestolower
  dateCols <- getDateColNames(DT.raw) %>% setdiff("purchase_date")
  ## purchase_date is standard format.  fulfillment_date and availability_date is another
  DT.raw[, purchase_date := as.POSIXct(purchase_date, tz=tz.guessed)]
  DT.raw[, paste0(dateCols) := lapply(.SD, as.POSIXct, origin=.origin.utc, format=frmt_time), .SDcols=dateCols]

  intCols <- c("season_number", "episode_number", "rev_share")
  numericCols <- c("retail_price", "unit_cost")
  DT.raw[, (intCols) := lapply(.SD, as.integer), .SDcols=intCols]
  DT.raw[, (numericCols) := lapply(.SD, as.numeric), .SDcols=numericCols]

  ## There is some issue with title
  DT.raw[, title := rep("A", 3000) %>% pasteC]
  DT.raw[, studio := rep("A", 500) %>% pasteC]
  DT.raw[, vendor := rep("A", 500) %>% pasteC]

  ## THE FILE HAS MULTIPLE FILE FORMATS.  NOT SURE WHAT TO DO ABOUT THAT
  ## TEMP - due to multiple date formats
  dateCols_that_will_be_strings <- c("fulfillment_date", "availability_date")
  DT.raw[, (dateCols_that_will_be_strings) := lapply(.SD, function(x) as.character(x + 3)), .SDcols=dateCols_that_will_be_strings] # the "+3" is because POSIX at 0hrs, drops the time
  setnames(DT.raw, dateCols_that_will_be_strings, paste0(dateCols_that_will_be_strings, "_string"))

  Qry_create <- makeSQLtable(DT.raw, table.name=tbl, dbname=dbname, schema=schema, replace=FALSE, datetime_type="TIMESTAMP_TZ", snowflake_inuse=TRUE, comment=comment_tbl, create_only=TRUE,  quiet=FALSE)

  setSnowflake(wh=wh, dbname=dbname, schema=schema, showWarnings=FALSE)
  ## CREATE THE SCHEMA IF NOT EXIST
  if (!(sfQry("SHOW SCHEMAS", verbose=FALSE, dbname=dbname) %>% {toupper(schema) %in% .$name}))
    paste0("CREATE SCHEMA ", dbschematbl(dbname=getSnowflakeDB(), schema=schema)) %>% sfQry(verbose=FALSE)

  ## SET TO SNOWFLAKE

  ## CREATE TABLE
  setSnowflake(wh=wh, dbname=dbname, schema=schema)
  sfQry(Qry_create, wh=wh, dbname=dbname, schema=schema)

  stopifnot(c(tbl) %>% toupper %in% sfShowTables(schema=schema, verbose=FALSE)$name)
}



## UPLOAD THE FILE & POPULATE THE TABLE
sfPopulateTableFromFile(file=file.unzip, tbl=tbl, stage_name=stage_name, validation_mode=FALSE, delete_files_after_load=FALSE, wh=wh, dbname=dbname, schema=schema, verbose.rows=TRUE, verbose=TRUE)


headDB(tbl=tbl, dbname=dbname, schema=schema, n=12)