Track every open role across 300 startups

import { existsSync } from 'node:fs';
import { readFile, writeFile } from 'node:fs/promises';
import { ConcurrencyLimitError, NoCapacityError, Outcrawl, QuotaExceededError } 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 NIGHT_CEILING = 2_000; // credits, per night
const AGENT_BUDGET = '50.00'; // about 3 minutes of a typical run
const WANTED = /\b(platform|infrastructure|site reliability|devops|backend)\b.*\bengineer/i;
const WANTED_WORDS = 'platform, infrastructure, site reliability, DevOps or backend engineer';
const NEEDS_AGENT = /myworkdayjobs\.com|icims\.com|taleo\.net|successfactors\.(com|eu)|oraclecloud\.com/;

interface Found {
  title: string;
  url: string;
  location?: string;
  workplace?: 'remote' | 'hybrid' | 'onsite' | 'unknown';
  salaryMin?: number;
  salaryMax?: number;
  currency?: string;
  period?: 'year' | 'month' | 'hour';
}
interface Role {
  company: string;
  title: string;
  location: string;
  workplace: string;
  salaryMin: number | null;
  salaryMax: number | null;
  currency: string | null;
  period: string | null;
  url: string;
  firstSeen: string;
}

const ROLE = {
  type: 'object',
  properties: {
    title: { type: 'string' },
    location: { type: 'string' },
    workplace: { type: 'string', enum: ['remote', 'hybrid', 'onsite', 'unknown'] },
    salaryMin: { type: 'number', description: 'Lowest posted pay. Omit when no pay is posted.' },
    salaryMax: { type: 'number', description: 'Highest posted pay. Omit when no pay is posted.' },
    currency: { type: 'string', description: 'ISO 4217 code of the posted pay' },
    period: { type: 'string', enum: ['year', 'month', 'hour'] },
    url: { type: 'string', description: 'Link to the posting itself' },
  },
  required: ['title', 'url'],
};
const ROLES = { type: 'object', properties: { roles: { type: 'array', items: ROLE } }, required: ['roles'] };

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.
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 > NIGHT_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 startups 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;
}

// One posting, one key: no fragment, no tracking parameters, no trailing slash.
function keyOf(url: string): string {
  const parsed = new URL(url);
  for (const name of [...parsed.searchParams.keys()]) {
    if (/^(utm_|gh_src$|lever-source|source$|ref$)/.test(name)) parsed.searchParams.delete(name);
  }
  return `${parsed.host.toLowerCase()}${parsed.pathname.replace(/\/$/, '')}${parsed.search}`;
}
function absolute(url: string, base: string): string | null {
  try {
    return new URL(url, base).href;
  } catch {
    return null;
  }
}

const today = new Date().toISOString().slice(0, 10);
const known = new Map<string, Role>(
  existsSync('roles.json') ? Object.entries(JSON.parse(await readFile('roles.json', 'utf8')) as Record<string, Role>) : [],
);
const startups: [string, string][] = (await readFile('startups.csv', 'utf8'))
  .split('\n')
  .slice(1)
  .filter((line) => line.includes(','))
  .map((line) => [line.slice(0, line.lastIndexOf(',')).trim(), line.slice(line.lastIndexOf(',') + 1).trim()]);

async function askAgent(name: string, careersUrl: string): Promise<Found[] | null> {
  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:
            `Open ${careersUrl}, the careers site of ${name}. Use its search box or filters to find every ` +
            `open ${WANTED_WORDS} role, and page through all results. Open each posting and return its ` +
            `title, location, whether it is remote, hybrid or onsite, the pay range if the posting shows one, ` +
            `and the link to the posting.`,
          caps: { budget: AGENT_BUDGET, duration: '5m' },
          schema: ROLES,
          rules: ['Never apply to a role, create an account 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),
  );
  // Not run, or did not finish: tonight says nothing about this company.
  if (run === undefined || run.status !== 'completed') return null;
  return (run.data as { roles?: Found[] } | undefined)?.roles ?? [];
}

// null means "could not read tonight": the company keeps yesterday's roles.
async function collect([name, careersUrl]: [string, string]): Promise<{ name: string; roles: Found[] | null }> {
  try {
    let listed: Found[] = [];
    if (!NEEDS_AGENT.test(careersUrl)) {
      const page = await withinBudget(1, () => withSlot(() => oc.scrape(careersUrl, { formats: ['json'], schema: ROLES })), () => 1);
      listed = (page?.json as { roles?: Found[] } | undefined)?.roles ?? [];
    }
    if (listed.length === 0) {
      const found = await askAgent(name, careersUrl);
      if (found === null) return { name, roles: null };
      return {
        name,
        roles: found.flatMap((role) => {
          const url = absolute(role.url, careersUrl);
          return url !== null && WANTED.test(role.title) ? [{ ...role, url }] : [];
        }),
      };
    }

    const roles: Found[] = [];
    for (const role of listed.filter((r) => WANTED.test(r.title))) {
      const url = absolute(role.url, careersUrl);
      if (url === null) continue;
      if (!known.has(keyOf(url))) {
        // New tonight: read the posting itself for pay and remote policy.
        const posting = await withinBudget(1, () => withSlot(() => oc.scrape(url, { formats: ['json'], schema: ROLE })), () => 1);
        Object.assign(role, (posting?.json ?? {}) as Partial<Found>);
      }
      roles.push({ ...role, url });
    }
    return { name, roles };
  } catch (error) {
    console.error(`${name}: ${String(error)}`);
    return { name, roles: null };
  }
}

const results = await inLanes(startups, collect);

// Normalise, dedupe, and compare with last night.
const open = new Map<string, Role>();
const fresh: Role[] = [];
for (const { name, roles } of results) {
  if (roles === null) {
    for (const [key, role] of known) if (role.company === name) open.set(key, role);
    continue;
  }
  const seen = new Set<string>(); // the same role listed twice, e.g. under two teams
  for (const found of roles) {
    const key = keyOf(found.url); // every url is absolute by now
    const sameRole = `${found.title.trim().toLowerCase()}|${(found.location ?? '').trim().toLowerCase()}`;
    if (open.has(key) || seen.has(sameRole)) continue;
    seen.add(sameRole);
    const role: Role = known.get(key) ?? {
      company: name,
      title: found.title.trim(),
      location: found.location?.trim() ?? '',
      workplace: found.workplace ?? 'unknown',
      salaryMin: found.salaryMin ?? null,
      salaryMax: found.salaryMax ?? null,
      currency: found.currency ?? null,
      period: found.period ?? null,
      url: found.url,
      firstSeen: today,
    };
    if (!known.has(key)) fresh.push(role);
    open.set(key, role);
  }
}
const closed = [...known.keys()].filter((key) => !open.has(key)).length;

const COLUMNS = ['company', 'title', 'location', 'workplace', 'salaryMin', 'salaryMax', 'currency', 'period', 'url', 'firstSeen'] as const;
const cell = (value: unknown) => `"${String(value ?? '').replaceAll('"', '""')}"`;
const csv = [COLUMNS.map(cell).join(','), ...[...open.values()].map((role) => COLUMNS.map((column) => cell(role[column])).join(','))];

await writeFile('roles.json', JSON.stringify(Object.fromEntries(open), null, 2));
await writeFile('roles.csv', `${csv.join('\n')}\n`);
await writeFile('new-roles.json', JSON.stringify(fresh, null, 2));
console.log(`${open.size} open roles at ${startups.length} startups: ${fresh.length} new tonight, ${closed} closed, ${spent.toFixed(2)} credits`);

import asyncio
import csv
import json
import os
import re
from datetime import date
from urllib.parse import parse_qsl, urlencode, urljoin, urlsplit

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
NIGHT_CEILING = 2_000  # credits, per night
AGENT_BUDGET = "50.00"  # about 3 minutes of a typical run
WANTED = re.compile(r"\b(platform|infrastructure|site reliability|devops|backend)\b.*\bengineer", re.I)
WANTED_WORDS = "platform, infrastructure, site reliability, DevOps or backend engineer"
NEEDS_AGENT = re.compile(r"myworkdayjobs\.com|icims\.com|taleo\.net|successfactors\.(com|eu)|oraclecloud\.com")

ROLE = {
    "type": "object",
    "properties": {
        "title": {"type": "string"},
        "location": {"type": "string"},
        "workplace": {"type": "string", "enum": ["remote", "hybrid", "onsite", "unknown"]},
        "salaryMin": {"type": "number", "description": "Lowest posted pay. Omit when no pay is posted."},
        "salaryMax": {"type": "number", "description": "Highest posted pay. Omit when no pay is posted."},
        "currency": {"type": "string", "description": "ISO 4217 code of the posted pay"},
        "period": {"type": "string", "enum": ["year", "month", "hour"]},
        "url": {"type": "string", "description": "Link to the posting itself"},
    },
    "required": ["title", "url"],
}
ROLES = {"type": "object", "properties": {"roles": {"type": "array", "items": ROLE}}, "required": ["roles"]}
COLUMNS = ["company", "title", "location", "workplace", "salaryMin", "salaryMax", "currency", "period", "url", "firstSeen"]

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."""
    if tally["spent"] + tally["held"] + maximum > NIGHT_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 key_of(url: str) -> str:
    """One posting, one key: no fragment, no tracking parameters, no trailing slash."""
    parts = urlsplit(url)
    query = [(k, v) for k, v in parse_qsl(parts.query) if not re.match(r"^(utm_|gh_src$|lever-source|source$|ref$)", k)]
    return f"{parts.netloc.lower()}{parts.path.rstrip('/')}{'?' + urlencode(query) if query else ''}"


async def main() -> None:
    today = date.today().isoformat()
    known: dict[str, dict] = {}
    if os.path.exists("roles.json"):
        with open("roles.json") as state:
            known = json.load(state)
    with open("startups.csv") as source:
        startups = [tuple(part.strip() for part in line.rsplit(",", 1))
                    for line in source.read().splitlines()[1:] if "," in line]

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

        async def ask_agent(name: str, careers_url: str) -> list | None:
            async def submit():
                job = oc.agent(
                    task=(
                        f"Open {careers_url}, the careers site of {name}. Use its search box or filters to find "
                        f"every open {WANTED_WORDS} role, and page through all results. Open each posting and "
                        f"return its title, location, whether it is remote, hybrid or onsite, the pay range if "
                        f"the posting shows one, and the link to the posting."
                    ),
                    caps={"budget": AGENT_BUDGET, "duration": "5m"},
                    schema=ROLES,
                    rules=["Never apply to a role, create an account 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"]))
            # Not run, or did not finish: tonight says nothing about this company.
            if run is None or run["status"] != "completed":
                return None
            return (run.get("data") or {}).get("roles") or []

        async def collect(name: str, careers_url: str) -> tuple[str, list | None]:
            """None means "could not read tonight": the company keeps yesterday's roles."""
            async with lanes:
                try:
                    listed: list = []
                    if not NEEDS_AGENT.search(careers_url):
                        page = await within_budget(
                            1,
                            lambda: with_slot(lambda: oc.scrape(careers_url, formats=["json"], schema=ROLES)),
                            lambda _: 1,
                        )
                        listed = ((page or {}).get("json") or {}).get("roles") or []
                    if not listed:
                        found = await ask_agent(name, careers_url)
                        if found is None:
                            return name, None
                        return name, [{**r, "url": urljoin(careers_url, r["url"])}
                                      for r in found if WANTED.search(r["title"])]

                    roles = []
                    for role in (r for r in listed if WANTED.search(r["title"])):
                        url = urljoin(careers_url, role["url"])
                        if key_of(url) not in known:
                            # New tonight: read the posting itself for pay and remote policy.
                            posting = await within_budget(
                                1,
                                lambda: with_slot(lambda: oc.scrape(url, formats=["json"], schema=ROLE)),
                                lambda _: 1,
                            )
                            role.update((posting or {}).get("json") or {})
                        roles.append({**role, "url": url})
                    return name, roles
                except Exception as error:
                    print(f"{name}: {error}")
                    return name, None

        results = await asyncio.gather(*(collect(name, url) for name, url in startups))

    # Normalise, dedupe, and compare with last night.
    open_roles: dict[str, dict] = {}
    fresh: list[dict] = []
    for name, roles in results:
        if roles is None:
            open_roles.update({k: r for k, r in known.items() if r["company"] == name})
            continue
        seen: set[str] = set()  # the same role listed twice, e.g. under two teams
        for found in roles:
            key = key_of(found["url"])
            same_role = f"{found['title'].strip().lower()}|{(found.get('location') or '').strip().lower()}"
            if key in open_roles or same_role in seen:
                continue
            seen.add(same_role)
            role = known.get(key) or {
                "company": name,
                "title": found["title"].strip(),
                "location": (found.get("location") or "").strip(),
                "workplace": found.get("workplace") or "unknown",
                "salaryMin": found.get("salaryMin"),
                "salaryMax": found.get("salaryMax"),
                "currency": found.get("currency"),
                "period": found.get("period"),
                "url": found["url"],
                "firstSeen": today,
            }
            if key not in known:
                fresh.append(role)
            open_roles[key] = role
    closed = sum(1 for key in known if key not in open_roles)

    with open("roles.json", "w") as state:
        json.dump(open_roles, state, indent=2, ensure_ascii=False)
    with open("roles.csv", "w", newline="") as sheet:
        writer = csv.DictWriter(sheet, fieldnames=COLUMNS, quoting=csv.QUOTE_ALL)
        writer.writeheader()
        writer.writerows(open_roles.values())
    with open("new-roles.json", "w") as news:
        json.dump(fresh, news, indent=2, ensure_ascii=False)
    print(f"{len(open_roles)} open roles at {len(startups)} startups: {len(fresh)} new tonight, "
          f"{closed} closed, {tally['spent']:.2f} credits")


asyncio.run(main())