## SoundCloudIssue 01 - load delivery_history into SF.r

setScience("DeNormalizing", subProj="soundcloudissue")
setGitBranchToSystem();  .g()

wh <- "SCIENCE"
dbname <- "prod"
schema.out <- "delivery_history"

qry.upc_store_deliverydate <- 
"
SELECT 
  dh.upc, 
  dh.customer_master_master_id, 
  min(dh.date_delivered) AS min_date_delivered, 
  max(dh.date_delivered) AS max_date_delivered
FROM art_relations.delivery_history dh
GROUP BY 1, 2
"

qry.upc_ingestiondate <- 
"
SELECT 
  r.upc, 
  r.ingestion_completed
FROM art_relations.releases r
"


setDBall("MySQL")
d_date <- as.Date("2016-04-27")


DT.upc_store_deliverydate <- runQry(qry.upc_store_deliverydate, key=NULL)
DT.upc_ingestiondate <- runQry(qry.upc_ingestiondate, key=NULL)

jesusForData(DT.upc_store_deliverydate, info="RAWFromMySQL", dir=data.p("SoundCloudIssue"))   %>% print
jesusForData(DT.upc_ingestiondate, info="RAWFromMySQL", dir=data.p("SoundCloudIssue"))   %>% print


## THERE IS SOMETHING BUGGY: 
##   Different UPCs are being treated the same, as though it was being rounded
##   eg: 600116813916
##       600116813923
## 
##  Thus, converting to strings
if (!is.character(DT.upc_store_deliverydate$upc))  
      DT.upc_store_deliverydate[, upc := as.character(upc)]
if (!is.character(DT.upc_ingestiondate$upc))       
      DT.upc_ingestiondate[, upc := as.character(upc)]

## CONFIRM NO ERRORS
## There should be exactly one row per each store-upc combination
stopifnot(0 == nrow(DT.upc_store_deliverydate[, .N, keyby=list(upc, customer_master_master_id)] [N != 1]))

## confirm key has been set
setkeyIfNot(DT.upc_store_deliverydate, c("upc", "customer_master_master_id"), verbose=FALSE)
setkeyIfNot(DT.upc_ingestiondate, "upc", verbose=FALSE)


if (!is.Date(DT.upc_ingestiondate$ingestion_completed))
  DT.upc_ingestiondate[, ingestion_completed := as.Date(ingestion_completed)]
if (!is.Date(DT.upc_store_deliverydate$min_date_delivered))
  DT.upc_store_deliverydate[, min_date_delivered := as.Date(min_date_delivered)]
if (!is.Date(DT.upc_store_deliverydate$max_date_delivered))
  DT.upc_store_deliverydate[, max_date_delivered := as.Date(max_date_delivered)]

### Add in T/F field; better to do it once and save it than over and over
DT.upc_ingestiondate[,   is_ingestion_date_before_20160427_inclusive := ingestion_completed <= d_date]

## Types of Offenses
DT.upc_store_deliverydate[, is_store_first_delivery_after_20160427 := min_date_delivered >= d_date & max_date_delivered >= d_date]
DT.upc_store_deliverydate[, is_all_store_first_delivery_after_20160427 := all(min_date_delivered >= d_date & max_date_delivered >= d_date), by=upc]
DT.upc_store_deliverydate[, is_any_store_first_delivery_after_20160427 := any(min_date_delivered >= d_date & max_date_delivered >= d_date), by=upc]
DT.upc_store_deliverydate[, is_some_store_before_and_some_after_20160427 := is_any_store_first_delivery_after_20160427 & !is_all_store_first_delivery_after_20160427]

## Calculate some stats on shipment to stores
DT.upc_store_deliverydate[, number_of_stores_shipped_to := lunique(customer_master_master_id), by=upc]
DT.upc_store_deliverydate[, number_of_stores_shipped_to_for_first_time_after_20160427 := sum(is_store_first_delivery_after_20160427), by=upc]
DT.upc_store_deliverydate[, perc_of_stores_shipped_to_for_first_time_after_20160427 := number_of_stores_shipped_to_for_first_time_after_20160427 / number_of_stores_shipped_to]

## VIEW
DT.upc_store_deliverydate[perc_of_stores_shipped_to_for_first_time_after_20160427 > .3 & perc_of_stores_shipped_to_for_first_time_after_20160427 < 1, 'A', keyby=list(upc, number_of_stores_shipped_to, perc_of_stores_shipped_to_for_first_time_after_20160427)]

# exportXLS.usingXLConnect(DT.upc_store_deliverydate)

## Again, there should not be any UPCs where the number of stores delivered to is different from the number of rows per UPC
stopifnot(0 == nrow(DT.upc_store_deliverydate[, .N != number_of_stores_shipped_to, by=upc][(V1)]))

## This was previously an erroneous UPC, before converting to CHARACTER
if (FALSE) {
    unique(DT.upc_store_deliverydate[.(600116813916)], by=NULL)

    DT.example_600116813916 <- runQry("SELECT * FROM art_relations.delivery_history where upc = 600116813916")
    DT.example_600116813916[, upc := as.character(upc)]
    DT.example_600116813916[customer_master_master_id == 4]

    DT.test <- 
    runQry("SELECT 
      dh.upc, 
      dh.customer_master_master_id, 
      min(dh.date_delivered) AS min_date_delivered, 
      max(dh.date_delivered) AS max_date_delivered
    FROM art_relations.delivery_history dh
    WHERE upc = 600116813916
    GROUP BY 1, 2
    order by customer_master_master_id
    ")

    print(DT.test)

    # delivery_id encoder_id          upc customer_master_master_id date_delivered package_size
    #    66261473         15 600116813916                         4     2012-10-16            0
    #    71858121         16 600116813916                         4     2013-06-20            0
    #    86887399         14 600116813916                         4     2014-05-06        77162
    #    89932776         18 600116813916                         4     2014-06-24           75

}


## Show the UPCs in question
DT.upc_store_deliverydate[(is_any_store_first_delivery_after_20160427), .N, keyby=upc][, .N] %>% formnumb(round=0) %>%
  catn("Number of releases offending at at-leats one store: ", .)
DT.upc_store_deliverydate[(is_all_store_first_delivery_after_20160427), .N, keyby=upc][, .N] %>% formnumb(round=0) %>%
  catn("Number of releases offending at every store it has been shipped to: ", .)
DT.upc_store_deliverydate[(perc_of_stores_shipped_to_for_first_time_after_20160427 > .3 ), .N, keyby=upc][, .N] %>% formnumb(round=0) %>%
  catn("Number of releases offending at at-least 30% of the stores they were shipped to: ", .)

jesusForData(DT.upc_store_deliverydate, DT.upc_ingestiondate, info="RAWFromMySQL", dir=data.p("SoundCloudIssue"))   %>% print

DT.upc_store_deliverydate[10000:100000][!is_all_store_first_delivery_after_20160427 & is_any_store_first_delivery_after_20160427]
DT.upc_store_deliverydate[.("15568247217")]



DT.out <- DT.upc_store_deliverydate[DT.upc_ingestiondate] [!is.na(is_ingestion_date_before_20160427_inclusive) & is_ingestion_date_before_20160427_inclusive & is_store_first_delivery_after_20160427]
setnames(DT.out, "customer_master_master_id", "store_id")
jesusForData(DT.out, info="RAWFromMySQL", dir=data.p("SoundCloudIssue"))   %>% print

f.out <- out.p("upcs_and_stores_from_soundcloud_issue", ext="xlsx")
exportXLS.usingXLConnect(f.out=f.out, DT=DT.out)


### LOAD INTO SNOWFLAKE
setSnowflake(wh=wh, dbname=dbname, start=TRUE)
sfWaitForWarehouse(wh=wh, N.seconds=30, max_iterations=100, verbose.waiting=TRUE)

DT.stores <- get_dim_store(refresh=TRUE)
DT.stores[storeid > 660]

sfCreateSchema(schema.out, comment="Cropped copies from reportsAR; initial purpose is SoundCloudIssue; Tables updated manually as of June 2016", showWarnings=FALSE, overwrite_if_exists=FALSE)

ingestIntoSQL(DT.upc_store_deliverydate, dont_take_copy_of_DT=TRUE, schema=schema.out, boolean_type="BOOLEAN", datetime_type='TIMESTAMP_NTZ', drop=TRUE, confirm=FALSE)
ingestIntoSQL(DT.upc_ingestiondate,      dont_take_copy_of_DT=TRUE, schema=schema.out, boolean_type="BOOLEAN", datetime_type='TIMESTAMP_NTZ', drop=TRUE, confirm=FALSE)


sourceSupportFns(proj="Looker")
create_lookml_model_from_schema(schema=schema.out)


## --------------------  PART 02 -------------------------
if (FALSE) {
    matchKey(DT.upc_store_deliverydate, DT.upc_ingestiondate, "upc")
    DT.upc_store_deliverydate[DT.upc_ingestiondate[is.na(ingestion_completed)]][!is.na(min_date_delivered)][!is.na(max_date_delivered)][, .N, keyby=list(upc, min_date_delivered, max_date_delivered)][, as.character(tail(upc))]

    ## This UPC has a null ingestion_completed
    sample_upc <- 9789709397239

    headDB(schema=schema.out, tbl= "UPC_INGESTIONDATE")[]
    headDB(schema=schema.out, tbl= "UPC_STORE_DELIVERYDATE")[]

    DT.offending_upcs <- sfQry({
    "
    SELECT 
      DH.*, 
      I.ingestion_completed
    FROM delivery_history.upc_ingestiondate I
    LEFT JOIN delivery_history.UPC_STORE_DELIVERYDATE DH
    ON DH.upc = I.upc
    "})

}