mirror of
https://gitlab.com/cief-data/data-management.git
synced 2026-08-20 12:54:05 +00:00
Yam ZhengLim
91 lines
2.0 KiB
Python
91 lines
2.0 KiB
Python
from ntpath import join
|
|
from operator import concat
|
|
import pandas as pd
|
|
import mysql.connector
|
|
import json
|
|
import os
|
|
import csv
|
|
|
|
def change_directory_to(path):
|
|
try:
|
|
current_directory = os.getcwd()
|
|
target_dir = concat(current_directory,path)
|
|
return os.chdir(target_dir)
|
|
|
|
except Exception as e:
|
|
print(e)
|
|
print("[FUNCTION_ERROR]-change_directory_to")
|
|
|
|
|
|
|
|
def read_json_file(name):
|
|
try:
|
|
with open(name) as json_file:
|
|
output = json.load(json_file)
|
|
return output
|
|
except Exception as e:
|
|
print(e)
|
|
print("[FUNCTION_ERROR]-read_json_file:" , name)
|
|
|
|
|
|
|
|
def connect_mysql(config):
|
|
mydb = mysql.connector.connect(
|
|
host = config["host"],
|
|
user = config["user"],
|
|
password = config["password"],
|
|
database = config["database"])
|
|
return mydb
|
|
|
|
|
|
|
|
def write_to_csv(path,dbname,tablename,cursor,data):
|
|
try:
|
|
filename = '/{}_{}.csv'.format(dbname, tablename)
|
|
|
|
headers = [col[0] for col in cursor.description] # get headers
|
|
data.insert(0, tuple(headers))
|
|
fp = open(concat(path,filename), 'w', newline = '')
|
|
myFile = csv.writer(fp)
|
|
myFile.writerows(data)
|
|
fp.close()
|
|
|
|
except Exception as e:
|
|
print(e)
|
|
print("[FUNCTION_ERROR]-write_to_csv:" , filename)
|
|
|
|
|
|
#Get all the Compulsory File
|
|
CURRENT_DIRECTORY_PATH = "\\python_el\\extract\\exchange_extract"
|
|
MYSQL_CONFIG_FILE = "EXCHANGE_CONFIG.json"
|
|
SQL_FILE = "EXCHANGE_EXTRACT_SQL.json"
|
|
CSV_STORAGE_PATH = "exchange_temp_extract_csv/"
|
|
|
|
#Change directory & open all file
|
|
change_directory_to(CURRENT_DIRECTORY_PATH)
|
|
MYSQL_CONFIG = read_json_file(MYSQL_CONFIG_FILE)
|
|
SQL_DICT = read_json_file(SQL_FILE)
|
|
|
|
|
|
#Connect DB
|
|
mydb = connect_mysql(MYSQL_CONFIG)
|
|
mycursor = mydb.cursor()
|
|
|
|
|
|
for tablename,sql in SQL_DICT.items():
|
|
|
|
try:
|
|
mycursor.execute(sql)
|
|
myresult = mycursor.fetchall()
|
|
dbname = mydb.database
|
|
write_to_csv(CSV_STORAGE_PATH,dbname,tablename,mycursor,myresult)
|
|
except Exception as e:
|
|
print(e)
|
|
print("[ERROR]-SQL ERROR",tablename,sql)
|
|
|
|
|
|
mydb.close()
|
|
|
|
|
|
|