Create table sitrep_messages

Author: AntoineAugusti

airflow/dags/download_secmar_json_ftp.py

@@ -128,6 +128,9 @@ def secmar_json_sql_task(dag, filename):
 insert_snosan_json_unique = secmar_json_sql_task(dag, "insert_snosan_json_unique")
 insert_snosan_json_unique.set_upstream(copy_json_data)
 
+insert_sitrep_messages = secmar_json_sql_task(dag, "insert_sitrep_messages")
+insert_sitrep_messages.set_upstream(insert_snosan_json_unique)
+
 start_create_codes_tables = DummyOperator(task_id="start_create_codes_tables", dag=dag)
 start_create_codes_tables.set_upstream(insert_snosan_json_unique)
 end_create_codes_tables = DummyOperator(task_id="end_create_codes_tables", dag=dag)

airflow/dags/replace_secmar_database.py

@@ -26,7 +26,7 @@
 )
 dag.doc_md = __doc__
 
-tables = SECMAR_TABLES + ["operations_stats", "moyens_snsm"]
+tables = SECMAR_TABLES + ["operations_stats", "moyens_snsm", "sitrep_messages"]
 
 template = "sudo -u postgres pg_dump -c --no-owner {tables} {schema} > {output}"
 dump_command = template.format(
@@ -35,9 +35,7 @@
     tables=" ".join(["-t " + t for t in tables]),
 )
 
-dump_local_database = BashOperator(
-    task_id="dump_local_database", bash_command=dump_command, dag=dag
-)
+dump_local_database = BashOperator(task_id="dump_local_database", bash_command=dump_command, dag=dag)
 
 template = "psql -U {user} -h {host} {schema} < {input}"
 target_connection = PostgresHook.get_connection("target_secmar")
@@ -55,7 +53,5 @@
 )
 import_remote_database.set_upstream(dump_local_database)
 
-delete_dump_file = BashOperator(
-    task_id="delete_dump_file", bash_command="rm " + OUTPUT_PATH, dag=dag
-)
+delete_dump_file = BashOperator(task_id="delete_dump_file", bash_command="rm " + OUTPUT_PATH, dag=dag)
 delete_dump_file.set_upstream(import_remote_database)

opendata/sql/secmar_json/insert_sitrep_messages.sql

@@ -0,0 +1,38 @@
+drop table if exists sitrep_messages;
+
+CREATE TABLE sitrep_messages (
+  operation_id bigint primary key,
+  sitrep text
+);
+
+INSERT INTO sitrep_messages
+  (operation_id, sitrep)
+select
+  sjoi.operation_id,
+  array_to_string(array_agg(value order by coalesce(mapping.dst, key) asc), ' ')
+from (
+  select
+    u.data->>'chrono' as chrono,
+    replace(key, 'paragraphe', '') as key,
+    value
+  from snosan_json_unique u, jsonb_each(u.data->'messages'->0) AS obj(key, value)
+  where key like 'paragraphe%'
+) t
+join snosan_json_operation_id sjoi on sjoi.chrono = t.chrono
+left join (
+  select '1' src, 'A' dst
+  union select '2' src, 'B' dst
+  union select '3' src, 'C' dst
+  union select '4' src, 'D' dst
+  union select '5' src, 'E' dst
+  union select '6' src, 'F' dst
+  union select '7' src, 'G' dst
+  union select '8' src, 'H' dst
+  union select '9' src, 'I' dst
+  union select '10' src, 'J' dst
+  union select '11' src, 'K' dst
+  union select '12' src, 'L' dst
+  union select '13' src, 'M' dst
+  union select '14' src, 'N' dst
+) mapping on mapping.src = key
+group by 1;