add search_mongodb.py and change seriesid in update_mongodb.py from a string to an int.
This commit is contained in:
Executable
+266
@@ -0,0 +1,266 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
import os, time, sys, datetime
|
||||
from db.functions import settype, dbmongo
|
||||
import json
|
||||
|
||||
|
||||
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 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, 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 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)
|
||||
|
||||
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, 'series_skipped': 0,
|
||||
'episodes_inserted': 0, 'episodes_updated': 0, 'episodes_skipped': 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")
|
||||
|
||||
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")})
|
||||
seriesid_str = str(seriesid)
|
||||
update_timestamp = updateDict.get(seriesid_str)
|
||||
if existing:
|
||||
# Check if data has changed by comparing update timestamp
|
||||
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}")
|
||||
# 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")})
|
||||
seriesid_str = str(seriesid)
|
||||
update_timestamp = updateDict.get(seriesid_str)
|
||||
if existing:
|
||||
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:
|
||||
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)
|
||||
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")})
|
||||
seriesid_str = str(seriesid)
|
||||
update_timestamp = updateDict.get(seriesid_str)
|
||||
if existing:
|
||||
if update_timestamp and int(update_timestamp) > existing.get("updated", 0):
|
||||
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")})
|
||||
seriesid_str = str(seriesid)
|
||||
update_timestamp = updateDict.get(seriesid_str)
|
||||
if existing:
|
||||
if update_timestamp and int(update_timestamp) > existing.get("updated", 0):
|
||||
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}")
|
||||
#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.")
|
||||
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
|
||||
|
||||
import settings.config
|
||||
|
||||
updateList = []
|
||||
updateDict = {}
|
||||
needed_updates = []
|
||||
|
||||
config_options = settings.config.Config(ROOTDIR)
|
||||
options = config_options.config_options
|
||||
mongo_query = dbmongo(options)
|
||||
|
||||
seriesid = 4
|
||||
query = { "id": int(seriesid)}
|
||||
for doc in mongo_query.mgseries.find(query):
|
||||
print(doc)
|
||||
query = { "seriesid": str(seriesid)}
|
||||
for doc in mongo_query.mgepisodes.find(query):
|
||||
print(doc)
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
|
||||
Reference in New Issue
Block a user