288 lines
10 KiB
Python
288 lines
10 KiB
Python
#!/usr/bin/env python3
|
|
|
|
from pymongo import MongoClient
|
|
import json
|
|
import psycopg2
|
|
import pandas as pd
|
|
from collections.abc import MutableMapping
|
|
from tqdm import tqdm
|
|
|
|
|
|
def flatten_dict(d: MutableMapping, sep: str = '.') -> MutableMapping:
|
|
[flat_dict] = pd.json_normalize(d, sep=sep).to_dict(orient='records')
|
|
return flat_dict
|
|
|
|
|
|
def delete_none(_dict):
|
|
"""
|
|
Delete None values recursively from all of the dictionaries, tuples,
|
|
lists, sets
|
|
"""
|
|
if isinstance(_dict, dict):
|
|
for key, value in list(_dict.items()):
|
|
if isinstance(value, (list, dict, tuple, set)):
|
|
_dict[key] = delete_none(value)
|
|
elif value is None or key is None:
|
|
del _dict[key]
|
|
|
|
elif isinstance(_dict, (list, set, tuple)):
|
|
_dict = type(_dict)(
|
|
delete_none(item) for item in _dict if item is not None
|
|
)
|
|
return _dict
|
|
|
|
|
|
def process_data(data, tablename):
|
|
inserts = []
|
|
failed_count = 0
|
|
seriesconflict = "ON CONFLICT (seriesid) DO UPDATE SET"
|
|
epconflict = "ON CONFLICT (episodeid, seriesid) DO UPDATE SET"
|
|
crewconflict = "ON CONFLICT (crewid) DO UPDATE SET"
|
|
actorconflict = "ON CONFLICT (actorid) DO UPDATE SET"
|
|
characterconflict = "ON CONFLICT (seriesid, characterid) DO UPDATE SET"
|
|
for document in data:
|
|
if document['name'].lower() == 'too many requests':
|
|
print("Skipping document due to 'Too Many Requests' in "
|
|
"document data")
|
|
continue
|
|
else:
|
|
new_document = delete_none(document)
|
|
if '_id' in new_document:
|
|
del new_document['_id']
|
|
try:
|
|
flatten_data = flatten_dict(new_document)
|
|
except TypeError as e:
|
|
print(
|
|
(
|
|
f"Unable to flatten data for current document. {e}\n"
|
|
"Exitting!"
|
|
)
|
|
)
|
|
failed_count += 1
|
|
else:
|
|
column_list = []
|
|
data_list = []
|
|
for key, value in flatten_data.items():
|
|
if isinstance(value, str):
|
|
value = value.replace('"', "'").replace("'", "''")
|
|
if key.lower() == '_links.self.href':
|
|
key = 'apilink'
|
|
if key.lower() == '_links.show.href':
|
|
key = 'apishowlink'
|
|
if key.lower() == '_links.show.name':
|
|
key = 'apishowname'
|
|
if '.' in key.lower():
|
|
key = key.replace('.', '')
|
|
if tablename == 'seriesdata' and (
|
|
key.lower() == 'name' or key.lower() == 'type'
|
|
):
|
|
key = f"series_{key}"
|
|
if tablename == 'seriesdata' and key.lower() == 'id':
|
|
key = "seriesid"
|
|
if tablename == 'crewdata' and key.lower() == 'id':
|
|
key = "crewid"
|
|
if (
|
|
tablename == 'actordata' and (
|
|
key.lower() == 'name' or key.lower() == 'number' or
|
|
key.lower() == 'type'
|
|
)
|
|
):
|
|
key = f"actor{key}"
|
|
if tablename == 'actordata' and key.lower() == 'id':
|
|
key = "actorid"
|
|
if tablename == 'characterdata' and key.lower() == 'id':
|
|
key = "characterid"
|
|
if tablename == 'characterdata' and key.lower() == 'name':
|
|
key = "charactername"
|
|
if tablename == 'characterdata' and key.lower() == 'id':
|
|
key = "seriesid, characterid"
|
|
if (
|
|
(
|
|
tablename == 'epdata' or
|
|
tablename == 'actordata'
|
|
) and (
|
|
key.lower() == 'name' or
|
|
key.lower() == 'number' or
|
|
key.lower() == 'type'
|
|
)
|
|
):
|
|
key = f"episode_{key}"
|
|
if key.lower() == 'language':
|
|
key = "language_name"
|
|
if tablename == 'epdata' and key.lower() == 'id':
|
|
key = "episodeid"
|
|
if tablename == 'actordata' and 'seriesid' in key.lower():
|
|
continue
|
|
else:
|
|
column_list.append(key)
|
|
if (
|
|
'genres' in key.lower() or
|
|
'schedule' in key.lower()
|
|
):
|
|
if isinstance(value, list):
|
|
value = ', '.join(value)
|
|
if isinstance(value, int) or isinstance(value, float):
|
|
value = str(json.dumps(value))
|
|
data_list.append(f"'{value}'")
|
|
columns = ", ".join(column_list)
|
|
column_data = ", ".join(data_list)
|
|
conflict_list = []
|
|
for indexer, column in enumerate(column_list):
|
|
conflict_list.append(
|
|
f"{column_list[indexer]} = {data_list[indexer]}"
|
|
)
|
|
conflict_data = ', '.join(conflict_list).replace("\\", "")
|
|
conflict_clause = ''
|
|
if tablename == 'seriesdata':
|
|
conflict_clause = seriesconflict
|
|
elif tablename == 'epdata':
|
|
conflict_clause = epconflict
|
|
elif tablename == 'crewdata':
|
|
conflict_clause = crewconflict
|
|
elif tablename == 'actordata':
|
|
conflict_clause = actorconflict
|
|
elif tablename == 'characterdata':
|
|
conflict_clause = characterconflict
|
|
insert_command = (
|
|
(
|
|
f"Insert Into updates.{tablename} ({columns}) "
|
|
f"values ({column_data}) "
|
|
f"{conflict_clause} {conflict_data};"
|
|
)
|
|
)
|
|
inserts.append(insert_command)
|
|
return inserts, failed_count
|
|
|
|
|
|
try:
|
|
connection_string = (
|
|
(
|
|
'postgres://postgres:Optimus0329@192.168.128.7:5432/media_dbsync'
|
|
'?options=-csearch_path%3Ddbo,public,updates'
|
|
)
|
|
)
|
|
pgclient = psycopg2.connect(connection_string)
|
|
pgcursor = pgclient.cursor()
|
|
# print(pgclient.get_dsn_parameters(), "\n")
|
|
print("Connected to postgresql database server.")
|
|
pgcursor.execute("SELECT version();")
|
|
record = pgcursor.fetchone()
|
|
print("You are connected to - ", record, "\n")
|
|
pgclient.set_isolation_level(0)
|
|
except Exception as e:
|
|
print(f"Unable to connect to postgresql database server:{e}\nExitting!")
|
|
exit()
|
|
|
|
|
|
host = '192.168.128.8'
|
|
port = 27017
|
|
print(f"Connecting to mongodb at {host}:{port}")
|
|
|
|
try:
|
|
connection_string = (
|
|
f"mongodb://{host}:{port}"
|
|
)
|
|
mgclient = MongoClient(connection_string)
|
|
mgdb = mgclient['test2']
|
|
except BaseException as e:
|
|
print(f"Unable to connect to mongodb.{e}\nExitting!")
|
|
exit()
|
|
else:
|
|
mgseries = mgdb["series"]
|
|
mgseries.create_index('id', unique=True)
|
|
mgepisodes = mgdb["episodes"]
|
|
mgepisodes.create_index('id', unique=True)
|
|
mgactors = mgdb["actors"]
|
|
mgactors.create_index('id', unique=True)
|
|
mgcharacters = mgdb["characters"]
|
|
mgcharacters.create_index('id', unique=True)
|
|
mgcrew = mgdb["crew"]
|
|
mgcrew.create_index('id', unique=True)
|
|
print(
|
|
f"Connection to mongodb at {host}:"
|
|
f"{port} successful."
|
|
)
|
|
|
|
print("Creating Insert commands from collection data.")
|
|
|
|
print("Collecting series data for series inserts.")
|
|
series_list = mgseries.find({})
|
|
inserts, failed_count = process_data(series_list, 'seriesdata')
|
|
print(f"Total number of series inserts: {len(inserts)}")
|
|
for i in tqdm(range(len(inserts))):
|
|
try:
|
|
insert = inserts[i]
|
|
pgcursor.execute(insert)
|
|
except Exception as e:
|
|
print(f"Unable to insert data into seriesdata table. {e}")
|
|
failed_count += 1
|
|
print(f"Total number of failed series inserts: {failed_count}")
|
|
pgclient.commit()
|
|
|
|
epinserts = []
|
|
failures = 0
|
|
episode_array = mgepisodes.find({})
|
|
print("Collecting episode data for episode inserts.")
|
|
inserts, failed_count = process_data(episode_array, 'epdata')
|
|
print(f"Total number of episode inserts: {len(inserts)}")
|
|
for i in tqdm(range(len(inserts))):
|
|
try:
|
|
insert = inserts[i]
|
|
pgcursor.execute(insert)
|
|
except Exception as e:
|
|
print(f"Unable to insert data into epdata table. {e}")
|
|
failed_count += 1
|
|
print(f"Total number of failed episode inserts: {failed_count}")
|
|
pgclient.commit()
|
|
|
|
failures = 0
|
|
actor_array = mgactors.find({})
|
|
print("Collecting actor data for actor inserts.")
|
|
inserts, failed_count = process_data(actor_array, 'actordata')
|
|
print(f"Total number of actor inserts: {len(inserts)}")
|
|
for i in tqdm(range(len(inserts))):
|
|
try:
|
|
insert = inserts[i]
|
|
pgcursor.execute(insert)
|
|
except Exception as e:
|
|
print(f"Unable to insert data into actors table. {e}")
|
|
failed_count += 1
|
|
print(f"Total number of failed actor inserts: {failed_count}")
|
|
pgclient.commit()
|
|
|
|
failures = 0
|
|
character_array = mgcharacters.find({})
|
|
print("Collecting actor data for actor inserts.")
|
|
inserts, failed_count = process_data(character_array, 'characterdata')
|
|
print(f"Total number of character inserts: {len(inserts)}")
|
|
for i in tqdm(range(len(inserts))):
|
|
try:
|
|
insert = inserts[i]
|
|
pgcursor.execute(insert)
|
|
except Exception as e:
|
|
print(f"Unable to insert data into character table. {e}")
|
|
failed_count += 1
|
|
print(f"Total number of failed character inserts: {failed_count}")
|
|
pgclient.commit()
|
|
|
|
failures = 0
|
|
crew_array = mgcrew.find({})
|
|
print("Collecting crew data for crew inserts.")
|
|
inserts, failed_count = process_data(crew_array, 'crewdata')
|
|
print(f"Total number of crew inserts: {len(inserts)}")
|
|
for i in tqdm(range(len(inserts))):
|
|
try:
|
|
insert = inserts[i]
|
|
pgcursor.execute(insert)
|
|
except Exception as e:
|
|
print(f"Unable to insert data into epdata table. {e}")
|
|
failed_count += 1
|
|
print(f"Total number of failed crew inserts: {failed_count}")
|
|
pgclient.commit()
|
|
|
|
pgclient.close()
|
|
print("PostgreSQL connection is closed")
|
|
|
|
|
|
exit()
|