116 lines
4.9 KiB
Python
Executable file
116 lines
4.9 KiB
Python
Executable file
#!/usr/bin/env python
|
|
# Script Name: strip-neg-patterns-from-unknown-cols
|
|
import os
|
|
import csv
|
|
import re
|
|
from pyfiglet import Figlet
|
|
from halo import Halo
|
|
|
|
def find_project_root(current_path):
|
|
"""Find the project root by locating the 'bin' directory."""
|
|
while current_path != os.path.dirname(current_path):
|
|
if os.path.basename(current_path) == 'bin':
|
|
return os.path.dirname(current_path)
|
|
current_path = os.path.dirname(current_path)
|
|
raise FileNotFoundError("Could not find 'bin' directory in the path hierarchy.")
|
|
|
|
# Define the paths based on the project details
|
|
script_path = os.path.abspath(__file__)
|
|
PROJECT_ROOT = find_project_root(script_path)
|
|
DATA_DIRECTORY = os.path.join(PROJECT_ROOT, "current-data", ".data")
|
|
STAGE_4_DIRECTORY = os.path.join(DATA_DIRECTORY, "stage-4")
|
|
|
|
# Define the pattern for foreign phone numbers
|
|
foreign_phone_patterns = [
|
|
re.compile(r'^\+\d{1,3}\s?\d{1,4}[\s.-]?\d{1,4}[\s.-]?\d{1,4}$'), # General pattern for most international numbers
|
|
re.compile(r'^\+\d{2,3}\s\d{1,2}\s\d{2,4}\s\d{2,4}$'), # Pattern for numbers with multiple spaces
|
|
re.compile(r'^\+\d{2,3}\s\d{3,4}-\d{3,4}-\d{3,4}$'), # Pattern for numbers with hyphens
|
|
re.compile(r'^\+\d{2,3}\s\d{2}\s\d{3}\s\d{2}\s\d{2}$'), # Pattern for numbers with multiple groups
|
|
re.compile(r'^\+\d{2,3}\s\d{3,4}\s\d{5,6}$'), # Pattern for numbers with two groups after country code
|
|
re.compile(r'^\+\d{1,3}\s\d{9,11}$'), # Pattern for numbers with country code and 9-11 digits
|
|
re.compile(r'^\+\d{2,3}\s\d{1,2}\s\d{4}\s\d{4}$'), # Pattern for numbers like +xx x xxxx xxxx
|
|
re.compile(r'^\+\d{2}\s\d\s\d{2}\s\d{2}\s\d{2}\s\d{2}$'), # Pattern for +33 1 72 67 69 14
|
|
re.compile(r'^\+\d{2}\s\d{3}\s\d{2}\s\d{2}\s\d{2}$'), # Pattern for +34 651 32 11 25
|
|
re.compile(r'^\+\d{2}\s\d{2}\s\d{2}\s\d{2}\s\d{2}$'), # Pattern for +46 31 23 06 50
|
|
re.compile(r'^\+91\s\d{5}\s\d{5}$'), # Pattern for Indian numbers like +91 62917 98998
|
|
]
|
|
|
|
# Define the pattern for websites to remove
|
|
website_patterns = [
|
|
re.compile(r'https://www\.google\.com/maps')
|
|
]
|
|
|
|
def process_csv_file(file_path):
|
|
"""Process a CSV file to remove negative patterns."""
|
|
with open(file_path, 'r') as file:
|
|
reader = csv.reader(file)
|
|
rows = list(reader)
|
|
|
|
if not rows:
|
|
return 0, 0
|
|
|
|
headers = rows[0]
|
|
data_rows = rows[1:]
|
|
modified_rows = [headers]
|
|
terms_removed = 0
|
|
websites_removed = 0
|
|
|
|
for row in data_rows:
|
|
for i, header in enumerate(headers):
|
|
if header.startswith("unknown-"):
|
|
# Check for phone numbers
|
|
for pattern in foreign_phone_patterns:
|
|
if pattern.match(row[i]):
|
|
row[i] = '' # Clear the cell if it contains a foreign phone number
|
|
terms_removed += 1
|
|
break
|
|
|
|
# Check for websites
|
|
for pattern in website_patterns:
|
|
if pattern.search(row[i]):
|
|
row[i] = '' # Clear the cell if it contains a matching website
|
|
websites_removed += 1
|
|
break
|
|
|
|
modified_rows.append(row)
|
|
|
|
with open(file_path, 'w', newline='') as file:
|
|
writer = csv.writer(file)
|
|
writer.writerows(modified_rows)
|
|
|
|
return terms_removed, websites_removed
|
|
|
|
def strip_negative_patterns():
|
|
"""Strip negative patterns from all CSV files in the stage 4 directory and tally the results."""
|
|
figlet = Figlet(font='slant')
|
|
print(figlet.renderText('Strip Negative Patterns'))
|
|
|
|
total_terms_removed = 0
|
|
total_websites_removed = 0
|
|
|
|
for state_dir in os.listdir(STAGE_4_DIRECTORY):
|
|
state_path = os.path.join(STAGE_4_DIRECTORY, state_dir)
|
|
if os.path.isdir(state_path):
|
|
spinner = Halo(text=f'Processing {state_dir}', spinner='dots')
|
|
spinner.start()
|
|
state_terms_removed = 0
|
|
state_websites_removed = 0
|
|
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)
|
|
terms, websites = process_csv_file(file_path)
|
|
state_terms_removed += terms
|
|
state_websites_removed += websites
|
|
total_terms_removed += state_terms_removed
|
|
total_websites_removed += state_websites_removed
|
|
spinner.succeed(f'Finished processing {state_dir}. Terms removed: {state_terms_removed}, Websites removed: {state_websites_removed}')
|
|
|
|
print(f"Total Terms Removed: {total_terms_removed}")
|
|
print(f"Total Websites Removed: {total_websites_removed}")
|
|
|
|
if __name__ == "__main__":
|
|
strip_negative_patterns()
|