← 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()