Refactor load_json_to_mongodb function to include updateDict for change detection and enhance logging with screenprint method

Makefile

@@ -1,15 +0,0 @@
-CC = gcc
-CFLAGS = -fPIC -O2 -I/usr/include
-LDFLAGS = -shared
-LIBS = -lpq
-
-TARGET = pg_vtab.so
-SRC = pg_vtab.c
-
-all: $(TARGET)
-
-$(TARGET): $(SRC)
-	$(CC) $(CFLAGS) $(LDFLAGS) -o $@ $^ $(LIBS)
-
-clean:
-	rm -f $(TARGET)

logs/Logger.py

@@ -79,6 +79,23 @@ class LPrint:
         else:
             logging.error(f'Unknown log level: {loglevel}. Message: {logmessage}')
 
+    def screenprint(self, loglevel, logmessage):
+        """Log a message with a timestamp."""
+        logstamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
+        log_message = f'{logstamp}: {logmessage}'
+        loglevel = loglevel.lower()
+        if loglevel == 'debug':
+            logging.debug(log_message)
+            print(f"{log_message}")
+        elif loglevel == 'warning':
+            logging.warning(log_message)
+            print(f"{log_message}")
+        elif loglevel == 'info':
+            logging.info(log_message)
+            print(f"{log_message}")
+        else:
+            logging.error(f'Unknown log level: {loglevel}. Message: {logmessage}')
+            
     def create_rotating_log(self, logpath, max_bytes=10_000_000, backup_count=5):
         """Create a rotating log handler."""
         logger = logging.getLogger("Rotating Log")

update_mongodb.py

@@ -3,17 +3,11 @@
 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):
@@ -47,13 +41,6 @@ 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.
@@ -72,18 +59,20 @@ def find_file_in_multiple_dirs(filename, directories):
                 
     return directories_found
         
-def load_json_to_mongodb(mongo_updater, directories, update_list, lprint, track_changes=True):
+def load_json_to_mongodb(mongo_updater, directories, update_list, updateDict, 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.
+    Automatically detects if documents are new, updated, or unchanged by comparing
+    updateDict timestamps with existing document 'updated' fields.
     
     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
+        updateDict: dict with seriesid as key and timestamp as value
         lprint: logger instance for logging operations
         track_changes: bool, if True tracks inserted vs updated documents (default: True)
     
@@ -93,9 +82,10 @@ def load_json_to_mongodb(mongo_updater, directories, update_list, lprint, track_
     """
     successful_count = 0
     failed_count = 0
-    stats = {'series_inserted': 0, 'series_updated': 0, 'episodes_inserted': 0, 
+    stats = {'series_inserted': 0, 'series_updated': 0, 'series_skipped': 0, 
-             'episodes_updated': 0, 'cast_inserted': 0, 'cast_updated': 0, 
+             'episodes_inserted': 0, 'episodes_updated': 0, 'episodes_skipped': 0, 
-             'crew_inserted': 0, 'crew_updated': 0}
+             'cast_inserted': 0, 'cast_updated': 0, 'cast_skipped': 0,
+             'crew_inserted': 0, 'crew_updated': 0, 'crew_skipped': 0}
     
     lprint.logprint("info", f"Starting to load {len(update_list)} series into MongoDB")
     
@@ -112,24 +102,22 @@ def load_json_to_mongodb(mongo_updater, directories, update_list, lprint, track_
                     if series_data:
                         if track_changes:
                             existing = mongo_updater.mgseries.find_one({"id": series_data.get("id")})
+                            seriesid_str = str(seriesid)
+                            update_timestamp = updateDict.get(seriesid_str)
                             if existing:
-                                # Check if data has changed
+                                # Check if data has changed by comparing update timestamp
-                                if existing.get("updated") != series_data.get("updated"):
+                                if update_timestamp and int(update_timestamp) > existing.get("updated", 0):
+                                    mongo_updater.bulk_upsert_by_id(mongo_updater.mgseries, [series_data])
                                     stats['series_updated'] += 1
                                     lprint.logprint("debug", f"Updated series data for ID {seriesid}")
                                 else:
                                     lprint.logprint("debug", f"Series {seriesid} unchanged, skipping")
+                                    stats['series_skipped'] += 1
                                     pass  # Document unchanged
                             else:
+                                mongo_updater.bulk_upsert_by_id(mongo_updater.mgseries, [series_data])
                                 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):
@@ -145,14 +133,20 @@ def load_json_to_mongodb(mongo_updater, directories, update_list, lprint, track_
                                             if track_changes:
                                                 for episode in episodes_data:
                                                     existing = mongo_updater.mgepisodes.find_one({"id": episode.get("id")})
+                                                    seriesid_str = str(seriesid)
+                                                    update_timestamp = updateDict.get(seriesid_str)
                                                     if existing:
-                                    if existing.get("updated") != episode.get("updated"):
+                                                        if update_timestamp and int(update_timestamp) > existing.get("updated", 0):
+                                                            mongo_updater.bulk_upsert_by_id(mongo_updater.mgepisodes, episodes_data)
                                                             stats['episodes_updated'] += 1
-                                    # else: unchanged
                                                         else: 
+                                                            lprint.logprint("debug", f"Episode Data for Series {seriesid} unchanged, skipping")
+                                                            pass  # Document unchanged
+                                                    else:
+                                                        mongo_updater.bulk_upsert_by_id(mongo_updater.mgepisodes, episodes_data)
                                                         stats['episodes_inserted'] += 1
                                             
-                        mongo_updater.bulk_upsert_by_id(mongo_updater.mgepisodes, episodes_data)
+                                            #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}")
@@ -178,8 +172,10 @@ def load_json_to_mongodb(mongo_updater, directories, update_list, lprint, track_
                                             if track_changes:
                                                 for actor in actors:
                                                     existing = mongo_updater.mgactors.find_one({"id": actor.get("id")})
+                                                    seriesid_str = str(seriesid)
+                                                    update_timestamp = updateDict.get(seriesid_str)
                                                     if existing:
-                                    if existing.get("updated") != actor.get("updated"):
+                                                        if update_timestamp and int(update_timestamp) > existing.get("updated", 0):
                                                             stats['cast_updated'] += 1
                                                     else:
                                                         stats['cast_inserted'] += 1
@@ -210,8 +206,10 @@ def load_json_to_mongodb(mongo_updater, directories, update_list, lprint, track_
                                             if track_changes:
                                                 for crew in crew_list:
                                                     existing = mongo_updater.mgcrew.find_one({"id": crew.get("id")})
+                                                    seriesid_str = str(seriesid)
+                                                    update_timestamp = updateDict.get(seriesid_str)
                                                     if existing:
-                                    if existing.get("updated") != crew.get("updated"):
+                                                        if update_timestamp and int(update_timestamp) > existing.get("updated", 0):
                                                             stats['crew_updated'] += 1
                                                     else:
                                                         stats['crew_inserted'] += 1
@@ -224,13 +222,16 @@ def load_json_to_mongodb(mongo_updater, directories, update_list, lprint, track_
                                 
                                 successful_count += 1
                                 lprint.logprint("info", f"Successfully processed series {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
         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"MongoDB loading complete.")
-    lprint.logprint("info", f"Change statistics: {stats}")
     return successful_count, failed_count, stats
 
 def main() -> int:
@@ -247,38 +248,36 @@ def main() -> int:
     #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
 
+    updateList = []
+    updateDict = {}
+    needed_updates = []
+    
+    new_updates = open(f'{tempdir}tvmaze_updates.json', 'r').readlines()
+    for line in new_updates:
+        line = line.strip().replace('\n', '').replace('\r', '')
+        linefield = line.split(":")
+        updateList.append(linefield[0].replace('"',''))
+        try:
+            updateDict[linefield[0].replace('"','')] = linefield[1].replace(',','').strip()
+        except IndexError:
+            lprint.logprint("warning", f"Failed to parse line: {line}")
+    # 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)
+            
     config_options = settings.config.Config(ROOTDIR)
     options = config_options.config_options
-    mongo_updater = dbmongo(options)  # Create an instance of dbmongo
+    mongo_updater = dbmongo(options)
     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.")
@@ -292,226 +291,12 @@ def main() -> int:
     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("<seriesid>", 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
     
+    lprint.screenprint("info", f"Preparing to load JSON files into MongoDB.")
+    lprint.screenprint("info", f"Loading JSON files into MongoDB.")
+    successful, failed , st= load_json_to_mongodb(mongo_updater, directories, updateList, updateDict, lprint)
+    lprint.screenprint("info", f"JSON loading complete. Successful: {successful}, Failed: {failed}")
+    lprint.screenprint("info", f"Stats: {st}")
 
 
 if __name__ == "__main__":