using MongoDB.Bson; using MongoDB.Bson.Serialization; using MongoDB.Driver; using Serilog; using System.Text.Json; public class MongoLoader { private readonly IMongoCollection _series; private readonly IMongoCollection _episodes; private readonly IMongoCollection _actors; private readonly IMongoCollection _characters; private readonly IMongoCollection _crew; public MongoLoader(IMongoDatabase db) { _series = db.GetCollection("series"); _episodes = db.GetCollection("episodes"); _actors = db.GetCollection("actors"); _characters = db.GetCollection("characters"); _crew = db.GetCollection("crew"); } public async Task<(int Success, int Failed)> LoadJsonToMongoAsync( Dictionary directories, List updateList, Dictionary 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 collection, BsonDocument doc, Dictionary updateDict, string seriesId) { var id = doc.GetValue("id", null); if (id == null) return; var filter = Builders.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 directories, Dictionary updateDict) { var file = Path.Combine(directories["EPISODEDIR"], $"{seriesId}.json"); if (!File.Exists(file)) return; var json = await File.ReadAllTextAsync(file); var array = JsonSerializer.Deserialize>>(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 directories, Dictionary 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(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 directories, Dictionary 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(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); } } } }