repo migration
This commit is contained in:
+605
@@ -0,0 +1,605 @@
|
||||
__author__ = "Wendell Jones"
|
||||
|
||||
import os
|
||||
import sqlalchemy
|
||||
from sqlalchemy import text
|
||||
from sqlalchemy import MetaData
|
||||
from sqlalchemy.ext.automap import automap_base
|
||||
from sqlalchemy.orm import Session
|
||||
from db.Tables import Createtables, CreateTempTables, Tables as dbTables
|
||||
from db.mssql.Merger import Charactermerge as MSCharactermerge
|
||||
from db.mssql.Merger import Jsonmerge as MSJsonmerge
|
||||
from db.mssql.Merger import SQLGenerate as MSSQLGenerate
|
||||
from db.mssql.Merger import Episodemerge as MSEpisodemerge
|
||||
from db.pgsql.Merger import Charactermerge as PGCharactermerge
|
||||
from db.pgsql.Merger import Updatemerge as PGUpdatemerge
|
||||
from db.mssql.Merger import Jsonmerge as PGJsonmerge
|
||||
from db.pgsql.Merger import SQLGenerate as PGSQLGenerate
|
||||
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 tqdm import tqdm
|
||||
|
||||
Createtables = Createtables
|
||||
CreateTempTables = CreateTempTables
|
||||
dbTables = dbTables
|
||||
|
||||
|
||||
class settype(object):
|
||||
|
||||
def __init__(self, dbtype, apitype, options):
|
||||
self.dbclass = {}
|
||||
self.dbengine = {}
|
||||
self.apiengine = {}
|
||||
self.load(dbtype)
|
||||
self.dbcreate(dbtype, options)
|
||||
self.setapi(apitype)
|
||||
self.get_data()
|
||||
|
||||
def load(self, dbtype):
|
||||
dbtype = dbtype.lower()
|
||||
if dbtype in ["pgsql", "mysql"]:
|
||||
Updatemerge = PGUpdatemerge
|
||||
Jsonmerge = PGJsonmerge
|
||||
Episodemerge = PGEpisodemerge
|
||||
Charactermerge = PGCharactermerge
|
||||
merger = PGSQLGenerate
|
||||
elif dbtype == "mssql":
|
||||
Jsonmerge = MSJsonmerge
|
||||
Charactermerge = MSCharactermerge
|
||||
merger = MSSQLGenerate
|
||||
Episodemerge = MSEpisodemerge
|
||||
|
||||
self.dbclass["jmerge"] = Jsonmerge()
|
||||
self.dbclass["cmerge"] = Charactermerge()
|
||||
self.dbclass["Createtables"] = Createtables
|
||||
self.dbclass["CreateTemptables"] = CreateTempTables
|
||||
self.dbclass["Tables"] = Createtables
|
||||
self.dbclass["merger"] = merger()
|
||||
|
||||
if dbtype == "pgsql":
|
||||
self.dbclass["Updatemerge"] = Updatemerge
|
||||
self.dbclass["Episodemerge"] = Episodemerge
|
||||
elif dbtype == "mssql":
|
||||
self.dbclass["Charactermerge"] = Charactermerge
|
||||
|
||||
def dbcreate(self, dbtype, options):
|
||||
updatesBase = automap_base()
|
||||
dboBase = automap_base()
|
||||
connection_string = None
|
||||
if dbtype.lower() == "mssql":
|
||||
connection_string = (
|
||||
"mssql+pymssql://"
|
||||
+ options["mssqlusername"]
|
||||
+ ":"
|
||||
+ options["mssqlpassword"]
|
||||
+ "@"
|
||||
+ options["hostname"]
|
||||
+ ":"
|
||||
+ options["mssqlport"]
|
||||
+ "/"
|
||||
+ options["dbname"]
|
||||
)
|
||||
elif dbtype.lower() == "pgsql":
|
||||
connection_string = (
|
||||
"postgresql+psycopg2://"
|
||||
+ options["pgsqlusername"]
|
||||
+ ":"
|
||||
+ options["pgsqlpassword"]
|
||||
+ "@"
|
||||
+ options["hostname"]
|
||||
+ ":"
|
||||
+ options["pgsqlport"]
|
||||
+ "/"
|
||||
+ options["dbname"]
|
||||
)
|
||||
elif dbtype.lower() == "mysql":
|
||||
connection_string = (
|
||||
"mysql+pymysql://"
|
||||
+ options["mysqlusername"]
|
||||
+ ":"
|
||||
+ options["mysqlpassword"]
|
||||
+ "@"
|
||||
+ options["hostname"]
|
||||
+ ":"
|
||||
+ options["mysqlport"]
|
||||
+ "/"
|
||||
+ options["dbname"]
|
||||
)
|
||||
engine = sqlalchemy.create_engine(
|
||||
connection_string, pool_size=20, max_overflow=20, pool_recycle=15
|
||||
)
|
||||
currentDir = os.path.dirname(os.path.abspath(__file__))
|
||||
currentDir = os.path.dirname(currentDir)
|
||||
ddl_file_path = f"{currentDir}/db/ddl/recreate_updates_schema.sql"
|
||||
with open(ddl_file_path, "r") as file:
|
||||
sql = file.read()
|
||||
file
|
||||
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")
|
||||
metadata.reflect(bind=engine)
|
||||
dboMetadata = sqlalchemy.MetaData("dbo")
|
||||
dboMetadata.reflect(bind=engine)
|
||||
session = Session(engine)
|
||||
self.dbengine["engine"] = engine
|
||||
self.dbengine["metadata"] = metadata
|
||||
self.dbengine["dbometadata"] = dboMetadata
|
||||
self.dbengine["updatesBase"] = updatesBase
|
||||
self.dbengine["dboBase"] = dboBase
|
||||
self.dbengine["session"] = session
|
||||
return self.dbengine
|
||||
|
||||
def setapi(self, apitype):
|
||||
if apitype.lower() == "tvmaze":
|
||||
self.apiengine["sprocess"] = api.tvmaze.series
|
||||
self.apiengine["cprocess"] = api.tvmaze.cast
|
||||
self.apiengine["eprocess"] = api.tvmaze.episodes
|
||||
return self.apiengine
|
||||
elif apitype.lower() == "thetvdb":
|
||||
pass
|
||||
|
||||
def reflect_tables(self, engine, Metadata):
|
||||
meta = MetaData()
|
||||
meta.reflect(bind=engine)
|
||||
return meta.tables
|
||||
|
||||
def get_data(self):
|
||||
return self.dbclass, self.dbengine, self.apiengine
|
||||
|
||||
def __del__(self):
|
||||
pass
|
||||
|
||||
|
||||
class dbmongo(object):
|
||||
|
||||
global mgclient
|
||||
global mgseries
|
||||
global mgepisodes
|
||||
global mgactors
|
||||
global mgcharacters
|
||||
global mgcrew
|
||||
|
||||
def __init__(self, options):
|
||||
host = options["mghostname"]
|
||||
port = options["mgport"]
|
||||
print(f"Connecting to mongodb at {host}:{port}")
|
||||
try:
|
||||
self.connection_string = (
|
||||
f"mongodb://{options['mghostname']}:{options['mgport']}"
|
||||
)
|
||||
self.mgclient = MongoClient(self.connection_string)
|
||||
self.mgdb = self.mgclient[options["mgdbname"]]
|
||||
except BaseException as e:
|
||||
print(f"Unable to connect to mongodb.{e}\nExitting!")
|
||||
exit()
|
||||
else:
|
||||
self.mgseries = self.mgdb["series"]
|
||||
self.mgseries.create_index("id", unique=True)
|
||||
self.mgepisodes = self.mgdb["episodes"]
|
||||
self.mgepisodes.create_index("id", unique=True)
|
||||
self.mgactors = self.mgdb["actors"]
|
||||
self.mgactors.create_index("id", unique=True)
|
||||
self.mgcharacters = self.mgdb["characters"]
|
||||
self.mgcharacters.create_index("id", unique=True)
|
||||
self.mgcrew = self.mgdb["crew"]
|
||||
self.mgcrew.create_index("id", unique=True)
|
||||
print(
|
||||
f"Connection to mongodb at {options['mghostname']}:"
|
||||
f"{options['mgport']} successful."
|
||||
)
|
||||
|
||||
def load_json_file(self, cachefile, jget, lprint, seriesid, datatype):
|
||||
try:
|
||||
with open(cachefile, "r") as file:
|
||||
jsondata = file.read()
|
||||
datastring = jget.load(jsondata)
|
||||
data = jget.jconvert(datastring)
|
||||
return data
|
||||
except Exception:
|
||||
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,
|
||||
seriesdata,
|
||||
jget,
|
||||
dbExec,
|
||||
dbengine,
|
||||
dboBase,
|
||||
lprint,
|
||||
countrydictionary,
|
||||
seriesid,
|
||||
):
|
||||
try:
|
||||
seriesresults, networkchannel, country, webchannel = (
|
||||
jget.processSeriesJson(seriesdata)
|
||||
)
|
||||
except Exception:
|
||||
lprint.logprint(
|
||||
"info", f"Unable to process series file for {seriesid}"
|
||||
)
|
||||
else:
|
||||
self.update_country_data(
|
||||
country,
|
||||
countrydictionary,
|
||||
dbExec,
|
||||
dbengine,
|
||||
dboBase,
|
||||
dbengine["dbometadata"],
|
||||
lprint,
|
||||
)
|
||||
self.update_network_data(
|
||||
networkchannel,
|
||||
countrydictionary,
|
||||
seriesresults,
|
||||
dbExec,
|
||||
dbengine,
|
||||
dboBase,
|
||||
lprint,
|
||||
)
|
||||
self.update_webchannel_data(
|
||||
webchannel,
|
||||
countrydictionary,
|
||||
seriesresults,
|
||||
dbExec,
|
||||
dbengine,
|
||||
dboBase,
|
||||
lprint,
|
||||
)
|
||||
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
|
||||
]
|
||||
|
||||
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
|
||||
]
|
||||
|
||||
def update_country_data(
|
||||
self, country, countrydictionary, dbExec, dbengine, dbBase,
|
||||
metadata, lprint
|
||||
):
|
||||
if len(country) > 0:
|
||||
try:
|
||||
selectquery = (
|
||||
f"select row_id, country_name from dbo.countrydata where "
|
||||
f"country_name = '{country['name'].replace("'", "''")}'"
|
||||
)
|
||||
countrylisting = dbExec.rawsql_select(
|
||||
dbengine["engine"], selectquery, lprint
|
||||
)
|
||||
for countryname in countrylisting:
|
||||
countrydictionary[countryname[1]] = countryname[0]
|
||||
except Exception as e:
|
||||
lprint.logprint(
|
||||
"info", (
|
||||
(
|
||||
f'Unable to process country data for '
|
||||
f'{country["name"]}. {e}'
|
||||
)
|
||||
)
|
||||
)
|
||||
else:
|
||||
if len(countrylisting) > 0:
|
||||
pass
|
||||
else:
|
||||
insert_sql = (
|
||||
(
|
||||
f"insert into dbo.countrydata "
|
||||
"(country_name, country_code, country_timezone) "
|
||||
f"values ('{country['name'].replace("'", "''")}', "
|
||||
f"'{country['code'].replace("'", "''")}', "
|
||||
f"'{country['timezone'].replace("'", "''")}')"
|
||||
)
|
||||
)
|
||||
dbExec.rawsql_insert(
|
||||
dbengine["engine"], insert_sql, lprint
|
||||
)
|
||||
# seriesresults['countryid'] = 'No Country'
|
||||
# return seriesresults['countryid']
|
||||
|
||||
def update_network_data(
|
||||
self,
|
||||
networkchannel,
|
||||
countrydictionary,
|
||||
seriesresults,
|
||||
dbExec,
|
||||
dbengine,
|
||||
dboBase,
|
||||
lprint,
|
||||
):
|
||||
if len(networkchannel) > 0:
|
||||
try:
|
||||
networkchannel["countryid"] = countrydictionary[
|
||||
networkchannel["country"]["name"]
|
||||
]
|
||||
seriesresults["networkid"] = networkchannel["id"]
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
def update_webchannel_data(
|
||||
self,
|
||||
webchannel,
|
||||
countrydictionary,
|
||||
seriesresults,
|
||||
dbExec,
|
||||
dbengine,
|
||||
dboBase,
|
||||
lprint,
|
||||
):
|
||||
if len(webchannel) > 0:
|
||||
try:
|
||||
webchannel["countryid"] = countrydictionary[
|
||||
webchannel["country"]["name"]
|
||||
]
|
||||
seriesresults["webchannelid"] = webchannel["id"]
|
||||
except Exception:
|
||||
pass
|
||||
seriesresults["seriesid"] = seriesresults["id"]
|
||||
del seriesresults["webchannel"]
|
||||
|
||||
def insert_or_update_mongo(self, collection, data, unique_key):
|
||||
try:
|
||||
collection.insert_one(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 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,
|
||||
dbExec,
|
||||
dbengine,
|
||||
dboBase,
|
||||
lprint,
|
||||
):
|
||||
print("Now inserting data into tables.")
|
||||
series_failed_count = 0
|
||||
episode_failed_count = 0
|
||||
cast_failed_count = 0
|
||||
crew_failed_count = 0
|
||||
processed_count = 0
|
||||
failed_file_count = 0
|
||||
|
||||
def process_data(
|
||||
file_path, data_type, process_method, failed_file_count, lprint
|
||||
):
|
||||
"""Helper function to load and process data."""
|
||||
data = self.load_json_file(
|
||||
file_path, jget, lprint, seriesid, data_type
|
||||
)
|
||||
if data:
|
||||
if "episode" in file_path:
|
||||
try:
|
||||
for item in data:
|
||||
item["seriesid"] = seriesid
|
||||
except Exception as e:
|
||||
lprint.logprint(
|
||||
"info", f"Unable to add {data_type} to data. {e}"
|
||||
)
|
||||
if "cast" in file_path:
|
||||
persondata = []
|
||||
characterdata = []
|
||||
for item in data:
|
||||
if "person" in item:
|
||||
persondata.append(item["person"])
|
||||
if "character" in item:
|
||||
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
|
||||
)
|
||||
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 = []
|
||||
except Exception as e:
|
||||
lprint.logprint(
|
||||
"info",
|
||||
f"Unable to add {data_type} data. {e}"
|
||||
)
|
||||
else:
|
||||
for item in data:
|
||||
item["seriesid"] = seriesid
|
||||
if "person" in item:
|
||||
crewdata.append(item["person"])
|
||||
try:
|
||||
process_method(crewdata, seriesid, jget, lprint)
|
||||
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
|
||||
|
||||
for series in tqdm(
|
||||
range(len(updateList)), desc="Update MongoDB", unit="series"
|
||||
):
|
||||
seriesid = updateList[series]
|
||||
seriescachefile = f'{directories["SERIESDIR"]}{seriesid}.json'
|
||||
episodecachefile = f'{directories["EPISODEDIR"]}{seriesid}.json'
|
||||
castcachefile = f'{directories["CASTDIR"]}{seriesid}.json'
|
||||
crewcachefile = f'{directories["CREWDIR"]}{seriesid}.json'
|
||||
|
||||
# Process series data
|
||||
seriesdata = self.load_json_file(
|
||||
seriescachefile, jget, lprint, seriesid, "series"
|
||||
)
|
||||
if seriesdata:
|
||||
try:
|
||||
self.process_series_data(
|
||||
seriesdata,
|
||||
jget,
|
||||
dbExec,
|
||||
dbengine,
|
||||
dboBase,
|
||||
lprint,
|
||||
countrydictionary,
|
||||
seriesid,
|
||||
)
|
||||
self.insert_or_update_mongo(
|
||||
self.mgseries, seriesdata, "id"
|
||||
)
|
||||
except Exception as e:
|
||||
series_failed_count += 1
|
||||
lprint.logprint(
|
||||
"info",
|
||||
(
|
||||
f"Unable to process series data for seriesid "
|
||||
f"{seriesid}. {e}"
|
||||
),
|
||||
)
|
||||
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
|
||||
fcount = process_data(
|
||||
castcachefile, "cast", self.process_cast_data,
|
||||
failed_file_count, lprint
|
||||
)
|
||||
if isinstance(fcount, int):
|
||||
cast_failed_count = cast_failed_count + fcount
|
||||
try:
|
||||
fcount = process_data(
|
||||
crewcachefile, "crew", self.process_crew_data,
|
||||
failed_file_count, lprint
|
||||
)
|
||||
except Exception:
|
||||
crew_failed_count = crew_failed_count + 1
|
||||
else:
|
||||
if isinstance(fcount, int):
|
||||
crew_failed_count = crew_failed_count + fcount
|
||||
processed_count += 1
|
||||
|
||||
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.")
|
||||
print(f"Failed to load {crew_failed_count} crew files.")
|
||||
print(f"Failed to load {cast_failed_count} cast files.")
|
||||
|
||||
def __del__(self):
|
||||
pass
|
||||
Reference in New Issue
Block a user