# After Downloading and Processing, there will be a collection of TSV files. # Read them, then save to a single TSV { setGitBranchToSystem(); .g() cls(12) rm(files_all) verboseFile <- FALSE debug_verbose <- TRUE .LOAD_FROM_BACKUP <- FALSE chunk_size <- 7500 ## The slowest part is rbinding, thus use Larger-sized chunks (and hence less chunks) REFRESH <- FALSE DROP_KNOWN_ERROR_FILES <- FALSE SAVE <- FALSE assignIfNotExist(minDate, value=today() - 1) minDate <- as.Date("2015-11-01") cat(" ~~~ STARTING ~~~\n") setScience("Spotify Playlist Webscrape", create=TRUE, subl=FALSE, load=.LOAD_FROM_BACKUP) options(warn = 1) options(verbose_minw = 80) verbose <- TRUE ## Drop these palylists for now -- they are fucking with the read. Later on put them back in. other_bad_playlists <- c("1kfaNVqsQhFo7gh7QJw81E", "1k1CSJziBIykIUFPBCfZlI", "1kfnBApzTgcu3lJUZyRwcJ") ## We are not capturing the whole of the TSV_file, just the part at api.spotify.com/.. .extract_tsv_files_from_DT.playlist_metadata <- function(DT.playlist_metadata=get("DT.playlist_metadata", envir=parent.frame())) { files <- unique(DT.playlist_metadata$tsv_file) if (!any(grepl("/git/", files[1:5]))) files %<>% out.p("parsed_json", .) return(files) } if (REFRESH) { message("Starting Fresh") suppressWarnings(rm(files_all, files_chunked, DT.playlist_metadata, tmp_DT, i)) } colClasses <- getColClassesForPlaylistTSV() if (!exists("DT.playlist_metadata")) DT.playlist_metadata <- getEmptyDT.playlist_metadata() if (!exists("files_all")) { message("Pulling files from system") files_all <- get_all_playlists_downloaded_to_outDir(subdir="parsed", return="file", minDate=minDate, full=TRUE) # files_all %<>% path.expand ## Save it. Since this part takes so long, in case of a crash or a manual break, we want to be able to bring it back online quickly jesusForData(files_all) } if (DROP_KNOWN_ERROR_FILES) files_all %<>% setdiff(get_known_error_tsv_files()) ## Drop the manually identified files_all <- files_all[!grepl(regOr(other_bad_playlists, escape=TRUE), files_all)] stopifnot(!length(files_all[grepl("1kfaNVqsQhFo7gh7QJw81E_", files_all)])) if (!length(files_all)) stop("There are no more files to process. If (a) your minDate is fine and (b) the scraping script has been running regularly, then Good job, nothing to do.\n\nOtherwise, check that there are files downlaoded on or after minDate '", format(minDate), "'") verboseMsg(verbose, "There are", length(files_all), "total files that will be fread() and rbind()'d -- (note that some might have already been processed into DT.playlist_metadata)") if (exists("DT.playlist_metadata") && nrow(DT.playlist_metadata) && length(files_all)) { files_all.bak <- files_all files_all %<>% setdiff(.extract_tsv_files_from_DT.playlist_metadata()) verboseMsg(verbose, "There are", length(files_all), "files remainging (not yet in DT.playlist_metadata)") } ## Break it up into chunks ## Note, always recreate the chunks. it takes no time { ## Separate out radial files and move them to the front so that they get done first if (!exists("radial_files")) radial_files <- files_all %>% {.[is_radial(.)]} files_all %<>% setdiff(radial_files) %>% c(radial_files, .) cat("chunking files into groups of size ", chunk_size, " ... ") L <- length(files_all) inds <- seq.int(L) group_by_inds <- 1 + ((inds-1) %/% chunk_size) noOfGroups <- ceiling(L / chunk_size) seqOfGroups <- seq.int(noOfGroups) files_chunked <- lapply(seqOfGroups, function(g) files_all[group_by_inds == g]) cat("There are", length(files_chunked), "groups of files\n") } ## We would like to use fread, however, that at times crashes. So use read.table when it does ## { read <- function(f, func) { if (identical(func, "read.table")) message("Using read.table() for file '", f, "'") {match.fun(func)}(f, header=TRUE, sep="\t", colClasses=colClasses, stringsAsFactors=FALSE) %>% as.data.table %>% {.[, tsv_file := sub("^/home/rsaporta/git/orch/out/Spotify Playlist Webscrape/parsed_json/", "", f)]} %>% {.} } F.read <- function(f, verbose=FALSE) { if (verbose) cat(f, "\n") # valueIfErr(fread(f, colClasses=colClasses, sep="\t"), NULL) valueIfErr(read(f, "fread") , valueIfErr(read(f, "read.table") , message("ERROR reading file '", f, "'") ) ) } # // end F.read } ## DOUBLE CHECK if (!all(unlist(files_chunked) == files_all)) warning("files_all and files_chunked have different files", call.=FALSE) L.chunked <- length(files_chunked) for (i in seq(files_chunked)) { verboseMsg(verbose, sprintf("processing %2i of %2i", i, L.chunked)) files <- files_chunked[[i]] # tmp_DT <- lapply(files, F.read) tmp_DT <- lapply(files, F.read, verbose=(verboseFile || debug_verbose)) %>% rbindlist(use.names=TRUE, fill=TRUE) verboseMsg(debug_verbose, "Done Reading ", i, " ---- now combining") if (anyDuplicated(names(tmp_DT))) warning("There are duplicate names in tmp_DT for i = ", i, call.=FALSE) if (!nrow(tmp_DT)) { warning("tmp_DT has no length.\nIf you are re-running the stragglers, this may be because all of the files already were attempted and previously returned Errors", call.=FALSE) next; } ## ---- start bad_cols ---------------------------------------------------------------------------------------------------- ## ## TEMP -- updated 2015-12-08. Some of the TSV files were written with an extra "palylist." in front of column names, but all NA ## This is a sign of a bug upstream, and we can reprocess the JSON files. For now, just remove those columns bad_cols <- c("playlist.playlist.track.isrc", "playlist.playlist.track.name", "playlist.playlist.track.id", "playlist.playlist.track.type", "playlist.playlist.track.uri", "playlist.playlist.track.datetimeadded", "playlist.playlist.track.artist.namecsv", "playlist.playlist.track.artist.idcsv", "playlist.playlist.track.artist.numberof", "playlist.playlist.track.album.name", "playlist.playlist.track.album.producttype", "playlist.playlist.track.album.id", "playlist.playlist.track.album.uri", "playlist.playlist.track.album.type", "playlist.playlist.track.album.markets", "playlist.playlist.track.markets", "playlist.playlist.track.is_local", "playlist.playlist.track.addedby.id", "playlist.playlist.track.addedby.type", "playlist.playlist.track.addedby.uri", "playlist.playlist.track.discnumber", "playlist.playlist.track.durationms", "playlist.playlist.track.is_explicit", "playlist.playlist.track.popularity", "playlist.playlist.track.previewurl", "playlist.playlist.track.tracknumber_on_album", "playlist.playlist.track.api.itemshref", "playlist.playlist.track.api.timestamp") # slow: # ## if isrc is in the tmp_DT. If it is not in the main DT, and it is all NA, drop the column. Otherwise create it in the main DT # slow: # if ("playlist.isrc" %in% names(tmp_DT)) { # slow: # message("Checking if we should clear out playlist.isrc for i=", i) # slow: # if ("playlist.isrc" %ni% DT.playlist_metadata && all(is.na(tmp_DT[["playlist.isrc"]]))) # slow: # tmp_DT[, playlist.isrc := NULL] # slow: # else if ("playlist.isrc" %ni% DT.playlist_metadata) # slow: # DT.playlist_metadata[, playlist.isrc := NA_character_] # slow: # ## if isrc is NOT in the tmp_DT and it IS in the main DT, create it in the temp DT, so that columns add up downstream # slow: # } else if ("playlist.isrc" %in% names(DT.playlist_metadata)) { # slow: # tmp_DT[, playlist.isrc := NA_character_] # slow: # } ## INSTEAD, just drop the column: if ("playlist.isrc" %in% names(tmp_DT)) tmp_DT[, playlist.isrc := NULL] ## these columns should be integers should_be_ints <- c("playlist.track_number_in_playlist", "playlist.playlist.tracks_per_album") for (col in should_be_ints) if (col %in% names(tmp_DT) && is.character(tmp_DT[[col]])) { message("Converting ", col, " from character to integer") tmp_DT[, (col) := as.integer(get(col))] } ## fix tmp_DT when it has double columns if (ncol(tmp_DT) - length(bad_cols) == ncol(DT.playlist_metadata) && all(bad_cols %in% names(tmp_DT))) { # sort(setdiff(names(tmp_DT), bad_cols)) == sort(names(DT.playlist_metadata)) # cbind(sort(setdiff(names(tmp_DT), bad_cols)) , sort(names(DT.playlist_metadata))) %>% {cbind(., .[,1] == .[,2])} verboseMsg(verbose, "Dropping duplicate columns (captured in bad_cols)", func="message") tmp_DT[, (bad_cols) := NULL] } ## ---- end bad_cols ---------------------------------------------------------------------------------------------------- ## classes <- desc(tmp_DT, quiet=TRUE, sort=FALSE, notClassy=TRUE) %>% {.[names(.) != "tsv_file"]} ## SANITY CHECK ## check for additional/missing columns if (length(classes) != length(colClasses)) { message(" ***** Unexpected number of columns for i = ", i, " ***** ") cbind(EXPECTED = length(colClasses), tmp_DT=length(classes)) %>% print ## check for different names } else if (any( names(DT.playlist_metadata) != names(tmp_DT) )) { message(" ***** Wrong names for i = ", i, " ***** ") cbind(EXPECTED = names(DT.playlist_metadata), tmp_DT=names(tmp_DT)) %>% print ## check for different classes } else if (length(wh.wrongclass <- nwhich(classes != colClasses))) { message(" ***** Wrong classes for i = ", i, " ***** ") cbind(EXPECTED = colClasses[wh.wrongclass], tmp_DT=classes[wh.wrongclass]) %>% print } ## Execute rbind() verboseMsg(verbose, "Beginning actual rbind()", func="message") s.t({DT.playlist_metadata <- rbind(DT.playlist_metadata, tmp_DT, use.names=TRUE, fill=TRUE)}, title="rbind") if (anyDuplicated(names(DT.playlist_metadata))) warning("There are duplicate names in DT.playlist_metadata for i = ", i, call.=FALSE) else rm(tmp_DT) if (i %% 3 == 0 || i == L.chunked) { verboseMsg(verbose, "Executing garbage collection since i=", i, func="message", sep="") gc() } ## Confirm that these two fields are NOT in DT.playlist_metadata print(any (c("playlist.track.upc", "playlist.track.is_orchard_dist") %in% names(DT.playlist_metadata))) # if (!(i %% 10)) # jesusForData(DT.playlist_metadata, info=sprintf("i_is_%03i", i)) %>% print } if (anyDuplicated(DT.playlist_metadata, by=c("playlist.id", "playlist.track.id", "playlist.track_number_in_playlist"))) { verboseMsg(TRUE, "There are duplicates by playlist-track-tracknumber\nUsing unique() for now", func="warning", call.=FALSE) ## TODO: (a) ideal: Capture this history (b) alternative: find the latest and use that (c) for now: use unique DT.playlist_metadata <- unique(DT.playlist_metadata, by=c("playlist.id", "playlist.track.id", "playlist.track_number_in_playlist")) } ## CONFIRM files_missing <- setdiff(files_all, .extract_tsv_files_from_DT.playlist_metadata()) if (length(files_missing)) { verboseMsg(verbose, "There are", length(files_missing), "files that were not processed - possibly due to error") } ## ~~~~~~~~~ UPDATED 2015-06-19: NOT NEEDED WITH THE NEW VERSION ~~~~~~~~~~~ # ~ ## Fix the names of the logical columns # ~ setnames(DT.playlist_metadata, "playlist.track.explicit", "playlist.track.is_explicit") # ~ setnames(DT.playlist_metadata, "playlist.track.islocal", "playlist.track.is_local") # ~ setnames(DT.playlist_metadata, "playlist.followers.total", "playlist.total_followers") # ~ # ~ ## ADD Track numbers by playlist # ~ DT.playlist_metadata[, `:=`(track_number_in_playlist=seq(.N), total_tracks_in_playlist=.N), by=playlist.uri] # ~ # ~ DT.playlist_metadata[, playlist.tracks_per_album := .N, by=list(playlist.uri, playlist.track.album.name)] # ~ ## ~~~~~~~~~ UPDATED 2015-06-19: NOT NEEDED WITH THE NEW VERSION ~~~~~~~~~~~ BackUpOrRestore("DT.playlist_metadata", force=TRUE) ## GET ISRC INFO DT.orch_isrc_upc <- get_DT.orch_isrc_upc(refresh=TRUE) DT.orch_isrc_upc[, track.is_orchard_dist := TRUE] DT.playlist_metadata[, c("playlist.track.upc", "playlist.track.is_orchard_dist") := list(as.idcol(NA), FALSE)] # DOESNT WORK: addColsFrom_(DT.playlist_metadata, DT.orch_isrc_upc, colsToBring=c("upc", "track.is_orchard_dist"), joinCols.rec="playlist.track.isrc", joinCols.giv="isrc") setkeyIfNot(DT.playlist_metadata, "playlist.track.isrc", organize=FALSE, verbose=FALSE) setkeyIfNot(DT.orch_isrc_upc, "isrc", superset.ok=TRUE, warnForColNameInEnv=FALSE, verbose=FALSE) s.t({ DT.playlist_metadata[DT.orch_isrc_upc, c("upc", "track.is_orchard_dist") := list(i.upc, i.track.is_orchard_dist), allow=TRUE] }, title="JOIN in track metadata") DT.playlist_metadata[is.na(track.is_orchard_dist), track.is_orchard_dist := FALSE] ## Add some additional information, namely user metadata and track counts per artist ## Update DT.usermeta DT.usermeta <- get_all_known_users() ## if DT.usermeta already exists, it will simply check for new updates since add_user_metadata_(DT.playlist_metadata, user_word="owner", preface="playlist", sep_preface=".", joinCols.rec=c("playlist.owner.id"), DT.usermeta=DT.usermeta) startCols.pl <- c("uri", "id", "name", "owner.id", "owner.name", "total_followers", "track.is_orchard_dist", "track.upc", "track.isrc") %>% paste0("playlist.", .) setcolorderpt(DT.playlist_metadata, startCols=startCols.pl) if (SAVE) { verboseMsg(verbose, "Saving as .RDS file (jesusForData)") print(jesusForData(DT.playlist_metadata, info="DONE", verbose=TRUE)) verboseMsg(verbose, "Saving as flat .tsv file (writeDT)") f.playlist_metadata.tsv <- writeDT(DT.playlist_metadata, subfolder="flattened_playlists", base.file.name=dateStamp("All_Playlists_"), append=FALSE, fileEncoding="UTF-8") print(f.playlist_metadata.tsv) # bucket <- s3_p("spotify", "playlists") # , prefix="flattened_playlists" # s3_upload(f.out, bucket=bucket) # saveImageTo() } ## Not sure about re-writing. Using this for now if (FALSE) { ingestIntoSQL(DT.playlist_metadata, tbl="playlist_metadata", schema="spotify", drop=FALSE, wh="LOOKER_WH_LARGE", dbname="prod", snowflake=TRUE) # ingestIntoSQL(DT.playlist_metadata, tbl="playlist_metadata", schema="spotify", drop=TRUE, wh="LOOKER_WH_LARGE", dbname="prod", snowflake=TRUE) } message("sourcing After II") source("~/git/orch/src/Spotify Playlist Webscrape/After Downloading II.r") } ## This is useful for when the reading crashes. Mostly for fread if (FALSE) { i <- 22:26 files <- unlist(files_chunked[i]) for (file in files) { cat(file, "\n") DT.okay <- fread(file, colClasses=colClasses, sep="\t", header=TRUE, stringsAsFactors=FALSE) # Sys.sleep(0.2) } }