matchmaking-v2/research/poc/browser-apply/apify_probe.py

"""
apify_probe.py — measure OFFSITE-APPLY-LINK YIELD across job-board actors.

Question: for an on-demand user search, how many jobs come back with a usable OFFSITE apply URL
(the employer's own ATS — Ashby/Greenhouse/Lever/Workday/...), i.e. the Tier-1 links our
autonomous apply engine can actually finish? We probe each actor with a small sample and report:
  total jobs · #with offsite link · #Tier-1 ATS · breakdown by ATS.

Small samples on purpose (this costs Apify credits). Tune SAMPLE below.

Run:  python apify_probe.py
"""
from __future__ import annotations

import os
import json
from collections import Counter
from urllib.parse import urlparse

import httpx
from dotenv import load_dotenv

load_dotenv()
TOKEN = os.environ["APIFY_TOKEN"]
SAMPLE = int(os.environ.get("SAMPLE", "25"))
ONLY = os.environ.get("ONLY", "").lower()                      # run only actors whose label matches
FOLLOW = os.environ.get("FOLLOW_REDIRECTS", "0") == "1"        # Indeed: resolve externalApplyLink

# role/geo to probe — tweak freely
TITLE = "Machine Learning Engineer"
COUNTRY = "IN"
LOCATION = "India"

# ATS domain -> (label, is_tier1_autonomous)
ATS = {
    "ashbyhq.com": ("Ashby", True), "jobs.ashbyhq.com": ("Ashby", True),
    "greenhouse.io": ("Greenhouse", True), "boards.greenhouse.io": ("Greenhouse", True),
    "lever.co": ("Lever", True), "jobs.lever.co": ("Lever", True),
    "smartrecruiters.com": ("SmartRecruiters", True), "recruitee.com": ("Recruitee", True),
    "personio.de": ("Personio", True), "personio.com": ("Personio", True),
    "workable.com": ("Workable", True), "breezy.hr": ("Breezy", True),
    "myworkdayjobs.com": ("Workday", False), "icims.com": ("iCIMS", False),
    "taleo.net": ("Taleo", False), "successfactors.com": ("SuccessFactors", False),
    "bamboohr.com": ("BambooHR", False), "jobvite.com": ("Jobvite", False),
}


def classify(url: str | None):
    """Return (ats_label, is_tier1) for an offsite URL, or (None, False) if no/empty url."""
    if not url or not isinstance(url, str) or not url.startswith("http"):
        return None, False
    host = (urlparse(url).hostname or "").lower()
    for dom, (label, t1) in ATS.items():
        if host == dom or host.endswith("." + dom) or dom in host:
            return label, t1
    return "other-ATS/site", False  # an offsite link, just not a known ATS


def offsite_url(kind: str, item: dict) -> str | None:
    if kind == "linkedin":
        am = item.get("applyMethod") or {}
        return am.get("companyApplyUrl") or item.get("companyApplyUrl")
    if kind == "indeed":
        return item.get("externalApplyLink") or item.get("applyUrl")
    if kind == "ats":
        return item.get("apply_url") or item.get("applyUrl") or item.get("url")
    if kind == "naukri":
        # applyRedirectUrl = the true external handoff. companyApplyUrl is Naukri's OWN apply API
        # (naukri.com/cloudgateway-apply/...), i.e. on-platform — not an external link.
        url = item.get("applyRedirectUrl") or item.get("companyApplyUrl")
        if url and "naukri.com" in url:
            return None
        return url
    return None


ACTORS = [
    {"label": "LinkedIn (harvestapi)", "kind": "linkedin",
     "id": "harvestapi~linkedin-job-search",
     "input": {"jobTitles": [TITLE], "locations": [LOCATION], "maxItems": SAMPLE, "sortBy": "date"}},
    {"label": "Indeed (misceres)", "kind": "indeed",
     "id": "misceres~indeed-scraper",
     "input": {"position": TITLE, "location": LOCATION, "country": COUNTRY,
               "maxItemsPerSearch": SAMPLE, "followApplyRedirects": FOLLOW,
               "saveOnlyUniqueItems": True}},
    {"label": "ATS-direct (bovi G/L/Ashby)", "kind": "ats",
     "id": "bovi~greenhouse-lever-ashby-job-scraper",
     "input": {"presetLists": ["ai-ml"], "maxJobsPerCompany": 2, "outputProfile": "compact"}},
    # muhammetakkurtt: 11k users, no full-permission gate, returns applyRedirectUrl (external)
    # + companyApplyUrl (Naukri-internal) + companyApplyJob flag. Uses maxJobs (NOT maxItems).
    # maxJobs has a min of 50; cities wants numeric city IDs (omit -> nationwide).
    {"label": "Naukri (muhammetakkurtt)", "kind": "naukri",
     "id": "muhammetakkurtt~naukri-job-scraper",
     "input": {"keyword": TITLE, "maxJobs": max(50, SAMPLE)}},
]


def run_actor(actor: dict) -> list[dict]:
    url = f"https://api.apify.com/v2/acts/{actor['id']}/run-sync-get-dataset-items?token={TOKEN}"
    r = httpx.post(url, json=actor["input"], timeout=300)
    r.raise_for_status()
    data = r.json()
    return data if isinstance(data, list) else data.get("items", [])


def main():
    print(f"Probe: '{TITLE}' / {LOCATION} · sample≈{SAMPLE} per actor\n" + "=" * 68)
    grand = {"jobs": 0, "offsite": 0, "tier1": 0}
    grand_ats = Counter()

    for a in ACTORS:
        if ONLY and ONLY not in a["label"].lower():
            continue
        print(f"\n▶ {a['label']}  [{a['id']}]")
        try:
            items = run_actor(a)
        except Exception as e:
            print(f"   ✗ failed: {type(e).__name__}: {str(e)[:160]}")
            continue
        n = len(items)
        ats_counter = Counter()
        offsite = tier1 = 0
        for it in items:
            label, t1 = classify(offsite_url(a["kind"], it))
            if label:
                offsite += 1
                ats_counter[label] += 1
                if t1:
                    tier1 += 1
        grand["jobs"] += n
        grand["offsite"] += offsite
        grand["tier1"] += tier1
        grand_ats.update(ats_counter)
        pct = (100 * offsite / n) if n else 0
        t1pct = (100 * tier1 / n) if n else 0
        print(f"   jobs={n}  offsite-link={offsite} ({pct:.0f}%)  Tier-1 autonomous={tier1} ({t1pct:.0f}%)")
        if ats_counter:
            print("   breakdown:", ", ".join(f"{k}={v}" for k, v in ats_counter.most_common()))

    print("\n" + "=" * 68)
    j, o, t = grand["jobs"], grand["offsite"], grand["tier1"]
    print(f"COMBINED: jobs={j}  offsite={o} ({(100*o/j if j else 0):.0f}%)  "
          f"Tier-1 autonomous={t} ({(100*t/j if j else 0):.0f}%)")
    if grand_ats:
        print("ATS mix:", ", ".join(f"{k}={v}" for k, v in grand_ats.most_common()))


if __name__ == "__main__":
    main()