96 lines
3.5 KiB
Python
Executable file
96 lines
3.5 KiB
Python
Executable file
#!/usr/bin/env python
|
|
# Script Name: sanitize-phone-data
|
|
|
|
import os
|
|
import csv
|
|
import re
|
|
import sys
|
|
from pyfiglet import Figlet
|
|
from halo import Halo
|
|
|
|
# Define the paths based on the project details
|
|
PROJECT_ROOT = "/home/ld/mgk-scrapes"
|
|
CURRENT_DATASET = os.path.join(PROJECT_ROOT, "current-data")
|
|
DATA_DIRECTORY = os.path.join(CURRENT_DATASET, ".data")
|
|
STAGE_2_DIRECTORY = os.path.join(DATA_DIRECTORY, "stage-2")
|
|
|
|
def sanitize_phone_data(file_path):
|
|
"""Sanitize phone data in the given CSV file."""
|
|
new_column_name = 'GBP Business Phone'
|
|
|
|
with open(file_path, 'r', newline='') as file:
|
|
reader = csv.DictReader(file)
|
|
rows = list(reader)
|
|
|
|
if not rows:
|
|
return
|
|
|
|
headers = reader.fieldnames
|
|
if new_column_name not in headers:
|
|
headers.append(new_column_name)
|
|
|
|
phone_patterns = [
|
|
re.compile(r'^\(\d{3}\) \d{3}-\d{4}$'), # (555) 555-5555
|
|
re.compile(r'^\+1 \d{3}-\d{3}-\d{4}$'), # +1 555-555-5555
|
|
re.compile(r'^\(\d{3}\) \d{3}-\d{4} ext\. \d+$') # (555) 555-5555 ext. 5
|
|
]
|
|
|
|
modified_rows = []
|
|
|
|
for row in rows:
|
|
if new_column_name not in row:
|
|
row[new_column_name] = ''
|
|
for key, cell in row.items():
|
|
if any(pattern.match(cell) for pattern in phone_patterns):
|
|
row[new_column_name] = cell
|
|
row[key] = ''
|
|
break
|
|
modified_rows.append(row)
|
|
|
|
for row in modified_rows:
|
|
if re.match(r'^\(\d{3}\) \d{3}-\d{4}$', row[new_column_name]):
|
|
row[new_column_name] = '+1 ' + row[new_column_name][1:4] + '-' + row[new_column_name][6:9] + '-' + row[new_column_name][10:]
|
|
elif re.match(r'^\(\d{3}\) \d{3}-\d{4} ext\. \d+$', row[new_column_name]):
|
|
row[new_column_name] = '+1 ' + row[new_column_name][1:4] + '-' + row[new_column_name][6:9] + '-' + row[new_column_name][10:14] + ' ext. ' + row[new_column_name][19:]
|
|
|
|
with open(file_path, 'w', newline='') as file:
|
|
writer = csv.DictWriter(file, fieldnames=headers)
|
|
writer.writeheader()
|
|
writer.writerows(modified_rows)
|
|
|
|
def process_csv_files(stage_directory):
|
|
"""Process all CSV files in the given stage directory to sanitize phone data."""
|
|
for state_dir in os.listdir(stage_directory):
|
|
state_path = os.path.join(stage_directory, state_dir)
|
|
if os.path.isdir(state_path):
|
|
spinner = Halo(text=f'Processing {state_dir}', spinner='dots')
|
|
spinner.start()
|
|
for county_dir in os.listdir(state_path):
|
|
county_path = os.path.join(state_path, county_dir)
|
|
if os.path.isdir(county_path):
|
|
csv_files = [file for file in os.listdir(county_path) if file.endswith('.csv')]
|
|
for file in csv_files:
|
|
file_path = os.path.join(county_path, file)
|
|
sanitize_phone_data(file_path)
|
|
spinner.succeed(f'Finished processing {state_dir}')
|
|
|
|
def main():
|
|
figlet = Figlet(font='slant')
|
|
script_name = "sanitize-phone-data".replace("-", " ").title()
|
|
print(figlet.renderText(script_name))
|
|
|
|
if not os.path.exists(STAGE_2_DIRECTORY):
|
|
print(f"Error: The directory {STAGE_2_DIRECTORY} does not exist.")
|
|
sys.exit(1)
|
|
|
|
print(f"Sanitizing phone data in {STAGE_2_DIRECTORY}...")
|
|
process_csv_files(STAGE_2_DIRECTORY)
|
|
|
|
final_spinner = Halo(spinner='dots', color='green')
|
|
final_spinner.start()
|
|
final_spinner.succeed("Phone data sanitized.")
|
|
|
|
if __name__ == "__main__":
|
|
main()
|
|
|