#!/usr/bin/env python3 """Everything needed to profile the seed customers: what GetLeads records about each one, plus its homepage as plain text. profile_seeds.py seeds.csv [--scrape-cmd 'firecrawl scrape {url} --format markdown'] > seeds_profile.json Read each profile, then assign the LinkedIn industry the website says the company belongs in and write the keyword list. A page a plain fetch can't read (a bot wall, a JavaScript site) goes through --scrape-cmd, the user's scraping tool. `homepage_status` other than live, or still-thin `homepage_text`, means read the site with a browser instead. """ from __future__ import annotations import argparse import json import sys from concurrent.futures import ThreadPoolExecutor from pathlib import Path from clean_companies import SCRAPE_CMD_HELP, THIN_PAGE_CHARS, fetch_homepage, merge, scrape_page def profile(path: Path, scrape_cmd: str | None = None) -> list[dict[str, str]]: seeds, _ = merge([path], label="") def one(item: tuple[str, dict[str, str]]) -> dict[str, str]: key, seed = item url = seed.get("website") or ("" if ":" in key else key) status, text = fetch_homepage(url) if url and scrape_cmd and status in ("live", "error") and len(text) < THIN_PAGE_CHARS: scraped = scrape_page(url, scrape_cmd) if scraped[0] != "error": status, text = scraped[:2] recorded = {k: seed[k] for k in ("name", "domain", "industry_filed", "industry", "headcount", "hq", "summary", "about") if seed.get(k)} return {**recorded, "homepage_status": status, "homepage_text": text} with ThreadPoolExecutor(max_workers=8) as pool: return list(pool.map(one, seeds.items())) if __name__ == "__main__": parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) parser.add_argument("seeds", type=Path) parser.add_argument("--scrape-cmd", help=SCRAPE_CMD_HELP) args = parser.parse_args() json.dump(profile(args.seeds, args.scrape_cmd), sys.stdout, indent=2, ensure_ascii=False)