Merge pull request #2832 from bcgov/2519-etl-responsible-party-property-owner

2519: ETL responsible party property owner

Author: trslater

bin/migrate-nris-data/ce_files/ce_files.py

@@ -5,7 +5,7 @@
 
 
 def etl(batch_size):
-    logger = setup_and_get_logger("ce-files-etl")
+    logger = setup_and_get_logger("ce-file-etl")
 
     batch_read_write(
         logger,
@@ -21,8 +21,8 @@ def obfuscate(batch_size):
 
     batch_read_write(
         logger,
-        batch_size,  # Obfuscation can be done in smaller batches to reduce transaction size.
-        ABS_PATH / "ce_files/sql/count.sql",
+        batch_size,
+        ABS_PATH / "ce_files/sql/obfuscate_count.sql",
         ABS_PATH / "ce_files/sql/obfuscate_get_rows.sql",
         ABS_PATH / "ce_files/sql/obfuscate_update.sql",
         row_processor=row_obfuscator(),

bin/migrate-nris-data/ce_files/sql/obfuscate_count.sql

@@ -0,0 +1,4 @@
+select
+    count(*)
+from
+    alcs.compliance_and_enforcement ace;

bin/migrate-nris-data/ce_files/sql/obfuscate_update.sql

@@ -8,4 +8,5 @@ set
 from
     (values %s) as v(uuid, intake_notes)
 where
-    ace.uuid::text = v.uuid::text;
+    ace.uuid::text = v.uuid::text
+    and ace.intake_notes <> '';

bin/migrate-nris-data/db.py

@@ -76,21 +76,20 @@ def batch_read_write(
         count_cursor.execute(count_sql)
         count_total = count_cursor.fetchone()[0]
 
-    logger.info(f"Total records to insert: {count_total}")
+    logger.info(f"Total records to write: {count_total}")
 
     num_successful = 0
 
     with read_conn.cursor(
-        name="read_cursor",
-        cursor_factory=RealDictCursor
+        name="read_cursor", cursor_factory=RealDictCursor
     ) as read_cursor, write_conn.cursor() as write_cursor:
         read_cursor.execute(read_sql)
 
         with Progress(SpinnerColumn(), BarColumn(), transient=True) as progress:
             task = progress.add_task("Progress:", total=count_total)
 
             while rows := read_cursor.fetchmany(batch_size):
-                num_to_insert = len(rows)
+                curr_batch_size = len(rows)
 
                 if row_processor:
                     rows = map(row_processor, rows)
@@ -103,19 +102,19 @@ def batch_read_write(
                     )
                     write_conn.commit()
 
-                    num_successful += num_to_insert
+                    num_successful += curr_batch_size
 
-                    progress.update(task, advance=num_to_insert)
+                    progress.update(task, advance=curr_batch_size)
                     logger.debug(
-                        f"{num_to_insert} records inserted successfully; {num_successful} of {count_total} records inserted so far."
+                        f"{curr_batch_size} records ran successfully; {num_successful} of {count_total} records run so far."
                     )
 
                 except Exception as err:
                     logger.exception(err)
                     write_conn.rollback()
-                    logger.info(f"Failed to insert batch of {num_to_insert}.")
+                    logger.info(f"Failed to run batch of {curr_batch_size}.")
 
-        logger.info(f"Successfully inserted {num_successful} of {count_total} records.")
+        logger.info(f"Successfully ran {num_successful} of {count_total} records.")
 
 
 def load_sql(file_path):

bin/migrate-nris-data/migrate.py

@@ -3,6 +3,7 @@
 
 from common.constants import BATCH_UPLOAD_SIZE
 import ce_files
+import responsible_parties
 import submitters
 
 console = Console()
@@ -28,6 +29,7 @@ def import_data(ctx, batch_size):
         # Order matters
         ctx.invoke(import_ce_files)
         ctx.invoke(import_submitters)
+        ctx.invoke(import_responsible_parties)
 
 
 @import_data.command("ce-files")
@@ -46,6 +48,14 @@ def import_submitters(ctx):
     console.log("Submitter import complete.")
 
 
+@import_data.command("responsible-parties")
+@click.pass_context
+def import_responsible_parties(ctx):
+    console.log("Start importing responsible parties...")
+    responsible_parties.etl(batch_size=ctx.parent.params["batch_size"])
+    console.log("Responsible party import complete.")
+
+
 @import_data.result_callback()
 def import_cleanup(results, batch_size):
     console.log("All imports complete. Cleaning up...")
@@ -68,6 +78,7 @@ def obfuscate(ctx, batch_size):
         # Order matters
         ctx.invoke(obfuscate_ce_files)
         ctx.invoke(obfuscate_submitters)
+        ctx.invoke(obfuscate_responsible_parties)
 
 
 @obfuscate.command("ce-files")
@@ -82,10 +93,18 @@ def obfuscate_ce_files(ctx):
 @click.pass_context
 def obfuscate_submitters(ctx):
     console.log("Start obfuscating submitters...")
-    submitters.obfuscate_submitters(batch_size=ctx.parent.params["batch_size"])
+    submitters.obfuscate(batch_size=ctx.parent.params["batch_size"])
     console.log("Submitter obfuscation complete.")
 
 
+@obfuscate.command("responsible-parties")
+@click.pass_context
+def obfuscate_responsible_parties(ctx):
+    console.log("Start obfuscating responsible parties...")
+    responsible_parties.obfuscate(batch_size=ctx.parent.params["batch_size"])
+    console.log("Responsible party obfuscation complete.")
+
+
 @obfuscate.result_callback()
 def obfuscation_cleanup(results, batch_size):
     console.log("All obfuscation complete. Cleaning up...")

bin/migrate-nris-data/responsible_parties/__init__.py

@@ -0,0 +1 @@
+from .responsible_parties import *

bin/migrate-nris-data/responsible_parties/responsible_parties.py

@@ -0,0 +1,40 @@
+from common.etl_logger import setup_and_get_logger
+from config import ABS_PATH
+from db import batch_read_write
+from faker import Faker
+
+
+def etl(batch_size):
+    logger = setup_and_get_logger("responsible-party-etl")
+
+    batch_read_write(
+        logger,
+        batch_size,
+        ABS_PATH / "responsible_parties/sql/count.sql",
+        ABS_PATH / "responsible_parties/sql/et.sql",
+        ABS_PATH / "responsible_parties/sql/l.sql",
+    )
+
+
+def obfuscate(batch_size):
+    logger = setup_and_get_logger("responsible-party-obfuscation")
+
+    batch_read_write(
+        logger,
+        batch_size,
+        ABS_PATH / "responsible_parties/sql/obfuscate_count.sql",
+        ABS_PATH / "responsible_parties/sql/obfuscate_get_rows.sql",
+        ABS_PATH / "responsible_parties/sql/obfuscate_update.sql",
+        row_processor=row_obfuscator(),
+    )
+
+
+def row_obfuscator():
+    faker = Faker("la")
+
+    def obfuscate_row(row):
+        row["individual_name"] = faker.name()
+
+        return row
+
+    return obfuscate_row

bin/migrate-nris-data/responsible_parties/sql/count.sql

@@ -0,0 +1,4 @@
+select
+    count(*)
+from
+    nris.complaint nc;

bin/migrate-nris-data/responsible_parties/sql/et.sql

@@ -0,0 +1,14 @@
+select
+  'Property Owner'::alcs."compliance_and_enforcement_responsible_party_party_type_enum" as party_type,
+  nc.foippa_type::alcs."compliance_and_enforcement_responsible_party_foippa_category_en" as foippa_category,
+  false as is_previous,
+  case
+    when nc.foippa_type = 'Organization' then nc.submission_property_owner::text
+  end as organization_name,
+  case
+    when nc.foippa_type <> 'Organization' then nc.submission_property_owner::text
+  end as individual_name,
+  ace.uuid as file_uuid
+from
+  nris.complaint nc
+  join alcs.compliance_and_enforcement ace on ace.file_number = nc.record_id

bin/migrate-nris-data/responsible_parties/sql/l.sql

@@ -0,0 +1,10 @@
+insert into
+    alcs.compliance_and_enforcement_responsible_party (
+        party_type,
+        foippa_category,
+        is_previous,
+        organization_name,
+        individual_name,
+        file_uuid
+    )
+values %s;