diff --git a/README.md b/README.md index fa545b3..faec563 100644 --- a/README.md +++ b/README.md @@ -1,8 +1,162 @@ -# DBSyncer +# DBSyncer - MongoDB to PostgreSQL Data Synchronization -This project has two parts: -1.) it uses the tvmaze API to download Television dat and store it in a mongodb database. -2.) Takes saved json data and inserts/updates data in a postgresql database. +A comprehensive data synchronization tool that: +1. Downloads Television data from the TVMaze API and stores it in MongoDB +2. Automatically generates PostgreSQL table schemas from MongoDB document structures +3. Efficiently inserts/updates MongoDB documents into PostgreSQL with intelligent type conversion +## Features -[Learn more about creating GitLab projects.](https://docs.gitlab.com/ee/gitlab-basics/create-project.html) +### Schema Generation (`db/schema_generator.py`) +- **Automatic Schema Inference**: Analyzes MongoDB documents to infer PostgreSQL types +- **Type Mapping**: Intelligent conversion between MongoDB and PostgreSQL types: + - Objects/Arrays → JSONB + - Dates → TIMESTAMP + - Numbers → BIGINT / DOUBLE PRECISION + - Strings → TEXT + +### Document Insertion +- **Batch Processing**: Efficiently inserts documents in configurable batches (default: 1000) +- **Type Conversion**: Automatic type conversion for: + - Epoch timestamps ↔ ISO datetime strings + - JSON strings ↔ Python dict/list + - ObjectId → Text +- **UPSERT Support**: Uses PostgreSQL `ON CONFLICT ... DO UPDATE SET` for automatic upserts +- **Error Handling**: Detailed logging with per-row error reporting + +## Quick Start + +### Prerequisites +```bash +pip install -r requirements.txt +``` + +### Configuration +Edit `settings/tvsync_settings.cfg`: +```ini +[dbsettings] +hostname = localhost +pgsqlport = 5432 +pgsqlusername = postgres +pgsqlpassword = your_password +dbname = media_dbsync +mghostname = localhost +mgport = 27017 +mgdbname = media + +[database] +dbtype = pgsql +updateschema = updates +``` + +### Main Scripts + +#### `update_mongodb.py` +Fetches TV data from TVMaze API and stores in MongoDB: +```bash +python3 update_mongodb.py +``` + +#### `mongodb2postgres.py` +Generates PostgreSQL schemas from MongoDB collections and inserts documents: +```bash +python3 mongodb2postgres.py +``` + +This script will: +1. Analyze each MongoDB collection (series, episodes, actors, characters, crew) +2. Automatically create corresponding PostgreSQL tables in the `updates` schema +3. Insert/upsert all documents with proper type conversions +4. Report progress and any errors + +## Usage Examples + +### Generate Table Schema from MongoDB Collection +```python +from db.schema_generator import MongoToPostgresSchemaGenerator + +generator = MongoToPostgresSchemaGenerator(sample_size=100) +schema = generator.analyze_collection(mongo_db, 'series') + +# View the SQL +sql = generator.generate_create_table_sql('seriesdata', schema_name='dbo', pk_field='id') +print(sql) + +# Create the table +generator.create_table_in_postgres(engine, 'seriesdata', schema_name='dbo', pk_field='id') +``` + +### Insert Documents with Type Conversion +```python +from db.schema_generator import MongoDocumentInserter + +inserter = MongoDocumentInserter(batch_size=1000) + +# From a collection +count = inserter.insert_from_collection( + engine=engine, + table_name='seriesdata', + collection=mongo_db['series'], + schema_name='dbo', + on_conflict="ON CONFLICT (id) DO UPDATE SET name=EXCLUDED.name, updated=EXCLUDED.updated", +) +print(f"Inserted {count} documents") +``` + +## Architecture + +### Database Classes (`db/functions.py`) +- `settype`: Manages database engine creation and API configuration +- `dbmongo`: Handles MongoDB connections and operations +- `Dbexec`: Executes raw SQL queries and batch updates + +### Schema Generator (`db/schema_generator.py`) +- `MongoToPostgresSchemaGenerator`: Infers and creates table schemas +- `MongoDocumentInserter`: Handles document insertion with type conversion + +### API Integration (`api/`) +- TVMaze API client for fetching show, episode, cast, and crew data +- TheTVDB API client (optional) + +## Performance Optimizations + +- **Batch Inserts**: Uses SQLAlchemy's parameterized queries for efficient batch operations +- **Transactional Inserts**: Groups inserts in transactions to reduce database overhead +- **Column Type Awareness**: Converts data to match target column types, avoiding casting overhead +- **Connection Pooling**: Reuses database connections (pool_size=20, max_overflow=20) +- **UPSERT Operations**: Uses PostgreSQL's `ON CONFLICT` for atomic insert-or-update + +## Troubleshooting + +### MetaData Schema Argument Error +If you see `sqlalchemy.exc.ArgumentError: Could not parse SQLAlchemy URL`, ensure: +- Use `MetaData(schema="updates")` instead of `MetaData("updates")` +- Pass full connection strings to `create_engine()`, not schema names + +### Type Conversion Errors +The inserter automatically handles: +- Datetime strings → epoch timestamps (for BIGINT columns) +- Epoch ints → datetime objects (for TIMESTAMP columns) +- JSON strings → Python dicts (for JSONB columns) + +If type errors persist, check that: +1. Table schema matches MongoDB document fields +2. Column types are correctly inferred from sample documents +3. Use appropriate `on_conflict` clauses for upserts + +### Column Name Case Sensitivity +PostgreSQL stores unquoted identifiers as lowercase. The inserter automatically: +- Lowercases column names in `ON CONFLICT` clauses +- Uses case-insensitive matching when mapping MongoDB fields to PostgreSQL columns + +## Contributing + +When modifying: +- **Schema inference**: Update `MongoToPostgresSchemaGenerator.TYPE_MAPPING` for new type support +- **Type conversion**: Extend `MongoDocumentInserter.convert_value()` for custom conversions +- **Database operations**: Use parameterized queries via `text()` to prevent SQL injection +- **Error handling**: Add specific exception types to improve debugging + +## License + +See LICENSE file for details. diff --git a/mongo2pgsql.py b/mongo2pgsql.py deleted file mode 100644 index 368fea1..0000000 --- a/mongo2pgsql.py +++ /dev/null @@ -1,205 +0,0 @@ -#!/usr/bin/env python3 -"""Convert a MongoDB collection into a PostgreSQL table. - -Contains `convert_mongo_to_pg(mongo_uri, mongo_db, collection_name, pg_dsn, pg_table=None, ...)` -and a small CLI. - -Notes: -- Uses `pymongo` to read documents and `psycopg2` to write to Postgres. -- Infers basic column types from a sample of documents; nested objects/arrays - are stored as `JSONB`. -""" - -from __future__ import annotations - -import argparse -import json -import logging -from typing import Any, Dict, Iterable, List, Optional - -import pymongo -import psycopg2 -import psycopg2.extras as pgextras -import psycopg2.sql as sql - -LOG = logging.getLogger("mongo2pgsql") - - -def _detect_type(value: Any) -> str: - 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 "real" - if isinstance(value, (dict, list)): - return "jsonb" - return "text" - - -def _merge_types(types: List[str]) -> str: - # priority: jsonb > text > real > integer > boolean > null - s = set(types) - if "jsonb" in s: - return "jsonb" - if "text" in s: - return "text" - if "real" in s: - return "real" - if "integer" in s: - return "integer" - if "boolean" in s: - return "boolean" - return "text" - - -def _pg_type_for(kind: str) -> str: - return { - "integer": "BIGINT", - "real": "DOUBLE PRECISION", - "boolean": "BOOLEAN", - "jsonb": "JSONB", - "text": "TEXT", - }.get(kind, "TEXT") - - -def convert_mongo_to_pg( - mongo_uri: str, - mongo_db: str, - collection_name: str, - pg_dsn: str, - pg_table: Optional[str] = None, - sample_size: int = 500, - batch_size: int = 1000, - create_table: bool = True, - pk_field: str = "_id", -) -> None: - """Copy a MongoDB collection to PostgreSQL. - - - `pg_table` defaults to `collection_name` if omitted. - - Nested documents/arrays are stored as JSONB. - - `_id` is stored as TEXT by default. - """ - if pg_table is None: - pg_table = collection_name - - client = pymongo.MongoClient(mongo_uri) - coll = client[mongo_db][collection_name] - - sample = list(coll.find({}, projection=None, limit=sample_size)) - if not sample: - raise ValueError("collection is empty or not found") - - # gather all field names and observed types - field_types: Dict[str, List[str]] = {} - for doc in sample: - for k, v in doc.items(): - t = _detect_type(v) - field_types.setdefault(k, []).append(t) - - # finalize types - final_types: Dict[str, str] = {} - for k, types in field_types.items(): - merged = _merge_types(types) - final_types[k] = _pg_type_for(merged) - - # ensure primary key present - if pk_field not in final_types: - final_types[pk_field] = "TEXT" - - cols = list(final_types.items()) - - # create table - if create_table: - # Build CREATE TABLE statement using psycopg2.sql for safe identifiers - col_defs = [] - for name, ptype in cols: - col_defs.append(sql.SQL("{} {}").format(sql.Identifier(name), sql.SQL(ptype))) - - create_body = sql.SQL(", ").join(col_defs) - if pk_field in final_types: - pk_clause = sql.SQL(", PRIMARY KEY ({})").format(sql.SQL(",").join(map(sql.Identifier, [pk_field]))) - create_body = create_body + pk_clause - - create_stmt = sql.SQL("CREATE TABLE IF NOT EXISTS {} ({})").format(sql.Identifier(pg_table), create_body) - with psycopg2.connect(pg_dsn) as pgconn: - with pgconn.cursor() as cur: - LOG.info("Creating table %s", pg_table) - cur.execute(create_stmt) - pgconn.commit() - - # stream documents and insert - insert_cols = [n for n, _ in cols] - - with psycopg2.connect(pg_dsn) as pgconn: - with pgconn.cursor() as cur: - # build INSERT statement for execute_values: it expects a single %s - base_insert = sql.SQL("INSERT INTO {} ({}) VALUES %s").format( - sql.Identifier(pg_table), - sql.SQL(', ').join(map(sql.Identifier, insert_cols)), - ).as_string(pgconn) - batch = [] - # Use an explicit session when using no_cursor_timeout to avoid pymongo warning - with client.start_session() as session: - cursor = coll.find({}, no_cursor_timeout=True, session=session).batch_size(batch_size) - try: - for doc in cursor: - row = [] - for col_name in insert_cols: - v = doc.get(col_name) - if v is None: - row.append(None) - continue - # convert ObjectId and other non-JSON types - if isinstance(v, (dict, list)): - row.append(pgextras.Json(v)) - else: - row.append(str(v) if col_name == pk_field else v) - batch.append(tuple(row)) - if len(batch) >= batch_size: - pgextras.execute_values(cur, base_insert, batch, template=None, page_size=batch_size) - pgconn.commit() - batch = [] - finally: - try: - cursor.close() - except Exception: - pass - if batch: - pgextras.execute_values(cur, base_insert, batch, template=None, page_size=len(batch)) - pgconn.commit() - - LOG.info("Finished copying collection %s.%s to %s", mongo_db, collection_name, pg_table) - - -def _cli() -> None: - parser = argparse.ArgumentParser(description="Copy a MongoDB collection to PostgreSQL") - parser.add_argument("mongo_uri", help="MongoDB connection URI, e.g. mongodb://user:pw@host:27017") - parser.add_argument("mongo_db", help="MongoDB database name") - parser.add_argument("collection", help="Collection name to copy") - parser.add_argument("pg_dsn", help="Postgres DSN, e.g. \"host=.. dbname=.. user=.. password=..\"") - parser.add_argument("--pg-table", help="Target Postgres table name (defaults to collection name)") - parser.add_argument("--sample", type=int, default=500, help="Sample size to infer schema") - parser.add_argument("--batch", type=int, default=1000, help="Insert batch size") - parser.add_argument("--no-create", dest="create", action="store_false", help="Don't attempt to create table in PG") - parser.add_argument("--verbose", "-v", action="store_true") - args = parser.parse_args() - - logging.basicConfig(level=logging.DEBUG if args.verbose else logging.INFO) - - convert_mongo_to_pg( - args.mongo_uri, - args.mongo_db, - args.collection, - args.pg_dsn, - pg_table=args.pg_table, - sample_size=args.sample, - batch_size=args.batch, - create_table=args.create, - ) - - -if __name__ == "__main__": - _cli() diff --git a/search mongodb.py b/search mongodb.py deleted file mode 100755 index 00dbd27..0000000 --- a/search mongodb.py +++ /dev/null @@ -1,266 +0,0 @@ -#!/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()) -