import os
import time
import requests

BASE = "https://batch.search.hereapi.com/v7/batch/jobs"
API_KEY = os.environ["HERE_API_KEY"]

# Selects the service that processes the job.
SERVICE_GEOCODE = "hrn:here:service::olp-here:search-geocode-7"


class EntitlementError(Exception): ...


def build_input(records: list[tuple[str, str, str]]) -> str:
    """
    Pipe-delimited, with a header row.
    records: [(recId, query, countryCode), ...]

    Deduplicate and normalize BEFORE you get here.
    """
    lines = ["recId|q|country"]
    lines += [f"{rid}|{q}|{c}" for rid, q, c in records]
    return "\n".join(lines)


def submit_job(payload: str, start_immediately: bool = True) -> dict:
    """
    POST the input as the request body. Returns {id, status, href}.

    PERSIST THE JOB ID BEFORE POLLING. If your worker dies between here and
    the first poll, the job is running on HERE's side regardless.
    Resubmitting a 4M-record job bills 4M records again.
    """
    params = {
        "serviceHrn": SERVICE_GEOCODE,
        "inputDelimiter": "|",
        "outputDelimiter": "|",
        "outputColumns": "|".join([
            "position", "title", "id",
            "addressLabel", "addressHouseNumber", "addressStreet",
            "addressCity", "addressPostalCode", "addressCountryCode",
        ]),
        "outputType": "csv",
        "startJob": str(start_immediately).lower(),
        "apiKey": API_KEY,
    }

    resp = requests.post(
        BASE, params=params,
        data=payload.encode("utf-8"),
        headers={"Content-Type": "text/plain", "Accept": "application/json"},
        timeout=120,
    )
    if resp.status_code == 403:
        raise EntitlementError("Batch not entitled")
    if resp.status_code == 429:
        # Concurrent job limit. QUEUE. Do not hammer.
        raise RuntimeError("Concurrency limit — queue this job")
    resp.raise_for_status()

    return resp.json()   # {"id": ..., "status": "submitted"|"queued", "href": ...}


def poll_status(job_id: str, max_wait_s: int = 7200) -> str:
    """Poll with backoff. Returns terminal status."""
    backoff, waited = 5.0, 0.0
    while waited < max_wait_s:
        resp = requests.get(f"{BASE}/{job_id}",
                            params={"apiKey": API_KEY}, timeout=30)
        resp.raise_for_status()
        status = resp.json()["status"]

        if status in ("success", "failed", "cancelled"):
            return status

        time.sleep(backoff)
        waited += backoff
        backoff = min(backoff * 1.4, 60.0)

    raise TimeoutError(f"Job {job_id} not terminal after {max_wait_s}s")


def fetch_errors(job_id: str) -> str | None:
    """
    204 means ZERO ERRORS. It is not a failure.

    Code that treats any non-200 as failure will log a perfectly successful
    job as broken, every night, until someone reads the spec.
    """
    resp = requests.get(f"{BASE}/{job_id}/errors",
                        params={"apiKey": API_KEY}, timeout=60)
    if resp.status_code == 204:
        return None       # Celebrate.
    resp.raise_for_status()
    return resp.text


def fetch_results(job_id: str, out_path: str) -> None:
    """
    404 means the job has not succeeded yet. Poll status first.
    Stream — 4M records do not fit in memory.
    """
    resp = requests.get(f"{BASE}/{job_id}/result",
                        params={"apiKey": API_KEY}, stream=True, timeout=300)
    if resp.status_code == 404:
        raise RuntimeError("Results not ready — job has not succeeded")
    resp.raise_for_status()

    with open(out_path, "wb") as f:
        for chunk in resp.iter_content(chunk_size=1 << 20):
            f.write(chunk)


def delete_job(job_id: str) -> None:
    """Step six. Jobs accumulate."""
    requests.delete(f"{BASE}/{job_id}", params={"apiKey": API_KEY}, timeout=30)


if __name__ == "__main__":
    records = [
        ("0001", "Invalidenstraße 116 10115 Berlin", "DEU"),
        ("0002", "425 W Randolph St Chicago IL 60606", "USA"),
        ("0003", "One Main Street Cambridge MA 02142", "USA"),
    ]

    job = submit_job(build_input(records))
    job_id = job["id"]
    # ---- PERSIST job_id HERE, before the next line. ----
    print(f"Job {job_id} — persist this")

    status = poll_status(job_id)
    if status != "success":
        raise SystemExit(f"Job ended: {status}")

    errors = fetch_errors(job_id)
    if errors:
        print("Partial failure — route these to the exception queue:")
        print(errors)   # Do NOT discard 2.9M good results over 100k bad ones

    fetch_results(job_id, "geocoded.csv")
    delete_job(job_id)

# 1. Create + start in one call (startJob=true)
resp=$(curl -sX POST \
  "https://batch.search.hereapi.com/v7/batch/jobs?\
serviceHrn=hrn:here:service::olp-here:search-geocode-7&\
inputDelimiter=|&outputDelimiter=|&\
outputColumns=position|title|id&\
outputType=csv&startJob=true&apiKey=${HERE_API_KEY}" \
  -H 'Accept: application/json' \
  -H 'Content-Type: text/plain' \
  --data 'recId|q|country
0001|Invalidenstraße 116 10115 Berlin|DEU
0002|425 W Randolph St Chicago IL 60606|USA')

jobId=$(echo "$resp" | jq -r '.id')
echo "PERSIST THIS: $jobId"

# 2. Poll status until "success"
curl -s "https://batch.search.hereapi.com/v7/batch/jobs/${jobId}?apiKey=${HERE_API_KEY}" | jq

# 3. Errors. HTTP 204 = zero errors. Not a failure.
curl -sw '%{http_code}' \
  "https://batch.search.hereapi.com/v7/batch/jobs/${jobId}/errors?apiKey=${HERE_API_KEY}"

# 4. Results. 404 until the job succeeds.
curl -s -o geocoded.csv \
  "https://batch.search.hereapi.com/v7/batch/jobs/${jobId}/result?apiKey=${HERE_API_KEY}"

# 5. Clean up
curl -sX DELETE \
  "https://batch.search.hereapi.com/v7/batch/jobs/${jobId}?apiKey=${HERE_API_KEY}"