From 96bd32356717300aa2552b868832b081e734f97d Mon Sep 17 00:00:00 2001 From: Wendell Jones Date: Thu, 11 Dec 2025 15:20:50 +0000 Subject: [PATCH] Implement bulk upsert functionality in dbmongo class and update configuration for MongoDB database --- db/functions.py | 191 ++++++++++++----------------------- settings/tvsync_settings.cfg | 2 +- update_mongo.py | 11 +- 3 files changed, 64 insertions(+), 140 deletions(-) diff --git a/db/functions.py b/db/functions.py index efc6677..5d39bb7 100644 --- a/db/functions.py +++ b/db/functions.py @@ -19,7 +19,7 @@ from db.pgsql.Merger import Episodemerge as PGEpisodemerge import api.tvmaze.episodes import api.tvmaze.cast import api.tvmaze.series -from pymongo import MongoClient +from pymongo import MongoClient, ReplaceOne, InsertOne from tqdm import tqdm from datetime import datetime @@ -121,11 +121,6 @@ class settype(object): with engine.connect() as conn: with conn as cursor: cursor.execute(text(sql)) - # conn.commit() - # with engine.connect() as conn: - # conn.execute(text("CREATE SCHEMA IF NOT EXISTS updates")) - # conn.execute(text("CREATE SCHEMA IF NOT EXISTS dbo")) - # conn.commit() updatesBase.prepare(engine, reflect=True, schema="updates") dboBase.prepare(engine, reflect=True, schema="dbo") metadata = sqlalchemy.MetaData("updates") @@ -171,6 +166,20 @@ class dbmongo(object): global mgcharacters global mgcrew + def bulk_upsert_by_id(self, collection, docs, batch_size=1000): + ops = [] + for doc in docs: + if "id" not in doc: + # skip documents without an `id` field + continue + # replace existing document with same `id`, or insert if it does not exist + ops.append(ReplaceOne({"id": doc["id"]}, doc, upsert=True)) + if len(ops) >= batch_size: + collection.bulk_write(ops, ordered=False) + ops = [] + if ops: + collection.bulk_write(ops, ordered=False) + def convert_updated_time(self, timestamp): new_timestamp = datetime.fromtimestamp(timestamp) update_time = str(new_timestamp).split(".")[0] @@ -216,8 +225,6 @@ class dbmongo(object): lprint.logprint( "info", f"Unable to open {datatype} file for {seriesid}" ) - # print(f'Unable to open {datatype} file for seriesid {seriesid}') - return None def process_series_data( self, @@ -268,79 +275,17 @@ class dbmongo(object): dboBase, lprint, ) + #self.bulk_upsert_by_id(self.mgseries, seriesdata) self.insert_or_update_mongo( - self.mgseries, seriesdata, "id" + self.mgseries, seriesdata, "id" ) def process_cast_data(self, persondata, characterdata, seriesid, lprint): - try: - aupserts = [ - self.mgactors.update_many( - {"id": x["id"]}, {"$setOnInsert": x}, upsert=True - ) - for x in persondata - ] - bupserts = [ - self.mgcharacters.update_many( - {"id": x["id"]}, {"$setOnInsert": x}, upsert=True - ) - for x in characterdata - ] - except Exception as e: - print(f"Skipping cast data.Reason: {str(e)}") - else: - [ - self.mgactors.update_many( - {"id": x["id"]}, {"$setOnInsert": x}, upsert=True - ) - for x in aupserts - ] - [ - self.mgcharacters.insert_many( - {"id": x["id"]}, {"$setOnInsert": x}, upsert=True - ) - for x in bupserts - ] - + self.bulk_upsert_by_id(self.mgactors, persondata) + self.bulk_upsert_by_id(self.mgcharacters, characterdata) + def process_crew_data(self, crewdata, seriesid, jget, lprint): - try: - upserts = [ - self.mgcrew.update_many( - {"id": x["id"]}, {"$setOnInsert": x}, upsert=True - ) - for x in crewdata - ] - except Exception as e: - print(f"Skipping crew data.Reason: {str(e)}") - else: - [ - self.mgcrew.insert_many( - {"id": x["id"]}, {"$setOnInsert": x}, upsert=True - ) - for x in upserts - ] - - def process_episode_data(self, episodedata, seriesid, lprint): - try: - upserts = [ - self.mgepisodes.update_many( - {"id": x["id"]}, {"$setOnInsert": x}, upsert=True - ) - for x in episodedata - ] - except Exception as e: - print( - f"file does not contain the correct json. " - f"Skipping. {str(e)}" - ) - else: - for episode in upserts: - [ - self.mgepisodes.insert_many( - {"id": x["id"]}, {"$setOnInsert": x}, upsert=True - ) - for x in upserts - ] + self.bulk_upsert_by_id(self.mgcrew, crewdata) def update_country_data( self, country, countrydictionary, dbExec, dbengine, dbBase, @@ -382,9 +327,7 @@ class dbmongo(object): dbExec.rawsql_insert( dbengine["engine"], insert_sql, lprint ) - # seriesresults['countryid'] = 'No Country' - # return seriesresults['countryid'] - + def update_network_data( self, networkchannel, @@ -434,19 +377,9 @@ class dbmongo(object): del new_values["$set"]["_id"] collection.update_one(filter_query, new_values) - def insert_or_update_many_mongo(self, collection, data, unique_key): - try: - collection.insert_many(data) - except Exception: - filter_query = {unique_key: data[unique_key]} - new_values = {"$set": data} - del new_values["$set"]["_id"] - collection.update_one(filter_query, new_values) - def update_mongo( self, updateList, - ROOTDIR, directories, jget, countrydictionary, @@ -464,7 +397,7 @@ class dbmongo(object): failed_file_count = 0 def process_data( - file_path, data_type, process_method, failed_file_count, lprint + file_path, data_type, failed_file_count, lprint ): """Helper function to load and process data.""" data = self.load_json_file( @@ -483,34 +416,25 @@ class dbmongo(object): persondata = [] characterdata = [] for item in data: - #update_time = convert_updated_time(item["person"]["updated"]) + update_time = self.convert_updated_time(item["person"]["updated"]) + item["person"]["updated"] = update_time if "person" in item: + item["person"]["seriesid"] = seriesid persondata.append(item["person"]) if "character" in item: + item["character"]["seriesid"] = seriesid characterdata.append(item["character"]) + try: - for item in persondata: - item["seriesid"] = seriesid - for item in characterdata: - item["seriesid"] = seriesid - except Exception as e: - lprint.logprint( - "info", f"Unable to add {data_type} data. {e}" - ) - - try: - process_method( - persondata, characterdata, seriesid, lprint - ) + self.bulk_upsert_by_id(self.mgactors, persondata) + self.bulk_upsert_by_id(self.mgcharacters, characterdata) failed_file_count = "N/A" - return failed_file_count except Exception as e: failed_file_count = 1 lprint.logprint( "info", f"Unable to process {data_type} data. {e}" ) - return failed_file_count if "crew" in file_path: try: crewdata = [] @@ -529,7 +453,7 @@ class dbmongo(object): if "person" in item: crewdata.append(item["person"]) try: - process_method(crewdata, seriesid, jget, lprint) + self.bulk_upsert_by_id(self.mgcrew, crewdata) failed_file_count = "N/A" return failed_file_count except Exception as e: @@ -565,9 +489,7 @@ class dbmongo(object): countrydictionary, seriesid, ) - self.insert_or_update_mongo( - self.mgseries, seriesdata, "id" - ) + self.bulk_upsert_by_id(self.mgseries, seriesdata) except Exception as e: series_failed_count += 1 lprint.logprint( @@ -580,45 +502,56 @@ class dbmongo(object): else: processed_count += 1 - # Process episode, cast, and crew data using the helper function fcount = process_data( episodecachefile, "episode", - self.process_episode_data, failed_file_count, lprint, ) if isinstance(fcount, int): episode_failed_count = episode_failed_count + fcount else: - for episode in fcount: - self.insert_or_update_mongo( - self.mgepisodes, episode, "id" + try: + self.bulk_upsert_by_id(self.mgepisodes, fcount) + # for episode in fcount: + # self.insert_or_update_mongo( + # self.mgepisodes, episode, "id" + # ) + except Exception as e: + lprint.logprint( + "info", + ( + f"Unable to insert episode data for seriesid " + f"{seriesid}. {e}" + ), ) fcount = process_data( - castcachefile, "cast", self.process_cast_data, - failed_file_count, lprint + castcachefile, "cast",failed_file_count, lprint ) - # Need to split cast between actors and characters collections - # if isinstance(fcount, int): - # cast_failed_count = cast_failed_count + fcount - # else: - # for cast in fcount: - # self.insert_or_update_mongo( - # self.mg, episode, "id" - # ) + if isinstance(fcount, int): + cast_failed_count = cast_failed_count + fcount if os.path.isfile(crewcachefile): fcount = process_data( - crewcachefile, "crew", self.process_crew_data, + crewcachefile, "crew", failed_file_count, lprint ) if isinstance(fcount, int): crew_failed_count = crew_failed_count + fcount else: - for crew in fcount: - self.insert_or_update_mongo( - self.mgcrew, crew, "id" + try: + for crew in fcount: + self.insert_or_update_mongo( + self.mgcrew, crew, "id" + ) + except Exception as e: + lprint.logprint( + "info", + ( + f"Unable to insert crew data for seriesid " + f"{seriesid}. {e}" + ), ) + print(f"Processing complete. {processed_count} series processed.") print(f"Failed to load {series_failed_count} series files.") print(f"Failed to load {episode_failed_count} episode files.") diff --git a/settings/tvsync_settings.cfg b/settings/tvsync_settings.cfg index 5e26cdd..fb15e3a 100644 --- a/settings/tvsync_settings.cfg +++ b/settings/tvsync_settings.cfg @@ -6,7 +6,7 @@ sqlserver_driver=ODBC Driver 13 for SQL Server hostname=192.168.128.7 mghostname=192.168.128.24 dbname=media_dbsync -mgdbname=test2 +mgdbname=tvdata mysqlusername=root pgsqlusername=postgres mssqlusername=sa diff --git a/update_mongo.py b/update_mongo.py index e7bc15e..736a2c2 100644 --- a/update_mongo.py +++ b/update_mongo.py @@ -113,20 +113,12 @@ def main() -> int: ts1 = str(datetime.now()).split(".")[0] lprint.logprint("info", "Sync start time: " + str(ts1)) print_starttime(lprint, ts1) - #updatesBase = dbengine["updatesBase"] dboBase = dbengine["dboBase"] - #session = dbengine["session"] dbExec = db.sql.Dbexec() createTables = dbclass["Createtables"] - #Tables = dbclass["Tables"] createTempTables = dbclass["CreateTemptables"] new_updates = {} - #sqlexec = db.sql.SQLGenerate() - #mt = createTables( - # dbengine["session"], - # dbengine["metadata"], - # dbengine["engine"] - #) + ct = createTempTables( dbengine["session"], dbengine["metadata"], @@ -160,7 +152,6 @@ def main() -> int: timeDifference = options["updatetype"] currentTimestamp = time.time() - #baseTime = str(currentTimestamp).split(".") if timeDifference.lower() == "full" or options["initload"] == "1": tdifference = 0 else: