Initial commit

2026-02-16 17:12:34 -05:00
parent 77eb3db44f
commit bdcb596e39
8 changed files with 646 additions and 0 deletions
@@ -0,0 +1,187 @@
+using MongoDB.Bson;
+using MongoDB.Bson.Serialization;
+using MongoDB.Driver;
+using Serilog;
+using System.Text.Json;
+
+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);
+            }
+        }
+    }
+}