← back to Marketing Command Center
scripts/crawl-csv-emails.py
150 lines
#!/usr/bin/env python3
"""
crawl-csv-emails.py — LOCAL, free email enrichment for a Google-Places CSV.
For each business row with a website: fetch the homepage, discover a
contact/about page, and scrape the best contact email. Writes a staged
"prospect group" JSON that the MCC Clients panel reads. Nothing is uploaded
to Constant Contact — this only stages a local dataset.
python3 scripts/crawl-csv-emails.py <input.csv> --slug la-architects [--limit 40] [--workers 12]
Output: public/data/prospects-<slug>.json
"""
import csv, sys, os, re, json, argparse, urllib.request, urllib.parse, datetime
from concurrent.futures import ThreadPoolExecutor, as_completed
csv.field_size_limit(10_000_000)
UA = "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124 Safari/537.36"
EMAIL_RE = re.compile(r"[A-Za-z0-9._%+\-]+@[A-Za-z0-9.\-]+\.[A-Za-z]{2,}")
JUNK_DOMAINS = ("sentry", "example.", "wix.com", "wixpress", "godaddy", "squarespace",
"schema.org", "w3.org", "googleapis", "gstatic", "jquery", "cloudflare",
"fontawesome", "sentry.io", "@2x", "@sentry", "core.min", "webpack",
"domain.com", "yourdomain", "yoursite", "email.com", "sentry-next",
"test.com", "mydomain")
JUNK_LOCAL = ("no-reply", "noreply", "donotreply", "user", "name", "email", "yourname",
"firstname", "lastname", "your-email", "youremail", "example", "sentry")
ROLE_PREFIXES = ("info", "contact", "hello", "sales", "studio", "design", "office", "admin")
CONTACT_HINTS = ("contact", "about", "team", "connect", "reach", "hello")
def norm_url(u):
u = (u or "").strip()
if not u: return None
if not u.startswith(("http://", "https://")): u = "http://" + u
return u
def host_of(u):
try: return urllib.parse.urlparse(u).netloc.lower().replace("www.", "")
except Exception: return ""
def fetch(url, timeout=8):
try:
req = urllib.request.Request(url, headers={"User-Agent": UA, "Accept": "text/html"})
with urllib.request.urlopen(req, timeout=timeout) as r:
ct = r.headers.get("Content-Type", "")
if "html" not in ct and "text" not in ct: return "", r.geturl()
return r.read(600_000).decode("utf-8", "replace"), r.geturl()
except Exception:
return "", url
def valid_email(e, site_host):
el = e.lower()
if el.count("@") != 1: return False
if any(j in el for j in JUNK_DOMAINS): return False
if el.split("@")[0] in JUNK_LOCAL: return False
if re.search(r"\.(png|jpg|jpeg|gif|svg|webp|css|js)$", el): return False
if len(el) > 60: return False
return True
def score_email(e, site_host):
el = e.lower(); dom = el.split("@")[1]; local = el.split("@")[0]
s = 0
if site_host and (dom == site_host or dom.endswith("." + site_host) or site_host.endswith("." + dom)): s += 10
if any(local.startswith(p) for p in ROLE_PREFIXES): s += 3
if dom.endswith((".com", ".net", ".co", ".studio", ".design")): s += 1
return s
def find_contact_links(html, base):
links = set()
for m in re.finditer(r'href=["\']([^"\']+)["\']', html, re.I):
href = m.group(1)
if any(h in href.lower() for h in CONTACT_HINTS):
links.add(urllib.parse.urljoin(base, href))
return list(links)[:3]
def emails_from(html, site_host):
found = set()
for m in re.finditer(r'mailto:([^"\'?>]+)', html, re.I):
found.add(m.group(1).strip())
for m in EMAIL_RE.finditer(html):
found.add(m.group(0))
return [e for e in found if valid_email(e, site_host)]
def enrich(row):
website = norm_url(row.get("website"))
biz = {"title": (row.get("title") or "").strip(),
"city": (row.get("city") or "").strip(),
# richer location fields kept from the Google-Places export so the
# Clients panel can show a full address + Maps link (see
# scripts/backfill-address-from-csv.py which backfills older JSONs).
"address": (row.get("address") or "").strip(),
"state": (row.get("state") or "").strip(),
"postal": (row.get("postalCode") or "").strip(),
"neighborhood": (row.get("neighborhood") or "").strip(),
"category": (row.get("categoryName") or "").strip(),
"mapsUrl": (row.get("url") or "").strip(),
"phone": (row.get("phone") or "").strip(),
"phoneUnformatted": (row.get("phoneUnformatted") or "").strip(),
"website": website, "email": "", "emails": []}
if not website: return biz
site_host = host_of(website)
html, final = fetch(website)
site_host = host_of(final) or site_host
emails = set(emails_from(html, site_host))
if len(emails) == 0 and html:
for link in find_contact_links(html, final):
chtml, _ = fetch(link)
emails |= set(emails_from(chtml, site_host))
if emails: break
ranked = sorted(emails, key=lambda e: score_email(e, site_host), reverse=True)
biz["emails"] = ranked[:5]
biz["email"] = ranked[0] if ranked else ""
return biz
def main():
ap = argparse.ArgumentParser()
ap.add_argument("csv"); ap.add_argument("--slug", required=True)
ap.add_argument("--limit", type=int, default=0); ap.add_argument("--workers", type=int, default=12)
ap.add_argument("--out-dir", default=os.path.join(os.path.dirname(__file__), "..", "public", "data"))
a = ap.parse_args()
rows = []
with open(a.csv, newline="", encoding="utf-8", errors="replace") as fh:
for row in csv.DictReader(fh):
if (row.get("website") or "").strip():
rows.append(row)
if a.limit: rows = rows[:a.limit]
print(f"[{a.slug}] {len(rows)} businesses with a website to crawl · workers={a.workers}", flush=True)
out, done, hits = [], 0, 0
with ThreadPoolExecutor(max_workers=a.workers) as ex:
futs = {ex.submit(enrich, r): r for r in rows}
for f in as_completed(futs):
b = f.result(); out.append(b); done += 1
if b["email"]: hits += 1
if done % 25 == 0 or done == len(rows):
print(f"\r {done}/{len(rows)} crawled · {hits} emails found", end="", flush=True)
print()
os.makedirs(a.out_dir, exist_ok=True)
path = os.path.join(a.out_dir, f"prospects-{a.slug}.json")
payload = {"source": os.path.basename(a.csv), "group": a.slug,
"updated": datetime.date.today().isoformat(),
"total": len(out), "with_email": hits, "businesses": out}
with open(path, "w") as fh:
json.dump(payload, fh)
print(f"✓ wrote {path} — {len(out)} businesses, {hits} with an email ({100*hits//max(1,len(out))}%) [$0 local]")
if __name__ == "__main__":
main()