563 lines
20 KiB
Python
563 lines
20 KiB
Python
__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, ReplaceOne, InsertOne
|
|
from tqdm import tqdm
|
|
from datetime import datetime
|
|
|
|
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))
|
|
updatesBase.prepare(engine, reflect=True, schema="updates")
|
|
dboBase.prepare(engine, reflect=True, schema="dbo")
|
|
metadata = sqlalchemy.MetaData(schema="updates")
|
|
metadata.reflect(bind=engine)
|
|
dboMetadata = sqlalchemy.MetaData(schema="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 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]
|
|
return update_time
|
|
|
|
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}"
|
|
)
|
|
|
|
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:
|
|
update_time = self.convert_updated_time(seriesdata["updated"])
|
|
seriesdata["updated"] = update_time
|
|
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.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):
|
|
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):
|
|
self.bulk_upsert_by_id(self.mgcrew, crewdata)
|
|
|
|
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
|
|
)
|
|
|
|
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 update_mongo(
|
|
self,
|
|
updateList,
|
|
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, 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:
|
|
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:
|
|
self.bulk_upsert_by_id(self.mgactors, persondata)
|
|
self.bulk_upsert_by_id(self.mgcharacters, characterdata)
|
|
failed_file_count = "N/A"
|
|
except Exception as e:
|
|
failed_file_count = 1
|
|
lprint.logprint(
|
|
"info",
|
|
f"Unable to process {data_type} data. {e}"
|
|
)
|
|
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["person"]["seriesid"] = seriesid
|
|
item["person"]["type"] = item["type"]
|
|
update_time =self.convert_updated_time(item["person"]["updated"])
|
|
if update_time:
|
|
item["person"]["updated"] = str(update_time).split(".")[0]
|
|
if "person" in item:
|
|
crewdata.append(item["person"])
|
|
try:
|
|
self.bulk_upsert_by_id(self.mgcrew, crewdata)
|
|
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
|
|
return data
|
|
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.bulk_upsert_by_id(self.mgseries, seriesdata)
|
|
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
|
|
|
|
fcount = process_data(
|
|
episodecachefile,
|
|
"episode",
|
|
failed_file_count,
|
|
lprint,
|
|
)
|
|
if isinstance(fcount, int):
|
|
episode_failed_count = episode_failed_count + fcount
|
|
else:
|
|
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",failed_file_count, lprint
|
|
)
|
|
if isinstance(fcount, int):
|
|
cast_failed_count = cast_failed_count + fcount
|
|
if os.path.isfile(crewcachefile):
|
|
fcount = process_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.")
|
|
print(f"Failed to load {crew_failed_count} crew files.")
|
|
print(f"Failed to load {cast_failed_count} cast files.")
|
|
|
|
def __del__(self):
|
|
pass
|