OpenAI Batch API JSONL Input Format & Per-Request Validation Errors

import io
import json
import uuid
from openai import OpenAI

client = OpenAI()


# ✅ Build the JSONL — one request per line
def build_batch_file(prompts: list[dict], model: str = "gpt-5.4") -> bytes:
    """prompts: list of {"id": str, "text": str, ...}
    Returns bytes ready to upload."""
    lines = []
    seen_ids = set()

    for p in prompts:
        cid = p["id"]
        if cid in seen_ids:
            raise ValueError(f"duplicate custom_id: {cid!r}")
        seen_ids.add(cid)
        if len(cid) > 64:
            raise ValueError(f"custom_id too long ({len(cid)} > 64): {cid[:40]}...")

        request = {
            "custom_id": cid,
            "method": "POST",
            "url": "/v1/chat/completions",     # SAME endpoint for every line
            "body": {
                "model": model,
                "messages": [
                    {"role": "user", "content": p["text"]},
                ],
                "max_completion_tokens": 2000,
                # Never include: "stream": true (batch is inherently non-stream)
            },
        }
        lines.append(json.dumps(request, ensure_ascii=False))

    return "\n".join(lines).encode("utf-8")


# ✅ Upload with purpose="batch"
def submit_batch(prompts: list[dict], endpoint: str = "/v1/chat/completions") -> str:
    """Returns batch_id."""
    payload = build_batch_file(prompts)

    if len(payload) > 200 * 1024 * 1024:
        raise ValueError(f"file too large: {len(payload):,} bytes (max 200MB)")

    if len(prompts) > 50_000:
        raise ValueError(f"too many requests: {len(prompts)} (max 50,000)")

    # Upload as file object
    upload = client.files.create(
        file=("batch_input.jsonl", io.BytesIO(payload)),
        purpose="batch",                        # ← required
    )

    # Create batch job
    batch = client.batches.create(
        input_file_id=upload.id,
        endpoint=endpoint,                       # must match every line's url
        completion_window="24h",                 # only "24h" is supported
        metadata={"submitted_by": "prod_pipeline", "batch_purpose": "nightly_classification"},
    )

    return batch.id


# ✅ Responses API in Batch
def build_responses_batch(prompts: list[dict]) -> bytes:
    lines = []
    for p in prompts:
        request = {
            "custom_id": p["id"],
            "method": "POST",
            "url": "/v1/responses",             # different endpoint
            "body": {
                "model": "gpt-5.4",
                "input": p["text"],
                "instructions": p.get("instructions", "You are a helpful assistant."),
                "max_output_tokens": 2000,
            },
        }
        lines.append(json.dumps(request))
    return "\n".join(lines).encode("utf-8")


# ✅ Embeddings batch
def build_embeddings_batch(texts: list[str], model: str = "text-embedding-3-small") -> bytes:
    lines = []
    for i, text in enumerate(texts):
        request = {
            "custom_id": f"emb-{i:07d}",
            "method": "POST",
            "url": "/v1/embeddings",
            "body": {
                "model": model,
                "input": text,
            },
        }
        lines.append(json.dumps(request))
    return "\n".join(lines).encode("utf-8")


# ✅ Auto-chunk large workloads into multiple batches
def submit_in_chunks(prompts: list[dict], chunk_size: int = 40_000) -> list[str]:
    """Split into ≤50K-line batches automatically."""
    batch_ids = []
    for i in range(0, len(prompts), chunk_size):
        chunk = prompts[i:i + chunk_size]
        batch_id = submit_batch(chunk)
        batch_ids.append(batch_id)
    return batch_ids


# ❌ ANTI-PATTERN — manual string building
# NEVER do this:
# line = f'{{"custom_id": "{cid}", "body": {{"messages": [{{"content": "{content}"}}]}}}}'
# ↑ Any quote, newline, or backslash in `content` breaks the JSON.
# Always use json.dumps().


# ✅ Deterministic custom_id generation from source data
def make_custom_id(row: dict) -> str:
    """Stable, unique custom_id derived from source row keys.
    Reproducible: same row → same custom_id, useful for idempotent re-submission."""
    return f"row-{row['user_id']}-{row['record_id']}"

import json
from typing import Any


# Known unsupported per-endpoint params
CHAT_UNSUPPORTED = {"stream", "stream_options"}
RESPONSES_UNSUPPORTED = {"stream"}
EMBED_UNSUPPORTED = {"stream"}
REASONING_MODELS = {"o3", "o3-mini", "o4-mini", "gpt-5.4", "gpt-5.5", "gpt-5.6"}
REASONING_ONLY = {"reasoning_effort"}   # only on reasoning models
NON_REASONING_ONLY = {"temperature", "top_p", "presence_penalty", "frequency_penalty",
                       "logit_bias", "logprobs", "top_logprobs"}


class BatchValidationError(Exception):
    pass


def validate_batch_file(jsonl_bytes: bytes) -> list[dict]:
    """Parse the JSONL, check every line, return list of validation issues.
    Empty list = valid."""
    issues = []
    seen_ids = set()
    seen_endpoints = set()

    lines = jsonl_bytes.decode("utf-8").split("\n")

    if len(lines) > 50_000:
        issues.append({"scope": "file", "issue": f"too many requests: {len(lines)}"})

    if len(jsonl_bytes) > 200 * 1024 * 1024:
        issues.append({"scope": "file", "issue": f"file too large: {len(jsonl_bytes):,} bytes"})

    for i, raw in enumerate(lines):
        if not raw.strip():
            continue
        line_no = i + 1

        try:
            req = json.loads(raw)
        except json.JSONDecodeError as e:
            issues.append({"line": line_no, "issue": f"invalid JSON: {e}"})
            continue

        # Envelope shape
        for field in ("custom_id", "method", "url", "body"):
            if field not in req:
                issues.append({"line": line_no, "issue": f"missing field: {field}"})

        if req.get("method") != "POST":
            issues.append({"line": line_no, "issue": f"method must be POST, got {req.get('method')!r}"})

        # custom_id uniqueness
        cid = req.get("custom_id")
        if cid:
            if cid in seen_ids:
                issues.append({"line": line_no, "issue": f"duplicate custom_id: {cid!r}"})
            seen_ids.add(cid)
            if len(cid) > 64:
                issues.append({"line": line_no, "issue": f"custom_id too long: {len(cid)}"})

        # Endpoint uniformity
        url = req.get("url")
        if url:
            seen_endpoints.add(url)

        # Per-endpoint body validation
        body = req.get("body", {})

        if url == "/v1/chat/completions":
            for p in CHAT_UNSUPPORTED:
                if p in body:
                    issues.append({"line": line_no, "issue": f"unsupported in batch: {p!r}"})

            model = body.get("model", "")
            if any(model.startswith(m) for m in REASONING_MODELS):
                for p in NON_REASONING_ONLY:
                    if p in body:
                        issues.append({
                            "line": line_no,
                            "issue": f"{p!r} not supported on reasoning model {model!r}",
                        })
                if "max_tokens" in body:
                    issues.append({
                        "line": line_no,
                        "issue": f"use max_completion_tokens (not max_tokens) on {model!r}",
                    })

        elif url == "/v1/responses":
            for p in RESPONSES_UNSUPPORTED:
                if p in body:
                    issues.append({"line": line_no, "issue": f"unsupported in batch: {p!r}"})

        elif url == "/v1/embeddings":
            if not body.get("input"):
                issues.append({"line": line_no, "issue": "embeddings request missing input"})

        elif url:
            issues.append({"line": line_no, "issue": f"unsupported batch endpoint: {url}"})

    # Endpoint uniformity (only one)
    if len(seen_endpoints) > 1:
        issues.append({
            "scope": "file",
            "issue": f"mixed endpoints not allowed in one batch: {seen_endpoints}",
        })

    return issues


# ✅ Fail-fast wrapper — refuse to upload if any issues
def submit_batch_safe(jsonl_bytes: bytes, client) -> str:
    issues = validate_batch_file(jsonl_bytes)
    if issues:
        report = "\n".join(f"  · {i.get('line', 'file')}: {i['issue']}" for i in issues[:20])
        more = f"\n  ... and {len(issues) - 20} more" if len(issues) > 20 else ""
        raise BatchValidationError(f"Found {len(issues)} issues:\n{report}{more}")

    import io
    upload = client.files.create(
        file=("batch_input.jsonl", io.BytesIO(jsonl_bytes)),
        purpose="batch",
    )
    batch = client.batches.create(
        input_file_id=upload.id,
        endpoint="/v1/chat/completions",
        completion_window="24h",
    )
    return batch.id


# ✅ Sample a few lines for smoke test before full batch
def smoke_test_sample(jsonl_bytes: bytes, sample_size: int = 3, client=None):
    """Run first N lines synchronously to catch shape errors early."""
    for raw in jsonl_bytes.decode("utf-8").split("\n")[:sample_size]:
        if not raw.strip():
            continue
        req = json.loads(raw)
        try:
            # Actually invoke the endpoint synchronously
            if req["url"] == "/v1/chat/completions":
                client.chat.completions.create(**req["body"])
            elif req["url"] == "/v1/responses":
                client.responses.create(**req["body"])
            print(f"  ✓ {req['custom_id']}")
        except Exception as e:
            print(f"  ✗ {req['custom_id']}: {e}")
            raise

import json


# ❌ ANTI-PATTERN — manual line building
def bad_build(prompts):
    lines = []
    for p in prompts:
        # If content has quotes or newlines, this produces INVALID JSON
        line = (
            '{"custom_id": "' + p["id"] + '", "method": "POST", "url": "/v1/chat/completions", "body": {"model": "gpt-5.4", "messages": [{"role": "user", "content": "'
            + p["text"] +
            '"}]}}'
        )
        lines.append(line)
    return "\n".join(lines).encode()


# ✅ CORRECT — json.dumps handles all escaping
def good_build(prompts):
    lines = []
    for p in prompts:
        req = {
            "custom_id": p["id"],
            "method": "POST",
            "url": "/v1/chat/completions",
            "body": {
                "model": "gpt-5.4",
                "messages": [{"role": "user", "content": p["text"]}],
                "max_completion_tokens": 2000,
            },
        }
        lines.append(json.dumps(req, ensure_ascii=False))
    return "\n".join(lines).encode("utf-8")


# ✅ Round-trip verify — parse what you produced
def build_and_verify(prompts) -> bytes:
    data = good_build(prompts)

    # Verify every line parses
    for i, raw in enumerate(data.decode("utf-8").split("\n")):
        if not raw.strip():
            continue
        try:
            parsed = json.loads(raw)
            assert "custom_id" in parsed, f"line {i+1}: missing custom_id"
        except json.JSONDecodeError as e:
            raise ValueError(f"line {i+1} invalid JSON: {e}")

    return data


# ✅ ensure_ascii=False keeps Unicode readable, still valid JSON
# Compare:
#   json.dumps({"content": "café ☕"})               → '{"content": "caf\u00e9 \u2615"}'
#   json.dumps({"content": "café ☕"}, ensure_ascii=False) → '{"content": "café ☕"}'
# Both are valid; the second is smaller and human-readable.


# ✅ Streaming write for very large batches — don't hold 200MB in memory
import io

def stream_write_batch(prompts_iterable, output_path: str):
    """Write JSONL incrementally to disk, then upload."""
    with open(output_path, "w", encoding="utf-8") as f:
        for p in prompts_iterable:
            req = {
                "custom_id": p["id"],
                "method": "POST",
                "url": "/v1/chat/completions",
                "body": {
                    "model": "gpt-5.4",
                    "messages": [{"role": "user", "content": p["text"]}],
                    "max_completion_tokens": 2000,
                },
            }
            f.write(json.dumps(req, ensure_ascii=False))
            f.write("\n")


# Then upload from file
def upload_from_disk(path: str, client):
    with open(path, "rb") as f:
        return client.files.create(file=f, purpose="batch")


# ✅ Compress oversized batches — Batch accepts uncompressed only, but you can
# split into 40K-line files instead of hitting the 200MB limit
def split_large_prompts(prompts, target_lines: int = 40_000):
    for i in range(0, len(prompts), target_lines):
        yield prompts[i:i + target_lines]