python_template.py

#! /usr/bin/env python3

import sys
import os
import argparse
import csv
import json
import time
from datetime import datetime
from dateutil.parser import parse as dateparse
import signal
import random
import hashlib


# =========================
class mapper:

 # ----------------------------------------
 def __init__(self):

 self.load_reference_data()
 self.stat_pack = {}

 # ----------------------------------------
 def map(self, raw_data, input_row_num=None):
 json_data = {}

 # --clean values
 for attribute in raw_data:
 raw_data[attribute] = self.clean_value(raw_data[attribute])

 # --place any filters needed here

 # --place any calculations needed here

 # --mandatory attributes
 json_data["DATA_SOURCE"] = "<supply>"

 # --the record_id should be unique, remove this mapping if there is not one
 json_data["RECORD_ID"] = "<remove_or_supply>"

 # --record type is not mandatory, but should be PERSON or ORGANIZATION
 # --json_data['RECORD_TYPE'] = 'PERSON'

 # --column mappings

 # --remove empty attributes and capture the stats
 json_data = self.remove_empty_tags(json_data)
 self.capture_mapped_stats(json_data)

 return json_data

 # ----------------------------------------
 def load_reference_data(self):

 # --garbage values
 self.variant_data = {}
 self.variant_data["GARBAGE_VALUES"] = ["NULL", "NUL", "N/A"]

 # -----------------------------------
 def clean_value(self, raw_value):
 if not raw_value:
 return ""
 new_value = " ".join(str(raw_value).strip().split())
 if new_value.upper() in self.variant_data["GARBAGE_VALUES"]:
 return ""
 return new_value

 # -----------------------------------
 def compute_record_hash(self, target_dict, attr_list=None):
 if attr_list:
 string_to_hash = ""
 for attr_name in sorted(attr_list):
 string_to_hash += (
 " ".join(str(target_dict[attr_name]).split()).upper()
 if attr_name in target_dict and target_dict[attr_name]
 else ""
 ) + "|"
 else:
 string_to_hash = json.dumps(target_dict, sort_keys=True)
 return hashlib.md5(bytes(string_to_hash, "utf-8")).hexdigest()

 # ----------------------------------------
 def format_date(self, raw_date):
 try:
 return datetime.strftime(dateparse(raw_date), "%Y-%m-%d")
 except:
 self.update_stat("!INFO", "BAD_DATE", raw_date)
 return ""

 # ----------------------------------------
 def remove_empty_tags(self, d):
 if isinstance(d, dict):
 for k, v in list(d.items()):
 if v is None or len(str(v).strip()) == 0:
 del d[k]
 else:
 self.remove_empty_tags(v)
 if isinstance(d, list):
 for v in d:
 self.remove_empty_tags(v)
 return d

 # ----------------------------------------
 def update_stat(self, cat1, cat2, example=None):

 if cat1 not in self.stat_pack:
 self.stat_pack[cat1] = {}
 if cat2 not in self.stat_pack[cat1]:
 self.stat_pack[cat1][cat2] = {}
 self.stat_pack[cat1][cat2]["count"] = 0

 self.stat_pack[cat1][cat2]["count"] += 1
 if example:
 if "examples" not in self.stat_pack[cat1][cat2]:
 self.stat_pack[cat1][cat2]["examples"] = []
 if example not in self.stat_pack[cat1][cat2]["examples"]:
 if len(self.stat_pack[cat1][cat2]["examples"]) < 5:
 self.stat_pack[cat1][cat2]["examples"].append(example)
 else:
 randomSampleI = random.randint(2, 4)
 self.stat_pack[cat1][cat2]["examples"][randomSampleI] = example
 return

 # ----------------------------------------
 def capture_mapped_stats(self, json_data):

 if "DATA_SOURCE" in json_data:
 data_source = json_data["DATA_SOURCE"]
 else:
 data_source = "UNKNOWN_DSRC"

 for key1 in json_data:
 if type(json_data[key1]) != list:
 self.update_stat(data_source, key1, json_data[key1])
 else:
 for subrecord in json_data[key1]:
 for key2 in subrecord:
 self.update_stat(data_source, key2, subrecord[key2])


# ----------------------------------------
def signal_handler(signal, frame):
 print("USER INTERRUPT! Shutting down ... (please wait)")
 global shut_down
 shut_down = True
 return


# ----------------------------------------
if __name__ == "__main__":
 proc_start_time = time.time()
 shut_down = False
 signal.signal(signal.SIGINT, signal_handler)

 input_file = "<input_file_name>"
 csv_dialect = "<dialect>"

 parser = argparse.ArgumentParser()
 parser.add_argument(
 "-i",
 "--input_file",
 dest="input_file",
 default=input_file,
 help="the name of the input file",
 )
 parser.add_argument(
 "-o", "--output_file", dest="output_file", help="the name of the output file"
 )
 parser.add_argument(
 "-l",
 "--log_file",
 dest="log_file",
 help="optional name of the statistics log file",
 )
 args = parser.parse_args()

 if not args.input_file or not os.path.exists(args.input_file):
 print("\nPlease supply a valid input file name on the command line\n")
 sys.exit(1)
 if not args.output_file:
 print("\nPlease supply a valid output file name on the command line\n")
 sys.exit(1)

 input_file_handle = open(args.input_file, "r")
 output_file_handle = open(args.output_file, "w", encoding="utf-8")
 mapper = mapper()

 input_row_count = 0
 output_row_count = 0
 for input_row in csv.DictReader(input_file_handle, dialect=csv_dialect):
 input_row_count += 1

 json_data = mapper.map(input_row, input_row_count)
 if json_data:
 output_file_handle.write(json.dumps(json_data) + "\n")
 output_row_count += 1

 if input_row_count % 1000 == 0:
 print(
 "%s rows processed, %s rows written"
 % (input_row_count, output_row_count)
 )
 if shut_down:
 break

 elapsed_mins = round((time.time() - proc_start_time) / 60, 1)
 run_status = (
 "completed in" if not shut_down else "aborted after"
 ) + " %s minutes" % elapsed_mins
 print(
 "%s rows processed, %s rows written, %s\n"
 % (input_row_count, output_row_count, run_status)
 )

 output_file_handle.close()
 input_file_handle.close()

 # --write statistics file
 if args.log_file:
 with open(args.log_file, "w") as outfile:
 json.dump(mapper.stat_pack, outfile, indent=4, sort_keys=True)
 print("Mapping stats written to %s\n" % args.log_file)

 sys.exit(0)