mirror of
https://gitlab.com/cief-data/data-management.git
synced 2026-08-19 04:13:57 +00:00
Yam ZhengLim
60 lines
2.2 KiB
Python
60 lines
2.2 KiB
Python
import os
|
|
from os import walk
|
|
import general_function
|
|
import exchange.extract.exchange_extract_function as ex_extract_func
|
|
import exchange.load.exchange_load_function as ex_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 = ex_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
|
|
ex_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 = ex_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)
|
|
ex_load_func.upload_to_bucket(ex_storage_client, file_name, file_path)
|
|
|
|
def linux_change_cron_directory():
|
|
script_dir = os.path.dirname(os.path.realpath(__file__))
|
|
print (os.getcwd())
|
|
os.chdir(script_dir)
|
|
|
|
def main():
|
|
#Make sure cron in the file directory
|
|
linux_change_cron_directory()
|
|
|
|
|
|
#Extract Load Exchange Data, and delete the temp csv
|
|
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)
|
|
general_function.delete_csv_in_path(EXCHANGE_CSV_TEMP_STORAGE_PATH)
|
|
|
|
|
|
if True:
|
|
main()
|