A python script to convert voxbeam rates to form that can be imported into astpp

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!')