using System; using System.Collections.Generic; using System.Linq; using NLog; using NzbDrone.Common.Crypto; using NzbDrone.Common.Extensions; using NzbDrone.Core.Configuration; using NzbDrone.Core.DecisionEngine; using NzbDrone.Core.Indexers; using NzbDrone.Core.Jobs; using NzbDrone.Core.Messaging.Events; using NzbDrone.Core.Parser; using NzbDrone.Core.Parser.Model; using NzbDrone.Core.Profiles.Delay; using NzbDrone.Core.Qualities; using NzbDrone.Core.Tv; using NzbDrone.Core.Tv.Events; namespace NzbDrone.Core.Download.Pending { public interface IPendingReleaseService { void Add(DownloadDecision decision); List GetPending(); List GetPendingRemoteEpisodes(int seriesId); List GetPendingQueue(); Queue.Queue FindPendingQueueItem(int queueId); void RemovePendingQueueItems(int queueId); RemoteEpisode OldestPendingRelease(int seriesId, IEnumerable episodeIds); } public class PendingReleaseService : IPendingReleaseService, IHandle, IHandle, IHandle { private readonly IIndexerStatusService _indexerStatusService; private readonly IPendingReleaseRepository _repository; private readonly ISeriesService _seriesService; private readonly IParsingService _parsingService; private readonly IDelayProfileService _delayProfileService; private readonly ITaskManager _taskManager; private readonly IConfigService _configService; private readonly IEventAggregator _eventAggregator; private readonly Logger _logger; public PendingReleaseService(IIndexerStatusService indexerStatusService, IPendingReleaseRepository repository, ISeriesService seriesService, IParsingService parsingService, IDelayProfileService delayProfileService, ITaskManager taskManager, IConfigService configService, IEventAggregator eventAggregator, Logger logger) { _indexerStatusService = indexerStatusService; _repository = repository; _seriesService = seriesService; _parsingService = parsingService; _delayProfileService = delayProfileService; _taskManager = taskManager; _configService = configService; _eventAggregator = eventAggregator; _logger = logger; } public void Add(DownloadDecision decision) { var alreadyPending = GetPendingReleases(); var episodeIds = decision.RemoteEpisode.Episodes.Select(e => e.Id); var existingReports = alreadyPending.Where(r => r.RemoteEpisode.Episodes.Select(e => e.Id) .Intersect(episodeIds) .Any()); if (existingReports.Any(MatchingReleasePredicate(decision.RemoteEpisode.Release))) { _logger.Debug("This release is already pending, not adding again"); return; } _logger.Debug("Adding release to pending releases"); Insert(decision); } public List GetPending() { var releases = _repository.All().Select(p => p.Release).ToList(); if (releases.Any()) { releases = FilterBlockedIndexers(releases); } return releases; } private List FilterBlockedIndexers(List releases) { var blockedIndexers = new HashSet(_indexerStatusService.GetBlockedIndexers().Select(v => v.IndexerId)); return releases.Where(release => !blockedIndexers.Contains(release.IndexerId)).ToList(); } public List GetPendingRemoteEpisodes(int seriesId) { return _repository.AllBySeriesId(seriesId).Select(GetRemoteEpisode).ToList(); } public List GetPendingQueue() { var queued = new List(); var nextRssSync = new Lazy(() => _taskManager.GetNextExecution(typeof(RssSyncCommand))); foreach (var pendingRelease in GetPendingReleases()) { foreach (var episode in pendingRelease.RemoteEpisode.Episodes) { var ect = pendingRelease.Release.PublishDate.AddMinutes(GetDelay(pendingRelease.RemoteEpisode)); if (ect < nextRssSync.Value) { ect = nextRssSync.Value; } else { ect = ect.AddMinutes(_configService.RssSyncInterval); } var queue = new Queue.Queue { Id = GetQueueId(pendingRelease, episode), Series = pendingRelease.RemoteEpisode.Series, Episode = episode, Quality = pendingRelease.RemoteEpisode.ParsedEpisodeInfo.Quality, Title = pendingRelease.Title, Size = pendingRelease.RemoteEpisode.Release.Size, Sizeleft = pendingRelease.RemoteEpisode.Release.Size, RemoteEpisode = pendingRelease.RemoteEpisode, Timeleft = ect.Subtract(DateTime.UtcNow), EstimatedCompletionTime = ect, Status = "Pending", Protocol = pendingRelease.RemoteEpisode.Release.DownloadProtocol }; queued.Add(queue); } } //Return best quality release for each episode var deduped = queued.GroupBy(q => q.Episode.Id).Select(g => { var series = g.First().Series; return g.OrderByDescending(e => e.Quality, new QualityModelComparer(series.Profile)) .ThenBy(q => PrioritizeDownloadProtocol(q.Series, q.Protocol)) .First(); }); return deduped.ToList(); } public Queue.Queue FindPendingQueueItem(int queueId) { return GetPendingQueue().SingleOrDefault(p => p.Id == queueId); } public void RemovePendingQueueItems(int queueId) { var targetItem = FindPendingRelease(queueId); var seriesReleases = _repository.AllBySeriesId(targetItem.SeriesId); var releasesToRemove = seriesReleases.Where( c => c.ParsedEpisodeInfo.SeasonNumber == targetItem.ParsedEpisodeInfo.SeasonNumber && c.ParsedEpisodeInfo.EpisodeNumbers.SequenceEqual(targetItem.ParsedEpisodeInfo.EpisodeNumbers)); _repository.DeleteMany(releasesToRemove.Select(c => c.Id)); } public RemoteEpisode OldestPendingRelease(int seriesId, IEnumerable episodeIds) { return GetPendingRemoteEpisodes(seriesId).Where(r => r.Episodes.Select(e => e.Id).Intersect(episodeIds).Any()) .OrderByDescending(p => p.Release.AgeHours) .FirstOrDefault(); } private List GetPendingReleases() { var result = new List(); foreach (var release in _repository.All()) { var remoteEpisode = GetRemoteEpisode(release); if (remoteEpisode == null) continue; release.RemoteEpisode = remoteEpisode; result.Add(release); } return result; } private RemoteEpisode GetRemoteEpisode(PendingRelease release) { var series = _seriesService.GetSeries(release.SeriesId); //Just in case the series was removed, but wasn't cleaned up yet (housekeeper will clean it up) if (series == null) return null; var episodes = _parsingService.GetEpisodes(release.ParsedEpisodeInfo, series, true); return new RemoteEpisode { Series = series, Episodes = episodes, ParsedEpisodeInfo = release.ParsedEpisodeInfo, Release = release.Release }; } private void Insert(DownloadDecision decision) { _repository.Insert(new PendingRelease { SeriesId = decision.RemoteEpisode.Series.Id, ParsedEpisodeInfo = decision.RemoteEpisode.ParsedEpisodeInfo, Release = decision.RemoteEpisode.Release, Title = decision.RemoteEpisode.Release.Title, Added = DateTime.UtcNow }); _eventAggregator.PublishEvent(new PendingReleasesUpdatedEvent()); } private void Delete(PendingRelease pendingRelease) { _repository.Delete(pendingRelease); _eventAggregator.PublishEvent(new PendingReleasesUpdatedEvent()); } private static Func MatchingReleasePredicate(ReleaseInfo release) { return p => p.Title == release.Title && p.Release.PublishDate == release.PublishDate && p.Release.Indexer == release.Indexer; } private int GetDelay(RemoteEpisode remoteEpisode) { var delayProfile = _delayProfileService.AllForTags(remoteEpisode.Series.Tags).OrderBy(d => d.Order).First(); var delay = delayProfile.GetProtocolDelay(remoteEpisode.Release.DownloadProtocol); var minimumAge = _configService.MinimumAge; return new[] { delay, minimumAge }.Max(); } private void RemoveGrabbed(RemoteEpisode remoteEpisode) { var pendingReleases = GetPendingReleases(); var episodeIds = remoteEpisode.Episodes.Select(e => e.Id); var existingReports = pendingReleases.Where(r => r.RemoteEpisode.Episodes.Select(e => e.Id) .Intersect(episodeIds) .Any()) .ToList(); if (existingReports.Empty()) { return; } var profile = remoteEpisode.Series.Profile.Value; foreach (var existingReport in existingReports) { var compare = new QualityModelComparer(profile).Compare(remoteEpisode.ParsedEpisodeInfo.Quality, existingReport.RemoteEpisode.ParsedEpisodeInfo.Quality); //Only remove lower/equal quality pending releases //It is safer to retry these releases on the next round than remove it and try to re-add it (if its still in the feed) if (compare >= 0) { _logger.Debug("Removing previously pending release, as it was grabbed."); Delete(existingReport); } } } private void RemoveRejected(List rejected) { _logger.Debug("Removing failed releases from pending"); var pending = GetPendingReleases(); foreach (var rejectedRelease in rejected) { var matching = pending.Where(MatchingReleasePredicate(rejectedRelease.RemoteEpisode.Release)); foreach (var pendingRelease in matching) { _logger.Debug("Removing previously pending release, as it has now been rejected."); Delete(pendingRelease); } } } private PendingRelease FindPendingRelease(int queueId) { return GetPendingReleases().First(p => p.RemoteEpisode.Episodes.Any(e => queueId == GetQueueId(p, e))); } private int GetQueueId(PendingRelease pendingRelease, Episode episode) { return HashConverter.GetHashInt31(string.Format("pending-{0}-ep{1}", pendingRelease.Id, episode.Id)); } private int PrioritizeDownloadProtocol(Series series, DownloadProtocol downloadProtocol) { var delayProfile = _delayProfileService.BestForTags(series.Tags); if (downloadProtocol == delayProfile.PreferredProtocol) { return 0; } return 1; } public void Handle(SeriesDeletedEvent message) { _repository.DeleteBySeriesId(message.Series.Id); } public void Handle(EpisodeGrabbedEvent message) { RemoveGrabbed(message.Episode); } public void Handle(RssSyncCompleteEvent message) { RemoveRejected(message.ProcessedDecisions.Rejected); } } }