package appleconsumeranalytics import ( "archive/zip" "bufio" "compress/gzip" "encoding/json" "io" "io/ioutil" "log" "os" "strconv" "strings" "time" "github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/aws/session" "github.com/aws/aws-sdk-go/service/s3" "github.com/aws/aws-sdk-go/service/s3/s3manager" "github.com/filtr/go-apollo/pkg/daterange" "github.com/google/uuid" ) type contentItem struct { AppleIdentifier string ISRC string Title string Artist string ArtistID string ItemType string MediaType string MediaDuration string VendorIdentifier string LabelNetworkStudio string Grid string } type streamRecord struct { Datestamp string IngestDatestamp string IngestTimestamp string AppleIdentifier string StorefrontName string AnonymizedPersonID string MembershipType string MembershipMode string MembershipPartner string PostalCode string DeviceType string OperatingSystem string UTCOffset string ActionType string EndReasonType string Offline string SourceOfStream string ContainerType string ContainerSubType string ContainerID string ContainerName string StreamTimestamp string StreamStartPosition string StreamDuration int Streams int } type summaryRecord struct { ReportDate string `json:"report_date"` ReportVendorID string `json:"report_vendor_id"` Date string `json:"date"` CountryCode string `json:"country_code"` Licensor string `json:"licensor"` ActionType1 int `json:"action_type_1"` DeviceType0 int `json:"device_type_0"` DeviceType1 int `json:"device_type_1"` DeviceType2 int `json:"device_type_2"` DeviceType3 int `json:"device_type_3"` DeviceOS0 int `json:"device_os_0"` DeviceOS1 int `json:"device_os_1"` DeviceOS2 int `json:"device_os_2"` DeviceOS3 int `json:"device_os_3"` DeviceOS4 int `json:"device_os_4"` DeviceOS5 int `json:"device_os_5"` DeviceOS6 int `json:"device_os_6"` EndReason0 int `json:"end_reason_0"` EndReason1 int `json:"end_reason_1"` EndReason2 int `json:"end_reason_2"` EndReason3 int `json:"end_reason_3"` EndReason4 int `json:"end_reason_4"` ContainerSubType0 int `json:"container_subtype_0"` ContainerSubType1 int `json:"container_subtype_1"` ContainerSubType2 int `json:"container_subtype_2"` ContainerSubType3 int `json:"container_subtype_3"` ContainerSubType4 int `json:"container_subtype_4"` ContainerSubType5 int `json:"container_subtype_5"` ContainerSubType6 int `json:"container_subtype_6"` ContainerSubType7 int `json:"container_subtype_7"` ContainerSubType8 int `json:"container_subtype_8"` ContainerSubType9 int `json:"container_subtype_9"` ContainerSubType10 int `json:"container_subtype_10"` ContainerSubType11 int `json:"container_subtype_11"` ContainerSubType12 int `json:"container_subtype_12"` ContainerSubType13 int `json:"container_subtype_13"` ContainerSubType14 int `json:"container_subtype_14"` ContainerSubType15 int `json:"container_subtype_15"` ContainerType0 int `json:"container_type_0"` ContainerType1 int `json:"container_type_1"` ContainerType2 int `json:"container_type_2"` ContainerType3 int `json:"container_type_3"` ContainerType4 int `json:"container_type_4"` Offline0 int `json:"offline_0"` Offline1 int `json:"offline_1"` Source0 int `json:"source_0"` Source1 int `json:"source_1"` Source2 int `json:"source_2"` Source3 int `json:"source_3"` Source4 int `json:"source_4"` Source5 int `json:"source_5"` Source6 int `json:"source_6"` Source7 int `json:"source_7"` LeanForward int `json:"lean_forward"` Streams int `json:"streams"` Listeners int `json:"listeners"` } type trackRecord struct { ReportDate string `json:"report_date"` ReportVendorID string `json:"report_vendor_id"` Date string `json:"date"` CountryCode string `json:"country_code"` Licensor string `json:"licensor"` ISRC string `json:"isrc"` ActionType1 int `json:"action_type_1"` DeviceType0 int `json:"device_type_0"` DeviceType1 int `json:"device_type_1"` DeviceType2 int `json:"device_type_2"` DeviceType3 int `json:"device_type_3"` DeviceOS0 int `json:"device_os_0"` DeviceOS1 int `json:"device_os_1"` DeviceOS2 int `json:"device_os_2"` DeviceOS3 int `json:"device_os_3"` DeviceOS4 int `json:"device_os_4"` DeviceOS5 int `json:"device_os_5"` DeviceOS6 int `json:"device_os_6"` EndReason0 int `json:"end_reason_0"` EndReason1 int `json:"end_reason_1"` EndReason2 int `json:"end_reason_2"` EndReason3 int `json:"end_reason_3"` EndReason4 int `json:"end_reason_4"` ContainerSubType0 int `json:"container_subtype_0"` ContainerSubType1 int `json:"container_subtype_1"` ContainerSubType2 int `json:"container_subtype_2"` ContainerSubType3 int `json:"container_subtype_3"` ContainerSubType4 int `json:"container_subtype_4"` ContainerSubType5 int `json:"container_subtype_5"` ContainerSubType6 int `json:"container_subtype_6"` ContainerSubType7 int `json:"container_subtype_7"` ContainerSubType8 int `json:"container_subtype_8"` ContainerSubType9 int `json:"container_subtype_9"` ContainerSubType10 int `json:"container_subtype_10"` ContainerSubType11 int `json:"container_subtype_11"` ContainerSubType12 int `json:"container_subtype_12"` ContainerSubType13 int `json:"container_subtype_13"` ContainerSubType14 int `json:"container_subtype_14"` ContainerSubType15 int `json:"container_subtype_15"` ContainerType0 int `json:"container_type_0"` ContainerType1 int `json:"container_type_1"` ContainerType2 int `json:"container_type_2"` ContainerType3 int `json:"container_type_3"` ContainerType4 int `json:"container_type_4"` Offline0 int `json:"offline_0"` Offline1 int `json:"offline_1"` Source0 int `json:"source_0"` Source1 int `json:"source_1"` Source2 int `json:"source_2"` Source3 int `json:"source_3"` Source4 int `json:"source_4"` Source5 int `json:"source_5"` Source6 int `json:"source_6"` Source7 int `json:"source_7"` LeanForward int `json:"lean_forward"` Streams int `json:"streams"` Listeners int `json:"listeners"` } type containerRecord struct { ReportDate string `json:"report_date"` ReportVendorID string `json:"report_vendor_id"` Date string `json:"date"` CountryCode string `json:"country_code"` Licensor string `json:"licensor"` ContainerName string `json:"container_name"` ContainerID string `json:"container_id"` ContainerSubType string `json:"container_subtype"` ActionType1 int `json:"action_type_1"` DeviceType0 int `json:"device_type_0"` DeviceType1 int `json:"device_type_1"` DeviceType2 int `json:"device_type_2"` DeviceType3 int `json:"device_type_3"` DeviceOS0 int `json:"device_os_0"` DeviceOS1 int `json:"device_os_1"` DeviceOS2 int `json:"device_os_2"` DeviceOS3 int `json:"device_os_3"` DeviceOS4 int `json:"device_os_4"` DeviceOS5 int `json:"device_os_5"` DeviceOS6 int `json:"device_os_6"` EndReason0 int `json:"end_reason_0"` EndReason1 int `json:"end_reason_1"` EndReason2 int `json:"end_reason_2"` EndReason3 int `json:"end_reason_3"` EndReason4 int `json:"end_reason_4"` ContainerType0 int `json:"container_type_0"` ContainerType1 int `json:"container_type_1"` ContainerType2 int `json:"container_type_2"` ContainerType3 int `json:"container_type_3"` ContainerType4 int `json:"container_type_4"` Offline0 int `json:"offline_0"` Offline1 int `json:"offline_1"` Source0 int `json:"source_0"` Source1 int `json:"source_1"` Source2 int `json:"source_2"` Source3 int `json:"source_3"` Source4 int `json:"source_4"` Source5 int `json:"source_5"` Source6 int `json:"source_6"` Source7 int `json:"source_7"` LeanForward int `json:"lean_forward"` Streams int `json:"streams"` Listeners int `json:"listeners"` } type containerTrackRecord struct { ReportDate string `json:"report_date"` ReportVendorID string `json:"report_vendor_id"` Date string `json:"date"` CountryCode string `json:"country_code"` Licensor string `json:"licensor"` ContainerName string `json:"container_name"` ContainerID string `json:"container_id"` ContainerSubType string `json:"container_subtype"` ISRC string `json:"isrc"` ActionType1 int `json:"action_type_1"` DeviceType0 int `json:"device_type_0"` DeviceType1 int `json:"device_type_1"` DeviceType2 int `json:"device_type_2"` DeviceType3 int `json:"device_type_3"` DeviceOS0 int `json:"device_os_0"` DeviceOS1 int `json:"device_os_1"` DeviceOS2 int `json:"device_os_2"` DeviceOS3 int `json:"device_os_3"` DeviceOS4 int `json:"device_os_4"` DeviceOS5 int `json:"device_os_5"` DeviceOS6 int `json:"device_os_6"` EndReason0 int `json:"end_reason_0"` EndReason1 int `json:"end_reason_1"` EndReason2 int `json:"end_reason_2"` EndReason3 int `json:"end_reason_3"` EndReason4 int `json:"end_reason_4"` ContainerType0 int `json:"container_type_0"` ContainerType1 int `json:"container_type_1"` ContainerType2 int `json:"container_type_2"` ContainerType3 int `json:"container_type_3"` ContainerType4 int `json:"container_type_4"` Offline0 int `json:"offline_0"` Offline1 int `json:"offline_1"` Source0 int `json:"source_0"` Source1 int `json:"source_1"` Source2 int `json:"source_2"` Source3 int `json:"source_3"` Source4 int `json:"source_4"` Source5 int `json:"source_5"` Source6 int `json:"source_6"` Source7 int `json:"source_7"` LeanForward int `json:"lean_forward"` Streams int `json:"streams"` Listeners int `json:"listeners"` } func isLeanForward(sourceOfStream, containerType string) bool { if sourceOfStream == "3" && containerType == "3" { return true } if sourceOfStream == "4" && containerType == "3" { return true } if sourceOfStream == "4" && containerType == "0" { return true } if sourceOfStream == "0" && containerType == "3" { return true } if sourceOfStream == "0" && containerType == "0" { return true } if sourceOfStream == "2" && containerType == "3" { return true } if sourceOfStream == "2" && containerType == "0" { return true } if sourceOfStream == "4" && containerType == "2" { return true } if sourceOfStream == "1" && containerType == "3" { return true } if sourceOfStream == "1" && containerType == "0" { return true } if sourceOfStream == "1" && containerType == "2" { return true } return false } func mutateSummary(s summaryRecord, streamRecord streamRecord) summaryRecord { s.Streams += streamRecord.Streams if streamRecord.ActionType == "1" { s.ActionType1++ } if isLeanForward(streamRecord.SourceOfStream, streamRecord.ContainerType) { s.LeanForward++ } switch streamRecord.ContainerSubType { case "0": s.ContainerSubType0++ case "1": s.ContainerSubType1++ case "2": s.ContainerSubType2++ case "3": s.ContainerSubType3++ case "4": s.ContainerSubType4++ case "5": s.ContainerSubType5++ case "6": s.ContainerSubType6++ case "7": s.ContainerSubType7++ case "8": s.ContainerSubType8++ case "9": s.ContainerSubType9++ case "10": s.ContainerSubType10++ case "11": s.ContainerSubType11++ case "12": s.ContainerSubType12++ case "13": s.ContainerSubType13++ case "14": s.ContainerSubType14++ case "15": s.ContainerSubType15++ } switch streamRecord.ContainerType { case "0": s.ContainerType0++ case "1": s.ContainerType1++ case "2": s.ContainerType2++ case "3": s.ContainerType3++ case "4": s.ContainerType4++ } switch streamRecord.DeviceType { case "0": s.DeviceType0++ case "1": s.DeviceType1++ case "2": s.DeviceType2++ case "3": s.DeviceType3++ } switch streamRecord.EndReasonType { case "0": s.EndReason0++ case "1": s.EndReason1++ case "2": s.EndReason2++ case "3": s.EndReason3++ case "4": s.EndReason4++ } switch streamRecord.Offline { case "0": s.Offline0++ case "1": s.Offline1++ } switch streamRecord.OperatingSystem { case "0": s.DeviceOS0++ case "1": s.DeviceOS1++ case "2": s.DeviceOS2++ case "3": s.DeviceOS3++ case "4": s.DeviceOS4++ case "5": s.DeviceOS5++ case "6": s.DeviceOS6++ } switch streamRecord.SourceOfStream { case "0": s.Source0++ case "1": s.Source1++ case "2": s.Source2++ case "3": s.Source3++ case "4": s.Source4++ case "5": s.Source5++ case "6": s.Source6++ case "7": s.Source7++ } return s } func mutateTrack(t trackRecord, streamRecord streamRecord) trackRecord { t.Streams += streamRecord.Streams if streamRecord.ActionType == "1" { t.ActionType1++ } if isLeanForward(streamRecord.SourceOfStream, streamRecord.ContainerType) { t.LeanForward++ } switch streamRecord.ContainerSubType { case "0": t.ContainerSubType0++ case "1": t.ContainerSubType1++ case "2": t.ContainerSubType2++ case "3": t.ContainerSubType3++ case "4": t.ContainerSubType4++ case "5": t.ContainerSubType5++ case "6": t.ContainerSubType6++ case "7": t.ContainerSubType7++ case "8": t.ContainerSubType8++ case "9": t.ContainerSubType9++ case "10": t.ContainerSubType10++ case "11": t.ContainerSubType11++ case "12": t.ContainerSubType12++ case "13": t.ContainerSubType13++ case "14": t.ContainerSubType14++ case "15": t.ContainerSubType15++ } switch streamRecord.ContainerType { case "0": t.ContainerType0++ case "1": t.ContainerType1++ case "2": t.ContainerType2++ case "3": t.ContainerType3++ case "4": t.ContainerType4++ } switch streamRecord.DeviceType { case "0": t.DeviceType0++ case "1": t.DeviceType1++ case "2": t.DeviceType2++ case "3": t.DeviceType3++ } switch streamRecord.EndReasonType { case "0": t.EndReason0++ case "1": t.EndReason1++ case "2": t.EndReason2++ case "3": t.EndReason3++ case "4": t.EndReason4++ } switch streamRecord.Offline { case "0": t.Offline0++ case "1": t.Offline1++ } switch streamRecord.OperatingSystem { case "0": t.DeviceOS0++ case "1": t.DeviceOS1++ case "2": t.DeviceOS2++ case "3": t.DeviceOS3++ case "4": t.DeviceOS4++ case "5": t.DeviceOS5++ case "6": t.DeviceOS6++ } switch streamRecord.SourceOfStream { case "0": t.Source0++ case "1": t.Source1++ case "2": t.Source2++ case "3": t.Source3++ case "4": t.Source4++ case "5": t.Source5++ case "6": t.Source6++ case "7": t.Source7++ } return t } func mutateContainer(c containerRecord, streamRecord streamRecord) containerRecord { c.Streams += streamRecord.Streams if streamRecord.ActionType == "1" { c.ActionType1++ } if isLeanForward(streamRecord.SourceOfStream, streamRecord.ContainerType) { c.LeanForward++ } switch streamRecord.ContainerType { case "0": c.ContainerType0++ case "1": c.ContainerType1++ case "2": c.ContainerType2++ case "3": c.ContainerType3++ case "4": c.ContainerType4++ } switch streamRecord.DeviceType { case "0": c.DeviceType0++ case "1": c.DeviceType1++ case "2": c.DeviceType2++ case "3": c.DeviceType3++ } switch streamRecord.EndReasonType { case "0": c.EndReason0++ case "1": c.EndReason1++ case "2": c.EndReason2++ case "3": c.EndReason3++ case "4": c.EndReason4++ } switch streamRecord.Offline { case "0": c.Offline0++ case "1": c.Offline1++ } switch streamRecord.OperatingSystem { case "0": c.DeviceOS0++ case "1": c.DeviceOS1++ case "2": c.DeviceOS2++ case "3": c.DeviceOS3++ case "4": c.DeviceOS4++ case "5": c.DeviceOS5++ case "6": c.DeviceOS6++ } switch streamRecord.SourceOfStream { case "0": c.Source0++ case "1": c.Source1++ case "2": c.Source2++ case "3": c.Source3++ case "4": c.Source4++ case "5": c.Source5++ case "6": c.Source6++ case "7": c.Source7++ } return c } func mutateContainerTrack(ct containerTrackRecord, streamRecord streamRecord) containerTrackRecord { ct.Streams += streamRecord.Streams if streamRecord.ActionType == "1" { ct.ActionType1++ } if isLeanForward(streamRecord.SourceOfStream, streamRecord.ContainerType) { ct.LeanForward++ } switch streamRecord.ContainerType { case "0": ct.ContainerType0++ case "1": ct.ContainerType1++ case "2": ct.ContainerType2++ case "3": ct.ContainerType3++ case "4": ct.ContainerType4++ } switch streamRecord.DeviceType { case "0": ct.DeviceType0++ case "1": ct.DeviceType1++ case "2": ct.DeviceType2++ case "3": ct.DeviceType3++ } switch streamRecord.EndReasonType { case "0": ct.EndReason0++ case "1": ct.EndReason1++ case "2": ct.EndReason2++ case "3": ct.EndReason3++ case "4": ct.EndReason4++ } switch streamRecord.Offline { case "0": ct.Offline0++ case "1": ct.Offline1++ } switch streamRecord.OperatingSystem { case "0": ct.DeviceOS0++ case "1": ct.DeviceOS1++ case "2": ct.DeviceOS2++ case "3": ct.DeviceOS3++ case "4": ct.DeviceOS4++ case "5": ct.DeviceOS5++ case "6": ct.DeviceOS6++ } switch streamRecord.SourceOfStream { case "0": ct.Source0++ case "1": ct.Source1++ case "2": ct.Source2++ case "3": ct.Source3++ case "4": ct.Source4++ case "5": ct.Source5++ case "6": ct.Source6++ case "7": ct.Source7++ } return ct } func parseStreamRow(headers []string, row []string) streamRecord { itemMap := make(map[string]string) for i, k := range headers { key := strings.ToLower(strings.TrimSpace(k)) value := strings.TrimSpace(row[i]) itemMap[key] = value } streamRecord := streamRecord{ Datestamp: itemMap["datestamp"], IngestDatestamp: itemMap["ingest datestamp"], IngestTimestamp: itemMap["ingest timestamp"], AppleIdentifier: itemMap["apple identifier"], StorefrontName: itemMap["storefront name"], AnonymizedPersonID: itemMap["anonymized person id"], MembershipType: itemMap["membership type"], MembershipMode: itemMap["membership mode"], MembershipPartner: itemMap["membership partner"], PostalCode: itemMap["postal code"], DeviceType: itemMap["device type"], OperatingSystem: itemMap["operating system"], UTCOffset: itemMap["utc offset"], ActionType: itemMap["action type"], EndReasonType: itemMap["end reason type"], Offline: itemMap["offline"], SourceOfStream: itemMap["source of stream"], ContainerType: itemMap["container type"], ContainerSubType: itemMap["container sub-type"], ContainerID: itemMap["container id"], ContainerName: itemMap["container name"], StreamTimestamp: itemMap["stream timestamp"], StreamStartPosition: itemMap["stream start position"], } var streamDuration int = 0; if _streamDuration, ok := itemMap["stream duration"]; ok { _streamDuration, err := strconv.Atoi(_streamDuration) if err != nil { log.Fatal(err) } streamDuration = _streamDuration } streamRecord.StreamDuration = streamDuration var streams int = 0 if _streams, ok := itemMap["streams"]; ok { _streams, err := strconv.Atoi(_streams) if err != nil { log.Println("Got unfilled streams counted as 0", err) } else { streams = _streams } } else { streams = 1 } streamRecord.Streams = streams if _, ok := itemMap["playlist id"]; ok { streamRecord.ContainerID = itemMap["playlist id"] } if _, ok := itemMap["playlist type"]; ok { streamRecord.ContainerSubType = itemMap["playlist type"] } if _, ok := itemMap["playlist name"]; ok { streamRecord.ContainerName = itemMap["playlist name"] } return streamRecord } func processStreams(s3SourceService *s3.S3, s3TargetService *s3.S3, s3SourceBucket, s3TargetBucket, date string, licensor string, vendorID string, content map[string]contentItem) { input := &s3.GetObjectInput{ Bucket: aws.String(s3SourceBucket), Key: aws.String(makeStreamsSourceS3Key(date, licensor, vendorID)), } result, err := s3SourceService.GetObject(input) if err != nil { log.Fatal(err) } reader, err := gzip.NewReader(result.Body) if err != nil { log.Fatal(err) } defer reader.Close() summaryMap := make(map[string]summaryRecord) trackMap := make(map[string]trackRecord) containerMap := make(map[string]containerRecord) containerTrackMap := make(map[string]containerTrackRecord) listenerSummaryMap := make(map[string]map[string]int) listenerTrackMap := make(map[string]map[string]int) listenerContainerMap := make(map[string]map[string]int) listenerContainerTrackMap := make(map[string]map[string]int) var headers []string firstItem := true scanner := bufio.NewScanner(reader) for scanner.Scan() { row := strings.Split(string(scanner.Text()), "\t") if firstItem == true { firstItem = false headers = row } else { streamRecord := parseStreamRow(headers, row) contentItem := content[streamRecord.AppleIdentifier] countryCode := strings.ToLower(streamRecord.StorefrontName) summaryKey := streamRecord.IngestDatestamp + "\xff" + streamRecord.StorefrontName ls, ok2 := listenerSummaryMap[summaryKey] if ok2 == false { m := make(map[string]int) m[streamRecord.AnonymizedPersonID] = 1 listenerSummaryMap[summaryKey] = m } else { ls[streamRecord.AnonymizedPersonID] = 1 listenerSummaryMap[summaryKey] = ls } s, ok := summaryMap[summaryKey] if ok == false { s = summaryRecord{ ReportDate: date, ReportVendorID: vendorID, Date: streamRecord.IngestDatestamp, CountryCode: countryCode, Licensor: licensor, } summaryMap[summaryKey] = s } summaryMap[summaryKey] = mutateSummary(s, streamRecord) if streamRecord.ContainerID != "" { containerKey := streamRecord.IngestDatestamp + "\xff" + streamRecord.StorefrontName + "\xff" + streamRecord.ContainerID lc, ok := listenerContainerMap[containerKey] if ok == false { m := make(map[string]int) m[streamRecord.AnonymizedPersonID] = 1 listenerContainerMap[containerKey] = m } else { lc[streamRecord.AnonymizedPersonID] = 1 listenerContainerMap[containerKey] = lc } c, ok := containerMap[containerKey] if ok == false { c = containerRecord{ ReportDate: date, ReportVendorID: vendorID, Date: streamRecord.IngestDatestamp, CountryCode: countryCode, Licensor: licensor, ContainerID: streamRecord.ContainerID, ContainerSubType: streamRecord.ContainerSubType, ContainerName: streamRecord.ContainerName, } containerMap[containerKey] = c } containerMap[containerKey] = mutateContainer(c, streamRecord) } if contentItem.ISRC != "" { trackKey := streamRecord.IngestDatestamp + "\xff" + streamRecord.StorefrontName + "\xff" + contentItem.ISRC lt, ok2 := listenerTrackMap[trackKey] if ok2 == false { m := make(map[string]int) m[streamRecord.AnonymizedPersonID] = 1 listenerTrackMap[trackKey] = m } else { lt[streamRecord.AnonymizedPersonID] = 1 listenerTrackMap[trackKey] = lt } t, ok := trackMap[trackKey] if ok == false { t = trackRecord{ ReportDate: date, ReportVendorID: vendorID, Date: streamRecord.IngestDatestamp, CountryCode: countryCode, Licensor: licensor, ISRC: contentItem.ISRC, } trackMap[trackKey] = t } trackMap[trackKey] = mutateTrack(t, streamRecord) if streamRecord.ContainerID != "" { containerTrackKey := streamRecord.IngestDatestamp + "\xff" + streamRecord.StorefrontName + "\xff" + streamRecord.ContainerID + "\xff" + contentItem.ISRC lct, ok2 := listenerContainerTrackMap[containerTrackKey] if ok2 == false { m := make(map[string]int) m[streamRecord.AnonymizedPersonID] = 1 listenerContainerTrackMap[containerTrackKey] = m } else { lct[streamRecord.AnonymizedPersonID] = 1 listenerContainerTrackMap[containerTrackKey] = lct } ct, ok := containerTrackMap[containerTrackKey] if ok == false { ct = containerTrackRecord{ ReportDate: date, ReportVendorID: vendorID, Date: streamRecord.IngestDatestamp, CountryCode: countryCode, Licensor: licensor, ContainerID: streamRecord.ContainerID, ContainerSubType: streamRecord.ContainerSubType, ContainerName: streamRecord.ContainerName, ISRC: contentItem.ISRC, } containerTrackMap[containerTrackKey] = ct } containerTrackMap[containerTrackKey] = mutateContainerTrack(ct, streamRecord) } } } } if err := scanner.Err(); err != nil { log.Fatal(err) } for k, v := range summaryMap { v.Listeners = len(listenerSummaryMap[k]) summaryMap[k] = v } for k, v := range trackMap { v.Listeners = len(listenerTrackMap[k]) trackMap[k] = v } for k, v := range containerMap { v.Listeners = len(listenerContainerMap[k]) containerMap[k] = v } for k, v := range containerTrackMap { v.Listeners = len(listenerContainerTrackMap[k]) containerTrackMap[k] = v } tracksSession := session.Must(session.NewSession(&aws.Config{Region: aws.String(s3TargetService.SigningRegion), S3UseAccelerate: aws.Bool(false)})) tracksUploader := s3manager.NewUploader(tracksSession) tracksReader, tracksWriter := io.Pipe() containersSession := session.Must(session.NewSession(&aws.Config{Region: aws.String(s3TargetService.SigningRegion), S3UseAccelerate: aws.Bool(false)})) containersUploader := s3manager.NewUploader(containersSession) containersReader, containersWriter := io.Pipe() containerTracksSession := session.Must(session.NewSession(&aws.Config{Region: aws.String(s3TargetService.SigningRegion), S3UseAccelerate: aws.Bool(false)})) containerTracksUploader := s3manager.NewUploader(containerTracksSession) containerTracksReader, containerTracksWriter := io.Pipe() summarySession := session.Must(session.NewSession(&aws.Config{Region: aws.String(s3TargetService.SigningRegion), S3UseAccelerate: aws.Bool(false)})) summaryUploader := s3manager.NewUploader(summarySession) summaryReader, summaryWriter := io.Pipe() go func() { gw := gzip.NewWriter(tracksWriter) for _, v := range trackMap { row, err := json.Marshal(v) if err != nil { log.Fatal(err) } r := append(row, []byte("\n")...) n1, err := gw.Write(r) if err != nil { log.Fatal(err) } _ = n1 } gw.Close() tracksWriter.Close() }() go func() { gw := gzip.NewWriter(containersWriter) for _, v := range containerMap { row, err := json.Marshal(v) if err != nil { log.Fatal(err) } r := append(row, []byte("\n")...) n1, err := gw.Write(r) if err != nil { log.Fatal(err) } _ = n1 } gw.Close() containersWriter.Close() }() go func() { gw := gzip.NewWriter(containerTracksWriter) for _, v := range containerTrackMap { row, err := json.Marshal(v) if err != nil { log.Fatal(err) } r := append(row, []byte("\n")...) n1, err := gw.Write(r) if err != nil { log.Fatal(err) } _ = n1 } gw.Close() containerTracksWriter.Close() }() go func() { gw := gzip.NewWriter(summaryWriter) for _, v := range summaryMap { row, err := json.Marshal(v) if err != nil { log.Fatal(err) } r := append(row, []byte("\n")...) n1, err := gw.Write(r) if err != nil { log.Fatal(err) } _ = n1 } gw.Close() summaryWriter.Close() }() tracksResult, err := tracksUploader.Upload(&s3manager.UploadInput{ Bucket: aws.String(s3TargetBucket), Key: aws.String(makeStreamsTracksTargetS3Key(date, licensor, vendorID)), Body: tracksReader, ContentType: aws.String("application/x-ndjson"), ContentEncoding: aws.String("gzip"), ACL: aws.String("bucket-owner-full-control"), }) if err != nil { log.Fatal(err) } _ = tracksResult containersResult, err := containersUploader.Upload(&s3manager.UploadInput{ Bucket: aws.String(s3TargetBucket), Key: aws.String(makeStreamsContainersTargetS3Key(date, licensor, vendorID)), Body: containersReader, ContentType: aws.String("application/x-ndjson"), ContentEncoding: aws.String("gzip"), ACL: aws.String("bucket-owner-full-control"), }) if err != nil { log.Fatal(err) } _ = containersResult containerTracksResult, err := containerTracksUploader.Upload(&s3manager.UploadInput{ Bucket: aws.String(s3TargetBucket), Key: aws.String(makeStreamsContainerTracksTargetS3Key(date, licensor, vendorID)), Body: containerTracksReader, ContentType: aws.String("application/x-ndjson"), ContentEncoding: aws.String("gzip"), ACL: aws.String("bucket-owner-full-control"), }) if err != nil { log.Fatal(err) } _ = containerTracksResult summaryResult, err := summaryUploader.Upload(&s3manager.UploadInput{ Bucket: aws.String(s3TargetBucket), Key: aws.String(makeStreamsSummaryTargetS3Key(date, licensor, vendorID)), Body: summaryReader, ContentType: aws.String("application/x-ndjson"), ContentEncoding: aws.String("gzip"), ACL: aws.String("bucket-owner-full-control"), }) if err != nil { log.Fatal(err) } _ = summaryResult } func processStreamsZIP(s3SourceService *s3.S3, s3TargetService *s3.S3, s3SourceBucket, s3TargetBucket, date string, licensor string, vendorID string, content map[string]contentItem) { input := &s3.GetObjectInput{ Bucket: aws.String(s3SourceBucket), Key: aws.String(makeStreamsSourceS3Key(date, licensor, vendorID)), } result, err := s3SourceService.GetObject(input) if err != nil { log.Fatal("Can't get object", input, err) } tempFileUUID := uuid.New() tempFile, err := ioutil.TempFile(os.TempDir(), tempFileUUID.String()) if err != nil { log.Fatal(err) } defer os.Remove(tempFile.Name()) // copy content of s3 object to temp file _, err = io.Copy(tempFile, result.Body) if err != nil { log.Fatal(err) } zipReader, err := zip.OpenReader(tempFile.Name()) if err != nil { log.Fatal(err) } defer zipReader.Close() summaryMap := make(map[string]summaryRecord) trackMap := make(map[string]trackRecord) containerMap := make(map[string]containerRecord) containerTrackMap := make(map[string]containerTrackRecord) listenerSummaryMap := make(map[string]map[string]int) listenerTrackMap := make(map[string]map[string]int) listenerContainerMap := make(map[string]map[string]int) listenerContainerTrackMap := make(map[string]map[string]int) for _, f := range zipReader.File { prefix := "AppleMusic_Streams_" + vendorID + "_" + strings.ReplaceAll(date, "-", "") summaryPrefix := "AppleMusic_SummaryStreams_" + vendorID + "_" + strings.ReplaceAll(date, "-", "") containerPrefix := "AppleMusic_Container_" + vendorID + "_" + strings.ReplaceAll(date, "-", "") suffix := ".txt.gz" if (strings.HasPrefix(f.Name, prefix) || strings.HasPrefix(f.Name, summaryPrefix) || strings.HasPrefix(f.Name, containerPrefix)) && strings.HasSuffix(f.Name, suffix) { reader, err := f.Open() if err != nil { log.Fatal(err) } gunzipReader, err := gzip.NewReader(reader) if err != nil { log.Fatal(err) } var headers []string firstItem := true scanner := bufio.NewScanner(gunzipReader) for scanner.Scan() { row := strings.Split(string(scanner.Text()), "\t") if firstItem == true { firstItem = false headers = row } else { streamRecord := parseStreamRow(headers, row) contentItem := content[streamRecord.AppleIdentifier] countryCode := strings.ToLower(streamRecord.StorefrontName) summaryKey := streamRecord.IngestDatestamp + "\xff" + streamRecord.StorefrontName ls, ok2 := listenerSummaryMap[summaryKey] if ok2 == false { m := make(map[string]int) m[streamRecord.AnonymizedPersonID] = 1 listenerSummaryMap[summaryKey] = m } else { ls[streamRecord.AnonymizedPersonID] = 1 listenerSummaryMap[summaryKey] = ls } s, ok := summaryMap[summaryKey] if ok == false { s = summaryRecord{ ReportDate: date, ReportVendorID: vendorID, Date: streamRecord.IngestDatestamp, CountryCode: countryCode, Licensor: licensor, } summaryMap[summaryKey] = s } summaryMap[summaryKey] = mutateSummary(s, streamRecord) if streamRecord.ContainerID != "" { containerKey := streamRecord.IngestDatestamp + "\xff" + streamRecord.StorefrontName + "\xff" + streamRecord.ContainerID lc, ok := listenerContainerMap[containerKey] if ok == false { m := make(map[string]int) m[streamRecord.AnonymizedPersonID] = 1 listenerContainerMap[containerKey] = m } else { lc[streamRecord.AnonymizedPersonID] = 1 listenerContainerMap[containerKey] = lc } c, ok := containerMap[containerKey] if ok == false { c = containerRecord{ ReportDate: date, ReportVendorID: vendorID, Date: streamRecord.IngestDatestamp, CountryCode: countryCode, Licensor: licensor, ContainerID: streamRecord.ContainerID, ContainerSubType: streamRecord.ContainerSubType, ContainerName: streamRecord.ContainerName, } containerMap[containerKey] = c } containerMap[containerKey] = mutateContainer(c, streamRecord) } if contentItem.ISRC != "" { trackKey := streamRecord.IngestDatestamp + "\xff" + streamRecord.StorefrontName + "\xff" + contentItem.ISRC lt, ok2 := listenerTrackMap[trackKey] if ok2 == false { m := make(map[string]int) m[streamRecord.AnonymizedPersonID] = 1 listenerTrackMap[trackKey] = m } else { lt[streamRecord.AnonymizedPersonID] = 1 listenerTrackMap[trackKey] = lt } t, ok := trackMap[trackKey] if ok == false { t = trackRecord{ ReportDate: date, ReportVendorID: vendorID, Date: streamRecord.IngestDatestamp, CountryCode: countryCode, Licensor: licensor, ISRC: contentItem.ISRC, } trackMap[trackKey] = t } trackMap[trackKey] = mutateTrack(t, streamRecord) if streamRecord.ContainerID != "" { containerTrackKey := streamRecord.IngestDatestamp + "\xff" + streamRecord.StorefrontName + "\xff" + streamRecord.ContainerID + "\xff" + contentItem.ISRC lct, ok2 := listenerContainerTrackMap[containerTrackKey] if ok2 == false { m := make(map[string]int) m[streamRecord.AnonymizedPersonID] = 1 listenerContainerTrackMap[containerTrackKey] = m } else { lct[streamRecord.AnonymizedPersonID] = 1 listenerContainerTrackMap[containerTrackKey] = lct } ct, ok := containerTrackMap[containerTrackKey] if ok == false { ct = containerTrackRecord{ ReportDate: date, ReportVendorID: vendorID, Date: streamRecord.IngestDatestamp, CountryCode: countryCode, Licensor: licensor, ContainerID: streamRecord.ContainerID, ContainerSubType: streamRecord.ContainerSubType, ContainerName: streamRecord.ContainerName, ISRC: contentItem.ISRC, } containerTrackMap[containerTrackKey] = ct } containerTrackMap[containerTrackKey] = mutateContainerTrack(ct, streamRecord) } } } } if err := scanner.Err(); err != nil { log.Fatal(err) } reader.Close() gunzipReader.Close() } } for k, v := range summaryMap { v.Listeners = len(listenerSummaryMap[k]) summaryMap[k] = v } for k, v := range trackMap { v.Listeners = len(listenerTrackMap[k]) trackMap[k] = v } for k, v := range containerMap { v.Listeners = len(listenerContainerMap[k]) containerMap[k] = v } for k, v := range containerTrackMap { v.Listeners = len(listenerContainerTrackMap[k]) containerTrackMap[k] = v } tracksSession := session.Must(session.NewSession(&aws.Config{Region: aws.String(s3TargetService.SigningRegion), S3UseAccelerate: aws.Bool(false)})) tracksUploader := s3manager.NewUploader(tracksSession) tracksReader, tracksWriter := io.Pipe() containersSession := session.Must(session.NewSession(&aws.Config{Region: aws.String(s3TargetService.SigningRegion), S3UseAccelerate: aws.Bool(false)})) containersUploader := s3manager.NewUploader(containersSession) containersReader, containersWriter := io.Pipe() containerTracksSession := session.Must(session.NewSession(&aws.Config{Region: aws.String(s3TargetService.SigningRegion), S3UseAccelerate: aws.Bool(false)})) containerTracksUploader := s3manager.NewUploader(containerTracksSession) containerTracksReader, containerTracksWriter := io.Pipe() summarySession := session.Must(session.NewSession(&aws.Config{Region: aws.String(s3TargetService.SigningRegion), S3UseAccelerate: aws.Bool(false)})) summaryUploader := s3manager.NewUploader(summarySession) summaryReader, summaryWriter := io.Pipe() go func() { gw := gzip.NewWriter(tracksWriter) for _, v := range trackMap { row, err := json.Marshal(v) if err != nil { log.Fatal(err) } r := append(row, []byte("\n")...) n1, err := gw.Write(r) if err != nil { log.Fatal(err) } _ = n1 } gw.Close() tracksWriter.Close() }() go func() { gw := gzip.NewWriter(containersWriter) for _, v := range containerMap { row, err := json.Marshal(v) if err != nil { log.Fatal(err) } r := append(row, []byte("\n")...) n1, err := gw.Write(r) if err != nil { log.Fatal(err) } _ = n1 } gw.Close() containersWriter.Close() }() go func() { gw := gzip.NewWriter(containerTracksWriter) for _, v := range containerTrackMap { row, err := json.Marshal(v) if err != nil { log.Fatal(err) } r := append(row, []byte("\n")...) n1, err := gw.Write(r) if err != nil { log.Fatal(err) } _ = n1 } gw.Close() containerTracksWriter.Close() }() go func() { gw := gzip.NewWriter(summaryWriter) for _, v := range summaryMap { row, err := json.Marshal(v) if err != nil { log.Fatal(err) } r := append(row, []byte("\n")...) n1, err := gw.Write(r) if err != nil { log.Fatal(err) } _ = n1 } gw.Close() summaryWriter.Close() }() tracksResult, err := tracksUploader.Upload(&s3manager.UploadInput{ Bucket: aws.String(s3TargetBucket), Key: aws.String(makeStreamsTracksTargetS3Key(date, licensor, vendorID)), Body: tracksReader, ContentType: aws.String("application/x-ndjson"), ContentEncoding: aws.String("gzip"), ACL: aws.String("bucket-owner-full-control"), }) if err != nil { log.Fatal(err) } _ = tracksResult containersResult, err := containersUploader.Upload(&s3manager.UploadInput{ Bucket: aws.String(s3TargetBucket), Key: aws.String(makeStreamsContainersTargetS3Key(date, licensor, vendorID)), Body: containersReader, ContentType: aws.String("application/x-ndjson"), ContentEncoding: aws.String("gzip"), ACL: aws.String("bucket-owner-full-control"), }) if err != nil { log.Fatal(err) } _ = containersResult containerTracksResult, err := containerTracksUploader.Upload(&s3manager.UploadInput{ Bucket: aws.String(s3TargetBucket), Key: aws.String(makeStreamsContainerTracksTargetS3Key(date, licensor, vendorID)), Body: containerTracksReader, ContentType: aws.String("application/x-ndjson"), ContentEncoding: aws.String("gzip"), ACL: aws.String("bucket-owner-full-control"), }) if err != nil { log.Fatal(err) } _ = containerTracksResult summaryResult, err := summaryUploader.Upload(&s3manager.UploadInput{ Bucket: aws.String(s3TargetBucket), Key: aws.String(makeStreamsSummaryTargetS3Key(date, licensor, vendorID)), Body: summaryReader, ContentType: aws.String("application/x-ndjson"), ContentEncoding: aws.String("gzip"), ACL: aws.String("bucket-owner-full-control"), }) if err != nil { log.Fatal(err) } _ = summaryResult } func getContent(s3Service *s3.S3, s3SourceBucket, date, licensor, vendorID string) map[string]contentItem { input := &s3.GetObjectInput{ Bucket: aws.String(s3SourceBucket), Key: aws.String(makeContentSourceS3Key(date, licensor, vendorID)), } result, err := s3Service.GetObject(input) if err != nil { log.Fatal(err) } reader, err := gzip.NewReader(result.Body) if err != nil { log.Fatal(err) } defer reader.Close() content := make(map[string]contentItem) var headers []string firstItem := true scanner := bufio.NewScanner(reader) for scanner.Scan() { row := strings.Split(string(scanner.Text()), "\t") if firstItem == true { firstItem = false headers = row } else { itemMap := make(map[string]string) for i, k := range headers { key := strings.ToLower(strings.TrimSpace(k)) value := strings.TrimSpace(row[i]) itemMap[key] = value } item := contentItem{ AppleIdentifier: itemMap["apple identifier"], ISRC: itemMap["isrc"], Title: itemMap["title"], Artist: itemMap["artist"], ArtistID: itemMap["artist id"], ItemType: itemMap["item type"], MediaType: itemMap["media type"], MediaDuration: itemMap["media duration"], VendorIdentifier: itemMap["vendor identifier"], LabelNetworkStudio: itemMap["label/studio/network"], Grid: itemMap["grid"], } content[item.AppleIdentifier] = item } } if err := scanner.Err(); err != nil { log.Fatal(err) } return content } func summaryExists(s3Service *s3.S3, s3TargetBucket, date, licensor, vendorID string) bool { key := makeStreamsSummaryTargetS3Key(date, licensor, vendorID) input := &s3.HeadObjectInput{ Bucket: aws.String(s3TargetBucket), Key: aws.String(key), } result, err := s3Service.HeadObject(input) if err != nil { return false } if result.LastModified.Unix() < 1523663640 { return false } log.Println("summary already exists", key) return true } func runStreamsTask(s3SourceService *s3.S3, s3TargetService *s3.S3, s3SourceBucket, s3TargetBucket, date, licensor, vendorID string) { if summaryExists(s3TargetService, s3TargetBucket, date, licensor, vendorID) == true { log.Println(date, licensor, vendorID, "already processed") return } streamsSource := makeStreamsSourceS3Key(date, licensor, vendorID) streamsExist := sourceFileExists(s3SourceService, s3SourceBucket, streamsSource) if streamsExist == false { log.Println(date, licensor, vendorID, "streams doesn't exist", streamsSource) return } contentSource := makeContentSourceS3Key(date, licensor, vendorID) contentExist := sourceFileExists(s3SourceService, s3SourceBucket, contentSource) if contentExist == false { log.Println(date, licensor, vendorID, "content doesn't exist", contentSource) return } log.Println(date, licensor, vendorID, "fetching content") content := getContent(s3SourceService, s3SourceBucket, date, licensor, vendorID) log.Println(date, licensor, vendorID, "starting streams") if (date < "2019-11-11" || date >= "2021-04-14") { processStreams(s3SourceService, s3TargetService, s3SourceBucket, s3TargetBucket, date, licensor, vendorID, content) } else { processStreamsZIP(s3SourceService, s3TargetService, s3SourceBucket, s3TargetBucket, date, licensor, vendorID, content) } } // RunStreams aggregates streams and uploads to S3 func RunStreams(s3SourceService *s3.S3, s3TargetService *s3.S3, s3SourceBucket, s3TargetBucket string, startDate time.Time, endDate time.Time, licensors []string) { dates := daterange.DateRange(startDate, endDate) log.Println("-----------------------STREAMS HAS STARTED-----------------------") for _, date := range dates { for _, licensor := range licensors { for _, vendorID := range VendorMap[licensor] { runStreamsTask(s3SourceService, s3TargetService, s3SourceBucket, s3TargetBucket, date, licensor, vendorID) } } } }