Files
new_dbsync/update_mongo.py
T

371 lines
14 KiB
Python
Executable File

#!/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_downloader
import json
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 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
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 = 'post_tvmaze_updates.json'
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 update 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()
print(f"Updating update table.")
dbExec.update_tvupdates(
dbengine["engine"],
downloaded_updates_dict,
"updates.tvupdates"
)
print("update complete.")
# 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"Processing {len(downloaded_updates_dict)} downloaded updates against {len(previous_listing_dict)} previous listings 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) == 8:
continue
else:
for directory in directory_list:
if directory not in found_path:
files_needed.append([f"{directory}", f"{seriesid}.json"])
else:
for directory in directory_list:
files_needed.append([f"{directory}", f"{seriesid}.json"])
except Exception:
for directory in directory_list:
files_needed.append([f"{directory}", f"{seriesid}.json"])
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)
# countrylisting = dbExec.rawsql_select(
# dbengine["engine"],
# "select row_id, country_name from dbo.countrydata",
# lprint
# )
# for country in countrylisting:
# country_name = country[1]
# country_id = country[0]
# countrydictionary[country_name] = country_id
# mongo_updater = dbmongo(options)
# mongo_updater.update_mongo(
# updateList,
# ROOTDIR,
# directories,
# jget,
# countrydictionary,
# dbExec,
# dbengine,
# dboBase,
# lprint,
# )
# return 0
if __name__ == "__main__":
sys.exit(main())