`web_li_scrape.py`

60 lines
#!/usr/bin/env python3
"""
web_li_scrape.py — $0 HEADLESS concurrent LinkedIn-company crawler.
Fetches each company's own website (homepage + a few common pages) and extracts the
linkedin.com/company/<slug> link it publishes (footer/social bar). Real + truthful
(the company itself linked it), fast (threads), no Google/bot-block, no rate limit.
Saves data/web_li.json {domain: linkedin_company_url}.
"""
import lib, json, re, os, ssl, urllib.request
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"}
LIC=re.compile(r'linkedin\.com/company/([A-Za-z0-9\-\_%\.]+)', re.I)
def fetch(u):
    try: return urllib.request.urlopen(urllib.request.Request(u,headers=UA),timeout=8,context=ctx).read(300000).decode("utf-8","ignore")
    except Exception: return ""
def li_of(domain):
    for path in ("","/contact","/about","/about-us","/team","/company","/contact-us"):
        for sch in ("https://","http://"):
            html=fetch(sch+domain+path)
            if not html: continue
            m=LIC.search(html)
            if m:
                slug=m.group(1).rstrip("/").split("?")[0]
                if slug.lower() not in ("unavailable","company"): return "https://www.linkedin.com/company/"+slug
        if path=="": pass
    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"); qi=hdr.index("LinkedIn (Company)")
    def g(r,i): return (r[i] if i<len(r) else "").strip()
    # domains that still lack a company LinkedIn
    domains=sorted({g(rows[r],wi) for r in range(1,len(rows)) if g(rows[r],wi) and not g(rows[r],qi)})
    print(f"{len(domains)} domains to crawl for LinkedIn",flush=True)
    cache=os.path.join(os.path.dirname(__file__),"data","web_li.json")
    out=json.load(open(cache)) if os.path.exists(cache) else {}
    todo=[d for d in domains if d not in out]
    done=0
    with ThreadPoolExecutor(max_workers=40) as ex:
        for d,li in zip(todo, ex.map(li_of, todo)):
            out[d]=li; done+=1
            if done%50==0:
                json.dump(out,open(cache,"w")); found=sum(1 for v in out.values() if v)
                print(f"  {done}/{len(todo)} crawled | {found} LinkedIns",flush=True)
    json.dump(out,open(cache,"w"))
    print(f"DONE crawl: {sum(1 for v in out.values() if v)} LinkedIns / {len(out)} domains",flush=True)
    # write to master by matching each row's domain -> its LinkedIn
    rows=lib.read_tab(tok,title)
    cells=[]
    for r in range(1,len(rows)):
        dom=g(rows[r],wi); li=out.get(dom)
        if dom and li and not g(rows[r],qi):
            cells.append({"row0":r,"col0":qi,"value":'=HYPERLINK("%s")'%li,"formula":True})
    res=lib.batch_fill(tok,GID,cells)
    print(f"wrote {res.get('totalUpdatedCells',0)} company-LinkedIn cells to master",flush=True)
if __name__=="__main__": main()