diff --git a/README.md b/README.md index 047477f..fa545b3 100644 --- a/README.md +++ b/README.md @@ -1,8 +1,8 @@ -# Sample GitLab Project +# DBSyncer -This sample project shows how a project in GitLab looks for demonstration purposes. It contains issues, merge requests and Markdown files in many branches, -named and filled with lorem ipsum. +This project has two parts: +1.) it uses the tvmaze API to download Television dat and store it in a mongodb database. +2.) Takes saved json data and inserts/updates data in a postgresql database. -You can look around to get an idea how to structure your project and, when done, you can safely delete this project. [Learn more about creating GitLab projects.](https://docs.gitlab.com/ee/gitlab-basics/create-project.html) diff --git a/tabledata_from_json.py b/json2postgres.py similarity index 62% rename from tabledata_from_json.py rename to json2postgres.py index ecdd3ac..dacbfc9 100644 --- a/tabledata_from_json.py +++ b/json2postgres.py @@ -12,8 +12,12 @@ 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""" + """ + 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)): @@ -22,31 +26,29 @@ def delete_none(_dict): del _dict[key] elif isinstance(_dict, (list, set, tuple)): - _dict = type(_dict)(delete_none(item) for item in _dict if item is not None) - + _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") + "document data") continue else: new_document = delete_none(document) if '_id' in new_document: del new_document['_id'] - # if tablename == 'epdata' and 'image' in document: - # del document['image'] - # if 'network' in document: - # if document['network'] is None: - # del document['network'] - # del document['image'] try: flatten_data = flatten_dict(new_document) except TypeError as e: @@ -61,16 +63,33 @@ def process_data(data, tablename): 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 == 'crewdata' and key.lower() == 'id': - key = "crewid" if ( - tablename == 'epdata' and ( + 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' ) @@ -91,33 +110,38 @@ def process_data(data, tablename): 'rating' in key.lower() ): key = key.replace('.', '_') - 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' + if tablename == 'actordata' and 'seriesid' in key.lower(): + continue else: - value = str(f"'{value}'") - data_list.append(value) + 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_list.append( + f"{column_list[indexer]} = {data_list[indexer]}" + ) + conflict_data = ', '.join(conflict_list) conflict_clause = '' if tablename == 'seriesdata': conflict_clause = seriesconflict @@ -125,6 +149,10 @@ def process_data(data, tablename): 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}) " @@ -132,6 +160,7 @@ def process_data(data, tablename): f"{conflict_clause} {conflict_data};" ) ) + inserts.append(insert_command) return inserts, failed_count @@ -145,7 +174,7 @@ try: ) pgclient = psycopg2.connect(connection_string) pgcursor = pgclient.cursor() - #print(pgclient.get_dsn_parameters(), "\n") + # print(pgclient.get_dsn_parameters(), "\n") print("Connected to postgresql database server.") pgcursor.execute("SELECT version();") record = pgcursor.fetchone() @@ -195,7 +224,7 @@ print("Creating Insert commands from collection data.") # try: # insert = inserts[i] # pgcursor.execute(insert) -# except Exception as e: +# 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}") @@ -211,12 +240,42 @@ print("Creating Insert commands from collection data.") # try: # insert = inserts[i] # pgcursor.execute(insert) -# except Exception as e: +# 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.") diff --git a/json_downloader.py b/json_downloader.py index e72d242..84b60a5 100644 --- a/json_downloader.py +++ b/json_downloader.py @@ -1,36 +1,38 @@ #!/usr/bin/env python3 import os -from pathlib import Path -from time import sleep -from tqdm import tqdm import json import requests + def load(self, rawdata): jdata = {} jdata = json.loads(rawdata) newdata = self.dump(jdata) - return(newdata) + return (newdata) + def dump(self, rawdata): jdata = json.dumps(rawdata, indent=4) - return(jdata) + return (jdata) + def jconvert(self, inputdata): outputdata = json.loads(inputdata) - return(outputdata) + return (outputdata) + def populate_cache(self, jdata, jfile): try: with open(jfile, 'w') as datafile: datafile.write(jdata) - except: + except Exception: print('Unable to save cache file.') else: datafile.close() -def fetch_json(self, weburl, seriesid, cachefile): + +def fetch_json(weburl, seriesid, cachefile): try: webdata = get_data(weburl) jdata = json.load(webdata) @@ -42,22 +44,27 @@ def fetch_json(self, weburl, seriesid, cachefile): else: populate_cache(jdata, cachefile) - def get_data(self,weburl): + def get_data(weburl): try: showresponse = requests.get(weburl) - except requests.exceptions.ConnectionError as e: + except requests.exceptions.ConnectionError: print('Unable to connect and get JSON data for url ' + str(weburl)) else: - return(showresponse.text) + return (showresponse.text) -def check_and_download(url, series_id, cachefile): + +def check_and_download(URL, series_id, cachefile): try: mazeurl = URL.replace('', str(series_id)) response = requests.get(mazeurl) data = json.loads(response.text) except FileNotFoundError: - #print(f'{cachefile} does not exist. Attempting to download.') - jget.fetch_json(url.replace(SERIESID, str(series_id)), series_id, cachefile) + # print(f'{cachefile} does not exist. Attempting to download.') + jget.fetch_json( + URL.replace('SERIESID', str(series_id)), + series_id, + cachefile + ) except Exception as e: print(f'Unable to download {cachefile}. {e}') else: @@ -65,12 +72,17 @@ def check_and_download(url, series_id, cachefile): json.dump(data, f, indent=4) f.close() + new_updates = [] -listing = open ('D:/Temp/filelist.txt', 'w') +listing = open('D:/Temp/filelist.txt', 'w') os.chdir('D:/Projects/new_dbsync/cache/series') -directory_path = 'D:/Projects/new_dbsync/cache/series' # replace with your directory path -files = [f for f in os.listdir(directory_path) if os.path.isfile(os.path.join(directory_path, f))] +# replace with your directory path +directory_path = 'D:/Projects/new_dbsync/cache/series' +files = [ + f for f in os.listdir(directory_path) + if os.path.isfile(os.path.join(directory_path, f)) +] for filename in files: with open(filename, 'r') as file: content = file.read() @@ -81,16 +93,3 @@ for filename in files: listing.close() exit() - -# new_updates_2 = [] -# listing = open ('D:/Temp/filelist.txt', 'r') -# for seriesid in listing: -# seriesid = seriesid.strip() -# new_updates_2.append(seriesid) -# listing.close() - -# for seriesid in tqdm(range(len(new_updates_2)), desc='Checking cache', unit='seriesid'): -# # cachefile = str(new_updates_2[seriesid]) + '.json' -# # URL ='http://api.tvmaze.com/shows//cast' -# # check_and_download(URL, str(new_updates_2[seriesid]), f'{new_updates_2[seriesid]}.json') -# os.remove(f'D:/Projects/new_dbsync/cache/cast/{new_updates_2[seriesid]}.json') \ No newline at end of file diff --git a/update.py b/update_mongo.py similarity index 100% rename from update.py rename to update_mongo.py diff --git a/tableddl_from_yaml.py b/yaml_ddl/tableddl_from_yaml.py similarity index 100% rename from tableddl_from_yaml.py rename to yaml_ddl/tableddl_from_yaml.py diff --git a/test_tableddl_from_yaml.py b/yaml_ddl/test_tableddl_from_yaml.py similarity index 100% rename from test_tableddl_from_yaml.py rename to yaml_ddl/test_tableddl_from_yaml.py