setScience("Spotify_API_V2", create=TRUE, subl=FALSE, load=FALSE)

# TEST_RUN <- TRUE
# minDate.using <- "2014-10-01"

library(doMC)
library(rjson)
# library(jsonlite)
registerDoMC(24)
envir.current <- environment()

## Modifiable Parameters
assignIfNotExist(TEST_RUN,              FALSE)
assignIfNotExist(minDate.using,         as.Date("2014-08-01"))
assignIfNotExist(processed.subfolder,   "processed") ## Where processed files will be stored, relative to the file
assignIfNotExist(raw_RDS.subfolder,     "raw_RDS")  ## Where...
assignIfNotExist(raw_DT.subfolder,      "raw_DT")   ## Where ... &&&
assignIfNotExist(verbose,               TRUE)



## Capture output to log file
if (!TEST_RUN) {
  sinkfile <- newLogFile(name="Automation_ETL+MGMT_LogFile", dir="~/git/orch/log/MgmtReport/")
  sinkOn(file=sinkfile, zarchive.old.files=TRUE)
}

## In case user forgot to call as.Date() when assigned
if (is.character(minDate.using))
  minDate.using <- as.Date(minDate.using, origin=.origin)

## File to record where the DTs are saved
f.record_of_DTs <- src.p("record_of_DTs", ext="txt")
if (!file.exists(f.record_of_DTs)) {
  file.create(f.record_of_DTs, showWarnings=TRUE)
}
if (!TEST_RUN)
  write.table(file=f.record_of_DTs, x=cbind("DT", "file", "last_updated", "status"), sep=",", append=TRUE, quote=TRUE, row.names=FALSE, col.names=FALSE)
readLines(f.record_of_DTs)

## Date info
Months.using <- seq(minDate.using, Sys.Date(), by="1 month")

## reload country info
if (!(exists("country_codes.using")))
  loadFromJesus("country_codes.using")

## Grab the folders
folder.streams <- data.p("streams", country_codes.using)
setattr(folder.streams, "names", country_codes.using)

## Inform user before starting
verboseMsg(TEST_RUN, "This is a TEST_RUN", func="message")
Sys.sleep(ifelse(TEST_RUN, 1.8, 0.001))

## for TEST_RUN take two countries and two months.  
if (TEST_RUN) 
{
  folder.streams <- folder.streams[c("AR", "AD")]
  Months.using  <- unique(c(max(Months.using), max(minDate.using, Months.using[Months.using != max(Months.using)])))

  folder <- folder.streams[1]
  Month  <- max(Months.using)
}

### break it down by country, process one country at a time. 
### Crate the DT, save it, remove it and the raw files. 
### Then rbind all the DTs after being saved. 

Months.using <- unique(Months.using)
ret <- foreach (folder.nm = names(folder.streams), .combine=rbind) %:%
        foreach (Month = Months.using, .combine=rbind) %dopar%
{

  folder <- setNames(nm=folder.nm, obj=folder.streams[[folder.nm]])

  print(names(folder))
  cat(" --------- FOLDER: ", folder, " MONTH: ", as.character(Month), " --------- \n")

  if (is.null(names(folder)))
    names(folder) <- timeStamp("Unknown")

  ## Find any zipped file and unzip them

  # f.streams.gz <- unlist(lapply(folder.streams, extractFilesFromFolder, ext="gz", full=TRUE, minSize=0), use.names=FALSE) &&& needs names, if using lapply
  f.streams.gz <- extractFilesFromFolder(folder, pattern=format(Month, "%Y%m\\d{2}"), ext="gz", full=TRUE, minSize=0)
  if (length(f.streams.gz)) {
    verboseMsg(verbose, length(f.streams.gz), " files found in", folder, " - needing to be unzipped")
    unzip.gz(f.streams.gz, force=TRUE, batch.size=20)
  }

  # f.streams <- unlist(lapply(folder.streams, extractFilesFromFolder, ext="", full=TRUE, minSize=0), use.names=FALSE) &&& needs names, if using lapply
  f.streams <- extractFilesFromFolder(folder, pattern=format(Month, "%Y%m\\d{2}"), ext="", full=TRUE, minSize=0)
  verboseMsg(verbose, length(f.streams), " files found in", folder, " - ready for processing")


  if (TEST_RUN) {
    f.streams <- f.streams[1:2]
    cat("Test Run - thus taking only 2 files in ", folder, "\n")
  }

  ## NEED TO MOVE FILES TO "processed"

  stopifnot(!is.null(names(f.streams)))

  # raw.objs <- paste0("raw.", gsub("\\.gz", "", names(f.streams.uncompressed)))
  raw.objs <- paste0("raw.", basename(f.streams))
  setattr(raw.objs, "names", names(f.streams))

  n.lines <- ifelse(TEST_RUN, 12, -1)
  for(nm in names(f.streams)) {
    cat("reading raw lines for", nm, "\n")
    # assign(paste0("DT.", gsub("\\.gz", "", nm)), fread(f.streams.uncompressed[[nm]], sep="\t"))
    if (!exists(raw.objs[[nm]]))
      assign(raw.objs[[nm]], readLines(f.streams[[nm]], n = n.lines), envir=envir.current)
  }

  if (!TEST_RUN) {
    jesusForData(objNames=raw.objs, dir=data.p(raw_RDS.subfolder))
    notifyAndEmail("Done saving the raw files and, beginning parallel work")
  }

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


  ## TODO:  This can be modified for streams / tracks / users
  nm.DT <- sprintf("DT.streams_%s%s", names(folder), format(Month, "%Y%m%d"))
  assign(nm.DT, setDT(
      foreach(raw=raw.objs, .combine=rbind) %dopar% {
          ccode_AND_date <- as.list(strsplit(raw, "_")[[1]] [2:3])
          {foreach(lin=get(raw), .combine=rbind) %dopar% setDT(rjson::fromJSON(lin))} [, c("country_code", "date") := ccode_AND_date ]
      }
  ), envir=envir.current)

  cat(nm.DT, "has dim", dim2txt(get(nm.DT)), "\n")
  f.jesus <- jesusForData(objNames=nm.DT, dir=data.p(raw_DT.subfolder), git=FALSE)

  if (fileExistsWithData(f.jesus)) {
    ret <- cbind(DT=nm.DT, file=f.jesus, last_updated=format(Sys.Date(), "%Y-%m-%d"), status="success")
    rm (list=c(nm.DT, raw.objs), envir=envir.current)
    if (!TEST_RUN)
      zArchive(f.streams, subfolder=processed.subfolder)
  } else {
    ret <- cbind(DT=nm.DT, file=f.jesus, last_updated=format(Sys.Date(), "%Y-%m-%d"), status="ERROR")
  }

  e <- try(cat ("writing info for file = ", pasteQ(ret[, "file"], w=""), "\n"))
  write.table(x=ret, file=f.record_of_DTs, sep=",", append=TRUE, quote=TRUE, row.names=FALSE, col.names=FALSE)
}

if (!TEST_RUN) {
  DT._DUMMY_FOR_JESUS <- data.table(a="DELETE ME. I exist just to kick off the git for jesusForData()")
  jesusForData(DT._DUMMY_FOR_JESUS, git=TRUE)
  rm(DT._DUMMY_FOR_JESUS)
}

saveImageTo()
sinkOff()