`phone_scrape.py`

71 lines
#!/usr/bin/env python3
"""
phone_scrape.py — $0 company-phone harvester.

Reads the master tab's Website column, fetches each unique company domain (homepage,
then /contact), extracts the main phone number (tel: links preferred, else the most
repeated US phone pattern — footer numbers repeat), and saves data/phones.json.
Concurrent (threads). No API cost. A separate step writes results to the sheet.
"""
import lib, enrich, json, re, os, urllib.request, ssl
from concurrent.futures import ThreadPoolExecutor

GID = 3823360
ctx = ssl.create_default_context(); ctx.check_hostname=False; ctx.verify_mode=ssl.CERT_NONE
UA = {"User-Agent":"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15) AppleWebKit/537.36"}
TEL = re.compile(r'tel:\+?1?[-.\s]?\(?(\d{3})\)?[-.\s]?(\d{3})[-.\s]?(\d{4})')
PHONE = re.compile(r'\(?\b(\d{3})\)?[-.\s]?(\d{3})[-.\s]?(\d{4})\b')
BAD = {"0000000000","1111111111","1234567890","8888888888"}

def fetch(url):
    try:
        req=urllib.request.Request(url, headers=UA)
        return urllib.request.urlopen(req, timeout=8, context=ctx).read(200000).decode("utf-8","ignore")
    except Exception:
        return ""

def phone_of(domain):
    for path in ("", "/contact", "/contact-us", "/about"):
        for scheme in ("https://","http://"):
            html=fetch(scheme+domain+path)
            if not html: continue
            m=TEL.search(html)
            if m:
                p="".join(m.groups())
                if p not in BAD and p[0] not in "01": return "-".join(m.groups())
            cands=["".join(g) for g in PHONE.findall(html)]
            cands=[c for c in cands if c not in BAD and c[0] not in "01" and c[3] not in "01"]
            if cands:
                best=max(set(cands), key=cands.count)  # most-repeated = footer/main line
                return f"{best[:3]}-{best[3:6]}-{best[6:]}"
        # only try /contact paths on the primary scheme round; break after homepage attempts
    return ""

def main():
    tok=lib.access_token()
    title={s["properties"]["sheetId"]:s["properties"]["title"] for s in lib.get_meta(tok)["sheets"]}[GID]
    rows=lib.read_tab(tok,title); hdr=rows[0]
    wi=hdr.index("Website")
    def g(r,i): return (r[i] if i<len(r) else "").strip()
    domains=sorted({g(rows[r],wi) for r in range(1,len(rows)) if g(rows[r],wi)})
    print(f"{len(domains)} unique domains to scrape", flush=True)
    out={}
    cache_path=os.path.join(os.path.dirname(__file__),"data","phones.json")
    if os.path.exists(cache_path): out=json.load(open(cache_path))
    todo=[d for d in domains if d not in out]
    print(f"{len(todo)} remaining (cache has {len(out)})", flush=True)
    done=0
    with ThreadPoolExecutor(max_workers=24) as ex:
        for d,ph in zip(todo, ex.map(phone_of, todo)):
            out[d]=ph; done+=1
            if done%50==0:
                json.dump(out,open(cache_path,"w"))
                found=sum(1 for v in out.values() if v)
                print(f"  {done}/{len(todo)} scraped | {found} phones so far", flush=True)
    json.dump(out,open(cache_path,"w"))
    found=sum(1 for v in out.values() if v)
    print(f"DONE: {len(out)} domains, {found} phones found -> data/phones.json", flush=True)

if __name__=="__main__":
    main()