working on json2postgres.py file
This commit is contained in:
@@ -0,0 +1,298 @@
|
||||
#!/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) DO UPDATE SET"
|
||||
crewconflict = "ON CONFLICT (crewid) DO UPDATE SET"
|
||||
actorconflict = "ON CONFLICT (actorid) DO UPDATE SET"
|
||||
characterconflict = "ON CONFLICT (characterid, actorid) 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 key.lower() == '_links.self.href':
|
||||
key = 'apilink'
|
||||
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 == '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 (
|
||||
'_links' in key.lower() or
|
||||
'image' in key.lower() or
|
||||
'schedule' in key.lower() or
|
||||
'externals' in key.lower() or
|
||||
'network' in key.lower() or
|
||||
'webchannel' in key.lower() or
|
||||
'country' in key.lower() or
|
||||
'rating' in key.lower()
|
||||
):
|
||||
key = key.replace('.', '_')
|
||||
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)
|
||||
else:
|
||||
value = str(value).replace("'", "''")
|
||||
if isinstance(value, str):
|
||||
value = value.replace("'", "''")
|
||||
if isinstance(value, int):
|
||||
value = f"'{value}'"
|
||||
elif isinstance(value, float):
|
||||
value = f"'{json.dumps(value)}'"
|
||||
elif isinstance(value, dict):
|
||||
value = f"'{json.dumps(value)}'"
|
||||
elif isinstance(value, bool):
|
||||
value = str(value).lower()
|
||||
elif value is None:
|
||||
value = 'NULL'
|
||||
else:
|
||||
value = str(f"'{value}'")
|
||||
data_list.append(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)
|
||||
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:
|
||||
# # 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:
|
||||
# # 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()
|
||||
Reference in New Issue
Block a user