Untitled
unknown
plain_text
2 years ago
2.0 kB
5
Indexable
import argparse import pandas as pd import mysql.connector import json from base64 import b64decode as base_b64decode import logging from pandas import read_sql as pd_read_sql import sys from sqlalchemy import create_engine def connect_to_database(json_data, replicationTarget): # ... (existing code for connecting to database) def write_sql(query_info): # ... (existing code for writing SQL query to the log file) def main(args): releaseId = args.release_id opId = args.op_id buId = args.bu_id replicationTarget = args.replication_target source = args.source replicationJobId = args.replication_job_id json_file_path = args.json_file_path sql_log_file = args.sql_log_file log_file = args.log_file # Set up logging logging.basicConfig( filename=log_file, level=logging.INFO, format='%(asctime)s - %(message)s', datefmt='%Y-%m-%d %H:%M:%S' ) # Open SQL log file for writing sql_log_file = open(sql_log_file, "w") # ... (rest of the code within the main function) if __name__ == '__main__': parser = argparse.ArgumentParser(description='PPM Product Catalog Replication Script') parser.add_argument('--release-id', required=True, help='Release ID') parser.add_argument('--op-id', required=True, help='Operation ID') parser.add_argument('--bu-id', required=True, help='Business Unit ID') parser.add_argument('--replication-target', required=True, help='Replication Target') parser.add_argument('--source', required=True, help='Source') parser.add_argument('--replication-job-id', required=True, help='Replication Job ID') parser.add_argument('--json-file-path', required=True, help='Path to JSON config file') parser.add_argument('--sql-log-file', required=True, help='Path to SQL log file') parser.add_argument('--log-file', required=True, help='Path to log file') args = parser.parse_args() main(args)
Editor is loading...