Files
new_dbsync/update_mongo.py
T

362 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 jprocessor.process
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, jget, 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"])
availableupdates = jget.jconvert(inputdata)
lprint.logprint("info", f"Retrieved {len(availableupdates)} rows for processing......")
dbExec.update_tvupdates(dbengine["engine"], availableupdates, updatetable)
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
# 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)
countrydictionary = {}
updateRange = None
lprint.logprint("info", "Creating JSON instance for data retrieval.")
jget = jprocessor.process.Jsonload()
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)
dboBase = dbengine["dboBase"]
dbExec = db.sql.Dbexec()
createTables = dbclass["Createtables"]
createTempTables = dbclass["CreateTemptables"]
new_updates = {}
ct = createTempTables(
dbengine["session"],
dbengine["metadata"],
dbengine["engine"]
)
tempTableList = ct.tablelist
if options["apitype"].lower() == "tvmaze":
try:
new_updates = set_apienv(
TVMAZE_URLS,
uprocess,
jget,
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
from cache.process import Cacher
cacher = Cacher()
try:
cacher.process(options, jget, new_updates, ROOTDIR, TVMAZE_URLS)
except Exception as exc:
lprint.logprint("error", f"Cache processing failed: {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'
# 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.")
#del downloaded_updates_list
previouslisting = dbExec.rawsql_select(
dbengine["engine"],
"select seriesid, timestamp from updates.tvupdates order by seriesid",
lprint
)
# 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)
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]
try:
if int(ts) == int(prev_ts):
found_path = find_file_in_multiple_dirs(f"{seriesid}.json", directory_list)
#print(f"Found path for series ID {seriesid}: {found_path}")
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:
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:
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)
# 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())