mirror of
https://gitlab.com/cief-data/data-management.git
synced 2026-08-19 04:13:57 +00:00
Yam ZhengLim
218 lines
9.3 KiB
Python
218 lines
9.3 KiB
Python
import os
|
|
from os import walk
|
|
import pandas as pd
|
|
import general_function
|
|
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
|
|
import crisp.extract.crisp_extract_function as crisp_extract_func
|
|
import crisp.load.crisp_load_function as crisp_load_func
|
|
import c2m.scrape.c2m_scrape_app as c2m_scrape_app
|
|
import c2m.load.c2m_load_function as c2m_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 = 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
|
|
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 = exchange_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)
|
|
exchange_load_func.upload_to_bucket(ex_storage_client, file_name, file_path)
|
|
|
|
def shipping_extract(SHIPPING_CSV_TEMP_STORAGE_PATH):
|
|
#Config file name
|
|
SHIPPING_MYSQL_CONFIG_FILE = "SHIPPING_CONFIG.json"
|
|
SHIPPING_SQL_FILE = "SHIPPING_EXTRACT_SQL.json"
|
|
SHIPPING_CSV_TEMP_STORAGE_PATH = SHIPPING_CSV_TEMP_STORAGE_PATH
|
|
|
|
#Read config file
|
|
SHIPPING_MYSQL_CONFIG = general_function.read_json_file("./config/",SHIPPING_MYSQL_CONFIG_FILE)
|
|
SHIPPING_SQL_DICT = general_function.read_json_file("./config/",SHIPPING_SQL_FILE)
|
|
|
|
#Connect to mysql
|
|
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
|
|
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):
|
|
#Config file name
|
|
CLOUDSTORAGE_CREDENTIAL_FILE = "./config/CLOUD_STORAGE_ADMIN_CREDENTIAL.json"
|
|
|
|
#Connect to cloudstorage
|
|
shipping_storage_client = shipping_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(SHIPPING_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)
|
|
shipping_load_func.upload_to_bucket(shipping_storage_client, file_name, file_path)
|
|
|
|
def crisp_extract(CRISP_CSV_TEMP_STORAGE_PATH):
|
|
#Config file name
|
|
CRISP_API_CONFIG_FILE = 'CRISP_CONFIG.json'
|
|
CRISP_CSV_TEMP_STORAGE_PATH = CRISP_CSV_TEMP_STORAGE_PATH
|
|
|
|
#Read config file
|
|
CRISP_API_CONFIG = general_function.read_json_file('./config/',CRISP_API_CONFIG_FILE)
|
|
|
|
#Get website_id
|
|
CRISP_WEBSITE_ID = CRISP_API_CONFIG['website_id']
|
|
|
|
#Connect to CRISP
|
|
CLIENT = crisp_extract_func.connect_crisp(CRISP_API_CONFIG)
|
|
|
|
#Get operators csv
|
|
crisp_extract_func.list_website_operators_json(CRISP_WEBSITE_ID, CLIENT, CRISP_CSV_TEMP_STORAGE_PATH, 'crisp_list_website_operators.json')
|
|
|
|
#Get users(people) csv
|
|
crisp_extract_func.list_people_profiles_json(CRISP_WEBSITE_ID, CLIENT, CRISP_CSV_TEMP_STORAGE_PATH, 'crisp_list_people_profiles.json')
|
|
|
|
#Get peoples manually marking by operator
|
|
crisp_extract_func.get_people_data_json(CRISP_WEBSITE_ID, CLIENT, CRISP_CSV_TEMP_STORAGE_PATH, 'crisp_get_people_data.json', './crisp/extract/crisp_temp_extract_csv/crisp_list_people_profiles.json')
|
|
|
|
#Get peoples subscription status
|
|
crisp_extract_func.get_people_subscription_status_json(CRISP_WEBSITE_ID, CLIENT, CRISP_CSV_TEMP_STORAGE_PATH, 'crisp_get_people_subscription_status.json', './crisp/extract/crisp_temp_extract_csv/crisp_list_people_profiles.json')
|
|
|
|
#Get current converstation list
|
|
crisp_extract_func.list_conversations_json(CRISP_WEBSITE_ID, CLIENT, CRISP_CSV_TEMP_STORAGE_PATH, 'crisp_list_conversations.json')
|
|
|
|
#Get message in conversations
|
|
crisp_extract_func.get_message_in_conversations_json(CRISP_WEBSITE_ID, CLIENT, CRISP_CSV_TEMP_STORAGE_PATH, 'crisp_get_message_in_conversations.json', './crisp/extract/crisp_temp_extract_csv/crisp_list_conversations.json')
|
|
|
|
def crisp_load_to_cloudstorage(CRISP_CSV_TEMP_STORAGE_PATH):
|
|
#Config file name
|
|
CLOUDSTORAGE_CREDENTIAL_FILE = "./config/CLOUD_STORAGE_ADMIN_CREDENTIAL.json"
|
|
|
|
#Connect to cloudstorage
|
|
cirsp_storage_client = crisp_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(CRISP_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)
|
|
crisp_load_func.upload_to_bucket(cirsp_storage_client, file_name, file_path)
|
|
|
|
def c2m_scrape(C2M_FILES_TEMP_STORAGE_PATH):
|
|
C2M_CONFIG_FILE = "C2M_CONFIG.json"
|
|
C2M_FILES_TEMP_STORAGE_PATH = C2M_FILES_TEMP_STORAGE_PATH
|
|
|
|
#Read config file
|
|
C2M_CONFIG = general_function.read_json_file("./config/",C2M_CONFIG_FILE)
|
|
|
|
#Run the c2m_scrape_app
|
|
c2m_scrape_app.run_app(C2M_FILES_TEMP_STORAGE_PATH, C2M_CONFIG)
|
|
|
|
def c2m_load_to_cloudstorage(C2M_FILES_TEMP_STORAGE_PATH):
|
|
#Config file name
|
|
CLOUDSTORAGE_CREDENTIAL_FILE = "./config/CLOUD_STORAGE_ADMIN_CREDENTIAL.json"
|
|
|
|
#Connect to cloudstorage
|
|
c2m_storage_client = c2m_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(C2M_FILES_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)
|
|
c2m_load_func.upload_to_bucket(c2m_storage_client, file_name, file_path)
|
|
|
|
|
|
def main():
|
|
print("RUNNING MAIN_PY")
|
|
|
|
# Exchange
|
|
#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/"
|
|
print("STARTING EXCHANGE EXTRACT")
|
|
exchange_extract(EXCHANGE_CSV_TEMP_STORAGE_PATH)
|
|
print("STARTING EXCHANGE LOAD")
|
|
exchange_load_to_cloudstorage(EXCHANGE_CSV_TEMP_STORAGE_PATH)
|
|
#Delete Temp CSV
|
|
print("DELETING EXCHANGE CSV")
|
|
general_function.delete_csv_in_path(EXCHANGE_CSV_TEMP_STORAGE_PATH)
|
|
|
|
# Shipping
|
|
#Make sure cron in the file directory
|
|
general_function.change_to_current_directory()
|
|
#Extract Load SHIPPING Data
|
|
SHIPPING_CSV_TEMP_STORAGE_PATH = "./shipping/extract/shipping_temp_extract_csv/"
|
|
print("STARTING SHIPPING EXTRACT")
|
|
shipping_extract(SHIPPING_CSV_TEMP_STORAGE_PATH)
|
|
print("STARTING SHIPPING LOAD")
|
|
shipping_load_to_cloudstorage(SHIPPING_CSV_TEMP_STORAGE_PATH)
|
|
#Delete Temp CSV
|
|
print("DELETING SHIPPING CSV")
|
|
general_function.delete_csv_in_path(SHIPPING_CSV_TEMP_STORAGE_PATH)
|
|
|
|
# C2M
|
|
#Make sure cron in the file directory
|
|
general_function.change_to_current_directory()
|
|
#Scrape C2M Data
|
|
C2M_FILES_TEMP_STORAGE_PATH = "./c2m/scrape/c2m_temp_scrape_files/"
|
|
print("STARTING C2M SCRAPE")
|
|
c2m_scrape(C2M_FILES_TEMP_STORAGE_PATH)
|
|
print("STARTING C2M LOAD")
|
|
c2m_load_to_cloudstorage(C2M_FILES_TEMP_STORAGE_PATH)
|
|
#Delete Temp CSV
|
|
print("DELETING C2M CSV or PNG")
|
|
general_function.delete_csv_or_png_in_path(C2M_FILES_TEMP_STORAGE_PATH)
|
|
|
|
# # CRISP
|
|
# #Make sure cron in the file directory
|
|
# general_function.change_to_current_directory()
|
|
# #Extract Load Exchange Data
|
|
# CRISP_CSV_TEMP_STORAGE_PATH = "./crisp/extract/crisp_temp_extract_csv/"
|
|
# print("STARTING CRISP EXTRACT")
|
|
# crisp_extract(CRISP_CSV_TEMP_STORAGE_PATH)
|
|
# print("STARTING CRISP LOAD")
|
|
# crisp_load_to_cloudstorage(CRISP_CSV_TEMP_STORAGE_PATH)
|
|
# #Delete Temp CSV
|
|
# print("DELETING CRISP CSV")
|
|
# general_function.delete_csv_in_path(CRISP_CSV_TEMP_STORAGE_PATH)
|
|
|
|
print("END MAIN_PY")
|
|
print("------------------------------------------------------------------------------------------------")
|
|
|
|
if True:
|
|
main() |