added rate limiter on the downloader and updated comparison to only look for files from updated or missing series ids.
This commit is contained in:
@@ -0,0 +1,168 @@
|
||||
using System.Text.Json;
|
||||
using System.Threading.RateLimiting;
|
||||
using Serilog;
|
||||
|
||||
namespace dbsync.Services;
|
||||
|
||||
public class Downloader
|
||||
{
|
||||
private static readonly HttpClient _httpClient = new();
|
||||
|
||||
// 20 requests every 10 seconds (global limiter)
|
||||
private static readonly RateLimiter _rateLimiter =
|
||||
new FixedWindowRateLimiter(new FixedWindowRateLimiterOptions
|
||||
{
|
||||
PermitLimit = 20,
|
||||
Window = TimeSpan.FromSeconds(10),
|
||||
QueueProcessingOrder = QueueProcessingOrder.OldestFirst,
|
||||
QueueLimit = 2000
|
||||
});
|
||||
|
||||
#region Public API
|
||||
|
||||
public async Task<List<string>> DownloadAsync(
|
||||
string url,
|
||||
string outputDir,
|
||||
string outputFile,
|
||||
ILogger log)
|
||||
{
|
||||
var episodeIds = new List<string>();
|
||||
|
||||
try
|
||||
{
|
||||
Directory.CreateDirectory(outputDir);
|
||||
|
||||
var jsonString = await GetStringWithRateLimitAsync(url);
|
||||
|
||||
if (url.Contains("episodes", StringComparison.OrdinalIgnoreCase))
|
||||
{
|
||||
episodeIds = ExtractEpisodeIds(jsonString);
|
||||
log.Information("Extracted {Count} episode IDs.", episodeIds.Count);
|
||||
}
|
||||
|
||||
await SaveFormattedJsonAsync(jsonString, outputDir, outputFile);
|
||||
|
||||
log.Information("Downloaded and formatted {OutputFile} successfully.", outputFile);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
log.Error(ex, "Failed to download {OutputFile} from {Url}", outputFile, url);
|
||||
}
|
||||
|
||||
return episodeIds;
|
||||
}
|
||||
|
||||
public async Task DownloadEpisodeExtrasAsync(
|
||||
List<string> episodeIds,
|
||||
Dictionary<string, string> directories,
|
||||
ILogger log)
|
||||
{
|
||||
var guestCastDir = directories["GUESTCAST"];
|
||||
var guestCrewDir = directories["GUESTCREW"];
|
||||
|
||||
Directory.CreateDirectory(guestCastDir);
|
||||
Directory.CreateDirectory(guestCrewDir);
|
||||
|
||||
var tasks = episodeIds.Select(async episodeId =>
|
||||
{
|
||||
try
|
||||
{
|
||||
var guestCastUrl = $"https://api.tvmaze.com/episodes/{episodeId}/guestcast";
|
||||
var guestCrewUrl = $"https://api.tvmaze.com/episodes/{episodeId}/guestcrew";
|
||||
|
||||
await DownloadAndSaveAsync(
|
||||
guestCastUrl,
|
||||
guestCastDir,
|
||||
$"{episodeId}_guestcast.json",
|
||||
log);
|
||||
|
||||
await DownloadAndSaveAsync(
|
||||
guestCrewUrl,
|
||||
guestCrewDir,
|
||||
$"{episodeId}_guestcrew.json",
|
||||
log);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
log.Error(ex, "Failed downloading guest data for episode {EpisodeId}", episodeId);
|
||||
}
|
||||
});
|
||||
|
||||
await Task.WhenAll(tasks);
|
||||
|
||||
log.Information("Completed guestcast and guestcrew downloads for {Count} episodes.", episodeIds.Count);
|
||||
}
|
||||
|
||||
#endregion
|
||||
|
||||
#region Core HTTP Logic (Rate Limited)
|
||||
|
||||
private async Task<string> GetStringWithRateLimitAsync(string url)
|
||||
{
|
||||
using var lease = await _rateLimiter.AcquireAsync();
|
||||
|
||||
if (!lease.IsAcquired)
|
||||
throw new Exception("Rate limit exceeded.");
|
||||
|
||||
var response = await _httpClient.GetAsync(url);
|
||||
response.EnsureSuccessStatusCode();
|
||||
|
||||
return await response.Content.ReadAsStringAsync();
|
||||
}
|
||||
|
||||
private async Task DownloadAndSaveAsync(
|
||||
string url,
|
||||
string outputDir,
|
||||
string outputFile,
|
||||
ILogger log)
|
||||
{
|
||||
try
|
||||
{
|
||||
var jsonString = await GetStringWithRateLimitAsync(url);
|
||||
await SaveFormattedJsonAsync(jsonString, outputDir, outputFile);
|
||||
|
||||
log.Information("Saved {File}", outputFile);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
log.Warning(ex, "Request failed for {Url}", url);
|
||||
}
|
||||
}
|
||||
|
||||
#endregion
|
||||
|
||||
#region JSON Helpers
|
||||
|
||||
private static List<string> ExtractEpisodeIds(string jsonString)
|
||||
{
|
||||
var episodeIds = new List<string>();
|
||||
|
||||
using var jsonDoc = JsonDocument.Parse(jsonString);
|
||||
|
||||
if (jsonDoc.RootElement.ValueKind == JsonValueKind.Array)
|
||||
foreach (var element in jsonDoc.RootElement.EnumerateArray())
|
||||
if (element.TryGetProperty("id", out var idProperty) &&
|
||||
idProperty.TryGetInt32(out var episodeId))
|
||||
episodeIds.Add(episodeId.ToString());
|
||||
|
||||
return episodeIds;
|
||||
}
|
||||
|
||||
private static async Task SaveFormattedJsonAsync(
|
||||
string jsonString,
|
||||
string outputDir,
|
||||
string outputFile)
|
||||
{
|
||||
using var jsonDoc = JsonDocument.Parse(jsonString);
|
||||
|
||||
var formattedJson = JsonSerializer.Serialize(
|
||||
jsonDoc.RootElement,
|
||||
new JsonSerializerOptions { WriteIndented = true });
|
||||
|
||||
var filePath = Path.Combine(outputDir, outputFile);
|
||||
|
||||
await File.WriteAllTextAsync(filePath, formattedJson);
|
||||
}
|
||||
|
||||
#endregion
|
||||
}
|
||||
@@ -0,0 +1,129 @@
|
||||
using dbsync.Config;
|
||||
using Serilog;
|
||||
using TvSyncApp;
|
||||
|
||||
namespace dbsync.Services
|
||||
{
|
||||
public class Loader
|
||||
{
|
||||
private readonly Downloader _downloader;
|
||||
|
||||
public Loader()
|
||||
{
|
||||
_downloader = new Downloader();
|
||||
}
|
||||
|
||||
public async Task<(int Success, int Failed)> LoadJsonAsync(
|
||||
Dictionary<string, string> directories,
|
||||
List<string> updateList,
|
||||
Dictionary<string, long> updateDict,
|
||||
Settings settings)
|
||||
|
||||
{
|
||||
int success = 0;
|
||||
int failed = 0;
|
||||
int completed = 0;
|
||||
int maxParallelDownloads = settings.Global!.MaxParallelDownloads!.Value;
|
||||
|
||||
var semaphore = new System.Threading.SemaphoreSlim(maxParallelDownloads);
|
||||
|
||||
var filesNeeded = new List<(string Directory, string FileName)>();
|
||||
|
||||
foreach (var seriesId in updateList)
|
||||
{
|
||||
foreach (var dir in directories.Values)
|
||||
{
|
||||
string filePath = Path.Combine(dir, $"{seriesId}.json");
|
||||
if (!File.Exists(filePath))
|
||||
filesNeeded.Add((dir, $"{seriesId}.json"));
|
||||
}
|
||||
}
|
||||
|
||||
Log.Information("Downloading {Count} files...", filesNeeded.Count);
|
||||
var totalFiles = filesNeeded.Count;
|
||||
var progress = new Progress<int>(value =>
|
||||
{
|
||||
completed += value;
|
||||
Console.Write($"\rDownloading files: {completed}/{totalFiles}");
|
||||
});
|
||||
|
||||
var tasks = new List<Task>();
|
||||
foreach (var file in filesNeeded)
|
||||
{
|
||||
tasks.Add(DownloadWithThrottleAsync(
|
||||
file.Directory,
|
||||
file.FileName,
|
||||
directories,
|
||||
settings,
|
||||
semaphore,
|
||||
progress));
|
||||
|
||||
}
|
||||
|
||||
await Task.WhenAll(tasks);
|
||||
Console.WriteLine("\nDownload complete.");
|
||||
|
||||
foreach (var file in filesNeeded)
|
||||
{
|
||||
string path = Path.Combine(file.Directory, file.FileName);
|
||||
if (File.Exists(path)) success++;
|
||||
else failed++;
|
||||
}
|
||||
|
||||
return (success, failed);
|
||||
}
|
||||
|
||||
private async Task DownloadWithThrottleAsync(
|
||||
string directory,
|
||||
string fileName,
|
||||
Dictionary<string, string> directories,
|
||||
Settings settings,
|
||||
SemaphoreSlim semaphore,
|
||||
IProgress<int> progress)
|
||||
|
||||
{
|
||||
List<string> episodeIds = new List<string>();
|
||||
await semaphore.WaitAsync();
|
||||
try
|
||||
{
|
||||
string seriesId = Path.GetFileNameWithoutExtension(fileName);
|
||||
string url;
|
||||
|
||||
if (directory.Contains("series", StringComparison.OrdinalIgnoreCase))
|
||||
url = settings.Json!.ShowApi!.Replace("<tvmazeid>", seriesId);
|
||||
else if (directory.Contains("episodes", StringComparison.OrdinalIgnoreCase))
|
||||
url = settings.Json!.EpisodeApi!.Replace("<tvmazeid>", seriesId);
|
||||
else if (directory.Contains("cast", StringComparison.OrdinalIgnoreCase))
|
||||
url = settings.Json!.CastApi!.Replace("<tvmazeid>", seriesId);
|
||||
else if (directory.Contains("crew", StringComparison.OrdinalIgnoreCase))
|
||||
url = settings.Json!.CrewApi!.Replace("<tvmazeid>", seriesId);
|
||||
else
|
||||
url = $"https://api.tvmaze.com/shows/{seriesId}";
|
||||
|
||||
// Download the file
|
||||
var epids = await Task.Run(() =>
|
||||
_downloader.DownloadAsync(url, directory, fileName, Log.Logger));
|
||||
|
||||
if (epids.Count > 0)
|
||||
{
|
||||
// await _downloader.DownloadEpisodeExtrasAsync(
|
||||
// epids,
|
||||
// directories,
|
||||
// Log.Logger);
|
||||
}
|
||||
|
||||
// Report progress
|
||||
progress?.Report(1);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
Log.Error(ex, "Failed to download {FileName}", fileName);
|
||||
progress?.Report(1); // still report so progress moves forward
|
||||
}
|
||||
finally
|
||||
{
|
||||
semaphore.Release();
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,189 @@
|
||||
using System.Text.Json;
|
||||
using MongoDB.Bson;
|
||||
using MongoDB.Bson.Serialization;
|
||||
using MongoDB.Driver;
|
||||
using Serilog;
|
||||
|
||||
namespace dbsync.Services;
|
||||
|
||||
public class MongoLoader
|
||||
{
|
||||
private readonly IMongoCollection<BsonDocument> _series;
|
||||
private readonly IMongoCollection<BsonDocument> _episodes;
|
||||
private readonly IMongoCollection<BsonDocument> _actors;
|
||||
private readonly IMongoCollection<BsonDocument> _characters;
|
||||
private readonly IMongoCollection<BsonDocument> _crew;
|
||||
|
||||
public MongoLoader(IMongoDatabase db)
|
||||
{
|
||||
_series = db.GetCollection<BsonDocument>("series");
|
||||
_episodes = db.GetCollection<BsonDocument>("episodes");
|
||||
_actors = db.GetCollection<BsonDocument>("actors");
|
||||
_characters = db.GetCollection<BsonDocument>("characters");
|
||||
_crew = db.GetCollection<BsonDocument>("crew");
|
||||
}
|
||||
|
||||
public async Task<(int Success, int Failed)> LoadJsonToMongoAsync(
|
||||
Dictionary<string, string> directories,
|
||||
List<string> updateList,
|
||||
Dictionary<string, long> updateDict)
|
||||
{
|
||||
int success = 0;
|
||||
int failed = 0;
|
||||
|
||||
var options = new ParallelOptions
|
||||
{
|
||||
MaxDegreeOfParallelism = Environment.ProcessorCount // or set manually (e.g. 4)
|
||||
};
|
||||
|
||||
await Parallel.ForEachAsync(updateList, options, async (seriesId, cancellationToken) =>
|
||||
{
|
||||
try
|
||||
{
|
||||
var seriesFile = Path.Combine(directories["SERIESDIR"], $"{seriesId}.json");
|
||||
if (!File.Exists(seriesFile))
|
||||
return;
|
||||
|
||||
var json = await File.ReadAllTextAsync(seriesFile, cancellationToken);
|
||||
var doc = BsonDocument.Parse(json);
|
||||
|
||||
await UpsertWithTimestamp(_series, doc, updateDict, seriesId);
|
||||
|
||||
await ProcessEpisodes(seriesId, directories, updateDict);
|
||||
await ProcessCast(seriesId, directories, updateDict);
|
||||
await ProcessCrew(seriesId, directories, updateDict);
|
||||
|
||||
Interlocked.Increment(ref success);
|
||||
|
||||
Log.Information("Processed series {SeriesId}", seriesId);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
Interlocked.Increment(ref failed);
|
||||
Log.Error(ex, "Failed processing series {SeriesId}", seriesId);
|
||||
}
|
||||
});
|
||||
|
||||
return (success, failed);
|
||||
}
|
||||
|
||||
|
||||
private async Task UpsertWithTimestamp(
|
||||
IMongoCollection<BsonDocument> collection,
|
||||
BsonDocument doc,
|
||||
Dictionary<string, long> updateDict,
|
||||
string seriesId)
|
||||
{
|
||||
var id = doc.GetValue("id", null);
|
||||
if (id == null) return;
|
||||
|
||||
var filter = Builders<BsonDocument>.Filter.Eq("id", id);
|
||||
var existing = await collection.Find(filter).FirstOrDefaultAsync();
|
||||
|
||||
if (updateDict.TryGetValue(seriesId, out var newTimestamp))
|
||||
{
|
||||
if (existing == null ||
|
||||
!existing.Contains("updated") ||
|
||||
existing["updated"].AsInt64 < newTimestamp)
|
||||
{
|
||||
await collection.ReplaceOneAsync(filter, doc,
|
||||
new ReplaceOptions { IsUpsert = true });
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private async Task ProcessEpisodes(
|
||||
string seriesId,
|
||||
Dictionary<string, string> directories,
|
||||
Dictionary<string, long> updateDict)
|
||||
{
|
||||
var file = Path.Combine(directories["EPISODEDIR"], $"{seriesId}.json");
|
||||
if (!File.Exists(file)) return;
|
||||
|
||||
var json = await File.ReadAllTextAsync(file);
|
||||
var array = JsonSerializer.Deserialize<List<Dictionary<string, object>>>(json);
|
||||
|
||||
if (array == null) return;
|
||||
|
||||
foreach (var episode in array)
|
||||
{
|
||||
episode["seriesid"] = int.Parse(seriesId);
|
||||
var doc = BsonDocument.Parse(JsonSerializer.Serialize(episode));
|
||||
await UpsertWithTimestamp(_episodes, doc, updateDict, seriesId);
|
||||
}
|
||||
}
|
||||
|
||||
private async Task ProcessCast(
|
||||
string seriesId,
|
||||
Dictionary<string, string> directories,
|
||||
Dictionary<string, long> updateDict)
|
||||
{
|
||||
var file = Path.Combine(directories["CASTDIR"], $"{seriesId}.json");
|
||||
if (!File.Exists(file)) return;
|
||||
|
||||
var json = await File.ReadAllTextAsync(file);
|
||||
|
||||
// ✅ Correct way to parse root JSON array
|
||||
var array = BsonSerializer.Deserialize<BsonArray>(json);
|
||||
|
||||
foreach (var item in array)
|
||||
{
|
||||
if (!item.IsBsonDocument)
|
||||
continue;
|
||||
|
||||
var doc = item.AsBsonDocument;
|
||||
|
||||
if (doc.Contains("person") && doc["person"].IsBsonDocument)
|
||||
{
|
||||
var person = doc["person"].AsBsonDocument;
|
||||
person["seriesid"] = int.Parse(seriesId);
|
||||
|
||||
await UpsertWithTimestamp(_actors, person, updateDict, seriesId);
|
||||
}
|
||||
|
||||
if (doc.Contains("character") && doc["character"].IsBsonDocument)
|
||||
{
|
||||
var character = doc["character"].AsBsonDocument;
|
||||
character["seriesid"] = int.Parse(seriesId);
|
||||
|
||||
await UpsertWithTimestamp(_characters, character, updateDict, seriesId);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private async Task ProcessCrew(
|
||||
string seriesId,
|
||||
Dictionary<string, string> directories,
|
||||
Dictionary<string, long> updateDict)
|
||||
{
|
||||
var file = Path.Combine(directories["CREWDIR"], $"{seriesId}.json");
|
||||
if (!File.Exists(file)) return;
|
||||
|
||||
var json = await File.ReadAllTextAsync(file);
|
||||
|
||||
// ✅ Correct way to parse root JSON array
|
||||
var array = BsonSerializer.Deserialize<BsonArray>(json);
|
||||
|
||||
foreach (var item in array)
|
||||
{
|
||||
if (!item.IsBsonDocument)
|
||||
continue;
|
||||
|
||||
var doc = item.AsBsonDocument;
|
||||
|
||||
if (doc.Contains("person") && doc["person"].IsBsonDocument)
|
||||
{
|
||||
var person = doc["person"].AsBsonDocument;
|
||||
|
||||
person["seriesid"] = int.Parse(seriesId);
|
||||
|
||||
// Safely get crew type
|
||||
person["crew_type"] = doc.Contains("type")
|
||||
? doc["type"]
|
||||
: "unknown";
|
||||
|
||||
await UpsertWithTimestamp(_crew, person, updateDict, seriesId);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user