#!/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()