mirror of
https://gitlab.com/cief-data/data-management.git
synced 2026-08-19 04:13:57 +00:00
111 lines
3.8 KiB
Python
111 lines
3.8 KiB
Python
import os
|
|
from os import walk
|
|
import general_function
|
|
import exchange.extract.exchange_extract_function as extract_func
|
|
import exchange.load.exchange_load_function as load_func
|
|
|
|
def exchange_extract(EXCHANGE_CSV_TEMP_STORAGE_PATH):
|
|
#Config file name
|
|
EX_MYSQL_CONFIG_FILE = "EXCHANGE_CONFIG.json"
|
|
EX_SQL_FILE = "EXCHANGE_EXTRACT_SQL.json"
|
|
EXCHANGE_CSV_TEMP_STORAGE_PATH = EXCHANGE_CSV_TEMP_STORAGE_PATH
|
|
|
|
#Read config file
|
|
EX_MYSQL_CONFIG = general_function.read_json_file("./config/",EX_MYSQL_CONFIG_FILE)
|
|
EX_SQL_DICT = general_function.read_json_file("./config/",EX_SQL_FILE)
|
|
|
|
#Connect to mysql
|
|
EX_MYSQLDB = extract_func.connect_mysql(EX_MYSQL_CONFIG)
|
|
EX_MYSQLCURSOR = EX_MYSQLDB.cursor()
|
|
|
|
#Run EX_SQL_DICT to get all data from mysql in csv format
|
|
extract_func.mysql_dict_sql_to_csv(EX_SQL_DICT, EXCHANGE_CSV_TEMP_STORAGE_PATH, EX_MYSQLDB, EX_MYSQLCURSOR)
|
|
|
|
def exchange_load_to_cloudstorage(EXCHANGE_CSV_TEMP_STORAGE_PATH):
|
|
#Config file name
|
|
CLOUDSTORAGE_CREDENTIAL_FILE = "./config/CLOUD_STORAGE_ADMIN_CREDENTIAL.json"
|
|
|
|
#Connect to cloudstorage
|
|
ex_storage_client = load_func.login_cloudstorage_credential(CLOUDSTORAGE_CREDENTIAL_FILE)
|
|
|
|
#Get the filename and path in temp csv storage
|
|
temp_file_list = []
|
|
for (dirpath, dirnames, filenames) in walk(EXCHANGE_CSV_TEMP_STORAGE_PATH):
|
|
temp_file_list.extend(filenames)
|
|
|
|
#Loop all the file and upload to cloudstorage
|
|
for file_name in temp_file_list:
|
|
file_path = os.path.join(dirpath,file_name)
|
|
load_func.upload_to_bucket(ex_storage_client, file_name, file_path)
|
|
|
|
|
|
|
|
def x1_extract(X1_CSV_TEMP_STORAGE_PATH):
|
|
#Config file name
|
|
X1_MYSQL_CONFIG_FILE = "X1_CONFIG.json"
|
|
X1_SQL_FILE = "X1_EXTRACT_SQL.json"
|
|
X1_CSV_TEMP_STORAGE_PATH = X1_CSV_TEMP_STORAGE_PATH
|
|
|
|
#Read config file
|
|
X1_MYSQL_CONFIG = general_function.read_json_file("./config/",X1_MYSQL_CONFIG_FILE)
|
|
X1_SQL_DICT = general_function.read_json_file("./config/",X1_SQL_FILE)
|
|
|
|
#Connect to mysql
|
|
X1_MYSQLDB = extract_func.connect_mysql(X1_MYSQL_CONFIG)
|
|
X1_MYSQLCURSOR = X1_MYSQLDB.cursor()
|
|
|
|
#Run EX_SQL_DICT to get all data from mysql in csv format
|
|
extract_func.mysql_dict_sql_to_csv(X1_SQL_DICT, X1_CSV_TEMP_STORAGE_PATH, X1_MYSQLDB, X1_MYSQLCURSOR)
|
|
|
|
|
|
def x1_load_to_cloudstorage(X1_CSV_TEMP_STORAGE_PATH):
|
|
#Config file name
|
|
CLOUDSTORAGE_CREDENTIAL_FILE = "./config/CLOUD_STORAGE_ADMIN_CREDENTIAL.json"
|
|
|
|
#Connect to cloudstorage
|
|
x1_storage_client = load_func.login_cloudstorage_credential(CLOUDSTORAGE_CREDENTIAL_FILE)
|
|
|
|
#Get the filename and path in temp csv storage
|
|
temp_file_list = []
|
|
for (dirpath, dirnames, filenames) in walk(X1_CSV_TEMP_STORAGE_PATH):
|
|
temp_file_list.extend(filenames)
|
|
|
|
#Loop all the file and upload to cloudstorage
|
|
for file_name in temp_file_list:
|
|
file_path = os.path.join(dirpath,file_name)
|
|
load_func.upload_to_bucket(x1_storage_client, file_name, file_path)
|
|
|
|
|
|
|
|
|
|
def main():
|
|
#Make sure cron in the file directory
|
|
general_function.change_to_current_directory()
|
|
|
|
#Extract Load Exchange Data
|
|
EXCHANGE_CSV_TEMP_STORAGE_PATH = "./exchange/extract/exchange_temp_extract_csv/"
|
|
exchange_extract(EXCHANGE_CSV_TEMP_STORAGE_PATH)
|
|
exchange_load_to_cloudstorage(EXCHANGE_CSV_TEMP_STORAGE_PATH)
|
|
|
|
#Delete Temp CSV
|
|
general_function.delete_csv_in_path(EXCHANGE_CSV_TEMP_STORAGE_PATH)
|
|
|
|
|
|
#Make sure cron in the file directory
|
|
general_function.change_to_current_directory()
|
|
|
|
#Extract Load X1 Data
|
|
X1_CSV_TEMP_STORAGE_PATH = "./exchange/extract/exchange_temp_extract_csv/"
|
|
x1_extract(X1_CSV_TEMP_STORAGE_PATH)
|
|
x1_load_to_cloudstorage(X1_CSV_TEMP_STORAGE_PATH)
|
|
|
|
#Delete Temp CSV
|
|
general_function.delete_csv_in_path(X1_CSV_TEMP_STORAGE_PATH)
|
|
|
|
|
|
|
|
|
|
|
|
if True:
|
|
main()
|