Research 500 companies into your CRM

import { readFile, writeFile } from 'node:fs/promises';
import {
  ConcurrencyLimitError,
  NoCapacityError,
  Outcrawl,
  QuotaExceededError,
  type CrawlPage,
} from '@outcrawl/sdk';

const oc = new Outcrawl();

// Every key on your account shares one browser slot count (Free 2, Builder 5,
// Pro 25, Scale 100). Keep LANES below yours so other work still gets a slot.
const LANES = 4;
const BATCH_CEILING = 10_000; // credits, for the whole list
const SEARCH_RESULTS = 5; // 5 credits per 10 results
const CRAWL_PAGES = 8; // 1 credit per page, and the crawl's hard limit
const AGENT_BUDGET = '40.00'; // about 2.5 minutes of a typical run
const DECIDES = /founder|chief|\bce?o\b|\bc[tfor]o\b|president|\bvp\b|head of|director/i;

interface Plan { name: string; price: string }
interface Person { name: string; title: string }
interface Facts { whatTheyDo?: string; plans?: Plan[]; people?: Person[]; openRoles?: number; teamSize?: string }
interface Company {
  name: string;
  domain: string;
  whatTheyDo: string;
  plans: Plan[];
  decisionMakers: Person[];
  openRoles: number | null;
  teamSize: string | null;
  news: { title: string; url: string }[];
}

const PLAN = { type: 'object', properties: { name: { type: 'string' }, price: { type: 'string' } }, required: ['name', 'price'] };
const PERSON = { type: 'object', properties: { name: { type: 'string' }, title: { type: 'string' } }, required: ['name', 'title'] };
const PAGE_SCHEMA = {
  type: 'object',
  properties: {
    whatTheyDo: { type: 'string', description: 'One sentence: what the company sells, and to whom. Empty if this page does not say.' },
    plans: { type: 'array', items: PLAN, description: 'Each plan with its price as shown, e.g. "$49 per seat per month" or "Contact sales"' },
    people: { type: 'array', items: PERSON, description: 'People this page names, with their job titles' },
    openRoles: { type: 'integer', description: 'How many open roles this page lists, if it is a careers page' },
    teamSize: { type: 'string', description: 'Any statement of team size, e.g. "a team of 40 across Berlin and Lisbon"' },
  },
};

const sleep = (ms: number) => new Promise((wake) => setTimeout(wake, ms));

// A refused call is retried; anything else is the caller's to see.
async function withSlot<T>(call: () => Promise<T>): Promise<T> {
  for (let attempt = 1; ; attempt++) {
    try {
      return await call();
    } catch (error) {
      if (attempt === 8) throw error;
      if (error instanceof ConcurrencyLimitError || error instanceof NoCapacityError) {
        // Refused, not queued, and no Retry-After: a slot frees when one of your calls ends.
        await sleep(20_000);
      } else if (error instanceof QuotaExceededError && error.ceiling === 'requests_per_minute') {
        await sleep(error.retryAfterSeconds * 1_000);
      } else {
        throw error; // out of monthly credits, or a bad request: retrying cannot help
      }
    }
  }
}

// Reserve a step's maximum before it starts, settle to what it really cost.
// A step that throws is counted at its maximum.
let spent = 0;
let held = 0;
async function withinBudget<T>(maximum: number, step: () => Promise<T>, cost: (value: T) => number): Promise<T | undefined> {
  if (spent + held + maximum > BATCH_CEILING) return undefined;
  held += maximum;
  let actual = maximum;
  try {
    const value = await step();
    actual = cost(value);
    return value;
  } finally {
    held -= maximum;
    spent += actual;
  }
}

// At most LANES companies in flight: each lane takes the next one when it is free.
async function inLanes<T, R>(items: readonly T[], work: (item: T) => Promise<R>): Promise<R[]> {
  const results: R[] = [];
  let next = 0;
  const lane = async () => {
    while (next < items.length) {
      const index = next++;
      results[index] = await work(items[index] as T);
    }
  };
  await Promise.all(Array.from({ length: Math.min(LANES, items.length) }, lane));
  return results;
}

// 0. Read the list, deduplicated by normalised domain.
const domain = (raw: string) => raw.trim().toLowerCase().replace(/^https?:\/\//, '').replace(/^www\./, '').split('/')[0] ?? '';
const targets = new Map<string, string>();
for (const line of (await readFile('companies.csv', 'utf8')).split('\n').slice(1)) {
  const cut = line.lastIndexOf(',');
  const key = domain(line.slice(cut + 1));
  if (cut > 0 && key !== '' && !targets.has(key)) targets.set(key, line.slice(0, cut).replace(/^"|"$/g, '').trim());
}

async function research([site, name]: [string, string]): Promise<Company> {
  const company: Company = { name, domain: site, whatTheyDo: '', plans: [], decisionMakers: [], openRoles: null, teamSize: null, news: [] };
  const people = new Map<string, Person>();
  const addPeople = (found: Person[] = []) => {
    for (const person of found) {
      if (DECIDES.test(person.title) && !people.has(person.name.toLowerCase())) people.set(person.name.toLowerCase(), person);
    }
  };

  try {
    // 1. Search: news that says growing or shrinking.
    const hits = await withinBudget(
      SEARCH_RESULTS * 0.5,
      () => withSlot(() => oc.search(`"${name}" funding OR hiring OR layoffs`, { limit: SEARCH_RESULTS })),
      (found) => found.length * 0.5,
    );
    company.news = (hits ?? []).filter((hit) => !hit.url.includes(site)).slice(0, 3).map((hit) => ({ title: hit.title, url: hit.url }));

    // 2. Crawl: the company's own pages, each read into the schema.
    const pages = await withinBudget(
      CRAWL_PAGES,
      () =>
        withSlot(async () => {
          const crawl = await oc.crawl(`https://${site}`, {
            limit: CRAWL_PAGES,
            maxDepth: 2,
            includePaths: ['/about*', '/about/**', '/company*', '/team*', '/leadership*', '/pricing*', '/plans*', '/careers*', '/jobs*'],
            formats: ['json'],
            schema: PAGE_SCHEMA,
          });
          const found: CrawlPage[] = [];
          for await (const page of crawl) found.push(page);
          return found;
        }),
      (found) => found.length,
    );
    for (const page of [...(pages ?? [])].sort((a, b) => a.depth - b.depth)) {
      const facts = (page.json ?? {}) as Facts;
      if (company.whatTheyDo === '' && facts.whatTheyDo) company.whatTheyDo = facts.whatTheyDo;
      if ((facts.plans?.length ?? 0) > company.plans.length) company.plans = facts.plans ?? [];
      if (facts.openRoles !== undefined) company.openRoles = Math.max(company.openRoles ?? 0, facts.openRoles);
      company.teamSize ??= facts.teamSize ?? null;
      addPeople(facts.people);
    }

    // 3. Agent: only for what the crawl could not find.
    const wanted = [
      company.plans.length === 0 && 'every plan and its price as shown (switch to monthly billing if there is a toggle; if prices are only on request, return one plan named "All plans" priced "Contact sales")',
      people.size === 0 && 'the founders and leaders the site names, with their job titles',
    ].filter((want): want is string => typeof want === 'string');
    if (wanted.length > 0) {
      const run = await withinBudget(
        Number(AGENT_BUDGET),
        async () => {
          // Only the submit is retried: a refusal arrives before any run exists.
          const { job } = await withSlot(async () => {
            const job = oc.agent({
              task: `On ${site}, the website of ${name}, find: ${wanted.join('; and ')}. Use only pages on ${site}.`,
              caps: { budget: AGENT_BUDGET, duration: '4m' },
              schema: { type: 'object', properties: { plans: { type: 'array', items: PLAN }, people: { type: 'array', items: PERSON } } },
              rules: [`Only visit ${site} and its subdomains.`, 'Never submit a form, book a demo, start a trial or sign in.'],
            });
            await job.submitted; // a refused submit throws here
            return { job }; // wrapped, because a job is awaitable and returning it bare waits for the run
          });
          return await job;
        },
        (settled) => Number(settled.credits),
      );
      const found = (run?.data ?? {}) as Facts;
      if (company.plans.length === 0) company.plans = found.plans ?? [];
      addPeople(found.people);
    }
  } catch (error) {
    console.error(`${site}: ${String(error)}`); // keep what was found before the error
  }
  company.decisionMakers = [...people.values()];
  return company;
}

const companies = await inLanes([...targets.entries()], research);
await writeFile('companies.json', JSON.stringify(companies, null, 2));

// 4. Write to HubSpot: update the company with this domain, or create it.
async function hubspot(path: string, method: string, body: unknown): Promise<{ results?: { id: string }[] }> {
  const response = await fetch(`https://api.hubapi.com/crm/v3/objects/companies${path}`, {
    method,
    headers: { authorization: `Bearer ${process.env.HUBSPOT_TOKEN}`, 'content-type': 'application/json' },
    body: JSON.stringify(body),
  });
  if (!response.ok) throw new Error(`HubSpot ${response.status}: ${await response.text()}`);
  return (await response.json()) as { results?: { id: string }[] };
}

let written = 0;
for (const company of companies) {
  if (company.whatTheyDo === '' && company.plans.length === 0) continue; // nothing worth writing
  const properties = {
    name: company.name,
    domain: company.domain,
    description: company.whatTheyDo,
    pricing_summary: company.plans.map((plan) => `${plan.name}: ${plan.price}`).join('\n'),
    decision_makers: company.decisionMakers.map((person) => `${person.name}, ${person.title}`).join('\n'),
    open_roles: company.openRoles === null ? '' : String(company.openRoles),
    team_size_signal: company.teamSize ?? '',
    recent_news: company.news.map((item) => `${item.title} ${item.url}`).join('\n'),
  };
  const found = await hubspot('/search', 'POST', {
    filterGroups: [{ filters: [{ propertyName: 'domain', operator: 'EQ', value: company.domain }] }],
    limit: 1,
  });
  const id = found.results?.[0]?.id;
  await (id === undefined ? hubspot('', 'POST', { properties }) : hubspot(`/${id}`, 'PATCH', { properties }));
  written += 1;
}
console.log(`${companies.length} companies researched, ${written} written to HubSpot, ${spent.toFixed(2)} credits`);

import asyncio
import json
import os
import re

import httpx
from outcrawl import ConcurrencyLimitError, NoCapacityError, Outcrawl, QuotaExceededError

# Every key on your account shares one browser slot count (Free 2, Builder 5,
# Pro 25, Scale 100). Keep LANES below yours so other work still gets a slot.
LANES = 4
BATCH_CEILING = 10_000  # credits, for the whole list
SEARCH_RESULTS = 5  # 5 credits per 10 results
CRAWL_PAGES = 8  # 1 credit per page, and the crawl's hard limit
AGENT_BUDGET = "40.00"  # about 2.5 minutes of a typical run
DECIDES = re.compile(r"founder|chief|\bce?o\b|\bc[tfor]o\b|president|\bvp\b|head of|director", re.I)

PLAN = {"type": "object", "properties": {"name": {"type": "string"}, "price": {"type": "string"}}, "required": ["name", "price"]}
PERSON = {"type": "object", "properties": {"name": {"type": "string"}, "title": {"type": "string"}}, "required": ["name", "title"]}
PAGE_SCHEMA = {
    "type": "object",
    "properties": {
        "whatTheyDo": {"type": "string", "description": "One sentence: what the company sells, and to whom. Empty if this page does not say."},
        "plans": {"type": "array", "items": PLAN, "description": 'Each plan with its price as shown, e.g. "$49 per seat per month" or "Contact sales"'},
        "people": {"type": "array", "items": PERSON, "description": "People this page names, with their job titles"},
        "openRoles": {"type": "integer", "description": "How many open roles this page lists, if it is a careers page"},
        "teamSize": {"type": "string", "description": 'Any statement of team size, e.g. "a team of 40 across Berlin and Lisbon"'},
    },
}

tally = {"spent": 0.0, "held": 0.0}


async def with_slot(call):
    """A refused call is retried; anything else is the caller's to see."""
    for attempt in range(1, 9):
        try:
            return await call()
        except (ConcurrencyLimitError, NoCapacityError):
            # Refused, not queued, and no Retry-After: a slot frees when one of your calls ends.
            if attempt == 8:
                raise
            await asyncio.sleep(20)
        except QuotaExceededError as error:
            # Out of monthly credits: retrying cannot help. Per-minute: wait as told.
            if error.ceiling != "requests_per_minute" or attempt == 8:
                raise
            await asyncio.sleep(error.retry_after_seconds)


async def within_budget(maximum, step, cost):
    """Reserve a step's maximum before it starts, settle to what it really cost.
    A step that raises is counted at its maximum."""
    if tally["spent"] + tally["held"] + maximum > BATCH_CEILING:
        return None
    tally["held"] += maximum
    actual = maximum
    try:
        value = await step()
        actual = cost(value)
        return value
    finally:
        tally["held"] -= maximum
        tally["spent"] += actual


def normalise(raw: str) -> str:
    raw = re.sub(r"^https?://", "", raw.strip().lower())
    return re.sub(r"^www\.", "", raw).split("/")[0]


async def main() -> None:
    # 0. Read the list, deduplicated by normalised domain.
    targets: dict[str, str] = {}
    with open("companies.csv") as source:
        for line in source.read().splitlines()[1:]:
            name, _, raw = line.rpartition(",")
            site = normalise(raw)
            if name and site and site not in targets:
                targets[site] = name.strip().strip('"')

    async with Outcrawl() as oc:
        lanes = asyncio.Semaphore(LANES)

        async def research(site: str, name: str) -> dict:
            company = {"name": name, "domain": site, "whatTheyDo": "", "plans": [], "decisionMakers": [],
                       "openRoles": None, "teamSize": None, "news": []}
            people: dict[str, dict] = {}

            def add_people(found) -> None:
                for person in found or []:
                    key = person["name"].lower()
                    if DECIDES.search(person["title"]) and key not in people:
                        people[key] = person

            async with lanes:
                try:
                    # 1. Search: news that says growing or shrinking.
                    hits = await within_budget(
                        SEARCH_RESULTS * 0.5,
                        lambda: with_slot(lambda: oc.search(f'"{name}" funding OR hiring OR layoffs', limit=SEARCH_RESULTS)),
                        lambda found: len(found) * 0.5,
                    )
                    company["news"] = [{"title": hit["title"], "url": hit["url"]}
                                       for hit in (hits or []) if site not in hit["url"]][:3]

                    # 2. Crawl: the company's own pages, each read into the schema.
                    async def crawl_site():
                        crawl = await oc.crawl(
                            f"https://{site}",
                            limit=CRAWL_PAGES,
                            maxDepth=2,
                            includePaths=["/about*", "/about/**", "/company*", "/team*", "/leadership*",
                                          "/pricing*", "/plans*", "/careers*", "/jobs*"],
                            formats=["json"],
                            schema=PAGE_SCHEMA,
                        )
                        return [page async for page in crawl]  # a refusal raises on the first read

                    pages = await within_budget(CRAWL_PAGES, lambda: with_slot(crawl_site), len)
                    for page in sorted(pages or [], key=lambda p: p.get("depth", 0)):
                        facts = page.get("json") or {}
                        if not company["whatTheyDo"] and facts.get("whatTheyDo"):
                            company["whatTheyDo"] = facts["whatTheyDo"]
                        if len(facts.get("plans") or []) > len(company["plans"]):
                            company["plans"] = facts["plans"]
                        if facts.get("openRoles") is not None:
                            company["openRoles"] = max(company["openRoles"] or 0, facts["openRoles"])
                        company["teamSize"] = company["teamSize"] or facts.get("teamSize")
                        add_people(facts.get("people"))

                    # 3. Agent: only for what the crawl could not find.
                    wanted = []
                    if not company["plans"]:
                        wanted.append('every plan and its price as shown (switch to monthly billing if there is a '
                                      'toggle; if prices are only on request, return one plan named "All plans" '
                                      'priced "Contact sales")')
                    if not people:
                        wanted.append("the founders and leaders the site names, with their job titles")
                    if wanted:
                        async def submit():
                            job = oc.agent(
                                task=f"On {site}, the website of {name}, find: {'; and '.join(wanted)}. "
                                     f"Use only pages on {site}.",
                                caps={"budget": AGENT_BUDGET, "duration": "4m"},
                                schema={"type": "object", "properties": {
                                    "plans": {"type": "array", "items": PLAN},
                                    "people": {"type": "array", "items": PERSON}}},
                                rules=[f"Only visit {site} and its subdomains.",
                                       "Never submit a form, book a demo, start a trial or sign in."],
                            )
                            await job.submitted  # a refused submit raises here
                            return job

                        async def run_agent():
                            # Only the submit is retried: a refusal arrives before any run exists.
                            job = await with_slot(submit)
                            return await job

                        run = await within_budget(float(AGENT_BUDGET), run_agent, lambda r: float(r["credits"]))
                        found = (run or {}).get("data") or {}
                        if not company["plans"]:
                            company["plans"] = found.get("plans") or []
                        add_people(found.get("people"))
                except Exception as error:
                    print(f"{site}: {error}")  # keep what was found before the error
            company["decisionMakers"] = list(people.values())
            return company

        companies = await asyncio.gather(*(research(site, name) for site, name in targets.items()))

    with open("companies.json", "w") as out:
        json.dump(companies, out, indent=2, ensure_ascii=False)

    # 4. Write to HubSpot: update the company with this domain, or create it.
    written = 0
    async with httpx.AsyncClient(
        base_url="https://api.hubapi.com/crm/v3/objects",
        headers={"authorization": f"Bearer {os.environ['HUBSPOT_TOKEN']}"},
    ) as hubspot:
        for company in companies:
            if not company["whatTheyDo"] and not company["plans"]:
                continue  # nothing worth writing
            properties = {
                "name": company["name"],
                "domain": company["domain"],
                "description": company["whatTheyDo"],
                "pricing_summary": "\n".join(f"{p['name']}: {p['price']}" for p in company["plans"]),
                "decision_makers": "\n".join(f"{p['name']}, {p['title']}" for p in company["decisionMakers"]),
                "open_roles": "" if company["openRoles"] is None else str(company["openRoles"]),
                "team_size_signal": company["teamSize"] or "",
                "recent_news": "\n".join(f"{n['title']} {n['url']}" for n in company["news"]),
            }
            found = await hubspot.post("/companies/search", json={
                "filterGroups": [{"filters": [{"propertyName": "domain", "operator": "EQ", "value": company["domain"]}]}],
                "limit": 1,
            })
            found.raise_for_status()
            existing = found.json().get("results") or []
            if existing:
                response = await hubspot.patch(f"/companies/{existing[0]['id']}", json={"properties": properties})
            else:
                response = await hubspot.post("/companies", json={"properties": properties})
            response.raise_for_status()
            written += 1

    print(f"{len(companies)} companies researched, {written} written to HubSpot, {tally['spent']:.2f} credits")


asyncio.run(main())