//using MoreLinq; //using MySql.Data.MySqlClient; //using Newtonsoft.Json; //using NLog; //using NodaTime.Serialization.JsonNet; //using NodaTime.TimeZones; //using PetaPoco.Business; //using Sony.Filtr.Contracts.Abstractions; //using Sony.Filtr.Contracts.Definitions; //using Sony.Filtr.Database; //using Sony.Filtr.SpotifyAnalytics; //using Sony.Filtr.SpotifyAnalytics.Data; //using Sony.Filtr.SpotifyAnalytics.Models; //using Sony.Filtr.Utility.Extensions; //using System; //using System.Collections.Concurrent; //using System.Collections.Generic; //using System.Globalization; //using System.IO; //using System.Linq; //using System.Runtime.Caching; //using System.Threading; //using System.Threading.Tasks; //using Sony.Filtr.Contracts.Entities; //using Sony.Filtr.Playlists.Spotify; //using Sony.Filtr.Utility.ConsoleUtility; //namespace Sony.Filtr.Tasks.Tasks.Spotify //{ // public class UpdateSpotifyAnalyticsAggregatedTask : IScheduledTask // { // private readonly SpotifyStreamingAggregatedReportApi _aggregatedReportApi; // private readonly IApplicationInstanceManager _applicationInstanceManager; // private readonly SpotifyAnalyticsManager _spotifyAnalyticsManager; // private readonly SpotifyPlaylistManager _spotifyPlaylistManager; // private readonly IBuzzAccountManager _buzzAccountManager; // private readonly Logger _logger; // public UpdateSpotifyAnalyticsAggregatedTask(SpotifyStreamingAggregatedReportApi aggregatedReportApi, IApplicationInstanceManager applicationInstanceManager, SpotifyAnalyticsManager spotifyAnalyticsManager, SpotifyPlaylistManager spotifyPlaylistManager, IBuzzAccountManager buzzAccountManager) // { // _aggregatedReportApi = aggregatedReportApi; // _applicationInstanceManager = applicationInstanceManager; // _spotifyAnalyticsManager = spotifyAnalyticsManager; // _spotifyPlaylistManager = spotifyPlaylistManager; // _buzzAccountManager = buzzAccountManager; // _logger = LogManager.GetLogger("SpotifyStreamingAggregatedReportApi"); // } // public async Task ExecuteAsync(Guid scheduledTaskLogId) // { // var taskLog = new ScheduledTaskLog(); // var fromDate = new DateTime(2018, 03, 21); // var endDate = DateTime.Today; // var dates = fromDate.GetDateRangeTo(endDate).Reverse().ToList(); // if (!Directory.Exists(TempFolder)) // { // Directory.CreateDirectory(TempFolder); // } // await dates.ForEachAsync(1, async date => // { // var files = await GetFilesToImport(date); // _logger.Debug($"Found {files.Count} files for {date.ToShortDateString()}"); // var playlistFiles = files.Where(p => p.FileType == SpotifyS3FileType.Playlists && p.Version == 2).ToList(); // if (playlistFiles.Any()) // { // bool importedPlaylistsForDay = false; // var playlistData = await GetFiltrPlaylistData(); // ConcurrentBag playlistUrisToAdd = new ConcurrentBag(); // await playlistFiles.ForEachAsync(2, async playlistFile => // { // var localPath = await DownloadFileAsync(playlistFile); // _logger.Debug($"Reading playlists streams from {playlistFile.FilePath}"); // var playlistStreams = ReadNJsonData(localPath, playlistFile.DistributorId).ToList(); // foreach (var playlistStream in playlistStreams) // { // var playlistUri = playlistStream.PlaylistUri; // var playlistLink = new SpotifyLink(playlistUri); // var playlistId = playlistLink.ExtractPlaylistID(); // if (!playlistData.AllPlaylists.Contains(playlistUri)) // { // var playlistUserName = playlistLink.ExtractSpotifyPlaylistUserName(); // if (playlistData.BuzzImportUsers.Contains(playlistUserName)) // { // playlistUrisToAdd.Add(playlistLink); // } // } // //Assuming we get the playlistUri in either format spotify:playlist:37i9dQZF1DXcBWIGoYBM5M or spotify:user:spotify:playlist:37i9dQZF1DXcBWIGoYBM5M we only want to store the Id part. // playlistStream.PlaylistUri = playlistId; // } // _logger.Debug($"Importing playlists streams from {playlistFile.FilePath}"); // PetaPocoRepository.Instance.ImportBulkFileLoader(playlistStreams, MySqlBulkLoaderConflictOption.Replace); // _logger.Debug($"Done importing playlists streams from {playlistFile.FilePath}"); // MarkAsProcessed(date, playlistFile); // importedPlaylistsForDay = true; // File.Delete(localPath); // }); // if (importedPlaylistsForDay) // { // var distinctPlaylistsToAdd = playlistUrisToAdd.Distinct().ToList(); // await ImportNewPlaylistsAsync(distinctPlaylistsToAdd, date); // var playlistsToAggregate = await GetPlaylistsWithStreams(date); // await AggregatePlaylistStreamsAsync(playlistsToAggregate, date); // await CalculatePlaylistCategoryStreamsAsync(date); // //try // //{ // // await SetPlaylistStreamsSummaryAsync(); // //} // //catch (Exception e) // //{ // // _logger.Error(e); // //} // } // } // var trackFiles = files.Except(playlistFiles).ToList(); // await trackFiles.Where(p => p.FileType != SpotifyS3FileType.Playlists).ForEachAsync(1, async trackFile => // { // try // { // var localPath = await DownloadFileAsync(trackFile); // if (trackFile.FileType == SpotifyS3FileType.Tracks) // { // _logger.Debug($"Reading tracks from {trackFile.FilePath}"); // var tracks = ReadNJsonData(localPath, trackFile.DistributorId); // var orderedTracks = tracks.OrderBy(p => p.country_code).ThenBy(p => p.report_country_code).ThenBy(p => p.date).ThenBy(p => p.isrc); // int batchIndex = 1; // foreach(var batch in orderedTracks.Batch(100000)) // { // _logger.Debug($"Importing batch {batchIndex} from {trackFile.FilePath}"); // await PetaPocoRepository.Instance.ImportBulkFileLoaderAsync(batch); // _logger.Debug($"Done batch {batchIndex} from {trackFile.FilePath}"); // batchIndex++; // } // MarkAsProcessed(date, trackFile); // } // else if (trackFile.FileType == SpotifyS3FileType.Demographics) // { // await ImportAsync(trackFile, localPath, date, "Demographics"); // } // else if (trackFile.FileType == SpotifyS3FileType.SaveSkips) // { // await ImportAsync(trackFile, localPath, date, "tracks save/skips"); // } // else if (trackFile.FileType == SpotifyS3FileType.Summary) // { // _logger.Debug($"Reading tracks country summary from {trackFile.FilePath}"); // var summary = ReadNJsonData(localPath, trackFile.DistributorId); // var oldStructure = await MapToOldStreamSummaryStructureAsync(trackFile.Date, trackFile.DistributorId, trackFile.Country, summary.ToList()); // await PetaPocoRepository.Instance.ImportBulkFileLoaderAsync(new[] { oldStructure }, MySqlBulkLoaderConflictOption.Replace); // MarkAsProcessed(date, trackFile); // } // else if (trackFile.FileType == SpotifyS3FileType.PlaylistTracks) // { // _logger.Debug($"Reading playlist-track streams from {trackFile.FilePath}"); // var playlistTrackStreams = ReadNJsonData(localPath, trackFile.DistributorId); // _logger.Debug($"Importing playlist-track streams from {trackFile.FilePath}"); // PetaPocoRepository.Instance.ImportBulkFileLoader(playlistTrackStreams); // _logger.Debug($"Done importing playlist-track streams from {trackFile.FilePath}"); // MarkAsProcessed(date, trackFile); // } // else if (trackFile.FileType == SpotifyS3FileType.ListenerSummaryTracks) // { // _logger.Debug($"Reading listener summary for track streams from {trackFile.FilePath}"); // var trackListeners = ReadNJsonData(localPath, trackFile.DistributorId); // _logger.Debug($"Importing listener summary for track streams from {trackFile.FilePath}"); // PetaPocoRepository.Instance.ImportBulkFileLoader(trackListeners); // _logger.Debug($"Done importing listener summary for track streams from {trackFile.FilePath}"); // MarkAsProcessed(date, trackFile); // } // else if (trackFile.FileType == SpotifyS3FileType.ListenerSummaryPlaylistTracks) // { // _logger.Debug($"Reading listener summary for playlist-track streams from {trackFile.FilePath}"); // var playlistTrackStreams = ReadNJsonData(localPath, trackFile.DistributorId); // _logger.Debug($"Importing listener summary for playlist-track streams from {trackFile.FilePath}"); // PetaPocoRepository.Instance.ImportBulkFileLoader(playlistTrackStreams); // _logger.Debug($"Done importing listener summary for playlist-track streams from {trackFile.FilePath}"); // MarkAsProcessed(date, trackFile); // } // else if (trackFile.FileType == SpotifyS3FileType.ListenerSummaryPlaylists) // { // _logger.Debug($"Reading listener summary for playlist streams from {trackFile.FilePath}"); // var playlistListeners = ReadNJsonData(localPath, trackFile.DistributorId); // _logger.Debug($"Importinglistener summary for playlist streams from {trackFile.FilePath}"); // PetaPocoRepository.Instance.ImportBulkFileLoader(playlistListeners); // _logger.Debug($"Done importing listener summary for playlist streams from {trackFile.FilePath}"); // MarkAsProcessed(date, trackFile); // } // File.Delete(localPath); // } // catch(Exception ex) // { // _logger.Error(ex); // } // }); // }); // return taskLog; // } // private async Task GetFiltrPlaylistData() // { // await _filtrPlaylistDataSemaphore.WaitAsync(); // try // { // if (_filtrPlaylistData == null) // { // var allPlaylists = _spotifyPlaylistManager.GetAllPlaylistUris().ToHashSet(); // var allBuzzUsers = await _buzzAccountManager.GetBuzzUsersAsync(musicServiceId: (int)MusicService.Spotify); // var buzzImportUsers = allBuzzUsers.Where(p => p.BuzzCategoryId == (int)StaticBuzzCategory.Spotify || p.BuzzCategoryId == (int)StaticBuzzCategory.SonyMusic).Select(p => p.Username).ToHashSet(StringComparer.InvariantCultureIgnoreCase); // _filtrPlaylistData = new FiltrPlaylistData // { // AllPlaylists = allPlaylists, // BuzzImportUsers = buzzImportUsers // }; // } // } // finally // { // _filtrPlaylistDataSemaphore.Release(); // } // return _filtrPlaylistData; // } // public async Task CalculatePlaylistCategoryStreamsAsync2(List spotifyPlaylistsWithStreamForDate, DateTime date) // { // var playlistInBuzzCategories = new List(); // using (var conn = await DatabaseHandler.GetOpenConnectionAsync()) // { // const string sql = "SELECT p.PlaylistUri, p.CountryCode, p.BuzzCategoryId " + // "FROM tblSpotifyPlaylist AS p " + // "WHERE p.BuzzCategoryId IN (@sonyCategoryId, @spotifyCategoryId) "; // MySqlCommand cmd = new MySqlCommand(sql, conn); // cmd.Parameters.AddWithValue("@spotifyMusicServiceId", MusicService.Spotify); // cmd.Parameters.AddWithValue("@sonyCategoryId", StaticBuzzCategory.SonyMusic); // cmd.Parameters.AddWithValue("@spotifyCategoryId", StaticBuzzCategory.Spotify); // var reader = await cmd.ExecuteReaderAsync(); // while (reader.Read()) // { // playlistInBuzzCategories.Add(new PlaylistStreamGroup() // { // PlaylistUri = reader.GetString("playlistUri"), // BuzzCategoryId = reader.GetInt32("buzzCategoryId"), // PlaylistCountry = reader.GetString("CountryCode") // }); // } // } // var spotifyPlaylistWithStreamsLookup = spotifyPlaylistsWithStreamForDate.ToHashSet(); // var playlistToProcess = playlistInBuzzCategories.Where(p => spotifyPlaylistWithStreamsLookup.Contains(p.PlaylistUri)).ToList(); // using (var progress = new ConsoleProgressBarCount(playlistToProcess.Count)) // { // await playlistToProcess.ForEachAsync(10, async playlistGroup => // { // try // { // using (var conn = await DatabaseHandler.GetOpenConnectionAsync()) // { // const string sql = "SELECT DistributerId, CountryCode, SUM(Streams), SUM(Listeners) " + // "FROM tblSpotifyPlaylistStream WHERE Date = @date AND PlaylistUri = @playlistUri " + // "GROUP BY DistributerId, Date, CountryCode"; // MySqlCommand cmd = new MySqlCommand(sql, conn); // cmd.Parameters.AddWithValue("@date", date); // cmd.Parameters.AddWithValue("@PlaylistUri", playlistGroup.PlaylistUri); // var reader = await cmd.ExecuteReaderAsync(); // while (reader.Read()) // { // playlistGroup.Streams.Add(new PlaylistStreams() // { // DistributerId = reader.GetInt32(0), // CountryCode = reader.GetString(1), // Streams = reader.GetInt32(2), // Listeners = reader.GetInt32(3), // }); // } // } // } // catch (Exception e) // { // _logger.Error(e); // } // }, progress); // } // var applications = _applicationInstanceManager.GetApplications().Where(a => !a.GlobalPushApplication && !a.WorkoutMarket).DistinctBy(a => a.SpotifyRegionCode); // var accounts = new List() { SpotifyAnalyticsAccount.Sony, SpotifyAnalyticsAccount.Orchard, SpotifyAnalyticsAccount.SonyMusicEntertainmentJapan, SpotifyAnalyticsAccount.SonyMusicEntertainmentJapanInternational }; // var accountIds = accounts.Select(p => (int)p); // //var buzzCategories = new List() { StaticBuzzCategory.SonyMusic, StaticBuzzCategory.Spotify }; // //var buzzCategoryIds = buzzCategories.Select(p => (int)p); // var applicationRegions = applications.ToDictionary(k => k.SpotifyRegionCode, v => _applicationInstanceManager.GetApplicationRegions(v).ToHashSet(StringComparer.InvariantCultureIgnoreCase), StringComparer.InvariantCultureIgnoreCase); // var flattPlaylistStreams = playlistToProcess.SelectMany(p => p.Streams, (group, streams) => new // { // BuzzCategoryId = group.BuzzCategoryId, // PlaylistCountryCode = group.PlaylistCountry, // AccountId = streams.DistributerId, // StreamCountryCode = streams.CountryCode, // Streams = streams.Streams, // Listeners = streams.Listeners // }).Where(p => accountIds.Contains(p.AccountId)); // var groupedPlaylistStreams = flattPlaylistStreams.GroupBy(p => new { p.BuzzCategoryId, p.AccountId, p.StreamCountryCode }).ToList(); // ConcurrentBag< PlaylistCategoryStreams2 > categoryStreams = new ConcurrentBag(); // foreach(var application in applicationRegions) // { // var applicationStreams = groupedPlaylistStreams.Where(s => application.Key == "_gl" || application.Key == s.Key.StreamCountryCode || application.Value.Contains(s.Key.StreamCountryCode)) // .Select(p => // new PlaylistCategoryStreams2() // { // BuzzCategoryId = p.Key.BuzzCategoryId, // Date = date, // AccountId = p.Key.AccountId, // Market = application.Key, // LocalListListeners = p.Where(s => application.Value.Contains(s.PlaylistCountryCode)).Sum(s => s.Listeners), // LocalListStreams = p.Where(s => application.Value.Contains(s.PlaylistCountryCode)).Sum(s => s.Streams), // TotalListListeners = p.Sum(s => s.Listeners), // TotalListStreams = p.Sum(s => s.Streams), // }).ToList(); // applicationStreams.ForEach(categoryStreams.Add); // } // await PetaPocoRepository.Instance.ImportBulkFileLoaderAsync(categoryStreams, MySqlBulkLoaderConflictOption.Replace); // } // private class PlaylistStreamGroup // { // public PlaylistStreamGroup() // { // Streams = new ConcurrentBag(); // } // public string PlaylistUri { get; set; } // public int BuzzCategoryId { get; set; } // public string PlaylistCountry { get; set; } // public ConcurrentBag Streams { get; set; } // } // private class PlaylistStreams // { // public int DistributerId { get; set; } // public string CountryCode { get; set; } // public int Streams { get; set; } // public int Listeners { get; set; } // } // } //}