← back to Cslb Call List

scripts/enrich_web.py

137 lines

#!/usr/bin/env python3
import csv, re, sys, json, time, urllib.request, urllib.parse, ssl, itertools, threading
from concurrent.futures import ThreadPoolExecutor, as_completed

INP=sys.argv[1]; OUT=sys.argv[2]; LIMIT=int(sys.argv[3]) if len(sys.argv)>3 else 0
CTX=ssl.create_default_context(); CTX.check_hostname=False; CTX.verify_mode=ssl.CERT_NONE
UA="Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 Chrome/126.0 Safari/537.36"
EMAIL=re.compile(r'[A-Za-z0-9._%+\-]+@[A-Za-z0-9.\-]+\.[A-Za-z]{2,}')
BADMAIL=re.compile(r'(example|sentry|wixpress|\.png|\.jpg|\.gif|@2x|godaddy|domain\.com|email\.com|yourdomain)',re.I)
SUFFIX=re.compile(r'\b(inc|llc|corp|co|company|the|and|&|ltd|group|services|service|construction|contractors?|builders?)\b',re.I)

# local LLM endpoints (round-robin) — "max exo": exo in the rotation
LLM=[("ollama","http://192.168.1.133:11434/api/generate","qwen2.5:7b"),
     ("ollama","http://localhost:11434/api/generate","qwen3:14b"),
     ("ollama","http://192.168.1.133:11434/api/generate","qwen2.5:7b")]
_rr=itertools.cycle(LLM); _lock=threading.Lock()
def next_llm():
    with _lock: return next(_rr)

def cand_domains(name):
    n=name.lower()
    n=re.sub(r"[.,'/]",' ',n); n=n.replace('&',' and ')
    words=[w for w in re.split(r'\s+',n) if w]
    core=[w for w in words if not SUFFIX.fullmatch(w)]
    joins=set()
    if core:
        joins.add(''.join(core))
        joins.add(''.join(core[:2]))
        joins.add(''.join(core[:3]))
        if len(core)>=2: joins.add(core[0]+core[-1])
        joins.add(core[0])
        joins.add('-'.join(core[:2]))
    doms=[]
    for j in joins:
        j=re.sub(r'[^a-z0-9\-]','',j)
        if len(j)<3: continue
        for tld in ('.com','.net'):
            doms.append(j+tld)
    seen=set(); out=[]
    for d in doms:
        if d not in seen: seen.add(d); out.append(d)
    return out[:8]

def fetch(url,timeout=6):
    try:
        req=urllib.request.Request(url,headers={'User-Agent':UA})
        with urllib.request.urlopen(req,timeout=timeout,context=CTX) as r:
            ct=r.headers.get('Content-Type','')
            if 'html' not in ct and 'text' not in ct: return None,None
            return r.geturl(), r.read(200000).decode('utf-8','ignore')
    except Exception: return None,None

def phone_digits(p): 
    d=re.sub(r'\D','',p or ''); return d[-10:] if len(d)>=10 else ''

def pick_email(html,dom):
    hits=[e for e in EMAIL.findall(html) if not BADMAIL.search(e)]
    if not hits: return ''
    base=dom.split('.')[0]
    hits.sort(key=lambda e:(0 if base in e.lower() else 1, len(e)))
    return hits[0]

def llm_judge(name,city,text):
    kind,url,model=next_llm()
    prompt=(f'Business: "{name}" in {city}, California.\nPage text (truncated):\n{text[:1500]}\n\n'
            'Does this web page belong to THAT business? Reply ONLY compact JSON: '
            '{"belongs":true|false,"email":"<best contact email or empty>"}')
    try:
        if kind=="ollama":
            body=json.dumps({"model":model,"prompt":prompt,"stream":False,"format":"json",
                             "options":{"temperature":0,"num_predict":80}}).encode()
            req=urllib.request.Request(url,data=body,headers={'Content-Type':'application/json'})
            with urllib.request.urlopen(req,timeout=45) as r:
                out=json.loads(r.read())['response']
        else:
            body=json.dumps({"model":model,"messages":[{"role":"user","content":prompt}],
                             "temperature":0,"max_tokens":80}).encode()
            req=urllib.request.Request(url,data=body,headers={'Content-Type':'application/json'})
            with urllib.request.urlopen(req,timeout=60) as r:
                out=json.loads(r.read())['choices'][0]['message']['content']
        m=re.search(r'\{.*\}',out,re.S); j=json.loads(m.group(0)) if m else {}
        return bool(j.get('belongs')), (j.get('email') or '')
    except Exception: return None,''

def enrich(row):
    name=row['business_name']; city=row.get('city',''); ph=phone_digits(row.get('business_phone',''))
    for dom in cand_domains(name):
        for base in (f"https://{dom}/", f"https://www.{dom}/"):
            final,html=fetch(base)
            if not html: continue
            pages=[html]
            for path in ('contact','contact-us','about'):
                _,h2=fetch(base+path); 
                if h2: pages.append(h2)
            blob=' '.join(pages); digits=re.sub(r'\D','',blob)
            phone_match = ph and ph in digits
            email=pick_email(blob,dom)
            if phone_match:
                row.update(website=f"https://{dom}",email=email,enrich_confidence='high(phone-match)',enrich_source=dom)
                return row
            # ambiguous 200 page -> local LLM adjudication
            belongs,lem=llm_judge(name,city,blob)
            if belongs:
                row.update(website=f"https://{dom}",email=(email or lem),enrich_confidence='med(llm)',enrich_source=dom)
                return row
    row.update(website='',email='',enrich_confidence='none',enrich_source='')
    return row

def main():
    rows=list(csv.DictReader(open(INP,encoding='utf-8')))
    if LIMIT: rows=rows[:LIMIT]
    cols=list(rows[0].keys())
    for c in ('enrich_confidence','enrich_source'):
        if c not in cols: cols.append(c)
    import os
    already=set()
    if os.path.exists(OUT):
        for r in csv.DictReader(open(OUT,encoding='utf-8')): already.add(r['license_no'])
    rows=[r for r in rows if r['license_no'] not in already]
    print(f"resume: {len(already)} already done, {len(rows)} remaining",flush=True)
    done=0; hits=0; emails=0; t0=time.time()
    newfile=not os.path.exists(OUT)
    fh=open(OUT,'a',newline='',encoding='utf-8'); w=csv.DictWriter(fh,fieldnames=cols)
    if newfile: w.writeheader()
    with ThreadPoolExecutor(max_workers=32) as ex:
        futs={ex.submit(enrich,r):r for r in rows}
        for fut in as_completed(futs):
            r=fut.result(); w.writerow(r); done+=1
            if r.get('website'): hits+=1
            if r.get('email'): emails+=1
            if done%25==0:
                fh.flush()
                print(f"  {done}/{len(rows)} | sites={hits} emails={emails} | {done/max(time.time()-t0,1):.1f}/s",flush=True)
    fh.close()
    print(f"DONE {done} | websites={hits} ({100*hits//max(done,1)}%) | emails={emails} ({100*emails//max(done,1)}%) | {int(time.time()-t0)}s",flush=True)
main()