"""Tokenhot documentation-scoped media examples, 2026-10-11. Python 3.13+.

Local preparation/decoding never opens a socket. Network commands need --execute.
No POST or failed GET is automatically retried. Not a production URL-fetch service,
a provider idempotency mechanism, or proof of real model/account behavior.
"""
from __future__ import annotations
import argparse
import base64
import binascii
import io
import ipaddress
import json
import math
import os
from pathlib import Path
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
import uuid
import wave

TTS = "gemini-2.5-flash-preview-tts"
GROK = "grok-imagine-video-1-5-preview"
VEO = "veo3.1"
VIDEO_URL = "https://api.tokenhot.ai/v1/video/generations"
TTS_URL = f"https://api.tokenhot.ai/v1beta/models/{TTS}:generateContent"
MAX_JSON = 64 * 1024 * 1024
MAX_VIDEO = 512 * 1024 * 1024  # Example memory limits, not provider quotas.

class WorkflowError(ValueError):
    pass

class NoRedirect(urllib.request.HTTPRedirectHandler):
    def redirect_request(self, req, fp, code, msg, headers, newurl):
        return None

def obj(value):
    if not isinstance(value, dict):
        raise WorkflowError("Expected a JSON object; inspect the response privately")
    return value

def text(value):
    if not isinstance(value, str) or not value.strip():
        raise WorkflowError("Expected nonempty text")
    return value

def load(path):
    return json.loads(Path(path).read_text(encoding="utf-8"))

def encoded(value):
    return (json.dumps(value, ensure_ascii=False, indent=2, allow_nan=False) + "\n").encode("utf-8")

def fresh_write(path, data):
    """Install complete bytes without replacing another file; hard links required."""
    path = Path(path)
    path.parent.mkdir(parents=True, exist_ok=True)
    temp = path.with_name(path.name + "." + uuid.uuid4().hex + ".part")
    try:
        with temp.open("xb") as stream:
            stream.write(data)
            stream.flush()
            os.fsync(stream.fileno())
        os.link(temp, path)
    finally:
        temp.unlink(missing_ok=True)
    return path

def public_https(url):
    text(url)
    if any(ord(c) <= 32 for c in url):
        raise WorkflowError("URL contains whitespace/control characters")
    p = urllib.parse.urlsplit(url)
    if p.scheme != "https" or not p.hostname or p.username is not None or p.password is not None:
        raise WorkflowError("Use HTTPS without embedded credentials")
    if p.fragment or p.port not in (None, 443):
        raise WorkflowError("Fragments and nonstandard ports are outside this example")
    host = p.hostname.lower().rstrip(".")
    if host == "localhost" or host.endswith((".localhost", ".local", ".internal")) or "." not in host:
        raise WorkflowError("Use a public hostname")
    try:
        address = ipaddress.ip_address(host)
    except ValueError:
        pass
    else:
        if not address.is_global:
            raise WorkflowError("Private/special-use IP URLs are not accepted")
    # Lexical guard only: this does not resolve DNS or prevent DNS rebinding.
    return url

def task_id(value):
    value = text(value)
    if len(value) > 512 or any(ord(c) < 33 for c in value):
        raise WorkflowError("Unexpected task ID shape")
    return value

def tts_request(spoken):
    return {"contents": [{"parts": [{"text": text(spoken)}]}],
            "generationConfig": {"responseModalities": ["AUDIO"],
                "speechConfig": {"voiceConfig": {"prebuiltVoiceConfig": {"voiceName": "Zephyr"}}}}}

def grok_request(image, prompt):
    return {"model": GROK, "input": {"image_urls": [public_https(image)],
        "prompt": text(prompt), "aspect_ratio": "16:9", "resolution": "720p", "duration": 8}}

def veo_request(prompt):
    return {"model": VEO, "prompt": text(prompt), "aspect_ratio": "16:9", "resolution": "1080p"}

def submission_id(response, model):
    response = obj(response)
    if response.get("model") != model:
        raise WorkflowError("Submission model absent/different; inspect response")
    # Documentation examples distinguish id from task_id. No guessed fallback.
    return task_id(response.get("task_id"))

def query_data(response, expected_id, model):
    response = obj(response)
    if response.get("code") != "success":
        raise WorkflowError("Query envelope is not successful")
    data = obj(response.get("data"))
    if data.get("task_id") != expected_id or data.get("model") != model:
        raise WorkflowError("Query task/model mismatch")
    state = text(data.get("status"))
    if data.get("fail_reason") or state == "FAILURE":
        raise WorkflowError("Query reports failure; inspect saved response")
    if model == VEO and state not in {"NOT_START", "IN_PROGRESS", "SUCCESS"}:
        raise WorkflowError("Undocumented Veo state")
    if state == "SUCCESS":
        public_https(data.get("result_url"))
    return data

def extension_request(parent_response, original_id, prompt):
    parent = query_data(parent_response, task_id(original_id), VEO)
    if parent["status"] != "SUCCESS":
        raise WorkflowError("This workflow extends only a confirmed successful parent")
    return {"model": VEO, "taskId": original_id, "prompt": text(prompt)}

def validate_request(kind, body):
    body = obj(body)
    if kind == "tts":
        try:
            spoken = body["contents"][0]["parts"][0]["text"]
        except (KeyError, IndexError, TypeError):
            raise WorkflowError("Unexpected TTS request shape") from None
        if body != tts_request(spoken):
            raise WorkflowError("Only the prepared single-voice request is accepted")
    elif kind == "grok":
        inp = obj(body.get("input"))
        images = inp.get("image_urls")
        if not isinstance(images, list) or len(images) != 1:
            raise WorkflowError("Exactly one image URL is required")
        if body != grok_request(images[0], inp.get("prompt")):
            raise WorkflowError("Unexpected Grok settings; recheck route before modifying")
    elif kind == "veo":
        if body != veo_request(body.get("prompt")):
            raise WorkflowError("Unexpected Veo generation settings")
    elif kind == "extend":
        task_id(body.get("taskId")); text(body.get("prompt"))
        if body.get("model") != VEO or set(body) != {"model", "taskId", "prompt"}:
            raise WorkflowError("Unexpected extension request shape")
    else:
        raise WorkflowError("Unknown request kind")

def transport(method, url, body=None, timeout=60, api=True, approved_host=None):
    """One HTTP attempt, redirects refused; preserve error bodies for local diagnosis."""
    headers = {"Accept": "application/json" if api else "video/*"}
    if api:
        if not (url == TTS_URL or url == VIDEO_URL or url.startswith(VIDEO_URL + "/")):
            raise WorkflowError("API URL outside fixed endpoints")
        key = os.environ.get("TOKENHOT_API_KEY")
        if not key:
            raise WorkflowError("Set TOKENHOT_API_KEY before network commands")
        headers["Authorization"] = "Bearer " + key
    else:
        public_https(url)
        if urllib.parse.urlsplit(url).hostname != approved_host:
            raise WorkflowError("Download hostname does not match --approve-host")
    if body is not None:
        headers["Content-Type"] = "application/json"
    request = urllib.request.Request(url, data=body, headers=headers, method=method)
    opener = urllib.request.build_opener(NoRedirect())
    try:
        response = opener.open(request, timeout=timeout)
    except urllib.error.HTTPError as exc:
        response = exc
    limit = MAX_JSON if api else MAX_VIDEO
    with response:
        raw = response.read(limit + 1)
        if len(raw) > limit:
            raise WorkflowError("Example response-size limit exceeded")
        return response.code, dict(response.headers), raw

def record_http(run, status, headers, raw):
    fresh_write(run / "response.json", raw)
    fresh_write(run / "http.json", encoded({"status": status,
        "content_type": next((v for k, v in headers.items() if k.lower() == "content-type"), None)}))
    if not 200 <= status < 300:
        raise WorkflowError(f"HTTP {status}; response saved, not retried")
    return json.loads(raw)

def submit(kind, request, run, execute=False, send=transport):
    if not execute:
        raise WorkflowError("Network action requires --execute")
    validate_request(kind, request)
    run = Path(run)
    run.parent.mkdir(parents=True, exist_ok=True)
    run.mkdir()  # Never reuse an existing run for another POST.
    fresh_write(run / "request.json", encoded(request))
    fresh_write(run / "attempt.json", encoded({"kind": kind, "state": "reserved_outcome_unknown"}))
    try:
        status, headers, raw = send("POST", TTS_URL if kind == "tts" else VIDEO_URL,
                                   body=encoded(request), timeout=180)
        result = record_http(run, status, headers, raw)
        if kind != "tts":
            model = GROK if kind == "grok" else VEO
            new_id = submission_id(result, model)
            if kind == "extend" and new_id == request["taskId"]:
                raise WorkflowError("Extension did not return a distinct child task ID")
            fresh_write(run / "task.json", encoded({"task_id": new_id, "model": model,
                "parent_task_id": request.get("taskId")}))
        return result
    except (OSError, ValueError, urllib.error.URLError) as exc:
        fresh_write(run / "outcome.json", encoded({"outcome": "inspect_saved_files_or_reconcile",
            "exception_type": type(exc).__name__}))
        raise WorkflowError("POST not retried; reconcile before deliberate resubmission") from None

def poll(submission, model, out, attempts=12, interval=5.0, execute=False, send=transport, sleep=time.sleep):
    if not execute:
        raise WorkflowError("Network action requires --execute")
    if model not in (GROK, VEO) or type(attempts) is not int or not 1 <= attempts <= 120:
        raise WorkflowError("Choose a supported model and 1..120 GET attempts")
    if not math.isfinite(interval) or not 0 <= interval <= 60:
        raise WorkflowError("Interval must be finite, between 0 and 60 seconds")
    tid = submission_id(submission, model)
    out = Path(out); out.mkdir(parents=True, exist_ok=False)
    for index in range(attempts):
        folder = out / f"query-{index + 1:03d}"; folder.mkdir()
        status, headers, raw = send("GET", VIDEO_URL + "/" + urllib.parse.quote(tid, safe=""), timeout=60)
        reply = record_http(folder, status, headers, raw)
        data = query_data(reply, tid, model)
        if data["status"] == "SUCCESS":
            fresh_write(out / "success.json", raw)
            return reply
        # Grok has no complete intermediate enum in the read documentation.
        # Record non-SUCCESS without inferring completion. Never send another POST.
        if index + 1 < attempts:
            sleep(interval)
    raise WorkflowError("GET attempt budget reached; keep original task")

def pcm_wav(raw):
    if raw[:4] != b"RIFF" or raw[8:12] != b"WAVE":
        raise WorkflowError("Expected RIFF/WAVE container")
    if len(raw) < 12 or int.from_bytes(raw[4:8], "little") + 8 != len(raw):
        raise WorkflowError("Inconsistent RIFF size")
    pos, sizes = 12, []
    while pos + 8 <= len(raw):
        size = int.from_bytes(raw[pos + 4:pos + 8], "little")
        end = pos + 8 + size
        if end > len(raw):
            raise WorkflowError("Truncated WAV chunk")
        if raw[pos:pos + 4] == b"data":
            sizes.append(size)
        pos = end + size % 2
    with wave.open(io.BytesIO(raw), "rb") as stream:
        align = stream.getnchannels() * stream.getsampwidth()
        frames = stream.getnframes()
        if not frames or len(sizes) != 1 or sizes[0] % align:
            raise WorkflowError("Empty or misaligned PCM frame data")
        if len(stream.readframes(frames + 1)) != frames * align:
            raise WorkflowError("WAV frames incomplete")
    return raw

def audio_bytes(response, raw_pcm=None):
    """Decode one audio part in candidate zero; never concatenate alternatives."""
    response = obj(response)
    try:
        candidate = obj(response["candidates"][0])
        if candidate.get("finishReason") != "STOP":
            raise WorkflowError("Audio candidate is not a completed STOP response")
        parts = candidate["content"]["parts"]
        audios = [p["inlineData"] for p in parts if isinstance(p, dict) and "inlineData" in p]
        if len(audios) != 1:
            raise WorkflowError("Expected one inline audio part")
        audio = obj(audios[0])
        mime = text(audio["mimeType"]).split(";", 1)[0].strip().lower()
        raw = base64.b64decode(text(audio["data"]), validate=True)
    except (KeyError, IndexError, TypeError, binascii.Error):
        raise WorkflowError("Missing/invalid candidate audio; inspect saved JSON") from None
    if not raw:
        raise WorkflowError("Empty decoded audio")
    if mime in {"audio/wav", "audio/x-wav", "audio/wave"}:
        return pcm_wav(raw), ".wav"
    if mime in {"audio/mp3", "audio/mpeg"}:
        # Signature sanity check only: neither a codec validation nor a listening test.
        if not (raw.startswith(b"ID3") and len(raw) > 10 or
                len(raw) > 4 and raw[0] == 255 and raw[1] & 224 == 224):
            raise WorkflowError("Audio labeled MP3 lacks a recognizable signature")
        return raw, ".mp3"
    if mime in {"audio/l16", "audio/pcm"}:
        if raw_pcm is None:
            raise WorkflowError("Raw audio needs separately confirmed signed-16 LE rate/channels")
        rate, channels = raw_pcm
        if type(rate) is not int or type(channels) is not int or not 8000 <= rate <= 192000 or channels not in (1, 2):
            raise WorkflowError("Invalid explicit raw PCM settings")
        if raw[:4] == b"RIFF" or len(raw) % (2 * channels):
            raise WorkflowError("Raw PCM bytes are already wrapped or incomplete")
        out = io.BytesIO()
        with wave.open(out, "wb") as stream:
            stream.setnchannels(channels); stream.setsampwidth(2)
            stream.setframerate(rate); stream.writeframes(raw)
        return out.getvalue(), ".wav"
    raise WorkflowError("Unsupported MIME type; do not guess an encoding")

def save_audio(response, output_stem, raw_pcm=None):
    raw, suffix = audio_bytes(response, raw_pcm)
    stem = Path(output_stem)
    if stem.suffix:
        raise WorkflowError("--output-stem must not contain an extension")
    return fresh_write(Path(str(stem) + suffix), raw)

def download(response, tid, model, output, approve_host, execute=False, send=transport):
    if not execute:
        raise WorkflowError("Network action requires --execute")
    data = query_data(response, task_id(tid), model)
    if data["status"] != "SUCCESS":
        raise WorkflowError("Download requires a successful task")
    url = public_https(data["result_url"])
    if urllib.parse.urlsplit(url).hostname != approve_host:
        raise WorkflowError("Confirm result hostname with --approve-host")
    if Path(output).exists():
        raise FileExistsError("Output exists")
    status, headers, raw = send("GET", url, timeout=60, api=False, approved_host=approve_host)
    if not 200 <= status < 300:
        raise WorkflowError(f"Download HTTP {status}; generation not repeated")
    if len(raw) < 16 or raw[4:8] != b"ftyp":
        raise WorkflowError("Missing expected MP4-family signature")
    return fresh_write(output, raw)

def main(argv=None):
    p = argparse.ArgumentParser(description=__doc__)
    sub = p.add_subparsers(dest="command", required=True)
    q = sub.add_parser("tts-request"); q.add_argument("--text-file", required=True); q.add_argument("--out", required=True)
    q = sub.add_parser("grok-request"); q.add_argument("--image-url", required=True); q.add_argument("--prompt-file", required=True); q.add_argument("--out", required=True)
    q = sub.add_parser("veo-request"); q.add_argument("--prompt-file", required=True); q.add_argument("--out", required=True)
    q = sub.add_parser("extend-request"); q.add_argument("--parent-success", required=True); q.add_argument("--parent-task", required=True); q.add_argument("--prompt-file", required=True); q.add_argument("--out", required=True)
    q = sub.add_parser("submit"); q.add_argument("--kind", choices=["tts", "grok", "veo", "extend"], required=True); q.add_argument("--request", required=True); q.add_argument("--run", required=True); q.add_argument("--execute", action="store_true")
    q = sub.add_parser("poll"); q.add_argument("--submission", required=True); q.add_argument("--model", choices=[GROK, VEO], required=True); q.add_argument("--out", required=True); q.add_argument("--attempts", type=int, default=12); q.add_argument("--interval", type=float, default=5); q.add_argument("--execute", action="store_true")
    q = sub.add_parser("audio"); q.add_argument("--response", required=True); q.add_argument("--output-stem", required=True); q.add_argument("--pcm-s16le", nargs=2, type=int, metavar=("RATE", "CHANNELS"))
    q = sub.add_parser("download"); q.add_argument("--success", required=True); q.add_argument("--task", required=True); q.add_argument("--model", choices=[GROK, VEO], required=True); q.add_argument("--output", required=True); q.add_argument("--approve-host", required=True); q.add_argument("--execute", action="store_true")
    a = p.parse_args(argv)
    read_text = lambda path: Path(path).read_text(encoding="utf-8")
    try:
        if a.command == "tts-request":
            fresh_write(a.out, encoded(tts_request(read_text(a.text_file))))
        elif a.command == "grok-request":
            fresh_write(a.out, encoded(grok_request(a.image_url, read_text(a.prompt_file))))
        elif a.command == "veo-request":
            fresh_write(a.out, encoded(veo_request(read_text(a.prompt_file))))
        elif a.command == "extend-request":
            fresh_write(a.out, encoded(extension_request(load(a.parent_success), a.parent_task, read_text(a.prompt_file))))
        elif a.command == "submit":
            submit(a.kind, load(a.request), a.run, a.execute)
        elif a.command == "poll":
            poll(load(a.submission), a.model, a.out, a.attempts, a.interval, a.execute)
        elif a.command == "audio":
            print(save_audio(load(a.response), a.output_stem, a.pcm_s16le))
        else:
            print(download(load(a.success), a.task, a.model, a.output, a.approve_host, a.execute))
        return 0
    except (WorkflowError, OSError, ValueError, wave.Error, urllib.error.URLError) as exc:
        print(f"Stopped ({type(exc).__name__}); inspect local files privately. No automatic resubmission.", file=sys.stderr)
        return 2

if __name__ == "__main__":
    raise SystemExit(main())