/* * Created by PyramIDE. * User: フランスのトトロ aka (David RENÉ) * Date: 09/05/2017 * Time: 15:27 * © David RENÉ */ read data_base/sqlite.anubis read data_base/read_csv.anubis read tools/function.anubis read hayamiki_lib/model/types/hk_database.anubis read calexium_lib/web/CXM_web_arg_utils.anubis read calexium_lib/web/CXM_web_session.anubis read calexium_lib/database/csv_import_status.anubis read lexical_analysis/fast_lexer_4.anubis define Maybe(SQLite3Bind) bind_column_value ( String col, String value ) = if col = "" then failure else success(bind_String(":"+col, value)). define List(Int) /* return a list of column index to get from CSV */ get_csv_columns ( List((Int, String)) columns, List(Int) so_far )= map(((Int, String) col) |-> since col is (idx, _), idx, columns) . define String dump_csv_db_col ( List((Int, String)) columns, String so_far )= if columns is { [] then so_far, [h . t] then if h is (csv_col, db_col) then dump_csv_db_col(t, so_far + "csv_col : "+csv_col+" db col name : "+db_col+"\r\n") }. define List(String) get_db_columns ( List((Int, String)) columns, List(String) so_far )= if columns is { [] then reverse(so_far), [h . t] then if h is (_, db_col) then get_db_columns(t, [db_col . so_far]) }. define List((Int csv_col, String db_col)) _get_columns_list ( List(Web_arg) lwa, Int max_col, //maximum columns in source side Int index, //current checking idx List((Int, String)) so_far ) = if index >= max_col then reverse(so_far) else if get_String(lwa, "col_" + index) is { failure then _get_columns_list(lwa, max_col, index + 1, so_far), success(db_col_name) then _get_columns_list(lwa, max_col, index + 1, [(index, db_col_name) . so_far]) } . define List((Int csv_col, String db_col)) /* get the list of source columns selected for importation process */ get_columns_list ( List(Web_arg) lwa ) = with max_col = get_Int(lwa, "hk_source_col", 0), _get_columns_list(lwa, max_col, 0, []) . define (Int, List(List(String))) extract_lines ( One -> ReadCsvResult cvs_line_reader, Int max_count, //number of line to extract List(String) columns, Int offset, List(List(String)) so_far ) = if max_count =< 0 then (offset, so_far) else if cvs_line_reader(unique) is { end_of_input then println("END OF INPUT "+offset);(offset, so_far) error(e) then //TODO report that error in importation report println("error ["+e+"]\n"); (offset, so_far), ok(_offset, all_values) then println("extracted line : \""+join("\", \"", all_values)+"\""); //if line_data is (line, all_values) then extract_lines(cvs_line_reader, max_count - 1, columns, _offset, [all_values . so_far]) } . define Result(String, One) update_data_to_table ( SQLite3Stmt update_data_stmt, List(String) columns, List(String) values )= if sqlite3_reset(update_data_stmt) is { error(err) then println(db_error(err, "Error reseting update data statement")); error("DB Error"), ok then //println("update reset ok"); if sqlite3_bind(update_data_stmt, map2_select(bind_column_value, columns, values)) is { error(err) then println(db_error(err, "Error binding update data statement"));error("DB Error"), ok then //println("update bind ok"); if sqlite3_step(update_data_stmt) is { error(err) then println(db_error(err, "Error executing update data statement")); error("DB Error"), no_more_row then ok(unique), row(_) then ok(unique) } } }. define Result(String, One) insert_or_update_data_to_table ( SQLite3Stmt insert_data_stmt, SQLite3Stmt update_data_stmt, Bool update, //if true make update on constraint violation List(String) columns, List(String) values )= if sqlite3_reset(insert_data_stmt) is { error(err) then println(db_error(err, "Error reseting insert data statement")); error("DB Error"), ok then if sqlite3_bind(insert_data_stmt, map2_select(bind_column_value, columns, values)) is { error(err) then println(db_error(err, "Error binding insert data statement"));error("DB Error"), ok then if sqlite3_step(insert_data_stmt) is { error(err) then println(db_error(err, "Error executing insert data statement")); forget(sqlite3_reset(insert_data_stmt)); if err.code = 19 & update then update_data_to_table(update_data_stmt, columns, values) else error("DB Error"), no_more_row then ok(unique) //println("insert no more row"); //update_data_to_table(update_data_stmt, columns, values), row(_) then ok(unique) //println("insert row"); //update_data_to_table(update_data_stmt, columns, values) } } }. define Maybe(One) store_data ( // SQLite3Stmt insert_data_stmt, // SQLite3Stmt update_data_stmt, // Bool update, List(String) -> Result(String, One) import_line_values, // SQLite3Stmt select_contact_id_opt_out_stmt, // SQLite3Stmt select_stmt, // SQLite3Stmt insert_list_contact_stmt, // SQLite3Stmt count_stmt, // List(String) columns, //list of column's name List(List(String)) rows //list of list of real data to store ) = if rows is { [] then println("store data row empty"); success(unique), [row . t] then if import_line_values(row) is //insert_or_update_data_to_table(insert_data_stmt, update_data_stmt, update, columns, row) is { error(msg) then println("data can't be added "+msg); //TODO as Julien's idea, copy error line in new csv file store_data(import_line_values, t), ok(_) then store_data(import_line_values, t) } } . define CSV_Import_Status main_loop ( SQLite3DataBase db, // Int import_id, */ // SQLite3Stmt insert_data_stmt, // SQLite3Stmt update_data_stmt, List(String) -> Result(String, One) import_line_values, One -> ReadCsvResult csv_line_reader, //One -> Int progression, UTime start_time, List(String) columns, Int line_num, Int _offset, Int offset_base ) = // with prog = progression(unique), // logDebug(debug_log, "Importing contacts into mailing in progress... " + abs_to_decimal(prog) + " Bytes"); // logDebug(debug_log, "Importing contacts into mailing in progress... Lines:" + line_num + " "+ (offset_base + offset) + " Bytes read"); //import_csv_set_status(db, list_id, csv_in_progress); with new_time = unow, since new_time - start_time is utime(seconds, _), //! import_task_update_progress(db, import_id, seconds, line_num, offset_base + offset); //get 500 lines from cvs since extract_lines(csv_line_reader, 500, columns, _offset, []) is (offset, so_far), with extract_length = length(so_far), //println("extraction: offset "+offset+" #lines "+extract_length); if extract_length = 0 then csv_finished else if db_do_transaction( db, (One _) |-> store_data(import_line_values, so_far), success((SQLite3Error err) |-> println(db_error(err,"import_csv_to_db"))), //success((SQLite3Error err) |-> logError("DB", db_error(err,"import_csv_to_db"))), 60000, // max 60s 100, // retry every 100 ms "CSV_Import_Status" ) is { error(_) then println("main loop db_do_transaction csv_db_error"); csv_db_error, ok(_) then main_loop(db, import_line_values, csv_line_reader, new_time, columns, line_num + extract_length, offset, offset_base) } . define CSV_Import_Status import_csv_to_db_transaction ( SQLite3DataBase db, String table_name, // Import_Task import_task, One -> ReadCsvResult csv_line_reader, //(One) -> Int progression, String import_type, List(String) columns, //columns selected for importation in same order from CSV file String column_key, UTime start_time, Int offset_base )= //since import_task is import_task(import_task_id, list_id, filename, file_size, csv_db_columns, separator, charset, amount_time, offset, current_line, lines_count, status, status_details, update_contact, opt_out_import), with current_line = (Int)0, with offset = (Int)0, with key = if column_key = "" then "id" else column_key, with filter_columns_esc = map_select((String s) |-> if s = "" then failure else success("\""+s+"\""), columns), with filter_columns = map_select((String s) |-> if s = "" then failure else success(s), columns), sql_columns = join(", ", filter_columns_esc), sql_values = join(", ", map((String s) |-> ":"+s, filter_columns)), with sql_update_values = join(", ", map((String s) |-> "\""+s+"\" = :"+s, filter_columns)), if sqlite3_prepare(db, "INSERT INTO "+table_name+" ("+sql_columns+") VALUES ("+sql_values+");") is { error(err) then //logError(debug_log, db_error(err, "preparing insert contact statement for import _csv_to_mailing_list")); println(db_error(err, "preparing insert data statement for import_csv_to_mailing_list")); csv_db_error, ok(insert_data_stmt) then // with update_query = "UPDATE "+table_name+" SET " + sql_update_values + " WHERE changes()=0 AND "+key+"=:"+key+";", with update_query = "UPDATE "+table_name+" SET " + sql_update_values + " WHERE \""+key+"\" = :"+key+";", println(" update query : "+update_query); if sqlite3_prepare(db, update_query) is { error(err) then //logError(debug_log, db_error(err, "preparing update contact statement for import_csv_to_mailing_list")); println(db_error(err, "preparing update data statement for import_csv_to_db_transaction")); csv_db_error, ok(update_data_stmt) then /* run the main loop */ with import_line_values = if import_type = "insert_or_update" then (List(String) row) |-> insert_or_update_data_to_table(insert_data_stmt, update_data_stmt, true, columns, row) else if import_type = "insert" then (List(String) row) |-> insert_or_update_data_to_table(insert_data_stmt, update_data_stmt, false, columns, row) else if import_type = "update" then (List(String) row) |-> update_data_to_table(update_data_stmt, columns, row) else (List(String) row) |-> insert_or_update_data_to_table(insert_data_stmt, update_data_stmt, true, columns, row), main_loop(db, import_line_values, csv_line_reader, start_time, columns, current_line, offset, offset_base) } } // } . public define Result(String, String) hkc_import_execute ( WEB_Session _session, SQLite3DataBase db, HK_Database hk_db )= //arguments list //table_name //file //csv_sep //col_x = 'target_col', where x is the column number in csv file and target_col is name of the column in the table //with _T = make_translate_function(_session.language), with lwa = *_session.web_request.lwa, with separator = get_String(lwa, "csv_sep", ";"), with csv_db_columns = get_columns_list(lwa), with columns = get_db_columns(csv_db_columns, []), column_key = get_String(lwa, "column_key", ""), table_name = get_String(lwa, "table_name", ""), import_type = get_String(lwa, "import_type", "insert_or_update"), with start_time = unow, offset_base = (Int)0, skip_first_line = get_Bool(lwa, "hk_remove_csv_header", true), file_name_full_path = get_String(_session.fields, "upl_file", ""), println(dump_csv_db_col(csv_db_columns,"")); if file(file_name_full_path, read) is { failure then error("Can't open CSV file '" + file_name_full_path + "'") //import_task_set_status(db, import_id, csv_file_not_found), success(f) then //with file_size = file_size(f), if make_lexing_stream("", f, 65536, 10) is { failure then error("csv_error Error while reading file"), success(ls) then with csv_line_reader = make_read_csv_line(ls, separator, get_csv_columns(csv_db_columns, [])), (if skip_first_line then forget(csv_line_reader(unique)) else unique); with status = import_csv_to_db_transaction(db, table_name, csv_line_reader, import_type, columns, column_key, start_time, offset_base), ok("imported") } } .