enabled upload of manual matches through the administrative interface

Author: sposerina

src/server/datasource_manager.py

@@ -21,7 +21,8 @@ def __clean_csv_headers(header):
                             'Account ID (18 digit)',
                             'Opportunity Name', 'Stage', 'Fiscal Period', 'Amount', 'Probability (%)', 'Age',
                             'Close Date', 'Created Date', 'Type', 'Primary Campaign Source',
-                            'Source', 'Contact ID (18 Digit)', 'Primary Contact']
+                            'Source', 'Contact ID (18 Digit)', 'Primary Contact'],
+    'manualmatches': ['salesforcecontacts', 'volgistics', 'shelterluvpeople']
 }
 
 DATASOURCE_MAPPING = {

src/server/pipeline/clean_and_load_data.py

@@ -17,6 +17,10 @@ def start(connection, pdp_contacts_df, file_path_list):
     for uploaded_file in file_path_list:
         file_path = os.path.join(CURRENT_SOURCE_FILES_PATH, uploaded_file)
         table_name = file_path.split('/')[-1].split('-')[0]
+        if table_name == 'manual_matches':
+            manual_matches_df = pd.read_csv((io.BytesIO(open(file_path, "rb").read())), encoding='iso-8859-1')
+            continue
+            
         current_app.logger.info('Running load_paws_data on: ' + uploaded_file)
 
         df = pd.read_csv((io.BytesIO(open(file_path, "rb").read())), encoding='iso-8859-1')
@@ -54,7 +58,7 @@ def start(connection, pdp_contacts_df, file_path_list):
 
         current_app.logger.info('   - Finish load_paws_data on: ' + uploaded_file)
 
-    return result, json_rows
+    return result, json_rows, manual_matches_df
 
 
 def create_normalized_df(df, normalized_df, table_name):

src/server/pipeline/flow_script.py

@@ -25,7 +25,7 @@ def start_flow():
             # Populate new records in secondary tables (donations, volunteer shifts)
             # input - existing files in path
             # output - normalized object of all entries, as well as the input json rows for primary sources
-            normalized_data, source_json = clean_and_load_data.start(connection, pdp_contacts_df, file_path_list)
+            normalized_data, source_json, manual_matches_df = clean_and_load_data.start(connection, pdp_contacts_df, file_path_list)
 
             # Standardize column data types via postgres (e.g. reading a csv column as int vs. str)
             # (If additional inconsistencies are encountered, may need to enforce the schema of
@@ -41,7 +41,7 @@ def start_flow():
 
             # Match new+updated records against previous version of pdp_contacts database, and
             # write these rows to the database.
-            match_data.start(connection, rows_classified)
+            match_data.start(connection, rows_classified, manual_matches_df)
 
             # Copy raw input rows to json fields in pdp_contacts,
             # using a temporary table to simplify the update code.

src/server/pipeline/match_data.py

@@ -15,13 +15,14 @@ def normalize_before_match(value):
     return result
 
 
-def start(connection, added_or_updated_rows):
+def start(connection, added_or_updated_rows, manual_matches_df):
     # Match new records to each other and existing pdp_contacts data.
     # Assigns matching ID's to records, as well.
     # WARNING: not thread-safe and could lead to concurrency issues if two users /execute simultaneously
     current_app.logger.info('Start record matching')
     # Will need to consider updating the existing row contents (filter by active), deactivate,
     # try to match, and merge previous matching groups if applicable
+    current_app.logger.info('Manual matches DF: ' + manual_matches_df)
 
     job_id = str(int(time.time()))
     log_db.log_exec_status(job_id, {'status': 'starting', 'at_row': 0, 'of_rows': 0})
@@ -63,13 +64,13 @@ def start(connection, added_or_updated_rows):
         # Exact matches based on specified columns
         row_matches = pdp_contacts[
             (
-                    ((pdp_contacts["first_name_normalized"] == row["first_name_normalized"]) &
-                     (pdp_contacts["last_name_normalized"] == row["last_name_normalized"]))
-                    |
-                    ((pdp_contacts["first_name_normalized"] == row["last_name_normalized"]) &
-                     (pdp_contacts["last_name_normalized"] == row["first_name_normalized"]))
-                    &
-                    ((pdp_contacts["email_normalized"] == row["email_normalized"]) | (pdp_contacts["mobile"] == row["mobile"]))
+                ((pdp_contacts["first_name_normalized"] == row["first_name_normalized"]) &
+                (pdp_contacts["last_name_normalized"] == row["last_name_normalized"]))
+                |
+                ((pdp_contacts["first_name_normalized"] == row["last_name_normalized"]) &
+                (pdp_contacts["last_name_normalized"] == row["first_name_normalized"]))
+                &
+                ((pdp_contacts["email_normalized"] == row["email_normalized"]) | (pdp_contacts["mobile"] == row["mobile"]))
             )
         ]
         if row_matches.empty:  # new record, no matching rows