"""Targeted EU/DACH ad fetch for one occupation (phase 2b). Deliberately NOT part of the default full-catalog crawl: 4 extra countries per occupation would blow the 33k lifetime JSearch budget. Use for the reference/flagship packages and future targeted runs. - Countries: DE, AT, CH, NL (kept as separate slice; US/UK slices untouched) - Queries: preferred label + first alt label, up to 2 pages/country - Budget: hard cap ~8-16 requests per occupation, all counted via progress.spend_request BEFORE each HTTP call - Raw pages cached under data/raw/jobs//jsearch___p.json (cache hits never re-spend quota) - Result: merged into data/raw/jobs/_ads.json (dedupe by job_id) — the raw file stays the single source the gate/extractor reads. Usage: python pipeline/eu_fetch.py [--pages 2] """ import argparse 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")) sys.path.insert(0, os.path.dirname(__file__)) import progress # noqa: E402 from db import connect # noqa: E402 JSEARCH_URL = "https://api.openwebninja.com/jsearch/search-v2" EU_COUNTRIES = ("de", "at", "ch", "nl") RAW_JOBS = os.path.join(BASE, "data", "raw", "jobs") EU_QUERIES_FILE = os.path.join(BASE, "data", "eu-queries.json") def _queries_for_country(slug, country): """Localized queries per country. The JSearch index is language-local — English titles return 0 hits in DACH. Priority: reviewed config (data/eu-queries.json) -> fallback English label. Empty list = skip.""" try: cfg = json.load(open(EU_QUERIES_FILE, encoding="utf-8")) if slug in cfg and country in cfg[slug]: return cfg[slug][country] except OSError: pass label = slug.replace("-", " ") try: m = json.load(open(os.path.join(BASE, "skills", slug, "manifest.json"), encoding="utf-8")) label = m.get("title", label) except Exception: pass return [label] def fetch_eu(slug, pages=2): key = os.environ.get("JSEARCH_API_KEY", "").strip() if not key: sys.exit("JSEARCH_API_KEY missing") state = progress.load() raw_dir = os.path.join(RAW_JOBS, slug) os.makedirs(raw_dir, exist_ok=True) ads_file = os.path.join(RAW_JOBS, f"{slug}_ads.json") existing = json.load(open(ads_file, encoding="utf-8")) if os.path.exists(ads_file) else [] seen = {a["job_id"] for a in existing} added = 0 for country in EU_COUNTRIES: for query in _queries_for_country(slug, country): safe_q = query.replace(" ", "_").replace("/", "-")[:60] cursor = None for page in range(1, pages + 1): cache = os.path.join(raw_dir, f"jsearch_{safe_q}_{country}_p{page}.json") if os.path.exists(cache): payload = json.load(open(cache, encoding="utf-8")) else: progress.spend_request(state, slug) params = {"query": query, "country": country} if cursor: params["cursor"] = cursor try: r = requests.get(JSEARCH_URL, params=params, headers={"x-api-key": key}, timeout=60) r.raise_for_status() except requests.RequestException as exc: print(f" {query}/{country} p{page}: {exc} — skip") break payload = r.json() json.dump(payload, open(cache, "w", encoding="utf-8"), ensure_ascii=False, indent=1) time.sleep(2) body = payload.get("data") or {} jobs = (body.get("jobs") if isinstance(body, dict) else body) or [] if not jobs: break for job in jobs: jid = job.get("job_id") if jid and jid not in seen and job.get("job_description"): seen.add(jid) existing.append({ "job_id": jid, "title": job.get("job_title"), "employer": job.get("employer_name"), "country": country, "description": job.get("job_description"), }) added += 1 cursor = (payload.get("cursor") or (body.get("cursor") if isinstance(body, dict) else None)) if not cursor: break print(f"{query}/{country}: +{added} kumuliert") json.dump(existing, open(ads_file, "w", encoding="utf-8"), ensure_ascii=False, indent=1) print(f"EU slice merged: +{added} new ads -> {ads_file} " f"(total {len(existing)})") if __name__ == "__main__": ap = argparse.ArgumentParser() ap.add_argument("slug") ap.add_argument("--pages", type=int, default=2) a = ap.parse_args() fetch_eu(a.slug, a.pages)