hk_rows_import.anubis
13.5 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
/*
* 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 xlib/web/CXM_web_arg_utils.anubis
read xlib/web/CXM_web_session.anubis
read xlib/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")
}
}
.