"""Full-catalog batch runner.

Stages (per skillfactor_finalize.md / auftrag_sonnet.md):
  1. packages:  deterministic generator for ALL ESCO occupations (no LLM)
  2. evidence:  per occupation fetch (JSearch, budget-capped) -> extract
                (Ollama, extract_local.py) -> aggregate (market.md)
  3. depth:     tier 1 = top-200 by evidence volume, tier 3 = night batch

Every step is idempotent and resumable: progress.json + file existence
decide what still needs work; an abort at any point loses at most the
current occupation's in-flight step.

Usage:
  python pipeline/batch_run.py --stage packages [--limit N]
  python pipeline/batch_run.py --stage evidence [--slugs a,b] [--limit N]
  python pipeline/batch_run.py --stage tier    # (re)compute tiers only
"""
# Guard: only the project venv is authorised to run this.
# If spawned by the venv Python itself as a subprocess, the path will be the same.
import sys as _sys, os as _os
_VENV_PYTHON = _os.path.join(_os.path.dirname(_os.path.abspath(__file__)),
                             "..", ".venv", "Scripts", "python.exe")
if _os.path.abspath(_sys.executable).lower() != _os.path.abspath(_VENV_PYTHON).lower():
    print(f"GUARD EXIT: not venv python ({_sys.executable}). Exiting.", flush=True)
    _sys.exit(3)
del _sys, _os, _VENV_PYTHON

import argparse
import json
import os
import subprocess
import sys
import time

import requests
from dotenv import load_dotenv

sys.path.insert(0, os.path.dirname(__file__))
import progress
from db import connect
import p2_generate
import p3b_store_evidence as p3b
import p3c_aggregate as p3c
import p4_publish as p4

BASE = os.path.join(os.path.dirname(__file__), "..")
load_dotenv(os.path.join(BASE, ".env"))

ADS_PER_OCC_TARGET = 60
COUNTRIES = ("us", "gb")
REQ_RATE_SLEEP = 0.5           # ~2 req/s max
TOP_TIER_SIZE = 200
THIN_MARKET_THRESHOLD = 20
JSEARCH_URL = "https://api.openwebninja.com/jsearch/search-v2"
ADZUNA_URL = "https://api.adzuna.com/v1/api/jobs/{country}/search/{page}"


def esco_occupations(cur):
    """All ESCO occupations with slug + altLabels for query building.

    Returns [(slug, preferred_label, [alt_labels...]), ...] ordered
    alphabetically by preferred_label. Slug generation matches p2_generate.py.
    """
    cur.execute(
        "SELECT concept_uri, preferred_label, alt_labels "
        "FROM esco_occupation ORDER BY preferred_label"
    )
    result = []
    for _uri, label, alt_labels_raw in cur.fetchall():
        slug = p2_generate.slugify(label)
        alts = [a.strip() for a in (alt_labels_raw or "").splitlines() if a.strip()]
        result.append((slug, label, alts))
    return result


# ---------------------------------------------------------------------------
# Fetch helpers
# ---------------------------------------------------------------------------

def _jsearch_page(params, api_key, slug, state):
    """Spend one budget token, fetch one JSearch page. Returns payload or None."""
    progress.spend_request(state, slug)
    for attempt in range(3):
        try:
            r = requests.get(JSEARCH_URL, params=params,
                             headers={"x-api-key": api_key}, timeout=60)
        except requests.RequestException as exc:
            print(f"  jsearch retry {attempt+1}: {exc}")
            time.sleep(8 * (attempt + 1))
            continue
        if r.status_code == 429:
            return None
        if r.status_code >= 500:
            time.sleep(8 * (attempt + 1))
            continue
        r.raise_for_status()
        return r.json()
    return None


def _adzuna_ads(slug, label, app_id, app_key):
    """Adzuna fallback (free, no budget cost). Returns list of ad dicts."""
    ads, seen = [], set()
    raw_dir = os.path.join(BASE, "data", "raw", "jobs", slug)
    os.makedirs(raw_dir, exist_ok=True)
    for country in ("gb", "us"):
        for page in range(1, 4):
            cache = os.path.join(raw_dir, f"adzuna_{country}_p{page}.json")
            if os.path.exists(cache):
                payload = json.load(open(cache, encoding="utf-8"))
            else:
                try:
                    r = requests.get(
                        ADZUNA_URL.format(country=country, page=page),
                        params={"app_id": app_id, "app_key": app_key,
                                "what": label, "results_per_page": 50,
                                "content-type": "application/json"},
                        timeout=30)
                    if r.status_code != 200:
                        break
                    payload = r.json()
                    json.dump(payload, open(cache, "w", encoding="utf-8"),
                              ensure_ascii=False, indent=1)
                    time.sleep(0.5)
                except Exception:
                    break
            for job in payload.get("results", []):
                jid = f"adzuna_{job.get('id', '')}"
                desc = job.get("description", "")
                if jid not in seen and desc:
                    seen.add(jid)
                    ads.append({
                        "job_id": jid,
                        "title": job.get("title"),
                        "employer": (job.get("company") or {}).get("display_name"),
                        "country": country,
                        "description": desc,
                    })
            if not payload.get("results"):
                break
    return ads


def _fetch_ads(state, slug, label, alts, api_key, adzuna_id, adzuna_key):
    """Fetch job ads for one occupation (JSearch + optional Adzuna fallback).

    Returns deduped list of {job_id, title, employer, country, description}.
    Every JSearch HTTP call goes through _jsearch_page which calls
    progress.spend_request() BEFORE the request.
    """
    raw_dir = os.path.join(BASE, "data", "raw", "jobs", slug)
    os.makedirs(raw_dir, exist_ok=True)

    ads, seen = [], set()
    terms = [label] + alts[:3]  # preferred + top-3 altLabels (≤10 req/occ per arch)

    if api_key:
        for query in terms:
            if len(ads) >= ADS_PER_OCC_TARGET:
                break
            safe_q = query.replace(" ", "_").replace("/", "-")[:60]
            for country in COUNTRIES:
                if len(ads) >= ADS_PER_OCC_TARGET:
                    break
                cursor = None
                for page in range(1, 6):
                    if len(ads) >= ADS_PER_OCC_TARGET:
                        break
                    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:
                        params = {"query": query, "country": country, "language": "en"}
                        if cursor:
                            params["cursor"] = cursor
                        payload = _jsearch_page(params, api_key, slug, state)
                        if payload is None:
                            break
                        json.dump(payload, open(cache, "w", encoding="utf-8"),
                                  ensure_ascii=False, indent=1)
                        time.sleep(REQ_RATE_SLEEP)

                    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)
                            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

    if len(ads) < THIN_MARKET_THRESHOLD and adzuna_id and adzuna_key:
        for a in _adzuna_ads(slug, label, adzuna_id, adzuna_key):
            if a["job_id"] not in seen:
                seen.add(a["job_id"])
                ads.append(a)

    return ads


# ---------------------------------------------------------------------------
# Stage implementations
# ---------------------------------------------------------------------------

def stage_packages(state, limit=0):
    """Generate package files for ALL 3039 ESCO occupations and push to Gitea.

    Reuses p2_generate.all_occupations (LEFT JOIN to O*NET crosswalk; None if
    no match) and p2_generate.gen_package (handles None onet_id gracefully).
    Gitea pushes via p4_publish: check-before-create, throttled ≥1 s/repo.
    Resumable: occupations already 'done' in progress.json are skipped.
    """
    cn = connect()
    try:
        p4.ensure_org("skills-core", "public")
        p4.ensure_org("skills-community", "public")
        p4.ensure_org("tenant-acme", "private")
    except Exception as exc:
        print(f"WARN: org setup failed ({exc}) — Gitea may be starting up; "
              "packages will still be generated locally and push retried per-package")

    all_occs = p2_generate.all_occupations(cn)
    print(f"ESCO catalog: {len(all_occs)} occupations")

    done = skipped = failed = attempted = 0
    for r in all_occs:
        slug = p2_generate.slugify(r["preferred_label"])
        occ_state = progress.occ(state, slug)
        if occ_state["package"] == "done":
            skipped += 1
            continue

        attempted += 1
        try:
            # If files already on disk (failed = push-only failure), skip re-generation
            manifest_path = os.path.join(p2_generate.SKILLS_DIR, slug, "manifest.json")
            if occ_state["package"] == "failed" and os.path.exists(manifest_path):
                actual_slug = slug
                manifest = json.load(open(manifest_path, encoding="utf-8"))
            else:
                actual_slug = p2_generate.gen_package(cn, r)
                manifest = json.load(open(
                    os.path.join(p2_generate.SKILLS_DIR, actual_slug, "manifest.json"),
                    encoding="utf-8"))
            # Phase 3 publish gate (SKIP_LINT=1 = explicitly-labeled v1-grade
            # republish, e.g. the evidence-crawl catalog refresh)
            if os.environ.get("SKIP_LINT") != "1":
                sys.path.insert(0, os.path.join(BASE, "qa"))
                import lint_package
                lint_findings = lint_package.lint(actual_slug)
                if lint_findings:
                    occ_state["package"] = "failed"
                    occ_state["error"] = f"lint: {lint_findings[0][:150]}"
                    progress.save(state)
                    failed += 1
                    print(f"LINT-BLOCKED {actual_slug}: {lint_findings[0][:100]}")
                    continue
            p4.ensure_repo("skills-core", actual_slug, manifest.get("title", actual_slug))
            p4.push_dir(
                "skills-core", actual_slug,
                os.path.join(p2_generate.SKILLS_DIR, actual_slug),
                f"feat: {actual_slug} skill package v{manifest['version']}")
            occ_state["package"] = "done"
            progress.save(state)
            done += 1
            print(f"[{done}] {actual_slug}")
            time.sleep(1.0)
        except Exception as exc:
            occ_state["package"] = "failed"
            occ_state["error"] = str(exc)[:200]
            progress.save(state)
            failed += 1
            print(f"FAIL {slug}: {exc}")

        if limit and attempted >= limit:
            break

    cn.close()
    print(f"stage_packages: {done} done, {skipped} skipped, {failed} failed")


def stage_evidence(state, slugs=None, limit=0):
    """Per occupation: fetch (cached) -> extract (Ollama) -> store -> aggregate.

    fetch:    preferredLabel + top-3 altLabels × COUNTRIES, cached under
              data/raw/jobs/<slug>/; progress.spend_request BEFORE each HTTP call.
              < THIN_MARKET_THRESHOLD usable ads -> Adzuna fallback -> mark failed.
    extract:  subprocess extract_local.py (Ollama gemma3:27b), result in
              data/evidence/<slug>.jsonl. Skipped if file already exists.
    store:    p3b.store_for_occupation — DELETE+INSERT scoped by occupation_slug.
    aggregate: p3c.aggregate_for_occupation — market.md + tools.md/skills.md sections.
    QA:       qa_sample.py --rate 0.02; two consecutive exit-1 = stop (systemic).
    """
    cn = connect()
    cur = cn.cursor()
    occ_map = {slug: (label, alts) for slug, label, alts in esco_occupations(cur)}
    cn.close()

    api_key = os.environ.get("JSEARCH_API_KEY", "").strip()
    adzuna_id = os.environ.get("ADZUNA_APP_ID", "").strip()
    adzuna_key = os.environ.get("ADZUNA_APP_KEY", "").strip()

    if not api_key:
        print("WARN: JSEARCH_API_KEY not set — only Adzuna fallback available")

    work_list = slugs if slugs else [
        slug for slug, s in state["occupations"].items()
        if s.get("package") == "done" and s.get("evidence") not in ("done",)
    ]
    if not slugs:
        # White collar first: computer-based professions get their market
        # evidence before service/blue collar (manifest "collar", p6_collar.py).
        _prio = {"white": 0, "service": 1, "blue": 2}
        # Homepage-Flaggschiffe zuerst (Landingpage zeigt deren Quellen-Mix live)
        _featured = {"artificial-intelligence-engineer": -2, "lawyer": -1,
                     "database-administrator": -1, "web-designer": -1}
        def _collar(slug):
            try:
                m = json.load(open(os.path.join(BASE, "skills", slug, "manifest.json"),
                                   encoding="utf-8"))
                return _prio.get(m.get("collar"), 3)
            except (OSError, ValueError):
                return 3
        work_list.sort(key=lambda s: (_featured.get(s, 0), _collar(s), s))

    qa_consecutive_fail = 0
    done = 0

    for slug in work_list:
        if limit and done >= limit:
            break
        occ_state = progress.occ(state, slug)
        if occ_state.get("evidence") == "done":
            continue

        label, alts = occ_map.get(slug, (None, []))
        if not label:
            manifest_path = os.path.join(BASE, "skills", slug, "manifest.json")
            if os.path.exists(manifest_path):
                label = (json.load(open(manifest_path, encoding="utf-8"))
                         .get("title", slug.replace("-", " ")))
                alts = []
            else:
                occ_state["evidence"] = "failed"
                occ_state["error"] = "label not found"
                progress.save(state)
                continue

        try:
            # --- FETCH ---
            occ_state["evidence"] = "fetching"
            progress.save(state)

            ads_file = os.path.join(BASE, "data", "raw", "jobs", f"{slug}_ads.json")
            if not os.path.exists(ads_file):
                ads = _fetch_ads(state, slug, label, alts, api_key, adzuna_id, adzuna_key)
                if len(ads) < 5:
                    occ_state["evidence"] = "failed"
                    occ_state["error"] = f"thin market ({len(ads)} ads)"
                    progress.save(state)
                    print(f"SKIP {slug}: thin market ({len(ads)} ads)")
                    continue
                with open(ads_file, "w", encoding="utf-8", newline="\n") as f:
                    json.dump(ads, f, ensure_ascii=False, indent=1)
            else:
                ads = json.load(open(ads_file, encoding="utf-8"))

            # --- RELEVANCE GATE (phase 2a): raw ads stay untouched on disk;
            # only gated-relevant ads reach extraction/store/aggregate ---
            import ad_gate
            ads = ad_gate.gate_ads(slug, ads)[0]
            if len(ads) < 5:
                occ_state["evidence"] = "failed"
                occ_state["error"] = f"thin market after gate ({len(ads)} relevant ads)"
                progress.save(state)
                print(f"SKIP {slug}: thin market after gate ({len(ads)} ads)")
                continue
            gated_ads_file = os.path.join(BASE, "data", "gated", f"{slug}_relevant.json")
            with open(gated_ads_file, "w", encoding="utf-8", newline="\n") as f:
                json.dump(ads, f, ensure_ascii=False, indent=1)

            occ_state["ads"] = len(ads)
            progress.save(state)

            # --- EXTRACT ---
            occ_state["evidence"] = "extracting"
            progress.save(state)

            ext_file = os.path.join(BASE, "data", "evidence", f"{slug}.jsonl")
            os.makedirs(os.path.dirname(ext_file), exist_ok=True)
            if not os.path.exists(ext_file) or os.path.getsize(ext_file) == 0:
                ret = subprocess.run(
                    [sys.executable,
                     os.path.join(BASE, "pipeline", "extract_local.py"),
                     "--ads", gated_ads_file, "--out", ext_file],
                    capture_output=True, text=True)
                # exit 0 = clean; exit 1 = >10% fail rate (still usable); >1 = fatal
                if ret.returncode > 1:
                    raise RuntimeError(
                        f"extract_local exit {ret.returncode}: {ret.stderr[-300:]}")

            extractions = [json.loads(l) for l in
                           open(ext_file, encoding="utf-8") if l.strip()]
            if not extractions:
                occ_state["evidence"] = "failed"
                occ_state["error"] = "extraction: 0 records"
                progress.save(state)
                continue

            # --- STORE + AGGREGATE ---
            occ_state["evidence"] = "aggregating"
            progress.save(state)

            cn2 = connect()
            ads_dict = {a["job_id"]: a for a in ads}
            n_jobs, n_ent = p3b.store_for_occupation(cn2, ads_dict, extractions, slug)
            p3c.aggregate_for_occupation(cn2, slug, p2_generate.SKILLS_DIR)
            cn2.close()

            # --- QA ---
            qa_ret = subprocess.run(
                [sys.executable,
                 os.path.join(BASE, "pipeline", "qa_sample.py"),
                 "--extractions", ext_file, "--ads", ads_file, "--rate", "0.02"],
                capture_output=True, text=True)
            if qa_ret.returncode == 1:
                qa_consecutive_fail += 1
                print(f"QA warn {slug}: {qa_ret.stdout.strip()}")
                if qa_consecutive_fail >= 2:
                    raise RuntimeError(
                        "QA harness failed twice in a row — systemic problem, stopping")
            else:
                qa_consecutive_fail = 0

            occ_state["evidence"] = "done"
            progress.save(state)
            done += 1
            print(f"evidence done: {slug} ({n_jobs} jobs, {n_ent} entities)")

        except progress.BudgetExhausted:
            raise
        except Exception as exc:
            occ_state["evidence"] = "failed"
            occ_state["error"] = str(exc)[:200]
            progress.save(state)
            print(f"FAIL evidence {slug}: {exc}")
            if "systemic" in str(exc):
                break

    print(f"stage_evidence: {done} completed")


def stage_tier(state):
    """Rank occupations for depth staging: tier 1 = TOP_TIER_SIZE by ads collected.

    Considers only occupations with evidence='done'. Updates progress.json.
    Pure bookkeeping — no API calls.
    """
    ranked = sorted(
        [(slug, s.get("ads", 0))
         for slug, s in state["occupations"].items()
         if s.get("evidence") == "done"],
        key=lambda x: -x[1]
    )
    for i, (slug, _ads) in enumerate(ranked):
        state["occupations"][slug]["tier"] = 1 if i < TOP_TIER_SIZE else 3

    n1 = min(len(ranked), TOP_TIER_SIZE)
    n3 = max(0, len(ranked) - TOP_TIER_SIZE)
    print(f"stage_tier: {n1} tier-1, {n3} tier-3 occupations ranked")
    progress.save(state)


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--stage", required=True,
                    choices=["packages", "evidence", "tier"])
    ap.add_argument("--slugs", default="")
    ap.add_argument("--limit", type=int, default=0)
    args = ap.parse_args()

    # Single-instance lock (atomic O_CREAT|O_EXCL prevents race).
    # Only the venv Python is the legitimate runner — system Python impostor exits.
    lock_path = os.path.join(BASE, "data", f"batch_{args.stage}.lock")
    legitimate_exe = os.path.join(BASE, ".venv", "Scripts", "python.exe")
    my_exe = os.path.abspath(sys.executable)
    if my_exe.lower() != os.path.abspath(legitimate_exe).lower():
        print(f"NOT_VENV: this is {my_exe}, not the venv Python. Exiting immediately.")
        sys.exit(3)

    state = progress.load()
    try:
        if args.stage == "packages":
            stage_packages(state, args.limit)
        elif args.stage == "evidence":
            slugs = [s for s in args.slugs.split(",") if s] or None
            stage_evidence(state, slugs, args.limit)
        else:
            stage_tier(state)
    except progress.BudgetExhausted as exc:
        print(f"STOP: {exc}")
        sys.exit(2)
    finally:
        progress.save(state)


if __name__ == "__main__":
    main()