ETL from LOCAL/MAIL to MENTDB/EXCEL.mql
ETL script with application 'supervision' compatibility
if (not (group exist "0001_folder1_folder2_folder3")) {
group add "0001_folder1_folder2_folder3";
};
script merge "folder1.folder2.folder3.step_1_source_connect.exe" true 5
(param
)
"Connect to the source"
{
# CONFIGURATION ;
-> "[CONF_NAME_OF_THE_FLOW]" "name_of_the_flow";
# INITIALIZATION ;
-> "[FLOW_PID]" [PID];
# HANDLE ;
try {
# Flow initialization;
stack flow_init [FLOW_PID] [CONF_NAME_OF_THE_FLOW] "{}";
stack flow_step [FLOW_PID] 1 "source_connect...";
# Local connection: don't use step 1;
log write "Source: Local connection" OK null null;
#Step 1 is valid;
stack flow_step [FLOW_PID] 1 "source_connect_ok";
include "folder1.folder2.folder3.step_2_extract.exe";
} {
#Step 1 is not valid;
stack flow_step [FLOW_PID] 1 "source_connect_KO";
# Give the error to the stack and stop the process;
exception (1) ([global_err]);
} "[global_err]";
} "Return nothing";
if (not (group is granted script "folder1.folder2.folder3.step_1_source_connect.exe" "0001_folder1_folder2_folder3")) {
group grant script "folder1.folder2.folder3.step_1_source_connect.exe" "0001_folder1_folder2_folder3";
};
script merge "folder1.folder2.folder3.step_2_extract.exe" true 1
(param
)
"Extract files from the source"
{
# CONFIGURATION ;
-> "[CONF_SOURCE_CM]" "demo_cm_imap";
-> "[CONF_SOURCE_RECEIVE_DIR]" (concat "home/" [FLOW_PID]);
# HANDLE ;
try {
# Mark the flow as step 2;
stack flow_step [FLOW_PID] 2 "extract...";
# Create the local directory PID;
file mkdir (concat "home/" [FLOW_PID]);
#Get mails;
json load "conf_mail" (cm get [CONF_SOURCE_CM]);
#json iobject "conf_mail" / "alternateSubDir" "tmp/PPJ" STR;
json iobject "conf_mail" / "partialfetch" "true" STR;
json iobject "conf_mail" / "max_buffer" "16000" STR;
json iobject "conf_mail" / "fetchsize" "512000" STR;
json iobject "conf_mail" / "sessionTimeout" "900000" STR;
#parameter add "0001_folder1_folder2_folder3_MAIL_MESSAGES_ORDERS" "{}";
json iobject "conf_mail" / "mark_msgs_order" (parameter get value "0001_folder1_folder2_folder3_MAIL_MESSAGES_ORDERS") OBJ;
json load "receive_state" (mail download imap_disk [CONF_SOURCE_RECEIVE_DIR]
3 unread "TERMINE|INBOX/atraiter" false true
"2018-01-01" null
(mql {string matches [imap_from] ".*digest-noreply@quora.com.*";})
(mql {string matches [imap_subject] ".*champ.*";})
{json doc "conf_mail"}
);
parameter update "0001_folder1_folder2_folder3_MAIL_MESSAGES_ORDERS" (json select "receive_state" /mark_msgs_order);
if (not (is null or empty (json select "receive_state" "/Error"))) {
};
if (> (json select "receive_state" "/NbReceived") 0) {
-> "[sub_receive_dir]" (json select "receive_state" "/Directory");
json load "sub_receive_dir" (file dir_list [sub_receive_dir]);
json parse_array "sub_receive_dir" "/" "[mail]" {
if (string ends_with [mail] ".json") {
json load "current_mail" (file load (concat [sub_receive_dir] "/" [mail]));
json parse_array "current_mail" "/Parts" "part" {
if (equal (json select "part" "/type") "file") {
if (string ends_with (json select "part" "/content_filename") ".csv") {
stack (date now) "folder1.folder2.folder3.step_3_transform.exe" "[FLOW_PID]" [FLOW_PID] "[filename]" (json select "part" "/content_filename");
};
};
};
};
};
};
#Step 2 is valid;
stack flow_step [FLOW_PID] 2 "extract_ok";
} {
#Step 2 is not valid;
stack flow_step [FLOW_PID] 2 "extract_KO";
# Give the error to the stack and stop the process;
exception (1) ([global_err]);
} "[global_err]";
} "Return nothing";
if (not (group is granted script "folder1.folder2.folder3.step_2_extract.exe" "0001_folder1_folder2_folder3")) {
group grant script "folder1.folder2.folder3.step_2_extract.exe" "0001_folder1_folder2_folder3";
};
script merge "folder1.folder2.folder3.step_3_transform.exe" true 1
(param
(var "[FLOW_PID]" {true} "The flow PID" is_null:true is_empty:true "1")
(var "[filename]" {true} "The filename" is_null:true is_empty:true "file.xxx")
)
"Transform the file"
{
# CONFIGURATION ;
-> "[CONF_SOURCE_CSV_SEPARATOR]" ",";
-> "[CONF_SOURCE_CSV_QUOTE]" "'";
-> "[CONF_SOURCE_CSV_FORCE_COLUMN]" "A,B,C";
-> "[CONF_DESTINATION_NEW_FILENAME]" (concat "new_filename_" (date systimestamp_min) ".xls");
-> "[CONF_DESTINATION_SHEETNAME]" "sheet1";
# HANDLE ;
try {
# Mark the flow as step 3;
stack flow_step [FLOW_PID] 3 "transform...";
excel load empty "flow";
excel sheet add "flow" [CONF_DESTINATION_SHEETNAME];
log write (concat "Build the file 'home/" [FLOW_PID] "/" [CONF_DESTINATION_NEW_FILENAME] "' from 'home/" [FLOW_PID] "/" [filename] "'...") OK null null;
excel cell set "flow" [CONF_DESTINATION_SHEETNAME] 0 0 "ColNameA" STR;
excel cell set "flow" [CONF_DESTINATION_SHEETNAME] 0 1 "ColNameB" STR;
excel cell set "flow" [CONF_DESTINATION_SHEETNAME] 0 2 "ColNameC" STR;
-> "[index]" 1;
csv parse (mql "T") (concat "home/" [FLOW_PID] "/" [filename]) (mql [CONF_SOURCE_CSV_SEPARATOR]) (mql [CONF_SOURCE_CSV_QUOTE]) (mql [CONF_SOURCE_CSV_FORCE_COLUMN]) {
excel cell set "flow" [CONF_DESTINATION_SHEETNAME] [index] 0 [T_A] STR;
excel cell set "flow" [CONF_DESTINATION_SHEETNAME] [index] 1 [T_B] STR;
excel cell set "flow" [CONF_DESTINATION_SHEETNAME] [index] 2 [T_C] STR;
++ "[index]";
};
excel save "flow" (concat "home/" [FLOW_PID] "/" [CONF_DESTINATION_NEW_FILENAME]);
excel close "flow";
log write (concat "Builded.") OK null null;
#Step 3 is valid;
stack flow_step [FLOW_PID] 3 "transform_ok";
stack (date now) "folder1.folder2.folder3.step_4_destination_connect.exe" "[FLOW_PID]" [FLOW_PID] "[filename]" [CONF_DESTINATION_NEW_FILENAME];
} {
try {excel close "flow";} {} "[err]";
#Step 3 is not valid;
stack flow_step [FLOW_PID] 3 "transform_KO";
# Give the error to the stack and stop the process;
exception (1) ([global_err]);
} "[global_err]";
} "Return nothing";
if (not (group is granted script "folder1.folder2.folder3.step_3_transform.exe" "0001_folder1_folder2_folder3")) {
group grant script "folder1.folder2.folder3.step_3_transform.exe" "0001_folder1_folder2_folder3";
};
script merge "folder1.folder2.folder3.step_4_destination_connect.exe" true 5
(param
(var "[FLOW_PID]" {true} "The flow PID" is_null:true is_empty:true "1")
(var "[filename]" {true} "The filename" is_null:true is_empty:true "file.xxx")
)
"Connect to the destination"
{
# CONFIGURATION ;
-> "[CONF_DESTINATION_CM]" "demo_cm";
# HANDLE ;
try {
# Flow initialization;
stack flow_step [FLOW_PID] 4 "destination_connect...";
tunnel connect "destination_sess" {cm get [CONF_DESTINATION_CM];};
#Step 4 is valid;
stack flow_step [FLOW_PID] 4 "destination_connect_ok";
include "folder1.folder2.folder3.step_5_load.exe";
} {
try {tunnel disconnect "destination_sess";} {} "[err]";
#Step 4 is not valid;
stack flow_step [FLOW_PID] 4 "destination_connect_KO";
# Give the error to the stack and stop the process;
exception (1) ([global_err]);
} "[global_err]";
} "Return nothing";
if (not (group is granted script "folder1.folder2.folder3.step_4_destination_connect.exe" "0001_folder1_folder2_folder3")) {
group grant script "folder1.folder2.folder3.step_4_destination_connect.exe" "0001_folder1_folder2_folder3";
};
script merge "folder1.folder2.folder3.step_5_load.exe" true 1
(param
)
"Load file to the destination"
{
# CONFIGURATION ;
-> "[CONF_DIR_DESTINATION]" "tmp/destination";
# HANDLE ;
try {
# Mark the flow as step 5;
stack flow_step [FLOW_PID] 5 "load...";
log write (concat "Send '" (concat "home/" [FLOW_PID] "/" [filename]) "' to the '" (concat [CONF_DIR_DESTINATION] "/" [filename]) "' ...") OK null null;
# Copy file into the destination directory;
#Open a reader;
file reader_open "r1" (concat "home/" [FLOW_PID] "/" [filename]) BINARY null;
tunnel execute "destination_sess" (concat
"-> \"[remote_destination]\" \"" (mql encode (concat [CONF_DIR_DESTINATION] "/" [filename])) "\";"
(mql {
file writer_open "w1" [remote_destination] true BINARY null;
})
);
#Parse the file;
while (is not null (-> "[bytes]" (file reader_get_bytes "r1" 402400));) {
tunnel execute "destination_sess" (concat
"-> \"[bytes]\" \"" (mql encode [bytes]) "\";"
(mql {
file writer_add_bytes "w1" [bytes];
file writer_flush "w1";
})
);
};
tunnel execute "destination_sess" (concat
(mql {
#Force to write;
file writer_flush "w1";
file writer_close "w1";
})
);
#Close the reader;
file reader_close "r1";
log write "Sended." OK null null;
#Step 5 is valid;
stack flow_step [FLOW_PID] 5 "load_ok";
try {tunnel disconnect "destination_sess";} {} "[err]";
} {
#Close objects;
try {file reader_close "r1";} {} "[sub_err]";
try {tunnel execute "destination_sess" (concat
(mql {
try {file writer_close "w1";} {} "[err]";
})
);} {} "[err]";
try {tunnel disconnect "destination_sess";} {} "[err]";
#Step 5 is not valid;
stack flow_step [FLOW_PID] 5 "load_KO";
# Give the error to the stack and stop the process;
exception (1) ([global_err]);
} "[global_err]";
} "Return nothing";
if (not (group is granted script "folder1.folder2.folder3.step_5_load.exe" "0001_folder1_folder2_folder3")) {
group grant script "folder1.folder2.folder3.step_5_load.exe" "0001_folder1_folder2_folder3";
};