"""Generate and download one image through a documented Tokenhot Nano Banana route. Each new generation needs unused output paths. The raw response is persisted before any image GET; repeated POSTs are never triggered automatically. """ from __future__ import annotations import argparse import http.client import json import os import socket import sys import urllib.error import urllib.parse import urllib.request from pathlib import Path from typing import Any API_BASE = "https://api.tokenhot.ai/v1beta/models" MODEL_IDS = ("nano-banana-pro", "nano-banana-2") MIME_EXTENSIONS = { "image/jpeg": ".jpg", "image/png": ".png", "image/webp": ".webp", } NETWORK_READ_ERRORS = ( TimeoutError, socket.timeout, urllib.error.URLError, OSError, http.client.HTTPException, ) class StageFailure(Exception): """A sanitized, reader-facing network or file stage failure.""" def save_new_bytes(path: Path, data: bytes) -> None: """Create a new artifact without replacing an existing file.""" created = False try: with path.open("xb") as file: created = True file.write(data) except OSError as error: if created: try: path.unlink(missing_ok=True) except OSError: pass raise StageFailure("could not save a new output artifact") from error def response_path_for(output_path: Path) -> Path: return output_path.with_name(output_path.name + ".response.json") def require_fresh_generation_paths(output_path: Path, response_path: Path) -> None: occupied = [path for path in (output_path, response_path) if path.exists()] if occupied: targets = ", ".join(str(path) for path in occupied) recovery_tip = ( "Choose a fresh --output path for a new generation, or use --resume-response " "for GET-only recovery if that saved response is the one you need." if response_path in occupied else "No saved response exists at the derived path; choose a fresh --output path." ) raise StageFailure( f"refusing a new POST because this output path already exists: {targets}. " + recovery_tip ) def response_image_url(document: object) -> str: """Find the HTTPS URL in the Tokenhot docs' documented text response.""" if not isinstance(document, dict): raise ValueError("response JSON is not an object") candidates = document.get("candidates") if not isinstance(candidates, list): raise ValueError("response has no candidates array") for candidate in candidates: if not isinstance(candidate, dict): continue content = candidate.get("content") if not isinstance(content, dict): continue parts = content.get("parts") if not isinstance(parts, list): continue for part in parts: if not isinstance(part, dict): continue text = part.get("text") if isinstance(text, str) and text.startswith("https://"): parsed = urllib.parse.urlsplit(text) if parsed.scheme == "https" and parsed.netloc: return text raise ValueError("no documented HTTPS image URL found in candidate text parts") def post_once( url: str, token: str, payload: dict[str, Any], response_path: Path ) -> bytes: """Make exactly one POST and save its response body without echoing it.""" request = urllib.request.Request( url, data=json.dumps(payload).encode("utf-8"), headers={ "Authorization": f"Bearer {token}", "Content-Type": "application/json", }, method="POST", ) try: response_context = urllib.request.urlopen(request, timeout=180) except urllib.error.HTTPError as error: status = error.code try: raw = error.read() except NETWORK_READ_ERRORS as read_error: partial = getattr(read_error, "partial", b"") saved = False if isinstance(partial, bytes): try: save_new_bytes(response_path, partial) saved = True except StageFailure: pass artifact_note = ( f" A partial response body is saved at {response_path}; inspect it " "locally and redact it before sharing." if saved else " No response body artifact was saved." ) raise StageFailure( f"POST received HTTP {status}, but reading the error body failed. " "The request outcome and charge are unknown; do not blindly retry." + artifact_note ) from read_error try: save_new_bytes(response_path, raw) except StageFailure as save_error: raise StageFailure( f"POST received HTTP {status}; its error body could not be saved. " "The request outcome and charge are unknown; do not blindly retry." ) from save_error raise StageFailure( f"POST received HTTP {status}. Its response body is saved at " f"{response_path}; inspect it locally and redact it before sharing. " "The request outcome and charge are unknown; do not blindly retry." ) from error except NETWORK_READ_ERRORS as error: raise StageFailure( "POST transport failed before a usable HTTP response body was received. " "The request outcome and charge are unknown; no response artifact was saved " "and no automatic retry was attempted." ) from error status = None try: with response_context as response: status = getattr(response, "status", None) raw = response.read() except NETWORK_READ_ERRORS as error: partial = getattr(error, "partial", b"") saved = False if isinstance(partial, bytes) and partial: try: save_new_bytes(response_path, partial) saved = True except StageFailure: pass status_text = f"HTTP {status}" if status is not None else "an HTTP response" artifact_note = ( f" A partial body is saved at {response_path}; inspect it locally and " "redact it before sharing." if saved else " No response body artifact was saved." ) raise StageFailure( f"POST received {status_text}, but reading its response body failed. " "The generation and charge outcome are unknown; do not blindly retry." + artifact_note ) from error try: save_new_bytes(response_path, raw) except StageFailure as error: status_text = f"HTTP {status}" if status is not None else "an HTTP response" raise StageFailure( f"POST received {status_text}, but the response body could not be saved. " "The generation and charge outcome are unknown; no image download was attempted." ) from error return raw def download_image(image_url: str, output_path: Path) -> None: parsed = urllib.parse.urlsplit(image_url) if parsed.scheme != "https" or not parsed.netloc: raise StageFailure("refusing a malformed or non-HTTPS image URL") # The image GET deliberately carries no Tokenhot bearer token. request = urllib.request.Request(image_url, method="GET") try: response_context = urllib.request.urlopen(request, timeout=60) except urllib.error.HTTPError as error: raise StageFailure( f"image GET received HTTP {error.code}; no image file was saved" ) from error except NETWORK_READ_ERRORS as error: raise StageFailure( "image GET transport failed before a usable response; no image file was saved" ) from error try: with response_context as response: image_bytes = response.read() mime_type = response.headers.get_content_type().lower() except NETWORK_READ_ERRORS as error: raise StageFailure( "image GET response body could not be read; no image file was saved" ) from error extension = MIME_EXTENSIONS.get(mime_type) if extension is None: raise StageFailure( f"image GET returned unsupported content type {mime_type}; no image file was saved" ) allowed_suffixes = {extension} if extension == ".jpg": allowed_suffixes.add(".jpeg") if output_path.suffix.lower() not in allowed_suffixes: raise StageFailure( f"image GET returned {mime_type}; choose an output path ending in {extension}" ) save_new_bytes(output_path, image_bytes) def main() -> int: parser = argparse.ArgumentParser( description="Generate one image with a documented Tokenhot Nano Banana route." ) parser.add_argument( "--model", choices=MODEL_IDS, default="nano-banana-2", help="Exact Tokenhot route ID (default: nano-banana-2)", ) parser.add_argument( "--prompt", default="A small red cabin beside a lake at sunrise, editorial illustration", help="Image prompt text", ) parser.add_argument( "--output", type=Path, default=Path("nano-banana-output.jpg"), help="Unused destination for a downloaded JPEG, PNG, or WebP", ) parser.add_argument( "--resume-response", type=Path, help="Reuse saved response JSON and make only its image GET; skips the POST", ) args = parser.parse_args() try: args.output.parent.mkdir(parents=True, exist_ok=True) except OSError: print("Cannot create the output directory.", file=sys.stderr) return 1 if args.resume_response: response_path = args.resume_response if not response_path.is_file(): print( f"Saved response file does not exist: {response_path}; no request was sent.", file=sys.stderr, ) return 1 if args.output.exists(): print( f"Refusing to overwrite existing image path {args.output}; " "choose a fresh --output path for GET-only recovery.", file=sys.stderr, ) return 1 try: raw = response_path.read_bytes() except OSError: print( f"Could not read saved response file {response_path}; no request was sent.", file=sys.stderr, ) return 1 print(f"Reusing saved response JSON; skipping POST: {response_path}") else: response_path = response_path_for(args.output) try: require_fresh_generation_paths(args.output, response_path) except StageFailure as error: print(str(error), file=sys.stderr) return 1 token = os.environ.get("TOKENHOT_API_KEY") if not token: parser.error( "set TOKENHOT_API_KEY in your shell; the script does not read a .env file" ) endpoint = f"{API_BASE}/{args.model}:generateContent" payload = { "contents": [ { "role": "user", "parts": [{"text": args.prompt}], } ], "generationConfig": { "responseModalities": ["TEXT", "IMAGE"], "imageConfig": { "aspectRatio": "16:9", "imageSize": "2K", }, }, } try: raw = post_once(endpoint, token, payload, response_path) except StageFailure as error: print(str(error), file=sys.stderr) return 1 try: document = json.loads(raw) image_url = response_image_url(document) except (json.JSONDecodeError, ValueError): print( f"No documented image URL could be read. The response body is preserved at " f"{response_path}; inspect it locally and redact it before sharing. " "No POST retry was made.", file=sys.stderr, ) return 1 try: download_image(image_url, args.output) except StageFailure as error: print( f"{error}. The API response remains at {response_path}; no generation POST " "was repeated. Use --resume-response to retry only the GET if its URL still works.", file=sys.stderr, ) return 1 print(f"Response JSON: {response_path}") print(f"Image saved: {args.output}") return 0 if __name__ == "__main__": raise SystemExit(main())