When literary discipline intersects with computational logic, experimentation takes on a distinctively methodical tone. That is precisely the path charted by the author behind a burgeoning venture, who personally took the helm to test, run, and benchmark an automated Python workflow developed with the assistance of artificial intelligence.
Conceived to support operations for the founder’s newly launched mini-enterprise, the project was executed inside the cloud-based Google Colaboratory environment. Rather than relying on theory, the author personally ran tests against two vast geographic landscapes: first across all 50 American states and the District of Columbia, followed by the states and mainland territories of Australia.
Stress-Testing Data Collection at Scale
The primary objective was to observe firsthand how AI-generated automation handles complex open datasets and dynamic web structures:
Dual-Continent Reach: The script systematically queries OpenStreetMap’s Overpass API for Christian places of worship with registered websites across both the United States and Australia.
Resilient Infrastructure: Network requests balance traffic across three distinct Overpass mirror endpoints to bypass regional server timeouts.
Multithreaded Inspection: An eight-worker thread pool visits each church domain, automatically parsing homepage and contact links to isolate operational email addresses while filtering out media file extensions.
Direct Cloud Export: All collected records—categorized by country, region, municipality, URL, and email—are compiled into a structured CSV file and immediately delivered for download.
The Complete Multi-Country Python Script
Run the following code directly inside a Google Colab notebook to collect directory data for both the United States and Australia:
import csv
import re
import time
import requests
from bs4 import BeautifulSoup
from urllib.parse import urljoin
from concurrent.futures import ThreadPoolExecutor
OVERPASS_SERVERS = [
"https://overpass.kumi.systems/api/interpreter",
"https://overpass-api.de/api/interpreter",
"https://maps.mail.ru/osm/tools/overpass/api/interpreter"
]
EMAIL_REGEX = r'[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+'
HEADERS = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64)"}
# Regions to process: (Country Name, List of States/Territories, OSM Admin Level)
TARGET_REGIONS = [
(
"United States",
[
"Alabama", "Alaska", "Arizona", "Arkansas", "California", "Colorado",
"Connecticut", "Delaware", "Florida", "Georgia", "Hawaii", "Idaho",
"Illinois", "Indiana", "Iowa", "Kansas", "Kentucky", "Louisiana",
"Maine", "Maryland", "Massachusetts", "Michigan", "Minnesota",
"Mississippi", "Missouri", "Montana", "Nebraska", "Nevada",
"New Hampshire", "New Jersey", "New Mexico", "New York",
"North Carolina", "North Dakota", "Ohio", "Oklahoma", "Oregon",
"Pennsylvania", "Rhode Island", "South Carolina", "South Dakota",
"Tennessee", "Texas", "Utah", "Vermont", "Virginia", "Washington",
"West Virginia", "Wisconsin", "Wyoming", "District of Columbia"
],
"4"
),
(
"Australia",
[
"New South Wales", "Victoria", "Queensland", "Western Australia",
"South Australia", "Tasmania", "Australian Capital Territory", "Northern Territory"
],
"4"
)
]
def query_overpass(query):
"""Sends query with automatic failover across mirrors."""
for server in OVERPASS_SERVERS:
try:
r = requests.post(server, data={'data': query}, timeout=120)
if r.status_code == 200:
return r.json()
except Exception:
continue
return None
def find_email_on_site(url):
"""Scrapes the domain and contact pages for valid email addresses."""
try:
res = requests.get(url, headers=HEADERS, timeout=5)
html = res.text
# Homepage regex extraction
matches = set(re.findall(EMAIL_REGEX, html))
clean = [e for e in matches if not e.lower().endswith(('.png', '.jpg', '.jpeg', '.webp', '.svg', '.js', '.css'))]
if clean:
return clean[0]
# Contact/About page fallback
soup = BeautifulSoup(html, 'html.parser')
contact_url = None
for a in soup.find_all('a', href=True):
txt = a.text.lower()
href = a['href'].lower()
if any(k in txt or k in href for k in ['contact', 'about', 'connect', 'reach']):
contact_url = urljoin(url, a['href'])
break
if contact_url:
res_c = requests.get(contact_url, headers=HEADERS, timeout=5)
matches_c = set(re.findall(EMAIL_REGEX, res_c.text))
clean_c = [e for e in matches_c if not e.lower().endswith(('.png', '.jpg', '.jpeg', '.webp', '.svg', '.js', '.css'))]
if clean_c:
return clean_c[0]
except Exception:
pass
return ""
def process_church(item, country_name, region_name):
"""Extracts metadata and initiates targeted email harvesting."""
tags = item.get("tags", {})
name = tags.get("name", "Unknown Place of Worship")
city = tags.get("addr:city", tags.get("addr:suburb", tags.get("addr:postcode", "")))
website = tags.get("website", "")
if not website:
return None
if not website.startswith("http"):
website = "http://" + website
email = find_email_on_site(website)
return [name, country_name, region_name, city, website, email]
output_file = "churches_usa_and_australia.csv"
# Initialize CSV structure
with open(output_file, mode="w", newline="", encoding="utf-8") as f:
writer = csv.writer(f)
writer.writerow(["Name", "Country", "State/Territory", "City", "Website", "Email"])
print("Starting extraction for USA and Australia...")
total_saved = 0
for country, states, admin_level in TARGET_REGIONS:
print(f"\n==================== Country: {country} ====================")
for state in states:
print(f"\n--- Processing: {state} ({country}) ---")
church_query = f"""
[out:json][timeout:90];
area["name"="{state}"]["admin_level"="{admin_level}"]->.region;
(
node["amenity"="place_of_worship"]["religion"="christian"]["website"](area.region);
way["amenity"="place_of_worship"]["religion"="christian"]["website"](area.region);
);
out tags;
"""
result = query_overpass(church_query)
if not result or "elements" not in result:
print(f" [!] Server busy or no response for {state}, skipping.")
continue
elements = result["elements"]
print(f" -> Found {len(elements)} entries with websites. Scanning domains...")
with ThreadPoolExecutor(max_workers=8) as executor:
futures = [executor.submit(process_church, item, country, state) for item in elements]
with open(output_file, mode="a", newline="", encoding="utf-8") as f:
writer = csv.writer(f)
for future in futures:
row = future.result()
if row:
writer.writerow(row)
total_saved += 1
if row[5]:
print(f" Identified: {row[0]} -> {row[5]}")
time.sleep(2) # Cooldown between regional runs
print(f"\nExecution complete. Total entries recorded: {total_saved}")
# Automated download for Google Colab users
try:
from google.colab import files
files.download(output_file)
except Exception:
print(f"Colab tool not detected. Access '{output_file}' directly from the left sidebar.")