🏷️ Tag Customer Reviews Overnight with the ZeroGPU Batch API

import os
from getpass import getpass

# Prompt for the ZeroGPU API key securely
zerogpu_api_key = getpass('Enter your ZeroGPU API key: ')
os.environ["ZEROGPU_API_KEY"] = zerogpu_api_key

# Optional: press Enter to skip and run unscoped
zerogpu_project_id = getpass('Enter your ZeroGPU Project ID (optional): ')
if zerogpu_project_id:
    os.environ["ZEROGPU_PROJECT_ID"] = zerogpu_project_id

import os, requests

headers = {
    "x-api-key":    os.environ["ZEROGPU_API_KEY"],
    "content-type": "application/json",
}
# x-project-id is optional; send it only when scoping to a project
if os.environ.get("ZEROGPU_PROJECT_ID"):
    headers["x-project-id"] = os.environ["ZEROGPU_PROJECT_ID"]

resp = requests.post(
    "https://api.zerogpu.ai/v1/chat/completions",
    headers=headers,
    json={
        "model":       "LFM2.5-1.2B-Instruct",
        "temperature": 0,
        "messages": [
            {"role": "system", "content": (
                "You are a review tagger. Respond with a single JSON object and "
                'nothing else, in this shape: { "sentiment": "...", "topics": ["..."] }. '
                'sentiment is one of: "positive", "neutral", "negative".'
            )},
            {"role": "user", "content": "Battery lasts two days, love it."},
        ],
    },
).json()

print(resp["choices"][0]["message"]["content"])

import time

def poll(batch_id: str, interval: int = 30) -> dict:
    while True:
        resp = requests.get(f"{BASE}/v1/batches/{batch_id}", headers=HEADERS)
        resp.raise_for_status()
        batch = resp.json()
        counts = batch.get("request_counts", {})
        print(
            f"{batch['status']}: "
            f"{counts.get('completed', 0)}/{counts.get('total', 0)} done, "
            f"{counts.get('failed', 0)} failed"
        )
        if batch["status"] in ("completed", "failed", "expired", "cancelled"):
            return batch
        time.sleep(interval)

import csv
import json
import os
import time

import requests

BASE    = "https://api.zerogpu.ai"
HEADERS = {"x-api-key": os.environ["ZEROGPU_API_KEY"]}
# x-project-id is optional; send it only when scoping to a project
if os.environ.get("ZEROGPU_PROJECT_ID"):
    HEADERS["x-project-id"] = os.environ["ZEROGPU_PROJECT_ID"]
MODEL = "LFM2.5-1.2B-Instruct"

SYSTEM_PROMPT = (
    "You are a review tagger. For each customer review, respond with a "
    "single JSON object and nothing else, in this exact shape:\n"
    '{ "sentiment": "...", "topics": ["..."] }\n'
    'sentiment is one of: "positive", "neutral", "negative". '
    "topics is a list of 1-3 short lowercase strings (for example: "
    '"battery", "shipping", "support", "build quality", "app", "price"). '
    "No prose, no code fences, no extra keys."
)


def build_jsonl(csv_path: str, jsonl_path: str) -> int:
    written = 0
    seen_ids: set[str] = set()

    with open(csv_path, encoding="utf-8") as src, \
         open(jsonl_path, "w", encoding="utf-8") as dst:

        reader = csv.DictReader(src)

        for row in reader:
            review_id = (row.get("review_id") or "").strip()

            if not review_id:
                print("skip: empty review_id")
                continue

            if review_id in seen_ids:
                print(f"skip {review_id}: duplicate review_id")
                continue

            seen_ids.add(review_id)

            review = (row.get("review") or "").strip()
            if not review:
                print(f"skip {review_id}: empty review text")
                continue

            line = {
                "custom_id": review_id,
                "method": "POST",
                "url": "/v1/chat/completions",
                "body": {
                    "model": MODEL,
                    "temperature": 0,
                    "messages": [
                        {"role": "system", "content": SYSTEM_PROMPT},
                        {"role": "user", "content": review},
                    ],
                },
            }

            dst.write(json.dumps(line) + "\n")
            written += 1

    return written


def upload(jsonl_path: str) -> str:
    with open(jsonl_path, "rb") as f:
        resp = requests.post(
            f"{BASE}/v1/files",
            headers=HEADERS,
            data={"purpose": "batch"},
            files={"file": (os.path.basename(jsonl_path), f, "application/jsonl")},
        )
    resp.raise_for_status()
    return resp.json()["id"]


def create_batch(file_id: str) -> dict:
    resp = requests.post(
        f"{BASE}/v1/batches",
        headers={**HEADERS, "content-type": "application/json"},
        json={
            "input_file_id":     file_id,
            "endpoint":          "/v1/chat/completions",
            "completion_window": "24h",
            "metadata":          {"job": "review-tagging"},
        },
    )
    resp.raise_for_status()
    return resp.json()


def poll(batch_id: str, interval: int = 30) -> dict:
    while True:
        resp = requests.get(f"{BASE}/v1/batches/{batch_id}", headers=HEADERS)
        resp.raise_for_status()
        batch = resp.json()
        counts = batch.get("request_counts", {})
        print(
            f"{batch['status']}: "
            f"{counts.get('completed', 0)}/{counts.get('total', 0)} done, "
            f"{counts.get('failed', 0)} failed"
        )
        if batch["status"] in ("completed", "failed", "expired", "cancelled"):
            return batch
        time.sleep(interval)


def download(file_id: str) -> str:
    resp = requests.get(f"{BASE}/v1/files/{file_id}/content", headers=HEADERS)
    resp.raise_for_status()
    return resp.text


def parse_tag(content: str) -> dict | None:
    """Model is asked for JSON only, but defensively strip code fences."""
    text = content.strip()
    if text.startswith("```"):
        text = text.strip("`")
        if text.startswith("json"):
            text = text[4:]
    try:
        obj = json.loads(text)
    except json.JSONDecodeError:
        return None
    if not isinstance(obj, dict):
        return None
    sentiment = obj.get("sentiment")
    topics    = obj.get("topics")
    if sentiment not in ("positive", "neutral", "negative"):
        return None
    if not isinstance(topics, list):
        return None
    return {"sentiment": sentiment, "topics": [str(t) for t in topics[:3]]}


def main() -> None:
    n = build_jsonl("reviews.csv", "input.jsonl")
    print(f"wrote {n} requests to input.jsonl")

    file_id = upload("input.jsonl")
    print(f"uploaded {file_id}")

    batch = create_batch(file_id)
    print(f"created {batch['id']}")

    batch = poll(batch["id"])
    if batch["status"] != "completed":
        raise SystemExit(f"batch ended with status {batch['status']}")

    # Output order is NOT input order. Key everything by custom_id.
    tagged: dict[str, dict] = {}
    if batch.get("output_file_id"):
        for line in download(batch["output_file_id"]).splitlines():
            if not line.strip():
                continue
            rec = json.loads(line)
            # Batch API returns OpenAI-style responses for /v1/chat/completions
            content = rec["response"]["body"]["choices"][0]["message"]["content"]
            tag = parse_tag(content)
            if tag is not None:
                tagged[rec["custom_id"]] = tag

    failures: list[dict] = []
    if batch.get("error_file_id"):
        for line in download(batch["error_file_id"]).splitlines():
            if not line.strip():
                continue
            rec = json.loads(line)
            failures.append({
                "review_id": rec["custom_id"],
                "code":      (rec.get("error") or {}).get("code", ""),
                "message":   (rec.get("error") or {}).get("message", ""),
            })

    # Merge: walk the CSV in its original order, attach tags by custom_id.
    with open("reviews.csv",  encoding="utf-8") as src, \
         open("tagged.csv", "w", encoding="utf-8", newline="") as dst:
        reader = csv.DictReader(src)
        writer = csv.writer(dst)
        writer.writerow(["review_id", "review", "sentiment", "topics"])
        for row in reader:
            tag = tagged.get(row["review_id"], {})
            writer.writerow([
                row["review_id"],
                row.get("review", ""),
                tag.get("sentiment", ""),
                "|".join(tag.get("topics", [])),
            ])

    if failures:
        with open("failed.csv", "w", encoding="utf-8", newline="") as f:
            writer = csv.DictWriter(f, fieldnames=["review_id", "code", "message"])
            writer.writeheader()
            writer.writerows(failures)
        print(f"{len(failures)} rows failed, see failed.csv")

    print(f"done. tagged {len(tagged)}/{n} reviews -> tagged.csv")


if __name__ == "__main__":
    main()

import json
import os

import requests

BASE    = "https://api.zerogpu.ai"
HEADERS = {"x-api-key": os.environ["ZEROGPU_API_KEY"]}
# x-project-id is optional; send it only when scoping to a project
if os.environ.get("ZEROGPU_PROJECT_ID"):
    HEADERS["x-project-id"] = os.environ["ZEROGPU_PROJECT_ID"]


def errored_ids(error_file_id: str) -> set[str]:
    resp = requests.get(
        f"{BASE}/v1/files/{error_file_id}/content",
        headers=HEADERS
    )
    resp.raise_for_status()

    text = resp.text

    return {
        json.loads(line)["custom_id"]
        for line in text.splitlines() if line.strip()
    }


def retry(input_jsonl: str, error_file_id: str) -> str:
    ids = errored_ids(error_file_id)
    with open(input_jsonl, encoding="utf-8") as src, \
         open("retry.jsonl", "w", encoding="utf-8") as dst:
        for line in src:
            rec = json.loads(line)
            if rec["custom_id"] in ids:
                dst.write(line)
    # ... upload retry.jsonl and create a new batch the same way as before.
    return "retry.jsonl"