Bulk LinkedIn Enrichment

import csv

def normalize(s):
    return (s or "").strip().lower()

seen, clean_rows = set(), []
with open("leads_raw.csv", newline="") as f:
    for row in csv.DictReader(f):
        name, company = row.get("full_name", ""), row.get("company", "")
        if not name.strip():
            continue                         # nothing to match on
        key = (normalize(name), normalize(company))
        if key in seen:
            continue                         # duplicate -> skip
        seen.add(key)
        clean_rows.append({"full_name": name.strip(), "company": company.strip()})

with open("leads.csv", "w", newline="") as f:
    w = csv.DictWriter(f, fieldnames=["full_name", "company"])
    w.writeheader(); w.writerows(clean_rows)

print(f"Kept {len(clean_rows)} unique leads.")

import os, requests

API_KEY = os.environ["LINKFINDER_API_KEY"]
BASE_URL = "https://api.linkfinderai.com"

def call(enrichment_type, input_data, **extra):
    """One request = one credit. Returns result or None."""
    resp = requests.post(
        BASE_URL,
        headers={"Authorization": f"Bearer {API_KEY}",
                 "Content-Type": "application/json"},
        json={"type": enrichment_type, "input_data": input_data, **extra},
        timeout=30,
    )
    resp.raise_for_status()
    data = resp.json()
    return data.get("result") if data.get("status") == "success" else None

def enrich_lead(full_name, company):
    lead = {"full_name": full_name, "company": company}
    url = call("lead_full_name_to_linkedin_url", f"{full_name} {company}")
    if not url:
        return lead                         # nothing to chain off of
    lead["linkedin"] = url
    lead["email"]   = call("linkedin_profile_to_email", url)
    lead["phone"]   = call("linkedin_profile_to_phone", url)
    lead["website"] = call("company_name_to_website", company)
    return lead

print(enrich_lead("Sarah Mitchell", "CloudCore"))

import csv, time

INPUT, OUTPUT = "leads.csv", "leads_enriched.csv"
REQS_PER_SEC = 5            # match your plan (Starter = 5/s)
DELAY = 1.0 / REQS_PER_SEC

with open(INPUT, newline="") as f_in, open(OUTPUT, "w", newline="") as f_out:
    reader = csv.DictReader(f_in)
    fields = ["full_name", "company", "linkedin", "email", "phone", "website"]
    writer = csv.DictWriter(f_out, fieldnames=fields, extrasaction="ignore")
    writer.writeheader()

    for i, row in enumerate(reader, 1):
        lead = enrich_lead(row["full_name"], row["company"])
        writer.writerow(lead)
        f_out.flush()                       # persist as you go
        print(f"[{i}] {lead['full_name']} -> {lead.get('email')}")
        time.sleep(DELAY)                   # stay under the rate limit

print("Done.")

import csv, time, threading
from concurrent.futures import ThreadPoolExecutor

REQS_PER_SEC = 5            # your plan's ceiling
_lock = threading.Lock()
_last = [0.0]

def rate_limited():
    """Simple shared throttle across worker threads."""
    with _lock:
        wait = (1.0 / REQS_PER_SEC) - (time.time() - _last[0])
        if wait > 0:
            time.sleep(wait)
        _last[0] = time.time()

def worker(row):
    rate_limited()
    return enrich_lead(row["full_name"], row["company"])

with open("leads.csv", newline="") as f:
    rows = list(csv.DictReader(f))

with open("leads_enriched.csv", "w", newline="") as f_out, \
     ThreadPoolExecutor(max_workers=REQS_PER_SEC) as pool:
    fields = ["full_name", "company", "linkedin", "email", "phone", "website"]
    writer = csv.DictWriter(f_out, fieldnames=fields, extrasaction="ignore")
    writer.writeheader()
    for lead in pool.map(worker, rows):
        writer.writerow(lead); f_out.flush()

print("Done (concurrent).")

import os, csv, time, requests

DONE_FILE = "done_keys.txt"

def load_done():
    if not os.path.exists(DONE_FILE):
        return set()
    with open(DONE_FILE) as f:
        return set(line.strip() for line in f)

def mark_done(key):
    with open(DONE_FILE, "a") as f:
        f.write(key + "\n")

def call(enrichment_type, input_data, max_retries=4, **extra):
    delay = 1.0
    for _ in range(max_retries):
        resp = requests.post(
            BASE_URL,
            headers={"Authorization": f"Bearer {API_KEY}",
                     "Content-Type": "application/json"},
            json={"type": enrichment_type, "input_data": input_data, **extra},
            timeout=30,
        )
        if resp.status_code in (429, 500):
            time.sleep(delay); delay *= 2      # 1s -> 2s -> 4s -> 8s
            continue
        if resp.status_code == 402:
            raise RuntimeError("Out of credits — top up to resume.")
        resp.raise_for_status()
        data = resp.json()
        return data.get("result") if data.get("status") == "success" else None
    raise RuntimeError("Exhausted retries after repeated rate limiting.")

done = load_done()
with open("leads.csv", newline="") as f:
    for row in csv.DictReader(f):
        key = f"{row['full_name']}|{row['company']}".lower()
        if key in done:
            continue                          # already enriched -> skip
        lead = enrich_lead(row["full_name"], row["company"])
        # ...write lead to output...
        mark_done(key)

rows = 12_000          # unique leads after de-duping
calls_per_lead = 4     # url + email + phone + website
worst_case = rows * calls_per_lead
print(f"Upper bound: {worst_case:,} credits")
# Real spend is lower: chains that miss the LinkedIn URL
# short-circuit and never make the 3 downstream calls.

# Zapier: "Webhooks by Zapier" -> POST action
URL:     https://api.linkfinderai.com
Method:  POST
Headers: Authorization: Bearer YOUR_API_KEY
         Content-Type: application/json
Body:    {"type": "lead_full_name_to_linkedin_url",
          "input_data": "{{full_name}} {{company}}"}