mirror of
https://gitlab.com/cief-data/data-management.git
synced 2026-08-19 04:13:57 +00:00
Add shipping ExtractLoad
This commit is contained in:
Yam ZhengLim
@@ -17,7 +17,7 @@
|
||||
"documents" : "SELECT * FROM documents;",
|
||||
"employees" : "SELECT * FROM employees;",
|
||||
"failed_jobs" : "SELECT * FROM failed_jobs;",
|
||||
"files" : "SELECT id, document_id, REPLACE(REPLACE(REPLACE(file, '\n', ''),'|', ''),'\\\\','') AS 'file', file_type, deleted_at, created_at, updated_at FROM files;",
|
||||
"files" : "SELECT id, document_id, REPLACE(REPLACE(REPLACE(file, '\n', '<br>'),'|', '|'),'\\\\','\\') AS 'file', file_type, deleted_at, created_at, updated_at FROM files;",
|
||||
"groups" : "SELECT * FROM groups;",
|
||||
"group_transactions" : "SELECT * FROM group_transactions;",
|
||||
"jobs" : "SELECT * FROM jobs;",
|
||||
@@ -34,7 +34,7 @@
|
||||
"service_types" : "SELECT * FROM service_types;",
|
||||
"states" : "SELECT * FROM states;",
|
||||
"transactions" : "SELECT * FROM transactions;",
|
||||
"transaction_detail" : "SELECT id, transaction_id, REPLACE(REPLACE(REPLACE(product_code, '\n', ''),'|', ''),'\\\\','') AS product_code, REPLACE(REPLACE(REPLACE(product_name, '\n', ''),'|', ''),'\\\\','') AS product_name, quantity, price, amount, deleted_at, created_at, updated_at FROM `transaction_detail`;",
|
||||
"transaction_detail" : "SELECT id, transaction_id, REPLACE(REPLACE(REPLACE(product_code, '\n', '<br>'),'|', '|'),'\\\\','\\') AS product_code, REPLACE(REPLACE(REPLACE(product_name, '\n', '<br>'),'|', '|'),'\\\\','\\') AS product_name, quantity, price, amount, deleted_at, created_at, updated_at FROM `transaction_detail`;",
|
||||
"users" : "SELECT id, name, email, type, status, remember_token, active_at, deleted_at, created_at, updated_at FROM users;",
|
||||
"user_email_verifications" : "SELECT * FROM user_email_verifications;",
|
||||
"wallets" : "SELECT * FROM wallets;"
|
||||
|
||||
@@ -16,7 +16,7 @@
|
||||
"districts" : "SELECT * FROM districts;",
|
||||
"documents" : "SELECT * FROM documents;",
|
||||
"failed_jobs" : "SELECT * FROM failed_jobs;",
|
||||
"files" : "SELECT id, document_id, REPLACE(REPLACE(REPLACE(file, '\n', ''),'|', ''),'\\\\','') AS 'file', file_type, deleted_at, created_at, updated_at FROM files;",
|
||||
"files" : "SELECT id, document_id, REPLACE(REPLACE(REPLACE(file, '\n', '<br>'),'|', '|'),'\\\\','\\') AS 'file', file_type, deleted_at, created_at, updated_at FROM files;",
|
||||
"jobs" : "SELECT * FROM jobs;",
|
||||
"migrations" : "SELECT * FROM migrations;",
|
||||
"orders" : "SELECT * FROM orders",
|
||||
@@ -24,7 +24,7 @@
|
||||
"packages" : "SELECT * FROM packages",
|
||||
"packing_lists" : "SELECT * FROM packing_lists",
|
||||
"password_resets" : "SELECT * FROM password_resets;",
|
||||
"remarks" : "SELECT id, owner_type, owner_id, commenter_id, REPLACE(REPLACE(REPLACE(content, '\n', ''),'|', ''),'\\\\','') AS 'content',, deleted_at, created_at, updated_at FROM remarks",
|
||||
"remarks" : "SELECT id, owner_type, owner_id, commenter_id, REPLACE(REPLACE(REPLACE(content, '\n', '<br>'),'|', '|'),'\\\\','\\') AS 'content',, deleted_at, created_at, updated_at FROM remarks",
|
||||
"schedules" : "SELECT * FROM schedules",
|
||||
"segments" : "SELECT * FROM segments;",
|
||||
"segment_constants" : "SELECT * FROM segment_constants;",
|
||||
|
||||
@@ -1,8 +1,10 @@
|
||||
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
|
||||
import exchange.extract.exchange_extract_function as exchange_extract_func
|
||||
import exchange.load.exchange_load_function as exchange_load_func
|
||||
import shipping.extract.shipping_extract_function as shipping_extract_func
|
||||
import shipping.load.shipping_load_function as shipping_load_func
|
||||
|
||||
def exchange_extract(EXCHANGE_CSV_TEMP_STORAGE_PATH):
|
||||
#Config file name
|
||||
@@ -15,18 +17,18 @@ def exchange_extract(EXCHANGE_CSV_TEMP_STORAGE_PATH):
|
||||
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_MYSQLDB = exchange_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)
|
||||
exchange_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)
|
||||
ex_storage_client = exchange_load_func.login_cloudstorage_credential(CLOUDSTORAGE_CREDENTIAL_FILE)
|
||||
|
||||
#Get the filename and path in temp csv storage
|
||||
temp_file_list = []
|
||||
@@ -36,7 +38,7 @@ def exchange_load_to_cloudstorage(EXCHANGE_CSV_TEMP_STORAGE_PATH):
|
||||
#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)
|
||||
exchange_load_func.upload_to_bucket(ex_storage_client, file_name, file_path)
|
||||
|
||||
|
||||
|
||||
@@ -51,11 +53,11 @@ def shipping_extract(SHIPPING_CSV_TEMP_STORAGE_PATH):
|
||||
SHIPPING_SQL_DICT = general_function.read_json_file("./config/",SHIPPING_SQL_FILE)
|
||||
|
||||
#Connect to mysql
|
||||
SHIPPING_MYSQLDB = extract_func.connect_mysql(SHIPPING_MYSQL_CONFIG)
|
||||
SHIPPING_MYSQLDB = shipping_extract_func.connect_mysql(SHIPPING_MYSQL_CONFIG)
|
||||
SHIPPING_MYSQLCURSOR = SHIPPING_MYSQLDB.cursor()
|
||||
|
||||
#Run EX_SQL_DICT to get all data from mysql in csv format
|
||||
extract_func.mysql_dict_sql_to_csv(SHIPPING_SQL_DICT, SHIPPING_CSV_TEMP_STORAGE_PATH, SHIPPING_MYSQLDB, SHIPPING_MYSQLCURSOR)
|
||||
shipping_extract_func.mysql_dict_sql_to_csv(SHIPPING_SQL_DICT, SHIPPING_CSV_TEMP_STORAGE_PATH, SHIPPING_MYSQLDB, SHIPPING_MYSQLCURSOR)
|
||||
|
||||
|
||||
def shipping_load_to_cloudstorage(SHIPPING_CSV_TEMP_STORAGE_PATH):
|
||||
@@ -63,7 +65,7 @@ def shipping_load_to_cloudstorage(SHIPPING_CSV_TEMP_STORAGE_PATH):
|
||||
CLOUDSTORAGE_CREDENTIAL_FILE = "./config/CLOUD_STORAGE_ADMIN_CREDENTIAL.json"
|
||||
|
||||
#Connect to cloudstorage
|
||||
shipping_storage_client = load_func.login_cloudstorage_credential(CLOUDSTORAGE_CREDENTIAL_FILE)
|
||||
shipping_storage_client = shipping_load_func.login_cloudstorage_credential(CLOUDSTORAGE_CREDENTIAL_FILE)
|
||||
|
||||
#Get the filename and path in temp csv storage
|
||||
temp_file_list = []
|
||||
@@ -73,7 +75,7 @@ def shipping_load_to_cloudstorage(SHIPPING_CSV_TEMP_STORAGE_PATH):
|
||||
#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(shipping_storage_client, file_name, file_path)
|
||||
shipping_load_func.upload_to_bucket(shipping_storage_client, file_name, file_path)
|
||||
|
||||
|
||||
|
||||
@@ -94,7 +96,7 @@ def main():
|
||||
general_function.change_to_current_directory()
|
||||
|
||||
#Extract Load SHIPPING Data
|
||||
SHIPPING_CSV_TEMP_STORAGE_PATH = "./exchange/extract/exchange_temp_extract_csv/"
|
||||
SHIPPING_CSV_TEMP_STORAGE_PATH = "./shipping/extract/shipping_temp_extract_csv/"
|
||||
shipping_extract(SHIPPING_CSV_TEMP_STORAGE_PATH)
|
||||
shipping_load_to_cloudstorage(SHIPPING_CSV_TEMP_STORAGE_PATH)
|
||||
|
||||
|
||||
Binary file not shown.
@@ -0,0 +1,52 @@
|
||||
import csv
|
||||
from operator import concat
|
||||
import mysql.connector
|
||||
import os
|
||||
|
||||
|
||||
def connect_mysql(config):
|
||||
try:
|
||||
mydb = mysql.connector.connect(
|
||||
host = config["host"],
|
||||
user = config["user"],
|
||||
password = config["password"],
|
||||
database = config["database"])
|
||||
return mydb
|
||||
except Exception as e:
|
||||
print(e)
|
||||
print("[FUNCTION_ERROR]-connect_mysql")
|
||||
return False
|
||||
|
||||
|
||||
def mysql_dict_sql_to_csv(SQL_DICT, path, mysqldb, mysqlcursor):
|
||||
temp_folder_path = path
|
||||
# Check whether the specified path exists or not
|
||||
temp_folder_path_isExist = os.path.exists(temp_folder_path)
|
||||
if not temp_folder_path_isExist:
|
||||
# Create a new directory because it does not exist
|
||||
os.makedirs(temp_folder_path)
|
||||
print("[Create Folder] shipping_temp_extract_csv")
|
||||
|
||||
for tablename,sql in SQL_DICT.items():
|
||||
|
||||
try:
|
||||
mysqlcursor.execute(sql)
|
||||
mysqlresult = mysqlcursor.fetchall()
|
||||
mysqldbname = mysqldb.database
|
||||
|
||||
filename = '/{}_{}.csv'.format(mysqldbname, tablename)
|
||||
headers = [col[0] for col in mysqlcursor.description] # get headers
|
||||
mysqlresult.insert(0, tuple(headers))
|
||||
fp = open(concat(path,filename), 'w', newline = '', encoding = 'utf-8-sig')
|
||||
myFile = csv.writer(fp,delimiter='|',quotechar='"', quoting=csv.QUOTE_MINIMAL)
|
||||
myFile.writerows(mysqlresult)
|
||||
fp.close()
|
||||
|
||||
except Exception as e:
|
||||
print(e)
|
||||
print("[FUNCTION_ERROR]-write_to_csv:" , filename)
|
||||
return False
|
||||
|
||||
finally:
|
||||
print("[Download Complete]",filename)
|
||||
|
||||
Binary file not shown.
@@ -0,0 +1,21 @@
|
||||
from google.cloud import storage
|
||||
|
||||
def login_cloudstorage_credential(credential):
|
||||
try:
|
||||
storage_client = storage.Client.from_service_account_json(credential)
|
||||
return storage_client
|
||||
except Exception as e:
|
||||
print(e)
|
||||
print("[FUNCTION_ERROR]-login_cloudstorage_credential")
|
||||
return False
|
||||
|
||||
def upload_to_bucket(storage_client, blob_name, file_path):
|
||||
try:
|
||||
exchange_bucket = storage_client.get_bucket('shipping_bucket_raw')
|
||||
blob = exchange_bucket.blob(blob_name)
|
||||
blob.upload_from_filename(file_path)
|
||||
return print("[Upload Complete]",file_path)
|
||||
except Exception as e:
|
||||
print(e)
|
||||
print("[FUNCTION_ERROR]-upload_to_bucket",blob_name)
|
||||
return False
|
||||
Reference in New Issue
Block a user