diff --git a/db/functions.py b/db/functions.py index 5d39bb7..ae26a11 100644 --- a/db/functions.py +++ b/db/functions.py @@ -114,18 +114,18 @@ class settype(object): ) currentDir = os.path.dirname(os.path.abspath(__file__)) currentDir = os.path.dirname(currentDir) - ddl_file_path = f"{currentDir}/db/ddl/recreate_updates_schema.sql" - with open(ddl_file_path, "r") as file: - sql = file.read() - file - with engine.connect() as conn: - with conn as cursor: - cursor.execute(text(sql)) + # ddl_file_path = f"{currentDir}/db/ddl/recreate_updates_schema.sql" + # with open(ddl_file_path, "r") as file: + # sql = file.read() + # file + # with engine.connect() as conn: + # with conn as cursor: + # cursor.execute(text(sql)) updatesBase.prepare(engine, reflect=True, schema="updates") dboBase.prepare(engine, reflect=True, schema="dbo") - metadata = sqlalchemy.MetaData("updates") + metadata = sqlalchemy.MetaData(schema="updates") metadata.reflect(bind=engine) - dboMetadata = sqlalchemy.MetaData("dbo") + dboMetadata = sqlalchemy.MetaData(schema="dbo") dboMetadata.reflect(bind=engine) session = Session(engine) self.dbengine["engine"] = engine diff --git a/db/schema_generator.py b/db/schema_generator.py new file mode 100644 index 0000000..91319fe --- /dev/null +++ b/db/schema_generator.py @@ -0,0 +1,654 @@ +"""Generate PostgreSQL table schemas from MongoDB document structures.""" + +__author__ = 'Wendell Jones' + +from typing import Any, Dict, List, Optional, Set +from pymongo import MongoClient +from sqlalchemy import text, MetaData, insert, Table +import logging +import json + +logger = logging.getLogger(__name__) + + +class MongoToPostgresSchemaGenerator: + """Infer PostgreSQL schema from MongoDB collection and create tables.""" + + # Map MongoDB/Python types to PostgreSQL types + TYPE_MAPPING = { + 'null': 'TEXT', + 'boolean': 'BOOLEAN', + 'integer': 'BIGINT', + 'float': 'DOUBLE PRECISION', + 'string': 'TEXT', + 'array': 'JSONB', + 'object': 'JSONB', + 'date': 'TIMESTAMP', + 'objectid': 'TEXT', + } + + def __init__(self, sample_size: int = 100): + """ + Initialize schema generator. + + Args: + sample_size: Number of documents to sample for type inference. + """ + self.sample_size = sample_size + self.field_types: Dict[str, Set[str]] = {} + self.final_schema: Dict[str, str] = {} + + def infer_type(self, value: Any) -> str: + """ + Infer the MongoDB type of a value. + + Args: + value: The value to inspect. + + Returns: + A type string suitable for TYPE_MAPPING. + """ + if value is None: + return 'null' + if isinstance(value, bool): + return 'boolean' + if isinstance(value, int) and not isinstance(value, bool): + return 'integer' + if isinstance(value, float): + return 'float' + if isinstance(value, str): + return 'string' + if isinstance(value, list): + return 'array' + if isinstance(value, dict): + return 'object' + # Handle MongoDB-specific types + if hasattr(value, '__class__'): + type_name = value.__class__.__name__.lower() + if 'objectid' in type_name: + return 'objectid' + if 'datetime' in type_name: + return 'date' + return 'string' + + def merge_types(self, types: Set[str]) -> str: + """ + Merge multiple observed types for a field into a single PostgreSQL type. + Priority: object/array > string > float > integer > boolean > null + + Args: + types: Set of type strings. + + Returns: + The merged PostgreSQL type. + """ + if not types: + return self.TYPE_MAPPING['null'] + + # Priority order + priority = ['array', 'object', 'string', 'float', 'integer', 'boolean', 'null', 'date'] + for ptype in priority: + if ptype in types: + return self.TYPE_MAPPING.get(ptype, self.TYPE_MAPPING['string']) + + return self.TYPE_MAPPING['string'] + + def analyze_collection( + self, + mongo_db: Any, + collection_name: Any + ) -> Dict[str, str]: + """ + Analyze a MongoDB collection and infer field types. + + Args: + mongo_db: A pymongo database object or None if collection_name is a collection. + collection_name: Name of the collection (str) or a pymongo Collection object directly. + + Returns: + Dictionary mapping field names to PostgreSQL types. + """ + self.field_types = {} + # Handle both cases: collection name string or collection object + if isinstance(collection_name, str): + collection = mongo_db[collection_name] + else: + # Assume it's already a pymongo Collection object + collection = collection_name + + # Sample documents from the collection + sample = list(collection.find({}).limit(self.sample_size)) + + if not sample: + logger.warning(f"Collection '{collection_name}' is empty; no schema inferred.") + return {} + + logger.info(f"Analyzing {len(sample)} documents from '{collection_name}'") + + # Gather field types + for doc in sample: + for field_name, value in doc.items(): + inferred = self.infer_type(value) + self.field_types.setdefault(field_name, set()).add(inferred) + + # Merge types + self.final_schema = {} + for field_name, types in self.field_types.items(): + merged_type = self.merge_types(types) + self.final_schema[field_name] = merged_type + logger.debug(f" {field_name}: {merged_type} (observed: {types})") + + return self.final_schema + + def generate_create_table_sql( + self, + table_name: str, + schema_name: str = 'public', + pk_field: Optional[str] = None, + exclude_fields: Optional[List[str]] = None, + ) -> str: + """ + Generate a CREATE TABLE statement from the inferred schema. + + Args: + table_name: Name of the table to create. + schema_name: PostgreSQL schema (default: 'public'). + pk_field: Field to use as primary key (e.g., 'id', '_id', 'seriesid'). + exclude_fields: List of field names to exclude. + + Returns: + A CREATE TABLE SQL statement. + """ + if not self.final_schema: + raise ValueError("No schema has been analyzed. Call analyze_collection first.") + + exclude_fields = exclude_fields or [] + columns = [] + + for field_name, pg_type in self.final_schema.items(): + if field_name in exclude_fields or field_name == '_id': + continue + columns.append(f" {field_name} {pg_type}") + + # Add primary key constraint if specified + constraints = [] + if pk_field and pk_field in self.final_schema: + constraints.append(f" PRIMARY KEY ({pk_field})") + + all_lines = columns + constraints + columns_str = ',\n'.join(all_lines) + + full_table_name = f"{schema_name}.{table_name}" + sql = f"""CREATE TABLE IF NOT EXISTS {full_table_name} ( +{columns_str} +);""" + + return sql + + def create_table_in_postgres( + self, + engine: Any, + table_name: str, + schema_name: str = 'public', + pk_field: Optional[str] = None, + exclude_fields: Optional[List[str]] = None, + drop_existing: bool = False, + ) -> None: + """ + Create a table in PostgreSQL based on the inferred schema. + + Args: + engine: SQLAlchemy engine connected to PostgreSQL, or a dict with 'engine' key. + table_name: Name of the table to create. + schema_name: PostgreSQL schema (default: 'public'). + pk_field: Field to use as primary key. + exclude_fields: List of field names to exclude. + drop_existing: If True, drop the table before creating. + """ + # Handle both dict (from db/functions.py) and engine object directly + if isinstance(engine, dict): + engine = engine['engine'] + + full_table_name = f"{schema_name}.{table_name}" + + create_sql = self.generate_create_table_sql( + table_name, + schema_name=schema_name, + pk_field=pk_field, + exclude_fields=exclude_fields, + ) + + with engine.begin() as conn: + if drop_existing: + logger.info(f"Dropping existing table {full_table_name}...") + conn.execute(text(f"DROP TABLE IF EXISTS {full_table_name};")) + + logger.info(f"Creating table {full_table_name}...") + conn.execute(text(create_sql)) + logger.info(f"Table {full_table_name} created successfully.") + + def create_table_from_collection( + self, + mongo_db: Any, + collection_name: str, + engine: Any, + table_name: Optional[str] = None, + schema_name: str = 'dbo', + pk_field: Optional[str] = None, + exclude_fields: Optional[List[str]] = None, + drop_existing: bool = False, + ) -> str: + """ + End-to-end: analyze a MongoDB collection and create a PostgreSQL table. + + Args: + mongo_db: A pymongo database object. + collection_name: Name of the MongoDB collection. + engine: SQLAlchemy engine connected to PostgreSQL. + table_name: Name of the PostgreSQL table (defaults to collection_name). + schema_name: PostgreSQL schema (default: 'dbo'). + pk_field: Field to use as primary key. + exclude_fields: List of field names to exclude. + drop_existing: If True, drop the table before creating. + + Returns: + The generated SQL statement. + """ + if table_name is None: + table_name = collection_name + + # Analyze the collection + self.analyze_collection(mongo_db, collection_name) + + # Create the table + self.create_table_in_postgres( + engine, + table_name, + schema_name=schema_name, + pk_field=pk_field, + exclude_fields=exclude_fields, + drop_existing=drop_existing, + ) + + # Return the SQL for reference + return self.generate_create_table_sql( + table_name, + schema_name=schema_name, + pk_field=pk_field, + exclude_fields=exclude_fields, + ) + + +# Convenience function +def create_postgres_table_from_mongo( + mongo_db: Any, + collection_name: str, + engine: Any, + table_name: Optional[str] = None, + schema_name: str = 'dbo', + pk_field: Optional[str] = None, + sample_size: int = 100, + drop_existing: bool = False, +) -> str: + """ + Convenience function to create a PostgreSQL table from a MongoDB collection. + + Example: + from db.schema_generator import create_postgres_table_from_mongo + + sql = create_postgres_table_from_mongo( + mongo_db=mgdb, + collection_name='series', + engine=pg_engine, + table_name='seriesdata', + schema_name='dbo', + pk_field='seriesid', + drop_existing=True, + ) + print(f"Created table with schema:\\n{sql}") + + Args: + mongo_db: A pymongo database object. + collection_name: Name of the MongoDB collection. + engine: SQLAlchemy engine connected to PostgreSQL. + table_name: Name of the PostgreSQL table (defaults to collection_name). + schema_name: PostgreSQL schema (default: 'dbo'). + pk_field: Field to use as primary key (optional). + sample_size: Number of documents to sample (default: 100). + drop_existing: If True, drop the table before creating. + + Returns: + The generated SQL statement. + """ + generator = MongoToPostgresSchemaGenerator(sample_size=sample_size) + return generator.create_table_from_collection( + mongo_db=mongo_db, + collection_name=collection_name, + engine=engine, + table_name=table_name, + schema_name=schema_name, + pk_field=pk_field, + drop_existing=drop_existing, + ) + + +class MongoDocumentInserter: + """Insert MongoDB documents into PostgreSQL tables with type conversion.""" + + def __init__(self, batch_size: int = 1000): + """ + Initialize the document inserter. + + Args: + batch_size: Number of documents to insert per batch (default: 1000). + """ + self.batch_size = batch_size + + def convert_value(self, value: Any) -> Any: + """ + Convert a MongoDB value to PostgreSQL-compatible type. + + - MongoDB ObjectId → string + - datetime → string (ISO format) + - list/dict → JSON string (will be stored as JSONB) + - None → None + + Args: + value: The value to convert. + + Returns: + The converted value. + """ + if value is None: + return None + + # Handle MongoDB ObjectId + if hasattr(value, '__class__') and 'ObjectId' in value.__class__.__name__: + return str(value) + + # Handle datetime + if hasattr(value, 'isoformat'): + return value.isoformat() + + # Handle lists and dicts (will be stored as JSONB) + if isinstance(value, (list, dict)): + return json.dumps(value) + + return value + + def prepare_documents( + self, + documents: List[Dict[str, Any]], + column_names: List[str], + exclude_fields: Optional[List[str]] = None, + ) -> List[Dict[str, Any]]: + """ + Prepare documents for insertion: convert types, exclude fields, etc. + + Args: + documents: List of documents (dicts) from MongoDB. + column_names: Column names in the target table. + exclude_fields: Fields to exclude (e.g., ['_id']). + + Returns: + List of prepared documents ready for insertion. + """ + exclude_fields = exclude_fields or ['_id'] + prepared = [] + + for doc in documents: + row = {} + for col in column_names: + if col not in exclude_fields and col in doc: + row[col] = self.convert_value(doc[col]) + elif col not in exclude_fields: + row[col] = None + prepared.append(row) + + return prepared + + def insert_documents( + self, + engine: Any, + table_name: str, + documents: List[Dict[str, Any]], + schema_name: str = 'dbo', + on_conflict: Optional[str] = None, + exclude_fields: Optional[List[str]] = None, + skip_null: bool = False, + ) -> int: + """ + Insert documents into a PostgreSQL table in batches. + + Args: + engine: SQLAlchemy engine (or dict with 'engine' key). + table_name: Target table name. + documents: List of documents (dicts) to insert. + schema_name: PostgreSQL schema (default: 'dbo'). + on_conflict: ON CONFLICT clause (e.g., "DO NOTHING" or + "DO UPDATE SET field=EXCLUDED.field"). + exclude_fields: Fields to exclude from insert (default: ['_id']). + skip_null: If True, skip None values in the INSERT (uses COALESCE). + + Returns: + Total number of documents inserted. + """ + if isinstance(engine, dict): + engine = engine['engine'] + + if not documents: + logger.warning("No documents to insert.") + return 0 + + exclude_fields = exclude_fields or ['_id'] + + full_table_name = f"{schema_name}.{table_name}" + total_inserted = 0 + + # Reflect target table to get column definitions and types + metadata = MetaData() + try: + table = Table(table_name, metadata, autoload_with=engine, schema=schema_name) + column_objs = {c.name: c for c in table.columns} + column_names = list(column_objs.keys()) + except Exception: + # Fallback: derive column names from first document + column_names = [k for k in documents[0].keys() if k not in exclude_fields] + column_objs = {} + + # Prepare and convert documents according to target column types + prepared: List[Dict[str, Any]] = [] + for doc in documents: + # create case-insensitive key map for doc + key_map = {k.lower(): k for k in doc.keys()} + row: Dict[str, Any] = {} + for col in column_names: + if exclude_fields and col in exclude_fields: + continue + # find matching key in document (case-insensitive) + val = None + if col in doc: + val = doc[col] + elif col.lower() in key_map: + val = doc[key_map[col.lower()]] + + # Convert based on column type if available + colobj = column_objs.get(col) + if colobj is not None and val is not None: + try: + from datetime import datetime + import sqlalchemy as sa + + ctype = colobj.type + # Integer target: convert ISO datetime strings to epoch + if isinstance(ctype, (sa.Integer, sa.BigInteger)): + if isinstance(val, str): + # try parse datetime string + try: + dt = datetime.fromisoformat(val) + val = int(dt.timestamp()) + except Exception: + # try numeric parse + try: + val = int(val) + except Exception: + pass + elif isinstance(val, float): + val = int(val) + # Datetime target: convert epoch ints to datetime + elif isinstance(ctype, (sa.DateTime, sa.TIMESTAMP)): + if isinstance(val, (int, float)): + val = datetime.fromtimestamp(val) + elif isinstance(val, str): + try: + val = datetime.fromisoformat(val) + except Exception: + pass + # JSON target: ensure Python dict/list + elif 'JSON' in type(ctype).__name__.upper() or 'JSON' in str(ctype).upper(): + if isinstance(val, str): + try: + val = json.loads(val) + except Exception: + pass + except Exception: + # ignore conversion errors and use original value + pass + + # Final conversion for general values + if val is None: + row[col] = None + else: + row[col] = self.convert_value(val) if col not in (exclude_fields or []) else None + + prepared.append(row) + + # Build INSERT statement template using reflected column order + cols_str = ', '.join(column_names) + placeholders = ', '.join([':' + col for col in column_names]) + insert_sql = f"INSERT INTO {full_table_name} ({cols_str}) VALUES ({placeholders})" + if on_conflict: + insert_sql += f" {on_conflict}" + + with engine.begin() as conn: + if on_conflict: + for doc in prepared: + try: + conn.execute(text(insert_sql), [doc]) + total_inserted += 1 + except Exception as e: + logger.error(f"Error inserting document into {full_table_name}: {e}") + raise + logger.info(f"Inserted {total_inserted} documents into {full_table_name}") + else: + for i in range(0, len(prepared), self.batch_size): + batch = prepared[i:i + self.batch_size] + try: + conn.execute(text(insert_sql), batch) + batch_count = len(batch) + total_inserted += batch_count + logger.info( + f"Inserted {batch_count} documents into {full_table_name} (total: {total_inserted})" + ) + except Exception as e: + logger.error(f"Error inserting batch into {full_table_name}: {e}") + raise + + return total_inserted + + def insert_from_collection( + self, + engine: Any, + table_name: str, + collection: Any, + schema_name: str = 'dbo', + on_conflict: Optional[str] = None, + exclude_fields: Optional[List[str]] = None, + query_filter: Optional[Dict] = None, + limit: Optional[int] = None, + ) -> int: + """ + Insert documents from a MongoDB collection into PostgreSQL. + + Args: + engine: SQLAlchemy engine (or dict with 'engine' key). + table_name: Target PostgreSQL table name. + collection: PyMongo collection object. + schema_name: PostgreSQL schema (default: 'dbo'). + on_conflict: ON CONFLICT clause. + exclude_fields: Fields to exclude (default: ['_id']). + query_filter: MongoDB query filter (default: {}). + limit: Maximum documents to insert (default: None for all). + + Returns: + Total number of documents inserted. + """ + query_filter = query_filter or {} + + logger.info(f"Fetching documents from MongoDB collection...") + cursor = collection.find(query_filter) + + if limit: + cursor = cursor.limit(limit) + + documents = list(cursor) + logger.info(f"Retrieved {len(documents)} documents.") + + return self.insert_documents( + engine=engine, + table_name=table_name, + documents=documents, + schema_name=schema_name, + on_conflict=on_conflict, + exclude_fields=exclude_fields, + ) + + +# Convenience function for inserting documents +def insert_mongo_documents_to_postgres( + engine: Any, + table_name: str, + documents: List[Dict[str, Any]], + schema_name: str = 'dbo', + on_conflict: Optional[str] = None, + exclude_fields: Optional[List[str]] = None, + batch_size: int = 1000, +) -> int: + """ + Convenience function to insert documents into PostgreSQL. + + Example: + from db.schema_generator import insert_mongo_documents_to_postgres + + count = insert_mongo_documents_to_postgres( + engine=dbengine, + table_name='seriesdata', + documents=series_docs, + schema_name='dbo', + on_conflict="DO NOTHING", + exclude_fields=['_id'], + ) + print(f"Inserted {count} documents") + + Args: + engine: SQLAlchemy engine (or dict with 'engine' key). + table_name: Target table name. + documents: List of documents to insert. + schema_name: PostgreSQL schema (default: 'dbo'). + on_conflict: ON CONFLICT clause (optional). + exclude_fields: Fields to exclude (default: ['_id']). + batch_size: Number of documents per batch (default: 1000). + + Returns: + Total number of documents inserted. + """ + inserter = MongoDocumentInserter(batch_size=batch_size) + return inserter.insert_documents( + engine=engine, + table_name=table_name, + documents=documents, + schema_name=schema_name, + on_conflict=on_conflict, + exclude_fields=exclude_fields, + ) + diff --git a/db/sql.py b/db/sql.py index c9a9e4c..efa3ef2 100644 --- a/db/sql.py +++ b/db/sql.py @@ -12,27 +12,32 @@ class Dbexec(): def update_tvupdates(self, engine, available_updates, tablename): print('Creating update transaction.') + # Build batch insert data updatelist = [ { - 'seriesid': seriesid, + 'seriesid': str(seriesid), 'timestamp': available_updates[seriesid] } for seriesid in tqdm( available_updates, - desc='Updating tvupdates table', + desc='Preparing tvupdates records', unit='record' ) ] - # Commented out code for applying updates for now. Will re-enable later after downloads are verified. print('Applying updates to table.') - with engine.connect() as conn: - conn.execute(text(f"truncate table {tablename}")) - for key, value in available_updates.items(): - seriesid = str(key) - timestamp = value - conn.execute(text(f"INSERT INTO {tablename} (seriesid, timestamp) VALUES (:seriesid, :timestamp)"), {'seriesid': seriesid, 'timestamp': timestamp}) - conn.commit() + # Use a transactional context; SQLAlchemy Connection.execute accepts + # a list of parameter mappings for bulk inserts (it will use the DBAPI + # executemany under the hood). + with engine.begin() as conn: + print(f'Truncating table {tablename} before insert.') + conn.execute(text(f"TRUNCATE TABLE {tablename}")) + print(f'Inserting {len(updatelist)} records into {tablename}.') + if updatelist: + conn.execute( + text(f"INSERT INTO {tablename} (seriesid, timestamp) VALUES (:seriesid, :timestamp)"), + updatelist, + ) print('Update of tvupdates table is complete.') def rawsql_select(self, engine, sqlquery, lprint): diff --git a/json2postgres.py b/json2postgres.py deleted file mode 100644 index d34d78d..0000000 --- a/json2postgres.py +++ /dev/null @@ -1,287 +0,0 @@ -#!/usr/bin/env python3 - -from pymongo import MongoClient -import json -import psycopg2 -import pandas as pd -from collections.abc import MutableMapping -from tqdm import tqdm - - -def flatten_dict(d: MutableMapping, sep: str = '.') -> MutableMapping: - [flat_dict] = pd.json_normalize(d, sep=sep).to_dict(orient='records') - return flat_dict - - -def delete_none(_dict): - """ - Delete None values recursively from all of the dictionaries, tuples, - lists, sets - """ - if isinstance(_dict, dict): - for key, value in list(_dict.items()): - if isinstance(value, (list, dict, tuple, set)): - _dict[key] = delete_none(value) - elif value is None or key is None: - del _dict[key] - - elif isinstance(_dict, (list, set, tuple)): - _dict = type(_dict)( - delete_none(item) for item in _dict if item is not None - ) - return _dict - - -def process_data(data, tablename): - inserts = [] - failed_count = 0 - seriesconflict = "ON CONFLICT (seriesid) DO UPDATE SET" - epconflict = "ON CONFLICT (episodeid, seriesid) DO UPDATE SET" - crewconflict = "ON CONFLICT (crewid) DO UPDATE SET" - actorconflict = "ON CONFLICT (actorid) DO UPDATE SET" - characterconflict = "ON CONFLICT (seriesid, characterid) DO UPDATE SET" - for document in data: - if document['name'].lower() == 'too many requests': - print("Skipping document due to 'Too Many Requests' in " - "document data") - continue - else: - new_document = delete_none(document) - if '_id' in new_document: - del new_document['_id'] - try: - flatten_data = flatten_dict(new_document) - except TypeError as e: - print( - ( - f"Unable to flatten data for current document. {e}\n" - "Exitting!" - ) - ) - failed_count += 1 - else: - column_list = [] - data_list = [] - for key, value in flatten_data.items(): - if isinstance(value, str): - value = value.replace('"', "'").replace("'", "''") - if key.lower() == '_links.self.href': - key = 'apilink' - if key.lower() == '_links.show.href': - key = 'apishowlink' - if key.lower() == '_links.show.name': - key = 'apishowname' - if '.' in key.lower(): - key = key.replace('.', '') - if tablename == 'seriesdata' and ( - key.lower() == 'name' or key.lower() == 'type' - ): - key = f"series_{key}" - if tablename == 'seriesdata' and key.lower() == 'id': - key = "seriesid" - if tablename == 'crewdata' and key.lower() == 'id': - key = "crewid" - if ( - tablename == 'actordata' and ( - key.lower() == 'name' or key.lower() == 'number' or - key.lower() == 'type' - ) - ): - key = f"actor{key}" - if tablename == 'actordata' and key.lower() == 'id': - key = "actorid" - if tablename == 'characterdata' and key.lower() == 'id': - key = "characterid" - if tablename == 'characterdata' and key.lower() == 'name': - key = "charactername" - if tablename == 'characterdata' and key.lower() == 'id': - key = "seriesid, characterid" - if ( - ( - tablename == 'epdata' or - tablename == 'actordata' - ) and ( - key.lower() == 'name' or - key.lower() == 'number' or - key.lower() == 'type' - ) - ): - key = f"episode_{key}" - if key.lower() == 'language': - key = "language_name" - if tablename == 'epdata' and key.lower() == 'id': - key = "episodeid" - if tablename == 'actordata' and 'seriesid' in key.lower(): - continue - else: - column_list.append(key) - if ( - 'genres' in key.lower() or - 'schedule' in key.lower() - ): - if isinstance(value, list): - value = ', '.join(value) - if isinstance(value, int) or isinstance(value, float): - value = str(json.dumps(value)) - data_list.append(f"'{value}'") - columns = ", ".join(column_list) - column_data = ", ".join(data_list) - conflict_list = [] - for indexer, column in enumerate(column_list): - conflict_list.append( - f"{column_list[indexer]} = {data_list[indexer]}" - ) - conflict_data = ', '.join(conflict_list).replace("\\", "") - conflict_clause = '' - if tablename == 'seriesdata': - conflict_clause = seriesconflict - elif tablename == 'epdata': - conflict_clause = epconflict - elif tablename == 'crewdata': - conflict_clause = crewconflict - elif tablename == 'actordata': - conflict_clause = actorconflict - elif tablename == 'characterdata': - conflict_clause = characterconflict - insert_command = ( - ( - f"Insert Into updates.{tablename} ({columns}) " - f"values ({column_data}) " - f"{conflict_clause} {conflict_data};" - ) - ) - inserts.append(insert_command) - return inserts, failed_count - - -try: - connection_string = ( - ( - 'postgres://postgres:Optimus0329@192.168.128.7:5432/media_dbsync' - '?options=-csearch_path%3Ddbo,public,updates' - ) - ) - pgclient = psycopg2.connect(connection_string) - pgcursor = pgclient.cursor() - # print(pgclient.get_dsn_parameters(), "\n") - print("Connected to postgresql database server.") - pgcursor.execute("SELECT version();") - record = pgcursor.fetchone() - print("You are connected to - ", record, "\n") - pgclient.set_isolation_level(0) -except Exception as e: - print(f"Unable to connect to postgresql database server:{e}\nExitting!") - exit() - - -host = '192.168.128.8' -port = 27017 -print(f"Connecting to mongodb at {host}:{port}") - -try: - connection_string = ( - f"mongodb://{host}:{port}" - ) - mgclient = MongoClient(connection_string) - mgdb = mgclient['test2'] -except BaseException as e: - print(f"Unable to connect to mongodb.{e}\nExitting!") - exit() -else: - mgseries = mgdb["series"] - mgseries.create_index('id', unique=True) - mgepisodes = mgdb["episodes"] - mgepisodes.create_index('id', unique=True) - mgactors = mgdb["actors"] - mgactors.create_index('id', unique=True) - mgcharacters = mgdb["characters"] - mgcharacters.create_index('id', unique=True) - mgcrew = mgdb["crew"] - mgcrew.create_index('id', unique=True) - print( - f"Connection to mongodb at {host}:" - f"{port} successful." - ) - -print("Creating Insert commands from collection data.") - -print("Collecting series data for series inserts.") -series_list = mgseries.find({}) -inserts, failed_count = process_data(series_list, 'seriesdata') -print(f"Total number of series inserts: {len(inserts)}") -for i in tqdm(range(len(inserts))): - try: - insert = inserts[i] - pgcursor.execute(insert) - except Exception as e: - print(f"Unable to insert data into seriesdata table. {e}") - failed_count += 1 -print(f"Total number of failed series inserts: {failed_count}") -pgclient.commit() - -epinserts = [] -failures = 0 -episode_array = mgepisodes.find({}) -print("Collecting episode data for episode inserts.") -inserts, failed_count = process_data(episode_array, 'epdata') -print(f"Total number of episode inserts: {len(inserts)}") -for i in tqdm(range(len(inserts))): - try: - insert = inserts[i] - pgcursor.execute(insert) - except Exception as e: - print(f"Unable to insert data into epdata table. {e}") - failed_count += 1 -print(f"Total number of failed episode inserts: {failed_count}") -pgclient.commit() - -failures = 0 -actor_array = mgactors.find({}) -print("Collecting actor data for actor inserts.") -inserts, failed_count = process_data(actor_array, 'actordata') -print(f"Total number of actor inserts: {len(inserts)}") -for i in tqdm(range(len(inserts))): - try: - insert = inserts[i] - pgcursor.execute(insert) - except Exception as e: - print(f"Unable to insert data into actors table. {e}") - failed_count += 1 -print(f"Total number of failed actor inserts: {failed_count}") -pgclient.commit() - -failures = 0 -character_array = mgcharacters.find({}) -print("Collecting actor data for actor inserts.") -inserts, failed_count = process_data(character_array, 'characterdata') -print(f"Total number of character inserts: {len(inserts)}") -for i in tqdm(range(len(inserts))): - try: - insert = inserts[i] - pgcursor.execute(insert) - except Exception as e: - print(f"Unable to insert data into character table. {e}") - failed_count += 1 -print(f"Total number of failed character inserts: {failed_count}") -pgclient.commit() - -failures = 0 -crew_array = mgcrew.find({}) -print("Collecting crew data for crew inserts.") -inserts, failed_count = process_data(crew_array, 'crewdata') -print(f"Total number of crew inserts: {len(inserts)}") -for i in tqdm(range(len(inserts))): - try: - insert = inserts[i] - pgcursor.execute(insert) - except Exception as e: - print(f"Unable to insert data into epdata table. {e}") - failed_count += 1 -print(f"Total number of failed crew inserts: {failed_count}") -pgclient.commit() - -pgclient.close() -print("PostgreSQL connection is closed") - - -exit() diff --git a/mongodb2postgres.py b/mongodb2postgres.py new file mode 100644 index 0000000..f662878 --- /dev/null +++ b/mongodb2postgres.py @@ -0,0 +1,120 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +import os +import time +import sys +from tqdm import tqdm +import logs.Logger +from db.functions import settype, dbmongo +from datetime import datetime +import json +from db.schema_generator import MongoToPostgresSchemaGenerator, MongoDocumentInserter +import settings.config + +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 + + +ROOTDIR = os.getcwd() +config_options = settings.config.Config(ROOTDIR) +options = config_options.config_options +dbtype, apitype = options["dbtype"], options["apitype"] +dbs = settype(dbtype, apitype, options) +dbclass, dbengine = dbs.dbclass, dbs.dbengine +mongo_updater = dbmongo(options) +generator = MongoToPostgresSchemaGenerator(sample_size=100) + +tableNames = ['seriesdata', 'episodesdata', 'actorsdata', 'charactersdata', 'crewdata'] + +for tableName in tableNames: + # Generate and print the schema for each collection + schema = generator.analyze_collection(mongo_updater, getattr(mongo_updater, f'mg{tableName[:-4]}')) + sql = generator.generate_create_table_sql( + tableName, + schema_name='updates', + pk_field='id' if tableName == 'seriesdata' else 'id', + ) + print(f"--- SQL Schema for {tableName} ---") + print(sql) + print("\n") + + # Or create it directly + generator.create_table_in_postgres( + dbengine, + tableName, + schema_name='updates', + #pk_field='seriesid' if tableName == 'seriesdata' else 'id', + pk_field='id', + drop_existing=True, + ) + + + +inserter = MongoDocumentInserter(batch_size=1000) + +# Map table names to their MongoDB collections and primary keys +collection_map = { + 'seriesdata': (mongo_updater.mgseries, 'id'), + 'episodesdata': (mongo_updater.mgepisodes, 'id'), + 'actorsdata': (mongo_updater.mgactors, 'id'), + 'charactersdata': (mongo_updater.mgcharacters, 'id'), + 'crewdata': (mongo_updater.mgcrew, 'id'), +} + +# Insert documents from collections with UPSERT +for tableName in tableNames: + if tableName in collection_map: + collection, pk_field = collection_map[tableName] + # Build ON CONFLICT ... DO UPDATE clause + # Re-analyze the specific collection so we have the correct fields + schema = generator.analyze_collection(mongo_updater, collection) + # Use lowercased column names to match PostgreSQL unquoted identifiers + update_cols = [col.lower() for col in schema.keys() if col.lower() != pk_field.lower() and col != '_id'] + update_clause = ', '.join([f"{col}=EXCLUDED.{col}" for col in update_cols]) + on_conflict = f"ON CONFLICT ({pk_field}) DO UPDATE SET {update_clause}" + + count = inserter.insert_from_collection( + engine=dbengine, + table_name=tableName, + collection=collection, + schema_name='updates', + on_conflict=on_conflict, + ) + print(f"Inserted/Updated {count} documents in {tableName}") + else: + print(f"Warning: No collection mapping for {tableName}") + +# # Or directly from MongoDB collection +# count = inserter.insert_from_collection( +# engine=dbengine, +# table_name='seriesdata', +# collection=mongo_db['series'], +# schema_name='dbo', +# on_conflict="DO NOTHING", +# ) +# if __name__ == "__main__": +# sys.exit(main()) + diff --git a/settings/tvsync_settings.cfg b/settings/tvsync_settings.cfg index 633f231..cafbc24 100644 --- a/settings/tvsync_settings.cfg +++ b/settings/tvsync_settings.cfg @@ -5,7 +5,7 @@ loglevel=info sqlserver_driver=ODBC Driver 13 for SQL Server hostname=192.168.128.3 mghostname=192.168.128.24 -dbname=media_dbsync +dbname=media_dbsync2 mgdbname=tvdata mysqlusername=root pgsqlusername=postgres diff --git a/temp/tvmaze_updates.json b/temp/tvmaze_updates.json index e1def04..e99a469 100644 --- a/temp/tvmaze_updates.json +++ b/temp/tvmaze_updates.json @@ -654,7 +654,7 @@ "672": 1770369680, "673": 1704794564, "674": 1756390245, - "675": 1770370407, + "675": 1770390069, "676": 1769358070, "677": 1739576696, "678": 1753836211, @@ -813,7 +813,7 @@ "836": 1764338899, "837": 1704794646, "838": 1732255155, - "839": 1770059557, + "839": 1770380111, "840": 1704794606, "841": 1728159086, "842": 1770083276, @@ -2043,7 +2043,7 @@ "2128": 1765286378, "2129": 1759734792, "2130": 1682854620, - "2131": 1770202701, + "2131": 1770389935, "2132": 1754081874, "2133": 1712680616, "2134": 1762462723, @@ -3305,7 +3305,7 @@ "3506": 1720796277, "3507": 1770263891, "3508": 1770366267, - "3509": 1768197125, + "3509": 1770384015, "3510": 1754868332, "3511": 1620925576, "3512": 1761558735, @@ -4748,7 +4748,7 @@ "5044": 1626604012, "5045": 1649023262, "5046": 1758134797, - "5047": 1770362960, + "5047": 1770389581, "5048": 1704795049, "5049": 1704794626, "5050": 1724707540, @@ -5284,7 +5284,7 @@ "5596": 1689448796, "5597": 1704795366, "5598": 1756382872, - "5599": 1766682039, + "5599": 1770383718, "5600": 1661599579, "5601": 1704795607, "5602": 1707581168, @@ -8058,7 +8058,7 @@ "8563": 1667774933, "8564": 1632672464, "8565": 1573754123, - "8566": 1770299022, + "8566": 1770383882, "8567": 1631188853, "8569": 1770238950, "8570": 1704795431, @@ -8977,7 +8977,7 @@ "9561": 1728600078, "9562": 1704794430, "9563": 1738102434, - "9564": 1770299127, + "9564": 1770383751, "9565": 1629446527, "9566": 1687701865, "9567": 1629446836, @@ -9061,7 +9061,7 @@ "9656": 1632926726, "9657": 1705154797, "9658": 1504835414, - "9659": 1770335206, + "9659": 1770380870, "9660": 1770342212, "9661": 1756100645, "9662": 1629180069, @@ -14343,8 +14343,8 @@ "15305": 1704795571, "15306": 1463223641, "15307": 1704795215, - "15308": 1770289880, - "15309": 1770307975, + "15308": 1770383287, + "15309": 1770390049, "15310": 1498199696, "15311": 1509849357, "15312": 1766427955, @@ -20300,7 +20300,7 @@ "21735": 1704795457, "21736": 1704793712, "21737": 1768578747, - "21738": 1770298180, + "21738": 1770383374, "21739": 1716253548, "21740": 1731204469, "21741": 1522661460, @@ -21807,7 +21807,7 @@ "23433": 1628631843, "23434": 1628637597, "23435": 1628637613, - "23436": 1481042161, + "23436": 1770382530, "23437": 1481041026, "23438": 1694104850, "23439": 1481049428, @@ -21928,7 +21928,7 @@ "23574": 1481573752, "23575": 1704795447, "23576": 1747320054, - "23577": 1724876760, + "23577": 1770390077, "23578": 1765392742, "23579": 1760210421, "23580": 1647202950, @@ -27247,7 +27247,7 @@ "29255": 1628522102, "29256": 1497558293, "29257": 1539116084, - "29259": 1770367925, + "29259": 1770387734, "29260": 1497572565, "29261": 1745446976, "29262": 1512674977, @@ -35175,7 +35175,7 @@ "37753": 1704793343, "37754": 1531699967, "37755": 1531748329, - "37756": 1770189666, + "37756": 1770380703, "37757": 1642223724, "37758": 1650994724, "37759": 1696148307, @@ -36423,7 +36423,7 @@ "39104": 1760378772, "39105": 1754132426, "39106": 1598397158, - "39107": 1764733647, + "39107": 1770381399, "39108": 1765038117, "39109": 1543156919, "39110": 1554972064, @@ -36474,7 +36474,7 @@ "39160": 1698029461, "39161": 1539905543, "39162": 1707811760, - "39163": 1769784522, + "39163": 1770389612, "39165": 1648137690, "39166": 1688320797, "39167": 1591725578, @@ -43816,7 +43816,7 @@ "46943": 1770243343, "46944": 1700089243, "46945": 1602438703, - "46946": 1735443941, + "46946": 1770381831, "46947": 1701103174, "46948": 1726280227, "46949": 1628841278, @@ -44823,7 +44823,7 @@ "47997": 1768856143, "47998": 1657728175, "47999": 1621762284, - "48000": 1760255385, + "48000": 1770384225, "48001": 1661583522, "48002": 1769958866, "48003": 1721922404, @@ -45172,7 +45172,7 @@ "48359": 1622723450, "48360": 1590765031, "48361": 1698407992, - "48362": 1769890084, + "48362": 1770389538, "48363": 1590860239, "48364": 1590674096, "48365": 1635803910, @@ -46082,7 +46082,7 @@ "49297": 1639764935, "49298": 1726141404, "49299": 1769555022, - "49300": 1770301348, + "49300": 1770386654, "49301": 1597529622, "49302": 1767610269, "49303": 1595897249, @@ -47584,7 +47584,7 @@ "50859": 1655261753, "50860": 1726997000, "50862": 1686923254, - "50863": 1770379277, + "50863": 1770379294, "50864": 1717432866, "50866": 1645129729, "50867": 1645316574, @@ -49699,7 +49699,7 @@ "53060": 1612246381, "53061": 1621109361, "53062": 1694741345, - "53063": 1770375548, + "53063": 1770384991, "53064": 1764157618, "53065": 1640595088, "53066": 1728813262, @@ -55856,7 +55856,7 @@ "59405": 1706814683, "59406": 1647860158, "59407": 1639197253, - "59408": 1745052135, + "59408": 1770380959, "59409": 1719835262, "59410": 1722798177, "59411": 1693816976, @@ -56047,7 +56047,7 @@ "59601": 1640111791, "59602": 1640108248, "59603": 1640110326, - "59604": 1770239960, + "59604": 1770379352, "59605": 1659182435, "59606": 1764671835, "59607": 1752655773, @@ -56664,7 +56664,7 @@ "60243": 1767887932, "60244": 1761404543, "60245": 1744968779, - "60246": 1768652340, + "60246": 1770386931, "60247": 1698265230, "60248": 1748359385, "60249": 1716059235, @@ -56821,7 +56821,7 @@ "60406": 1691253620, "60407": 1644413859, "60408": 1765306589, - "60409": 1764020578, + "60409": 1770380036, "60410": 1766837754, "60411": 1730521941, "60412": 1644607682, @@ -57968,7 +57968,7 @@ "61602": 1649850192, "61603": 1765317022, "61604": 1649860658, - "61605": 1770365336, + "61605": 1770383926, "61606": 1733481464, "61607": 1744813312, "61608": 1761742933, @@ -60007,7 +60007,7 @@ "63709": 1661465758, "63710": 1766237079, "63711": 1661730024, - "63712": 1672145630, + "63712": 1770379986, "63713": 1662157637, "63714": 1726920075, "63715": 1662157134, @@ -60861,7 +60861,7 @@ "64590": 1754943885, "64591": 1761422838, "64592": 1736339460, - "64593": 1770362466, + "64593": 1770386107, "64594": 1675519996, "64595": 1721990770, "64596": 1737673546, @@ -62313,7 +62313,7 @@ "66081": 1672763556, "66082": 1672325952, "66083": 1674572636, - "66084": 1770330926, + "66084": 1770384812, "66085": 1749380610, "66087": 1672339301, "66088": 1725004398, @@ -62544,7 +62544,7 @@ "66316": 1695678339, "66317": 1674291993, "66318": 1739522228, - "66319": 1770299300, + "66319": 1770383945, "66320": 1673467669, "66321": 1680546872, "66323": 1729544285, @@ -64946,7 +64946,7 @@ "68799": 1704793557, "68800": 1768128357, "68801": 1766505280, - "68802": 1761928983, + "68802": 1770389875, "68803": 1752508624, "68804": 1718623922, "68805": 1700369742, @@ -65842,7 +65842,7 @@ "69717": 1769721254, "69718": 1688666206, "69719": 1693388160, - "69720": 1770376501, + "69720": 1770388104, "69721": 1704709157, "69722": 1746639900, "69723": 1688686825, @@ -65920,7 +65920,7 @@ "69796": 1710435465, "69797": 1692859855, "69798": 1707370233, - "69799": 1694346042, + "69799": 1770387958, "69800": 1691237888, "69802": 1688922786, "69803": 1699391479, @@ -67614,7 +67614,7 @@ "71521": 1704794680, "71522": 1762797711, "71523": 1696364305, - "71524": 1770225202, + "71524": 1770385531, "71525": 1697874991, "71526": 1726402987, "71527": 1770225880, @@ -67925,7 +67925,7 @@ "71843": 1766886700, "71844": 1727895790, "71845": 1699105768, - "71846": 1766242116, + "71846": 1770380430, "71847": 1701450871, "71848": 1701105079, "71849": 1727140020, @@ -68533,7 +68533,7 @@ "72463": 1750671873, "72464": 1699107113, "72465": 1699108042, - "72466": 1770303379, + "72466": 1770388117, "72467": 1760248398, "72468": 1699190921, "72469": 1699122789, @@ -68568,7 +68568,7 @@ "72499": 1763421745, "72500": 1769685541, "72501": 1714488170, - "72503": 1770365206, + "72503": 1770389446, "72504": 1768137892, "72505": 1699650170, "72506": 1738182845, @@ -69249,7 +69249,7 @@ "73205": 1701776542, "73206": 1704576025, "73207": 1701877011, - "73208": 1770327598, + "73208": 1770386825, "73209": 1731774599, "73211": 1719693165, "73212": 1701801487, @@ -72914,7 +72914,7 @@ "76966": 1768761465, "76967": 1768758519, "76968": 1769954755, - "76969": 1766257916, + "76969": 1770381586, "76970": 1755010600, "76971": 1755019739, "76972": 1739111156, @@ -73142,7 +73142,7 @@ "77203": 1746727348, "77204": 1755015647, "77205": 1752312979, - "77206": 1766259087, + "77206": 1770382671, "77207": 1746807200, "77208": 1746539572, "77209": 1746633383, @@ -73841,7 +73841,7 @@ "77916": 1743014955, "77917": 1760026625, "77918": 1718830237, - "77919": 1770374260, + "77919": 1770380750, "77920": 1718877983, "77921": 1734218388, "77922": 1718880242, @@ -75837,7 +75837,7 @@ "79981": 1735325096, "79982": 1763020185, "79983": 1728159496, - "79984": 1770374849, + "79984": 1770387819, "79985": 1729468424, "79986": 1746436252, "79987": 1728239280, @@ -76501,7 +76501,7 @@ "80666": 1745811599, "80667": 1761644015, "80668": 1737713346, - "80669": 1770015814, + "80669": 1770386507, "80670": 1731086096, "80671": 1764146510, "80672": 1731692207, @@ -76912,7 +76912,7 @@ "81083": 1742124425, "81084": 1764461940, "81085": 1770082562, - "81086": 1769792537, + "81086": 1770388358, "81087": 1760112115, "81088": 1770148494, "81089": 1770147915, @@ -77045,7 +77045,7 @@ "81218": 1734746700, "81219": 1743943483, "81220": 1733188497, - "81221": 1746017882, + "81221": 1770388233, "81222": 1754669600, "81223": 1733646901, "81224": 1733688682, @@ -77132,7 +77132,7 @@ "81306": 1769476047, "81307": 1743013970, "81308": 1738312206, - "81309": 1770064272, + "81309": 1770380362, "81310": 1770180582, "81311": 1758417980, "81312": 1762305124, @@ -77382,8 +77382,8 @@ "81565": 1734595294, "81566": 1734595676, "81567": 1743703666, - "81568": 1769770251, - "81569": 1770326396, + "81568": 1770389726, + "81569": 1770388981, "81570": 1759413046, "81571": 1734598275, "81572": 1746684532, @@ -78247,7 +78247,7 @@ "82451": 1738831592, "82452": 1738832974, "82453": 1770188859, - "82454": 1769518011, + "82454": 1770386319, "82455": 1738854177, "82456": 1756458272, "82457": 1765148055, @@ -80323,7 +80323,7 @@ "84567": 1746894728, "84568": 1748265563, "84569": 1746877112, - "84570": 1750181288, + "84570": 1770383969, "84571": 1752166593, "84572": 1768731011, "84573": 1746943367, @@ -80608,7 +80608,7 @@ "84863": 1751387076, "84864": 1752068734, "84865": 1748011964, - "84867": 1750525495, + "84867": 1770382699, "84868": 1749156589, "84869": 1748014475, "84870": 1748326620, @@ -81265,7 +81265,7 @@ "85536": 1750608288, "85537": 1767478722, "85538": 1765910274, - "85539": 1769517959, + "85539": 1770386286, "85540": 1750633652, "85541": 1750629832, "85542": 1768666861, @@ -81917,7 +81917,7 @@ "86198": 1753625431, "86199": 1753721893, "86200": 1756201319, - "86201": 1753643053, + "86201": 1770382711, "86202": 1753643223, "86203": 1756041317, "86204": 1754558819, @@ -82258,7 +82258,7 @@ "86546": 1755495978, "86547": 1755491967, "86548": 1756799789, - "86549": 1769738565, + "86549": 1770386260, "86550": 1768752858, "86551": 1755513073, "86552": 1755513106, @@ -82655,7 +82655,7 @@ "86954": 1760947445, "86956": 1757246884, "86957": 1765576391, - "86958": 1770102835, + "86958": 1770386411, "86959": 1757263721, "86960": 1768084646, "86961": 1760804995, @@ -82850,7 +82850,7 @@ "87154": 1757981408, "87155": 1757996181, "87156": 1766798833, - "87157": 1769738465, + "87157": 1770386976, "87158": 1761729663, "87159": 1766675860, "87160": 1758010168, @@ -83011,7 +83011,7 @@ "87317": 1759126381, "87318": 1770352035, "87319": 1761950006, - "87320": 1769738498, + "87320": 1770386217, "87321": 1764720650, "87322": 1758795932, "87323": 1766660961, @@ -83037,7 +83037,7 @@ "87344": 1765138889, "87345": 1758883690, "87346": 1761241280, - "87347": 1770188277, + "87347": 1770386464, "87348": 1763664142, "87349": 1758885260, "87350": 1758885266, @@ -84304,7 +84304,7 @@ "88640": 1768330728, "88641": 1767045759, "88642": 1765734444, - "88643": 1770057169, + "88643": 1770380166, "88644": 1769594243, "88645": 1768640171, "88646": 1764262894, @@ -84901,7 +84901,7 @@ "89247": 1766623135, "89248": 1766624188, "89249": 1770125566, - "89250": 1769965178, + "89250": 1770386346, "89251": 1766671413, "89252": 1766694552, "89253": 1767337981, @@ -85309,7 +85309,7 @@ "89663": 1770376457, "89664": 1767778012, "89665": 1769937659, - "89666": 1770221040, + "89666": 1770386379, "89667": 1768024493, "89668": 1767817646, "89669": 1767797039, @@ -85404,7 +85404,7 @@ "89762": 1769791747, "89763": 1768170314, "89764": 1770289534, - "89765": 1770298563, + "89765": 1770380209, "89766": 1768413494, "89767": 1769473263, "89768": 1769362187, @@ -85599,7 +85599,7 @@ "89963": 1768903757, "89964": 1768903900, "89965": 1768826000, - "89966": 1769020186, + "89966": 1770379684, "89967": 1768750815, "89968": 1768753730, "89969": 1769616921, @@ -85660,7 +85660,7 @@ "90024": 1769186410, "90025": 1769186332, "90026": 1769172583, - "90027": 1770033309, + "90027": 1770380036, "90028": 1770284294, "90029": 1769204606, "90030": 1769187207, @@ -85788,7 +85788,7 @@ "90153": 1770040645, "90154": 1770057517, "90155": 1770044140, - "90156": 1770366182, + "90156": 1770381576, "90157": 1770059065, "90158": 1770058997, "90159": 1770366240, @@ -85822,7 +85822,7 @@ "90187": 1770239660, "90188": 1770251847, "90189": 1770226358, - "90190": 1770378900, + "90190": 1770379946, "90191": 1770372888, "90192": 1770297108, "90193": 1770281201, diff --git a/update_json.py b/update_json.py index 1cf1ae6..1dca40d 100755 --- a/update_json.py +++ b/update_json.py @@ -16,7 +16,7 @@ import json_downloader response = None -def set_apienv(urls, uprocess, dbengine, dbExec, updatesBase, lprint): +def set_apienv(dbengine, dbExec, lprint): """Populate updates table from the API and return available updates. Args: @@ -86,7 +86,6 @@ 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 @@ -108,10 +107,9 @@ def main() -> int: 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 + dbclass, dbengine = dbs.dbclass, dbs.dbengine uprocess = url.process.Loadurl() TVMAZEURLS = url.urls.Tvmazeurls() TVMAZE_URLS = get_tvmaze(TVMAZEURLS) @@ -146,12 +144,9 @@ def main() -> int: if options["apitype"].lower() == "tvmaze": try: new_updates = set_apienv( - TVMAZE_URLS, - uprocess, dbengine, dbExec, - tempTableList, - lprint, + lprint ) except Exception as exc: # noqa: PLC0301 - broad handler to log failures lprint.logprint("error", f"Failed to set API env: {exc}") @@ -337,32 +332,7 @@ def main() -> int: 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) - - - - # 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 + lprint.logprint("info", "JSON download complete.") if __name__ == "__main__":