#!/usr/bin/env python3 # -*- coding: utf-8 -*- import os import time import sys import url.urls import url.process from tqdm import tqdm import db.sql import logs.Logger from db.functions import settype, dbmongo from datetime import datetime import json import requests import json_downloader response = None def set_apienv(urls, uprocess, dbengine, dbExec, updatesBase, lprint): """Populate updates table from the API and return available updates. Args: urls: URL configuration with an "updatesurl" key. uprocess: URL loader instance with `get_data`. jget: JSON processor with `jconvert`. dbengine: database engine dict (expects key "engine"). dbExec: database execution helper with `update_tvupdates` and `rawsql_select`. updatesBase: list-like containing tables (uses index 8 for updates table). lprint: logger instance with `logprint`. Returns: List of tuples (seriesid, timestamp) representing available updates. """ #updatetable = updatesBase[8] #inputdata = uprocess.get_data(urls["updatesurl"]) #lprint.logprint("info", f"Retrieved {len(availableupdates)} rows for processing......") # newupdates = dbExec.rawsql_select( dbengine["engine"], "select seriesid,timestamp from updates.tvupdates", lprint ) return newupdates def print_starttime(lprint, ts1): lprint.logprint("info", f"Sync start time: {str(ts1)}") def get_tvdb(tvdburls): return tvdburls.get_data() def get_tvmaze(tvmazeurls): return tvmazeurls.get_data() def find_file_in_multiple_dirs(filename, directories): """ Checks if a file exists in any of the provided directories. Args: filename (str): The name of the file to search for. directories (list): A list of directory paths to check. Returns: Path or None: The full path to the file if found, otherwise None. """ directories_found = [] for directory in directories: if os.path.isfile(f'{directory}{filename}'): directories_found.append(directory) return directories_found def load_json_to_mongodb(mongo_updater, directories, update_list, lprint, track_changes=True): """ Load downloaded JSON files into MongoDB collections with change detection. Processes series, episodes, cast, and crew data from cached JSON files and inserts or updates them in the appropriate MongoDB collections. Automatically detects if documents are new, updated, or unchanged. Args: mongo_updater: dbmongo instance for MongoDB operations directories: dict containing cache directory paths for different data types update_list: list of series IDs to process lprint: logger instance for logging operations track_changes: bool, if True tracks inserted vs updated documents (default: True) Returns: tuple: (successful_count, failed_count, stats_dict) for series processed stats_dict contains: {'inserted': count, 'updated': count} """ successful_count = 0 failed_count = 0 stats = {'series_inserted': 0, 'series_updated': 0, 'episodes_inserted': 0, 'episodes_updated': 0, 'cast_inserted': 0, 'cast_updated': 0, 'crew_inserted': 0, 'crew_updated': 0} lprint.logprint("info", f"Starting to load {len(update_list)} series into MongoDB") for seriesid in tqdm(update_list, desc="Loading series data into MongoDB", unit="series"): try: # Load series data series_file = f"{directories['SERIESDIR']}{seriesid}.json" if os.path.exists(series_file): try: with open(series_file, 'r') as f: series_data = json.load(f) # Insert or update series in MongoDB if series_data: if track_changes: existing = mongo_updater.mgseries.find_one({"id": series_data.get("id")}) if existing: # Check if data has changed if existing.get("updated") != series_data.get("updated"): stats['series_updated'] += 1 lprint.logprint("debug", f"Updated series data for ID {seriesid}") else: lprint.logprint("debug", f"Series {seriesid} unchanged, skipping") pass # Document unchanged else: stats['series_inserted'] += 1 lprint.logprint("debug", f"Inserted new series data for ID {seriesid}") mongo_updater.bulk_upsert_by_id(mongo_updater.mgseries, [series_data]) except Exception as e: lprint.logprint("warning", f"Failed to load series file for {seriesid}: {e}") failed_count += 1 continue # Load episodes data episodes_file = f"{directories['EPISODEDIR']}{seriesid}.json" if os.path.exists(episodes_file): try: with open(episodes_file, 'r') as f: episodes_data = json.load(f) if isinstance(episodes_data, list) and episodes_data: # Add seriesid to each episode for episode in episodes_data: episode['seriesid'] = seriesid if track_changes: for episode in episodes_data: existing = mongo_updater.mgepisodes.find_one({"id": episode.get("id")}) if existing: if existing.get("updated") != episode.get("updated"): stats['episodes_updated'] += 1 # else: unchanged else: stats['episodes_inserted'] += 1 mongo_updater.bulk_upsert_by_id(mongo_updater.mgepisodes, episodes_data) lprint.logprint("debug", f"Loaded {len(episodes_data)} episodes for series {seriesid}") except Exception as e: lprint.logprint("warning", f"Failed to load episodes file for {seriesid}: {e}") # Load cast data cast_file = f"{directories['CASTDIR']}{seriesid}.json" if os.path.exists(cast_file): try: with open(cast_file, 'r') as f: cast_data = json.load(f) if isinstance(cast_data, list) and cast_data: actors = [] characters = [] for item in cast_data: if 'person' in item and item['person']: item['person']['seriesid'] = seriesid actors.append(item['person']) if 'character' in item and item['character']: item['character']['seriesid'] = seriesid characters.append(item['character']) if track_changes: for actor in actors: existing = mongo_updater.mgactors.find_one({"id": actor.get("id")}) if existing: if existing.get("updated") != actor.get("updated"): stats['cast_updated'] += 1 else: stats['cast_inserted'] += 1 if actors: mongo_updater.bulk_upsert_by_id(mongo_updater.mgactors, actors) if characters: mongo_updater.bulk_upsert_by_id(mongo_updater.mgcharacters, characters) lprint.logprint("debug", f"Loaded cast data for series {seriesid}") except Exception as e: lprint.logprint("warning", f"Failed to load cast file for {seriesid}: {e}") # Load crew data crew_file = f"{directories['CREWDIR']}{seriesid}.json" if os.path.exists(crew_file): try: with open(crew_file, 'r') as f: crew_data = json.load(f) if isinstance(crew_data, list) and crew_data: crew_list = [] for item in crew_data: if 'person' in item and item['person']: item['person']['seriesid'] = seriesid item['person']['crew_type'] = item.get('type', 'unknown') crew_list.append(item['person']) if track_changes: for crew in crew_list: existing = mongo_updater.mgcrew.find_one({"id": crew.get("id")}) if existing: if existing.get("updated") != crew.get("updated"): stats['crew_updated'] += 1 else: stats['crew_inserted'] += 1 if crew_list: mongo_updater.bulk_upsert_by_id(mongo_updater.mgcrew, crew_list) lprint.logprint("debug", f"Loaded crew data for series {seriesid}") except Exception as e: lprint.logprint("warning", f"Failed to load crew file for {seriesid}: {e}") successful_count += 1 lprint.logprint("info", f"Successfully processed series {seriesid}") except Exception as e: failed_count += 1 lprint.logprint("error", f"Unexpected error processing series {seriesid}: {e}") lprint.logprint("info", f"MongoDB loading complete. Successful: {successful_count}, Failed: {failed_count}") lprint.logprint("info", f"Change statistics: {stats}") return successful_count, failed_count, stats def main() -> int: ROOTDIR = os.getcwd() directories = {} # Use cross-platform path builders and ensure directories exist LOGDIR = os.path.join(ROOTDIR, "log") directories["SERIESDIR"] = os.path.join(ROOTDIR, "cache", "series") + os.sep directories["EPISODEDIR"] = os.path.join(ROOTDIR, "cache", "episodes") + os.sep directories["CASTDIR"] = os.path.join(ROOTDIR, "cache", "cast") + os.sep directories["GUESTCASTDIR"] = os.path.join(ROOTDIR, "cache", "guestcast") + os.sep directories["CREWDIR"] = os.path.join(ROOTDIR, "cache", "crew") + os.sep #directories["GUESTCREWDIR"] = os.path.join(ROOTDIR, "cache", "guestcrew") + os.sep #directories["CREDITSDIR"] = os.path.join(ROOTDIR, "cache", "credits") + os.sep #directories["ALIASDIR"] = os.path.join(ROOTDIR, "cache", "aliases") + os.sep tempdir = os.path.join(ROOTDIR, 'temp') + os.sep updatetable = 'updates.tvupdates' skip_file = f"{tempdir}skip_ids.txt" # Create directories if missing so later code doesn't fail try: os.makedirs(LOGDIR, exist_ok=True) os.makedirs(os.path.join(ROOTDIR, "cache", "series"), exist_ok=True) os.makedirs(os.path.join(ROOTDIR, "cache", "episodes"), exist_ok=True) os.makedirs(os.path.join(ROOTDIR, "cache", "cast"), exist_ok=True) os.makedirs(os.path.join(ROOTDIR, "cache", "guestcast"), exist_ok=True) os.makedirs(os.path.join(ROOTDIR, "cache", "crew"), exist_ok=True) os.makedirs(os.path.join(ROOTDIR, "cache", "guestcrew"), exist_ok=True) except Exception: # If directory creation fails, let the logging initialization surface the error pass lprint = logs.Logger.LPrint(LOGDIR, False) lprint.start_log() import settings.config config_options = settings.config.Config(ROOTDIR) options = config_options.config_options mongo_updater = dbmongo(options) # Create an instance of dbmongo dbtype, apitype = options["dbtype"], options["apitype"] dbs = settype(dbtype, apitype, options) dbclass, dbengine, apiengine = dbs.dbclass, dbs.dbengine, dbs.apiengine uprocess = url.process.Loadurl() TVMAZEURLS = url.urls.Tvmazeurls() TVMAZE_URLS = get_tvmaze(TVMAZEURLS) TVDBURLS = url.urls.TVDBurls() TVDB_URLS = get_tvdb(TVDBURLS) updateRange = None lprint.logprint("info", "Creating JSON instance for data retrieval.") lprint.logprint("info", "JSON instance created.") lprint.logprint("info", f"Update type is {options['updatetype']}") timedifference = options["updatetype"] currenttimestamp = time.time() if timedifference.lower() == "full": timedifference = int(currenttimestamp) basetime = str(currenttimestamp).split(".") currenttimestamp = int(basetime[0]) ts1 = str(datetime.now()).split(".")[0] lprint.logprint("info", "Sync start time: " + str(ts1)) print_starttime(lprint, ts1) dbExec = db.sql.Dbexec() createTempTables = dbclass["CreateTemptables"] new_updates = {} ct = createTempTables( dbengine["session"], dbengine["metadata"], dbengine["engine"] ) tempTableList = ct.tablelist new_updates = [] if options["apitype"].lower() == "tvmaze": try: new_updates = set_apienv( TVMAZE_URLS, uprocess, dbengine, dbExec, tempTableList, lprint, ) except Exception as exc: # noqa: PLC0301 - broad handler to log failures lprint.logprint("error", f"Failed to set API env: {exc}") return 1 timeDifference = options["updatetype"] currentTimestamp = time.time() if timeDifference.lower() == "full" or options["initload"] == "1": tdifference = 0 else: tdifference = int(currentTimestamp) - int( options[options["updatetype"]] ) needed_updates = [] updateList = [] for result in new_updates: updateid = result[0] updatestamp = int(result[1]) tdif = int(tdifference) if updatestamp > tdif: needed_updates.append(updateid) updateList.append(updateid) lprint.logprint("info", "Update Type: " + str(options["updatetype"])) updateRange = len(updateList) lprint.logprint("info", "Number of updates: " + str(updateRange)) # Save a downloaded copy of the updates JSON for debugging/inspection try: os.makedirs(tempdir, exist_ok=True) out_file = 'tvmaze_updates.json' prev_file = f'post_tvmaze_updates.json.{currentTimestamp}' previous_updates_file = f"{tempdir}{prev_file}" # TVMAZE_URLS is a dict of urls; use the updatesurl if available updates_url = TVMAZE_URLS.get('updatesurl') if isinstance(TVMAZE_URLS, dict) else None if updates_url: json_downloader.download_json_to_file(updates_url, tempdir, out_file, lprint) # load and convert saved JSON into a dict (id -> timestamp) try: downloaded_updates_dict = {} downloaded_updates_list = [] if os.path.exists(f"{tempdir}{out_file}"): with open(f"{tempdir}{out_file}", 'r') as jf: loaded = json.load(jf) if isinstance(loaded, dict): for k, v in loaded.items(): try: kid = int(k) except Exception: kid = k try: vt = int(v) except Exception: vt = v downloaded_updates_dict[kid] = vt downloaded_updates_list = list(downloaded_updates_dict.items()) elif isinstance(loaded, list): for item in loaded: if isinstance(item, list) and len(item) >= 2: k = item[0]; v = item[1] elif isinstance(item, dict): if 'id' in item and 'timestamp' in item: k = item['id']; v = item['timestamp'] elif 'seriesid' in item and 'timestamp' in item: k = item['seriesid']; v = item['timestamp'] else: downloaded_updates_list.append(item) continue else: k = item; v = None try: kid = int(k) except Exception: kid = k try: vt = int(v) if v is not None and str(v).isdigit() else v except Exception: vt = v downloaded_updates_dict[kid] = vt downloaded_updates_list.append((kid, vt)) else: downloaded_updates_list = [loaded] lprint.logprint("info", f"Loaded {len(downloaded_updates_list)} entries from tvmaze_updates.json") else: downloaded_updates_dict = {} downloaded_updates_list = [] except Exception: downloaded_updates_dict = {} downloaded_updates_list = [] except Exception: # non-fatal; continue lprint.logprint("warning", "Failed to save or load downloaded updates JSON.") previouslisting = dbExec.rawsql_select( dbengine["engine"], "select seriesid, timestamp from updates.tvupdates order by seriesid", lprint ) print(f"Saving previous updates to file.") with open(f"{previous_updates_file}", 'w') as pjf: for pl in previouslisting: pjf.write(f"{pl[0]}: {pl[1]}\n") pjf.close() # convert previouslisting (list of tuples) into a dict: row_id -> country_name try: _previouslisting_list = previouslisting previous_listing_dict = {row[0]: row[1] for row in _previouslisting_list} except Exception: # leave previouslisting as-is on any failure pass # Ensure dictionaries exist try: downloaded_updates_dict except NameError: downloaded_updates_dict = {} try: previous_listing_dict except NameError: previous_listing_dict = {} files_needed = [] directory_list = [] for value in directories.values(): directory_list.append(value) skip_ids = {} skipfile = open(skip_file, 'r') for line in skipfile: field = line.split(';') skip_ids[field[0]] = field[1] print(f"Updating update table.") dbExec.update_tvupdates( dbengine["engine"], downloaded_updates_dict, "updates.tvupdates" ) print("update complete.") print(f"Processing {len(downloaded_updates_dict)} downloaded updates against {len(previous_listing_dict)} previous updates for files needed.") for seriesid, ts in downloaded_updates_dict.items(): if seriesid in previous_listing_dict: prev_ts = previous_listing_dict[seriesid] # Check to see if all json files are present for seriesid try: if int(ts) == int(prev_ts): found_path = find_file_in_multiple_dirs(f"{seriesid}.json", directory_list) if len(found_path) == 5: continue else: for directory in directory_list: if directory not in found_path: files_needed.append([f"{directory}", f"{seriesid}.json"]) elif int(ts) > int(prev_ts): for directory in directory_list: files_needed.append([f"{directory}", f"{seriesid}.json"]) except Exception: lprint.logprint("debug", f"Unable to process Series {seriesid} timestamps: downloaded {ts} vs previous {prev_ts}") else: # Add needed json files to needed file list. for directory in directory_list: files_needed.append([f"{directory}", f"{seriesid}.json"]) lprint.logprint("debug", f"Series {seriesid} present in downloaded updates but missing in previous listing") lprint.logprint("info", f"Downloadinng JSON for {len(files_needed)} series IDs.") print(f"Downloadinng JSON for {len(files_needed)} files.") for i in tqdm(range(len(files_needed)), desc="Downloading JSON files", unit="filename"): for directory, file in files_needed[i:i+1]: if 'credits' in directory: if file.replace('.json', '') in skip_ids and 'castcredits' in skip_ids[file.replace('.json', '')].strip(): lprint.logprint("info", f"Skipping credits download for series ID {file.replace('.json', '')} as per skip_ids.txt") continue else: url_template = TVMAZE_URLS["creditsurl"] elif 'crew' in directory: url_template = TVMAZE_URLS["crewurl"] elif 'cast' in directory: url_template = TVMAZE_URLS["casturl"] elif 'episodes' in directory: url_template = TVMAZE_URLS["episodesurl"] elif 'aliases' in directory: url_template = TVMAZE_URLS["aliasurl"] else: url_template = TVMAZE_URLS["showurl"] seriesid = file.replace('.json', '') json_downloader.download_json_to_file(url_template.replace("", str(seriesid)), directory, f'{seriesid}.json', lprint) dbExec.update_tvupdates(dbengine["engine"], downloaded_updates_dict, updatetable) print(f"Loading JSON files into MongoDB.") successful, failed = load_json_to_mongodb(mongo_updater, directories, updateList, lprint) # mongo_updater = dbmongo(options) # mongo_updater.update_mongo( # updateList, # directories, # dbExec, # dbengine, # lprint, # ) # return 0 if __name__ == "__main__": sys.exit(main())