from config import Config
from airflow.hooks.base import BaseHook
import io
import json
from config.fips import FipsCodes
"""This util file is used to clean the data before it is loaded into the data lake.
1. Remove special characters from the company names
"""
[docs]
class CleanDataUtil:
"""A series of functions to clean the data before it is loaded into the data lake"""
[docs]
@staticmethod
def clean_company_names(
option: str = "all", special_char: str = """!#$%^*".,()-"""
) -> str:
"""
This function is returns a sql string that you can add to the sql query
to remove those special characters in a company name. Default removing all special characters.
:param: option (str): the option to remove the special characters from the company name. The options are:
- company_suffix: remove the company suffixes like LLP, LLC, LTD, LP, INC, CO, IL, LIMITED
- common_words: remove the company suffixes and state abbreviations
- all: remove all above mentioned character and the words like COMMUNICATIONS, COMMUNICATION, COOPERATIVE, CORPORATION, TELECOMMUNICATIONS, COMM, TELECOM, TELCOM, TELEPHONE, FIBER, INTERNET, COOP, COMPANY, NETWORK, NETWORKS, ELECTRIC, POWER, WIRELESS, TELEVISION, CABLE, ELECTRICITY, CONNECT, CONNECTION
:param: special_char (str): the special characters to remove from the company name. Default: !#$%^*".,()-
:return: a function called clean_company_names that removes special characters from a company name
"""
# Remove all state abbrevation as well
STATE_ABBR = [k for k, v in FipsCodes.STATES.items()]
company_suffix_ori = [
"LLP",
"LLC",
"LTD",
"LP",
"INC",
"CO",
"IL",
"LIMITED",
"Incorporated",
]
company_suffix = [x + "." for x in company_suffix_ori] + company_suffix_ori
common_words = [
"COMMUNICATIONS",
"COMMUNICATION",
"COOPERATIVE",
"CORPORATION",
"TELECOMMUNICATIONS",
"COMM",
"TELECOM",
"TELCOM",
"TELEPHONE",
"FIBER",
"INTERNET",
"COOP",
"COMPANY",
"NETWORK",
"NETWORKS",
"ELECTRIC",
"POWER",
"WIRELESS",
"TELEVISION",
"CABLE",
"ELECTRICITY",
"CONNECT",
"CONNECTION",
]
if option == "company_suffix":
to_remove = company_suffix
elif option == "common_words":
to_remove = STATE_ABBR + company_suffix
elif option == "all":
to_remove = STATE_ABBR + company_suffix + common_words
else:
raise ValueError(
"Invalid option. The options are: company_suffix, common_words, all"
)
layer_0 = """trim(regexp_replace({input_text}, '[{special_char}]', ''))"""
construct_full_line = layer_0.format(
special_char=special_char, input_text="{input_text}"
)
for i in to_remove:
construct_full_line = (
f"""trim(regexp_replace({construct_full_line}, '(?i){" "+i}$', ''))"""
)
return construct_full_line