Files
skillfactor-pipeline/pipeline/p3a_fetch_jobs.py

122 lines
4.8 KiB
Python

"""Phase 3a โ€” fetch real recruiter job ads via the JSearch API (openwebninja).
Official API only (no HTML scraping). Raw responses are cached in
data/raw/jobs/ so repeated runs never re-spend API quota. Target: 50-100
English-language ads (us + gb) for the occupation "recruiter".
Fallback: Adzuna (if ADZUNA_APP_ID/KEY are set and JSearch quota is exhausted).
"""
import json
import os
import sys
import time
import requests
from dotenv import load_dotenv
BASE = os.path.join(os.path.dirname(__file__), "..")
load_dotenv(os.path.join(BASE, ".env"))
RAW_JOBS = os.path.join(BASE, "data", "raw", "jobs")
JSEARCH_URL = "https://api.openwebninja.com/jsearch/search-v2"
TARGET = 500
MAX_REQUESTS = 55 # hartes Budget (Basic-Plan: 200/Monat); Cache-Hits kosten nichts
# Query-Varianten = ESCO-altLabels des Berufs -> maximale Marktabdeckung
QUERIES = [
("recruiter", "us"), ("recruiter", "gb"),
("talent acquisition specialist", "us"), ("talent acquisition specialist", "gb"),
("technical recruiter", "us"),
("recruiting coordinator", "us"),
("headhunter", "gb"),
]
def fetch_jsearch():
key = os.environ.get("JSEARCH_API_KEY", "").strip()
if not key:
print("SKIP: JSEARCH_API_KEY missing (.env) โ€” documented as TODO in README")
return []
os.makedirs(RAW_JOBS, exist_ok=True)
ads, seen = [], set()
spent = [0]
for query, country in QUERIES:
cursor, page = None, 0
while len(ads) < TARGET and page < 10:
page += 1
cache = os.path.join(RAW_JOBS, f"jsearch_{query.replace(' ', '_')}_{country}_p{page}.json")
if os.path.exists(cache):
payload = json.load(open(cache, encoding="utf-8"))
elif spent[0] >= MAX_REQUESTS:
print(f"request budget ({MAX_REQUESTS}) reached โ€” stopping")
break
else:
spent[0] += 1
params = {"query": query, "country": country, "language": "en"}
if cursor:
params["cursor"] = cursor
# transiente API-Fehler (5xx/Timeout) mit Backoff abfangen,
# statt den ganzen Lauf zu verlieren
payload = None
for attempt in range(3):
try:
r = requests.get(JSEARCH_URL, params=params,
headers={"x-api-key": key}, timeout=60)
except requests.RequestException as exc:
print(f" {query}/{country} p{page}: {exc} (retry {attempt+1})")
time.sleep(8 * (attempt + 1))
continue
if r.status_code == 429:
print(f"rate limit on {query}/{country} p{page}; stopping this query")
break
if r.status_code >= 500:
print(f" {query}/{country} p{page}: HTTP {r.status_code} (retry {attempt+1})")
time.sleep(8 * (attempt + 1))
continue
r.raise_for_status()
payload = r.json()
break
if payload is None:
print(f" {query}/{country} p{page}: aufgegeben, naechste Query")
break
json.dump(payload, open(cache, "w", encoding="utf-8"),
ensure_ascii=False, indent=1)
time.sleep(2) # be gentle with the free tier
body = payload.get("data") or {}
# search-v2 liefert {"data": {"jobs": [...]}, "cursor": "..."}
data = body.get("jobs") if isinstance(body, dict) else body
data = data or []
if not data:
break
for job in data:
jid = job.get("job_id")
if jid and jid not in seen and job.get("job_description"):
seen.add(jid)
ads.append({
"job_id": jid,
"title": job.get("job_title"),
"employer": job.get("employer_name"),
"country": country,
"description": job.get("job_description"),
})
cursor = (payload.get("cursor")
or (body.get("cursor") if isinstance(body, dict) else None)
or (payload.get("meta") or {}).get("cursor"))
if not cursor:
break
print(f"{query}/{country}: kumuliert {len(ads)} Anzeigen")
return ads
def main():
ads = fetch_jsearch()
out = os.path.join(RAW_JOBS, "recruiter_ads.json")
if ads:
json.dump(ads, open(out, "w", encoding="utf-8"), ensure_ascii=False, indent=1)
print(f"OK: {len(ads)} unique ads -> {out}")
else:
print("No ads fetched.")
if __name__ == "__main__":
main()