wjones/new_dbsync
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Enhance load_json_to_mongodb function to support change detection and statistics tracking for inserted and updated documents.

Browse Source
This commit is contained in:
Wendell Jones
2026-02-05 10:37:17 -05:00
parent df89aad7e6
commit 332ad8e5d8
1 changed files with 55 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
update_mongodb.py
+55 -6
View File
@@ -72,24 +72,30 @@ def find_file_in_multiple_dirs(filename, directories):
return directories_found
def load_json_to_mongodb(mongo_updater, directories, update_list, lprint):
def load_json_to_mongodb(mongo_updater, directories, update_list, lprint, track_changes=True):
"""
Load downloaded JSON files into MongoDB collections.
Load downloaded JSON files into MongoDB collections with change detection.
Processes series, episodes, cast, and crew data from cached JSON files
and inserts them into the appropriate MongoDB collections.
and inserts or updates them in the appropriate MongoDB collections.
Automatically detects if documents are new, updated, or unchanged.
Args:
mongo_updater: dbmongo instance for MongoDB operations
directories: dict containing cache directory paths for different data types
update_list: list of series IDs to process
lprint: logger instance for logging operations
track_changes: bool, if True tracks inserted vs updated documents (default: True)
Returns:
tuple: (successful_count, failed_count) for series processed
tuple: (successful_count, failed_count, stats_dict) for series processed
stats_dict contains: {'inserted': count, 'updated': count}
"""
successful_count = 0
failed_count = 0
stats = {'series_inserted': 0, 'series_updated': 0, 'episodes_inserted': 0,
'episodes_updated': 0, 'cast_inserted': 0, 'cast_updated': 0,
'crew_inserted': 0, 'crew_updated': 0}
lprint.logprint("info", f"Starting to load {len(update_list)} series into MongoDB")
@@ -104,8 +110,21 @@ def load_json_to_mongodb(mongo_updater, directories, update_list, lprint):
# Insert or update series in MongoDB
if series_data:
if track_changes:
existing = mongo_updater.mgseries.find_one({"id": series_data.get("id")})
if existing:
# Check if data has changed
if existing.get("updated") != series_data.get("updated"):
stats['series_updated'] += 1
lprint.logprint("debug", f"Updated series data for ID {seriesid}")
else:
lprint.logprint("debug", f"Series {seriesid} unchanged, skipping")
pass # Document unchanged
else:
stats['series_inserted'] += 1
lprint.logprint("debug", f"Inserted new series data for ID {seriesid}")
mongo_updater.bulk_upsert_by_id(mongo_updater.mgseries, [series_data])
lprint.logprint("debug", f"Loaded series data for ID {seriesid}")
except Exception as e:
lprint.logprint("warning", f"Failed to load series file for {seriesid}: {e}")
failed_count += 1
@@ -122,6 +141,17 @@ def load_json_to_mongodb(mongo_updater, directories, update_list, lprint):
# Add seriesid to each episode
for episode in episodes_data:
episode['seriesid'] = seriesid
if track_changes:
for episode in episodes_data:
existing = mongo_updater.mgepisodes.find_one({"id": episode.get("id")})
if existing:
if existing.get("updated") != episode.get("updated"):
stats['episodes_updated'] += 1
# else: unchanged
else:
stats['episodes_inserted'] += 1
mongo_updater.bulk_upsert_by_id(mongo_updater.mgepisodes, episodes_data)
lprint.logprint("debug", f"Loaded {len(episodes_data)} episodes for series {seriesid}")
except Exception as e:
@@ -145,6 +175,15 @@ def load_json_to_mongodb(mongo_updater, directories, update_list, lprint):
item['character']['seriesid'] = seriesid
characters.append(item['character'])
if track_changes:
for actor in actors:
existing = mongo_updater.mgactors.find_one({"id": actor.get("id")})
if existing:
if existing.get("updated") != actor.get("updated"):
stats['cast_updated'] += 1
else:
stats['cast_inserted'] += 1
if actors:
mongo_updater.bulk_upsert_by_id(mongo_updater.mgactors, actors)
if characters:
@@ -168,6 +207,15 @@ def load_json_to_mongodb(mongo_updater, directories, update_list, lprint):
item['person']['crew_type'] = item.get('type', 'unknown')
crew_list.append(item['person'])
if track_changes:
for crew in crew_list:
existing = mongo_updater.mgcrew.find_one({"id": crew.get("id")})
if existing:
if existing.get("updated") != crew.get("updated"):
stats['crew_updated'] += 1
else:
stats['crew_inserted'] += 1
if crew_list:
mongo_updater.bulk_upsert_by_id(mongo_updater.mgcrew, crew_list)
lprint.logprint("debug", f"Loaded crew data for series {seriesid}")
@@ -182,7 +230,8 @@ def load_json_to_mongodb(mongo_updater, directories, update_list, lprint):
lprint.logprint("error", f"Unexpected error processing series {seriesid}: {e}")
lprint.logprint("info", f"MongoDB loading complete. Successful: {successful_count}, Failed: {failed_count}")
return successful_count, failed_count
lprint.logprint("info", f"Change statistics: {stats}")
return successful_count, failed_count, stats
def main() -> int:
ROOTDIR = os.getcwd()