example.py

"""Website contact scraper example: a list of domains -> contacts.csv.

Reads domains from domains.txt (one per line) or uses a small default list,
runs the rel8ble/website-contact-extractor actor on Apify, then writes:
  contacts.csv     one row per website (best email, all emails, phone, socials)
  a fill-rate summary (email / phone / LinkedIn / contact page)
  a "form only" list: sites that loaded but publish no email

Usage:
  pip install -r requirements.txt
  set APIFY_TOKEN=<YOUR_APIFY_TOKEN>   (or export on macOS/Linux)
  python example.py
"""
import csv
import os
import sys

from apify_client import ApifyClient

ACTOR_ID = "rel8ble/website-contact-extractor"
DEFAULT_DOMAINS = ["apify.com", "hetzner.com", "thomann.de", "kinsta.com", "allbirds.com"]
SOCIALS = ["linkedin", "facebook", "instagram", "twitter", "youtube", "tiktok"]


def load_domains(path="domains.txt"):
    if not os.path.exists(path):
        return DEFAULT_DOMAINS
    with open(path, encoding="utf-8") as f:
        return [line.strip() for line in f if line.strip() and not line.startswith("#")]


def main():
    token = os.environ.get("APIFY_TOKEN")
    if not token:
        sys.exit("Set APIFY_TOKEN first (Apify Console > Settings > Integrations).")

    domains = load_domains()
    client = ApifyClient(token)
    run = client.actor(ACTOR_ID).call(run_input={
        "websites": domains,
        "maxPagesPerWebsite": 6,
        "scanTextForPhones": True,
        "onlySameDomainEmails": False,
    })
    rows = list(client.dataset(run["defaultDatasetId"]).iterate_items())

    ok = [r for r in rows if r.get("status") == "ok"]
    failed = [r for r in rows if r.get("status") != "ok"]

    fields = ["domain", "companyName", "email", "emails", "phone", "phones",
              *SOCIALS, "contactPageUrl", "pagesCrawled"]
    with open("contacts.csv", "w", newline="", encoding="utf-8") as f:
        w = csv.DictWriter(f, fieldnames=fields)
        w.writeheader()
        for r in ok:
            row = {k: r.get(k) for k in fields}
            row["emails"] = "; ".join(r.get("emails") or [])
            row["phones"] = "; ".join(r.get("phones") or [])
            w.writerow(row)

    def rate(key):
        return f"{sum(1 for r in ok if r.get(key)) / max(len(ok), 1):.0%}"

    print(f"{len(ok)} websites loaded, {len(failed)} failed (failed rows are not charged)")
    print(f"email {rate('email')} | phone {rate('phone')} | "
          f"LinkedIn {rate('linkedin')} | contact page {rate('contactPageUrl')}")

    form_only = [r for r in ok if not r.get("email") and r.get("contactPageUrl")]
    if form_only:
        print("\nNo published email (use the contact form or the phone):")
        for r in form_only:
            print(f"  {r['domain']:<28} {r.get('phone') or '-':<18} {r['contactPageUrl']}")
    for r in failed:
        print(f"  FAILED {r.get('website')}: {r.get('error')}")
    print("\nWrote contacts.csv")


if __name__ == "__main__":
    main()