Job Scraper: Public Job Listings via API

import json, re, requests
from urllib.parse import urlsplit, parse_qs
from bs4 import BeautifulSoup

API, HEAD = "https://api.proxyfield.io/v1", {"Authorization": "Bearer <api_key>"}
JOB_LINK = re.compile(r"greenhouse\.io|lever\.co|myworkdayjobs\.com|/jobs?/|/careers?/")

def target(href):  # Google wraps some links as /url?q=<target>
    return parse_qs(urlsplit(href).query).get("q", [""])[0] if href.startswith("/url?") else href

serp = requests.post(f"{API}/google/search", headers=HEAD, timeout=180,
                     json={"q": "data engineer jobs austin", "gl": "us", "output": "html"}).text
links = {target(a["href"]) for a in BeautifulSoup(serp, "html.parser").find_all("a", href=True)}
links = sorted(u for u in links if u.startswith("http") and "google." not in u and JOB_LINK.search(u))

for url in links:
    page = requests.post(f"{API}/scrape", headers=HEAD, timeout=180,
                         json={"url": url, "render": False, "output": "html"}).text
    for tag in BeautifulSoup(page, "html.parser").find_all("script", type="application/ld+json"):
        try:
            data = json.loads(tag.string or "")
        except json.JSONDecodeError:
            continue
        for job in data if isinstance(data, list) else [data]:
            if isinstance(job, dict) and job.get("@type") == "JobPosting":
                print(url, "|", job.get("title"), "|", (job.get("hiringOrganization") or {}).get("name"))

https://boards.greenhouse.io/examplecompany/jobs/1234567 | Data Engineer | Example Inc.
https://jobs.lever.co/examplestartup/0b6c3e2a-5f1d-4c8e-9a7b-2d4f6e8a1c3b | Senior Data Engineer | Example Startup

import hashlib, json, re
from datetime import date
from difflib import SequenceMatcher
from urllib.parse import urlsplit, urlunsplit, parse_qsl, urlencode
import requests
from bs4 import BeautifulSoup

PROXY = "http://<login>-country-us:<password>@gate.proxyfield.io:7000"
TODAY = date.today().isoformat()
TRACKING = re.compile(r"^(utm_\w+|gh_src|lever-source(\[\])?|lever-origin)$")  # add source, ref ... only where a board uses them for tracking
COMPANY_SUFFIX = re.compile(r"\b(inc|llc|ltd|gmbh|corp)\b")
ABBR = {"sr": "senior", "jr": "junior", "mgr": "manager", "eng": "engineer"}

def canonical_url(url):
    p = urlsplit(url)
    query = [(k, v) for k, v in parse_qsl(p.query) if not TRACKING.match(k)]
    return urlunsplit((p.scheme, p.netloc.lower(), p.path.rstrip("/"), urlencode(query), ""))

def norm(text, company=False):
    text = (text or "").lower()
    if company:
        text = COMPANY_SUFFIX.sub(" ", text)
    return " ".join(ABBR.get(w, w) for w in re.sub(r"[^a-z0-9]+", " ", text).split())

def clean_description(html):
    return " ".join(BeautifulSoup(html or "", "html.parser").get_text(" ").lower().split())

def ats_key(url):
    m = (re.search(r"greenhouse\.io/([^/]+)/jobs/(\d+)", url)
         or re.search(r"jobs\.lever\.co/([^/]+)/([0-9a-f-]{36})", url))
    return ("ats", m.group(1).lower(), m.group(2)) if m else None

def postings(url):
    """Fetch one job page through a rotating residential IP; yield its JobPosting records."""
    html = requests.get(url, proxies={"http": PROXY, "https": PROXY},
                        headers={"User-Agent": "Mozilla/5.0"}, timeout=30).text
    for tag in BeautifulSoup(html, "html.parser").find_all("script", type="application/ld+json"):
        try:
            data = json.loads(tag.string or "")
        except json.JSONDecodeError:
            continue
        for item in data if isinstance(data, list) else [data]:
            if not (isinstance(item, dict) and item.get("@type") == "JobPosting"):
                continue
            org = item.get("hiringOrganization") or {}
            place = item.get("jobLocation") or {}
            place = place[0] if isinstance(place, list) and place else place
            addr = place.get("address") or {} if isinstance(place, dict) else {}
            if not isinstance(addr, dict):
                addr = {"addressLocality": str(addr)}
            ident = item.get("identifier")
            u = canonical_url(url)
            yield {"url": u, "title": item.get("title"),
                   "company": org.get("name") if isinstance(org, dict) else org,
                   "location": ", ".join(filter(None, [addr.get("addressLocality"), addr.get("addressRegion")])),
                   "salary": item.get("baseSalary"), "employment_type": item.get("employmentType"),
                   "req_id": ident.get("value") if isinstance(ident, dict) else ident,
                   "date_posted": item.get("datePosted"), "valid_through": item.get("validThrough"),
                   "description": clean_description(item.get("description")),
                   "first_seen": TODAY, "last_seen": TODAY, "sources": [u]}

def keys(rec):  # strongest signal first
    yield ("url", rec["url"])
    if ats_key(rec["url"]):
        yield ats_key(rec["url"])
    if rec.get("req_id"):
        yield ("req", norm(rec["company"], True), str(rec["req_id"]))
    if len(rec.get("description", "")) >= 300:  # short texts are shared by many jobs
        yield ("desc", hashlib.sha1(rec["description"].encode()).hexdigest())
    yield ("job", norm(rec["company"], True), norm(rec["title"]), norm(rec["location"]))

def conflict(a, b):
    """Two different openings: both carry an ATS job ID or a requisition ID, and the IDs differ."""
    ka, kb = ats_key(a["url"]), ats_key(b["url"])
    if ka and kb and ka != kb:
        return True
    return bool(a.get("req_id") and b.get("req_id") and str(a["req_id"]) != str(b["req_id"]))

def merge(records):
    feed, index = [], {}
    for rec in records:
        ks = list(keys(rec))
        hit = next((index[k] for k in ks if k in index and not conflict(index[k], rec)), None)
        if hit is None:
            feed.append(rec)
            hit = rec
        else:
            seen = (min(hit["first_seen"], rec["first_seen"]), max(hit["last_seen"], rec["last_seen"]))
            sources = sorted(set(hit["sources"]) | set(rec["sources"]))
            if ats_key(rec["url"]) and not ats_key(hit["url"]):
                hit.update(rec)  # the employer's own page becomes the master record
            hit.update(first_seen=seen[0], last_seen=seen[1], sources=sources)
        for k in ks:
            index[k] = hit
    return feed

def shingles(text, n=5):
    words = text.split()
    return {" ".join(words[i:i + n]) for i in range(max(1, len(words) - n + 1))}

def jaccard(a, b):
    sa, sb = shingles(a), shingles(b)
    return len(sa & sb) / len(sa | sb) if sa | sb else 0.0

def maybe_same(a, b):
    """Review tier: same company and city, and similar titles or descriptions. Flag, never merge blind."""
    if (norm(a["company"], True), norm(a["location"])) != (norm(b["company"], True), norm(b["location"])):
        return False
    ta, tb = (" ".join(sorted(norm(r["title"]).split())) for r in (a, b))
    return (SequenceMatcher(None, ta, tb).ratio() >= 0.9
            or jaccard(a.get("description", ""), b.get("description", "")) >= 0.8)

urls = [  # replace with real job pages from boards and careers sites
    "https://boards.greenhouse.io/examplecompany/jobs/1234567",
    "https://www.example.com/careers/data-engineer",
]
try:
    with open("jobs.json") as f:
        old = json.load(f)
except FileNotFoundError:
    old = []
feed = merge(old + [rec for url in urls for rec in postings(url)])
review = [(a["url"], b["url"]) for i, a in enumerate(feed) for b in feed[i + 1:] if maybe_same(a, b)]
with open("jobs.json", "w") as f:
    json.dump(feed, f, indent=2)
closed = [job for job in feed if job["last_seen"] < TODAY or (job["valid_through"] or "9999") < TODAY]
print(len(feed), "jobs,", len(review), "pairs to review,", len(closed), "possibly closed")
# on the three copies in the before-and-after table: 1 jobs, 0 pairs to review, 0 possibly closed