dedayoa · February 3, 2019 20:03
diff --git a/vox2astpp.py b/vox2astpp.py
 import csv
 import os
 import sys
 from decimal import Decimal
 import logging

 stdout_handler = logging.StreamHandler(sys.stdout)
 logging.basicConfig(
    level=logging.DEBUG, 
    format='[%(asctime)s] %(levelname)s - %(message)s',
    handlers=[stdout_handler]
 )
 logger = logging.getLogger(__name__)

 class ASTPPConverto(object):

    def __init__(self, in_file_loc = "", fx_rate = 1, profit_margin = 1, exclude_rules = [], logging_active = False):
        self.fx_rate = fx_rate
        self.profit_margin = profit_margin
        self.in_file_loc = in_file_loc
        self.exclude_rules = exclude_rules
        self.excluded_rows = []
        self.logging_active = logging_active

    def exclude_filter(self, row):
        """ Function returns false if row should be excluded, else returns true.
        Based on exclude rule
        """
        is_a_match = False
        for rule in self.exclude_rules:
            if len(rule) != 3:
                raise ValueError("The rule requires a 3 element tuple")
            try:
                rule_value, rule_colname, rule_matchby = rule
                if rule_matchby == "CONTAINS" and (rule_value.lower() in (row[rule_colname]).lower()):
                    self.excluded_rows.append(row)
                    is_a_match = True
                elif rule_matchby == "STARTSWITH" and (rule_value == row[rule_colname][0:len(str(rule_value))].strip()):
                    self.excluded_rows.append(row)
                    is_a_match = True
            except IndexError:
                raise IndexError(f"Column name {rule_colname} does not exist")
        if self.logging_active and is_a_match:
            logger.info((" | ").join(row.values()))
        return is_a_match

    def generate_output_file(self, rate_type, out_file_suffix = "output"):
        """ Function to generate ASTPP compatible file
        Args:
            rate_type (str): The rate type. Options are ORIG for origination rate and
            TERM for termination rate
            row_exclude (dict): The 
        """
        with open(self.in_file_loc, 'r') as inputfile:
            pfn, ext = os.path.splitext(self.in_file_loc)
            self.excluded_rows = []
            rows_written = 0
            with open("{}_{}.csv".format(pfn, out_file_suffix), 'w+', newline='') as outputfile:
                field_names = ['Code','Destination','Connect Cost','Included Seconds',\
                               'Per Minute Cost','Initial Increment','Increment']
                if rate_type == "TERM":
                    field_names += ['Priority','Strip','Prepend']

                if_reader = csv.DictReader(inputfile)
                out_writer = csv.DictWriter(outputfile, fieldnames = field_names, extrasaction='ignore')
                out_writer.writeheader()
                for row in if_reader:
                    if self.exclude_rules:
                        if self.exclude_filter(row):
                            continue
                    to_write = {
                            'Code': row['Prefix'],
                            'Destination': row['Named Route Name'],
                            'Connect Cost': 0,
                            'Included Seconds': 0,
                            'Per Minute Cost': str(Decimal(str(row['New Rate'])) * self.fx_rate * self.profit_margin),
                            'Initial Increment': row['Connection Unit'],
                            'Increment': row['Bill Unit']
                    }
                    if rate_type == "TERM":
                            to_write.update({
                                    'Per Minute Cost': str(Decimal(str(row['New Rate'])) * self.fx_rate),
                                    'Priority': 0,
                                    'Strip': '',
                                    'Prepend': ''
                                    })
                    rows_written += 1
                    out_writer.writerow(to_write)
            
            if self.logging_active:
                logger.info(f"{rows_written} rows written")
                logger.info(f"{len(self.excluded_rows)} rows excluded")

 if __name__ == "__main__":
    print('Starting...')
    astppImportable = ASTPPConverto(in_file_loc=r'C:\Users\dayo\Downloads\VOXBEAM_GOLD_incoming.csv', 
                                    fx_rate = 370, logging_active = True,
                                    exclude_rules=[("234","Prefix","STARTSWITH"), ("South Africa", "Named Route Name", "CONTAINS")])
    astppImportable.generate_output_file("TERM")
    print('Done!')
	import csv
	import os
	import sys
	from decimal import Decimal
	import logging

	stdout_handler = logging.StreamHandler(sys.stdout)
	logging.basicConfig(
	level=logging.DEBUG,
	format='[%(asctime)s] %(levelname)s - %(message)s',
	handlers=[stdout_handler]
	)
	logger = logging.getLogger(__name__)

	class ASTPPConverto(object):

	def __init__(self, in_file_loc = "", fx_rate = 1, profit_margin = 1, exclude_rules = [], logging_active = False):
	self.fx_rate = fx_rate
	self.profit_margin = profit_margin
	self.in_file_loc = in_file_loc
	self.exclude_rules = exclude_rules
	self.excluded_rows = []
	self.logging_active = logging_active

	def exclude_filter(self, row):
	""" Function returns false if row should be excluded, else returns true.
	Based on exclude rule
	"""
	is_a_match = False
	for rule in self.exclude_rules:
	if len(rule) != 3:
	raise ValueError("The rule requires a 3 element tuple")
	try:
	rule_value, rule_colname, rule_matchby = rule
	if rule_matchby == "CONTAINS" and (rule_value.lower() in (row[rule_colname]).lower()):
	self.excluded_rows.append(row)
	is_a_match = True
	elif rule_matchby == "STARTSWITH" and (rule_value == row[rule_colname][0:len(str(rule_value))].strip()):
	self.excluded_rows.append(row)
	is_a_match = True
	except IndexError:
	raise IndexError(f"Column name {rule_colname} does not exist")
	if self.logging_active and is_a_match:
	logger.info((" \| ").join(row.values()))
	return is_a_match

	def generate_output_file(self, rate_type, out_file_suffix = "output"):
	""" Function to generate ASTPP compatible file
	Args:
	rate_type (str): The rate type. Options are ORIG for origination rate and
	TERM for termination rate
	row_exclude (dict): The
	"""
	with open(self.in_file_loc, 'r') as inputfile:
	pfn, ext = os.path.splitext(self.in_file_loc)
	self.excluded_rows = []
	rows_written = 0
	with open("{}_{}.csv".format(pfn, out_file_suffix), 'w+', newline='') as outputfile:
	field_names = ['Code','Destination','Connect Cost','Included Seconds',\
	'Per Minute Cost','Initial Increment','Increment']
	if rate_type == "TERM":
	field_names += ['Priority','Strip','Prepend']

	if_reader = csv.DictReader(inputfile)
	out_writer = csv.DictWriter(outputfile, fieldnames = field_names, extrasaction='ignore')
	out_writer.writeheader()
	for row in if_reader:
	if self.exclude_rules:
	if self.exclude_filter(row):
	continue
	to_write = {
	'Code': row['Prefix'],
	'Destination': row['Named Route Name'],
	'Connect Cost': 0,
	'Included Seconds': 0,
	'Per Minute Cost': str(Decimal(str(row['New Rate'])) * self.fx_rate * self.profit_margin),
	'Initial Increment': row['Connection Unit'],
	'Increment': row['Bill Unit']
	}
	if rate_type == "TERM":
	to_write.update({
	'Per Minute Cost': str(Decimal(str(row['New Rate'])) * self.fx_rate),
	'Priority': 0,
	'Strip': '',
	'Prepend': ''
	})
	rows_written += 1
	out_writer.writerow(to_write)

	if self.logging_active:
	logger.info(f"{rows_written} rows written")
	logger.info(f"{len(self.excluded_rows)} rows excluded")

	if __name__ == "__main__":
	print('Starting...')
	astppImportable = ASTPPConverto(in_file_loc=r'C:\Users\dayo\Downloads\VOXBEAM_GOLD_incoming.csv',
	fx_rate = 370, logging_active = True,
	exclude_rules=[("234","Prefix","STARTSWITH"), ("South Africa", "Named Route Name", "CONTAINS")])
	astppImportable.generate_output_file("TERM")
	print('Done!')