Implement bulk upsert functionality in dbmongo class and update configuration for MongoDB database
This commit is contained in:
+55
-122
@@ -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"
|
||||
)
|
||||
|
||||
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,8 +327,6 @@ class dbmongo(object):
|
||||
dbExec.rawsql_insert(
|
||||
dbengine["engine"], insert_sql, lprint
|
||||
)
|
||||
# seriesresults['countryid'] = 'No Country'
|
||||
# return seriesresults['countryid']
|
||||
|
||||
def update_network_data(
|
||||
self,
|
||||
@@ -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:
|
||||
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.")
|
||||
|
||||
@@ -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
|
||||
|
||||
+1
-10
@@ -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:
|
||||
|
||||
Reference in New Issue
Block a user