skillfactor-pipeline/pipeline/eu_fetch.py

"""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/<slug>/jsearch_<q>_<cc>_p<n>.json
  (cache hits never re-spend quota)
- Result: merged into data/raw/jobs/<slug>_ads.json (dedupe by job_id) —
  the raw file stays the single source the gate/extractor reads.

Usage:  python pipeline/eu_fetch.py <slug> [--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)