#!/usr/bin/env python3 """Bomber Battle asset-pack creator CLI. Python 3.9+, standard library only.""" import argparse import getpass import hashlib import json import os import pathlib import re import shutil import sys import time import urllib.error import urllib.parse import urllib.request import uuid import zipfile CLI_VERSION = "2.0.0" SCHEMA_V1 = "boooooom.asset-pack/v1" SCHEMA_V2 = "boooooom.asset-pack/v2" SCHEMAS = (SCHEMA_V1, SCHEMA_V2) CATEGORIES = ("bomberman_character", "academy", "company", "small_monster", "boss_monster") GENERATABLE = ("small-monster", "boss-monster") ASSET_ROLES = ("preview", "portrait", "sprite", "sprite_sheet", "animation", "reference", "source", "audio", "data") SEMVER_RE = re.compile(r"(?:0|[1-9]\d*)\.(?:0|[1-9]\d*)\.(?:0|[1-9]\d*)(?:-[0-9A-Za-z.-]+)?(?:\+[0-9A-Za-z.-]+)?") PACKAGE_ID_RE = re.compile(r"[a-z0-9]+(?:[._-][a-z0-9]+)+") ASSET_ID_RE = re.compile(r"[a-z][a-z0-9_-]{0,63}") CONFIG_DIR = pathlib.Path(os.environ.get("BOMBER_ASSET_HOME", pathlib.Path.home() / ".boooooom")) CONFIG_FILE = CONFIG_DIR / "credentials.json" DEFAULT_API = "https://boooooom.shop" class CliError(RuntimeError): pass def load_credentials(): if not CONFIG_FILE.is_file(): return {"api": DEFAULT_API, "token": ""} try: data = json.loads(CONFIG_FILE.read_text(encoding="utf-8")) return {"api": str(data.get("api") or DEFAULT_API).rstrip("/"), "token": str(data.get("token") or "")} except (OSError, json.JSONDecodeError) as exc: raise CliError(f"Cannot read credentials: {exc}") from exc def save_credentials(api, token): CONFIG_DIR.mkdir(parents=True, exist_ok=True) CONFIG_FILE.write_text(json.dumps({"api": api.rstrip("/"), "token": token}, indent=2), encoding="utf-8") try: os.chmod(CONFIG_FILE, 0o600) except OSError: pass def request_json(method, path, data=None, token=None, api=None, timeout=90, headers=None): api = (api or load_credentials()["api"]).rstrip("/") body = None if data is None else json.dumps(data, ensure_ascii=False).encode("utf-8") request_headers = {"Accept": "application/json"} if body is not None: request_headers["Content-Type"] = "application/json" if token: request_headers["Authorization"] = f"Bearer {token}" request_headers.update(headers or {}) req = urllib.request.Request(api + path, data=body, method=method, headers=request_headers) try: with urllib.request.urlopen(req, timeout=timeout) as response: return json.loads(response.read().decode("utf-8")) except urllib.error.HTTPError as exc: try: payload = json.loads(exc.read().decode("utf-8")) message = payload.get("error") or f"HTTP {exc.code}" except Exception: message = f"HTTP {exc.code}" raise CliError(message) from exc except (urllib.error.URLError, TimeoutError) as exc: raise CliError(f"Cannot reach {api}: {exc}") from exc def require_token(args): token = str(getattr(args, "token", "") or load_credentials()["token"]) if not token: raise CliError("No Creator Token. Run: bomber_asset_cli.py auth set-token") return token def read_manifest(root): path = pathlib.Path(root) / "manifest.json" if not path.is_file(): raise CliError("manifest.json is missing from the pack root") try: return json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as exc: raise CliError(f"Invalid manifest.json: {exc}") from exc def normalized_asset_path(value): raw = str(value or "").replace("\\", "/") path = pathlib.PurePosixPath(raw) if not raw or path.is_absolute() or ".." in path.parts: raise CliError(f"unsafe asset path: {raw}") normalized = path.as_posix() if normalized in {"manifest.json", "checksums.sha256"}: raise CliError(f"reserved asset path: {normalized}") return normalized def validate_manifest(manifest, editable=True): errors = [] if not isinstance(manifest, dict): return ["manifest root must be an object"] schema = str(manifest.get("schema", "")) if schema not in SCHEMAS: errors.append(f"schema must be {SCHEMA_V1} or {SCHEMA_V2}") package_id = str(manifest.get("id", "")) if not PACKAGE_ID_RE.fullmatch(package_id) or len(package_id) > 128: errors.append("id must use lowercase namespace segments, for example creator.name.character") if editable and (package_id.startswith("official.") or manifest.get("read_only") is True): errors.append("official and read-only packs cannot be modified by this CLI") version = str(manifest.get("version", "")) if not SEMVER_RE.fullmatch(version) or len(version) > 32: errors.append("version must be valid Semantic Versioning") if manifest.get("category") not in CATEGORIES: errors.append("category is unsupported") if not str(manifest.get("name", "")).strip() or len(str(manifest.get("name", "")).strip()) > 128: errors.append("name must contain 1-128 characters") if len(str(manifest.get("description", ""))) > 2000: errors.append("description must not exceed 2000 characters") if len(str(manifest.get("author", ""))) > 128: errors.append("author must not exceed 128 characters") if len(str(manifest.get("license", ""))) > 64: errors.append("license must not exceed 64 characters") assets = manifest.get("assets") if not isinstance(assets, list): errors.append("assets must be an array") assets = [] if schema == SCHEMA_V2 and not assets: errors.append("v2 packs must declare at least one asset") asset_ids = set() asset_paths = set() for index, asset in enumerate(assets): if not isinstance(asset, dict): errors.append(f"assets[{index}] must be an object") continue try: path = normalized_asset_path(asset.get("path")) except CliError as exc: errors.append(str(exc)) continue if path in asset_paths: errors.append(f"duplicate asset path: {path}") asset_paths.add(path) role = str(asset.get("role", "")).strip() if not role: errors.append(f"assets[{index}].role is required") if schema == SCHEMA_V2: asset_id = str(asset.get("id", "")).strip() if role not in ASSET_ROLES: errors.append(f"assets[{index}].role is unsupported") if not ASSET_ID_RE.fullmatch(asset_id): errors.append(f"assets[{index}].id is invalid") elif asset_id in asset_ids: errors.append(f"duplicate asset id: {asset_id}") asset_ids.add(asset_id) declared_sha = str(asset.get("sha256", "")).strip() if declared_sha and not re.fullmatch(r"[a-f0-9]{64}", declared_sha): errors.append(f"assets[{index}].sha256 is invalid") if schema == SCHEMA_V2: package_type = str(manifest.get("packageType", "")) if package_type not in {"concept", "runtime", "reference"}: errors.append("packageType must be concept, runtime or reference") if package_type == "runtime": engine = manifest.get("engine") if not isinstance(engine, dict) or not SEMVER_RE.fullmatch(str(engine.get("minVersion", ""))): errors.append("runtime packs require a valid engine.minVersion") gameplay = manifest.get("gameplay") hp = gameplay.get("hitPoints") if isinstance(gameplay, dict) else None speed = gameplay.get("speedMultiplier") if isinstance(gameplay, dict) else None behavior = str(gameplay.get("behavior", "")) if isinstance(gameplay, dict) else "" if isinstance(hp, bool) or not isinstance(hp, int) or not 1 <= hp <= 100000: errors.append("gameplay.hitPoints must be an integer from 1 to 100000") if isinstance(speed, bool) or not isinstance(speed, (int, float)) or not 0.05 <= speed <= 10: errors.append("gameplay.speedMultiplier must be between 0.05 and 10") if not re.fullmatch(r"[a-z][a-z0-9_-]{0,63}", behavior): errors.append("gameplay.behavior is invalid") render = manifest.get("render") tile_size = render.get("tileSize") if isinstance(render, dict) else None directions = render.get("directions") if isinstance(render, dict) else None if isinstance(tile_size, bool) or not isinstance(tile_size, int) or not 16 <= tile_size <= 512: errors.append("render.tileSize must be an integer from 16 to 512") if directions not in {1, 2, 4, 8}: errors.append("render.directions must be 1, 2, 4 or 8") collision = manifest.get("collision") if not isinstance(collision, dict) or collision.get("shape") not in {"circle", "rectangle"}: errors.append("runtime packs require a circle or rectangle collision shape") animations = manifest.get("animations") if not isinstance(animations, dict) or not animations: errors.append("runtime packs require at least one animation") else: for clip_name, clip in animations.items(): if not ASSET_ID_RE.fullmatch(str(clip_name)) or not isinstance(clip, dict): errors.append("animation name or value is invalid") continue if str(clip.get("asset", "")) not in asset_ids: errors.append(f"animation {clip_name} references an unknown asset") frames, fps = clip.get("frames"), clip.get("fps") if isinstance(frames, bool) or not isinstance(frames, int) or not 1 <= frames <= 256: errors.append(f"animation {clip_name}.frames must be 1-256") if isinstance(fps, bool) or not isinstance(fps, (int, float)) or not 1 <= fps <= 60: errors.append(f"animation {clip_name}.fps must be 1-60") return errors def pack_files(root): root = pathlib.Path(root).resolve() files = [] for path in root.rglob("*"): if path.is_file() and ".git" not in path.parts and path.name != "checksums.sha256" and path.suffix != ".zip": files.append(path) return sorted(files, key=lambda item: item.relative_to(root).as_posix()) def command_auth_set(args): token = args.value or getpass.getpass("Creator Token: ").strip() if not token.startswith("bb_creator_"): raise CliError("Creator Token must start with bb_creator_") save_credentials(args.api or DEFAULT_API, token) print(f"Saved Creator Token to {CONFIG_FILE}") def command_auth_whoami(args): creds = load_credentials() payload = request_json("GET", "/api/v1/creator/profile", token=require_token(args), api=args.api or creds["api"]) print(f"{payload.get('name')} <{payload.get('email')}>") print(f"Creator namespace: {payload.get('creatorNamespace', 'not assigned')}") print(f"Packages: {len(payload.get('packages', []))} Generations: {len(payload.get('generations', []))}") def command_auth_logout(_args): if CONFIG_FILE.exists(): CONFIG_FILE.unlink() print(f"Removed local Creator Token: {CONFIG_FILE}") def command_new(args): root = pathlib.Path(args.directory).resolve() if root.exists() and any(root.iterdir()): raise CliError(f"Directory is not empty: {root}") (root / "assets" / "sprites").mkdir(parents=True, exist_ok=True) (root / "prompts").mkdir(parents=True, exist_ok=True) schema = SCHEMA_V2 if args.schema == "v2" else SCHEMA_V1 manifest = {"schema": schema, "id": args.id, "version": args.version, "category": args.category, "name": args.name, "description": args.description or "", "author": args.author or "", "license": args.license, "read_only": False, "assets": []} if schema == SCHEMA_V2: manifest["packageType"] = args.package_type if args.package_type == "runtime": manifest.update({ "engine": {"minVersion": "1.0.0"}, "gameplay": {"hitPoints": 1, "speedMultiplier": 1.0, "behavior": "basic"}, "render": {"tileSize": 96, "directions": 4}, "collision": {"shape": "circle", "radius": 28}, "animations": {}, }) (root / "manifest.json").write_text(json.dumps(manifest, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") description = args.description or "Describe the character, motion and gameplay behavior here." (root / "README.md").write_text(f"# {args.name}\n\n{description}\n", encoding="utf-8") (root / "LICENSE.txt").write_text(args.license + "\n", encoding="utf-8") (root / "prompts" / "design.md").write_text("# Design prompt\n\nRecord the final prompt and references here.\n", encoding="utf-8") print(f"Created asset pack: {root}") def command_asset_add(args): root = pathlib.Path(args.directory).resolve() manifest = read_manifest(root) source = pathlib.Path(args.file).resolve() if not source.is_file(): raise CliError(f"Asset file does not exist: {source}") relative = args.path or f"assets/{source.name}" relative = normalized_asset_path(relative) destination = (root / relative).resolve() if not destination.is_relative_to(root): raise CliError(f"Asset destination is unsafe: {relative}") if destination != source: if destination.exists() and not args.force: raise CliError(f"Asset already exists; use --force to replace it: {destination}") destination.parent.mkdir(parents=True, exist_ok=True) shutil.copy2(source, destination) asset_id = args.id or re.sub(r"[^a-z0-9_-]+", "-", destination.stem.lower()).strip("-") if not ASSET_ID_RE.fullmatch(asset_id): raise CliError("Asset id must start with a letter and use lowercase letters, numbers, _ or -") assets = manifest.setdefault("assets", []) if not isinstance(assets, list): raise CliError("manifest.assets must be an array") assets[:] = [item for item in assets if isinstance(item, dict) and item.get("id") != asset_id and item.get("path") != relative] entry = {"id": asset_id, "path": relative, "role": args.role} if manifest.get("schema") == SCHEMA_V2: entry["sha256"] = hashlib.sha256(destination.read_bytes()).hexdigest() assets.append(entry) (root / "manifest.json").write_text( json.dumps(manifest, ensure_ascii=False, indent=2) + "\n", encoding="utf-8" ) print(f"Added asset {asset_id} ({args.role}): {relative}") def command_validate(args): root = pathlib.Path(args.directory).resolve() manifest = read_manifest(root) errors = validate_manifest(manifest) for asset in manifest.get("assets", []): rel = str(asset.get("path", "")) if isinstance(asset, dict) else "" path = (root / rel).resolve() if not rel or not path.is_relative_to(root) or not path.is_file(): errors.append(f"asset file is missing or unsafe: {rel}") continue declared_sha = str(asset.get("sha256", "")).strip() if declared_sha and hashlib.sha256(path.read_bytes()).hexdigest() != declared_sha: errors.append(f"asset SHA-256 does not match: {rel}") checksum_path = root / "checksums.sha256" if checksum_path.is_file() and getattr(args, "verify_checksums", True): try: verify_directory_checksums(root) except CliError as exc: errors.append(str(exc)) if errors: raise CliError("Validation failed:\n- " + "\n- ".join(errors)) print(f"Valid {manifest['schema']} pack: {manifest['id']}@{manifest['version']}") def verify_directory_checksums(root): root = pathlib.Path(root).resolve() checksum_path = root / "checksums.sha256" declared = {} try: lines = checksum_path.read_text(encoding="ascii").splitlines() except (OSError, UnicodeDecodeError) as exc: raise CliError(f"Cannot read checksums.sha256: {exc}") from exc for line in lines: match = re.fullmatch(r"([a-f0-9]{64}) (.+)", line) if not match or match.group(2) in declared: raise CliError("checksums.sha256 is malformed or contains duplicate paths") declared[match.group(2)] = match.group(1) expected = {path.relative_to(root).as_posix() for path in pack_files(root)} if set(declared) != expected: raise CliError("checksums.sha256 does not cover every pack file") for rel, expected_sha in declared.items(): path = (root / rel).resolve() if not path.is_relative_to(root) or not path.is_file(): raise CliError(f"checksum path is missing or unsafe: {rel}") if hashlib.sha256(path.read_bytes()).hexdigest() != expected_sha: raise CliError(f"checksum mismatch: {rel}") def command_build(args): root = pathlib.Path(args.directory).resolve() command_validate(argparse.Namespace(directory=str(root), verify_checksums=False)) manifest = read_manifest(root) files = pack_files(root) checksums = [] for path in files: digest = hashlib.sha256(path.read_bytes()).hexdigest() checksums.append(f"{digest} {path.relative_to(root).as_posix()}") checksum_path = root / "checksums.sha256" checksum_path.write_text("\n".join(checksums) + "\n", encoding="ascii") output = pathlib.Path(args.output or f"{manifest['id']}-{manifest['version']}.zip").resolve() output.parent.mkdir(parents=True, exist_ok=True) with zipfile.ZipFile(output, "w", zipfile.ZIP_DEFLATED, compresslevel=9) as archive: for path in pack_files(root) + [checksum_path]: name = path.relative_to(root).as_posix() info = zipfile.ZipInfo(name, date_time=(1980, 1, 1, 0, 0, 0)) info.compress_type = zipfile.ZIP_DEFLATED info.external_attr = 0o100644 << 16 archive.writestr(info, path.read_bytes(), compress_type=zipfile.ZIP_DEFLATED, compresslevel=9) print(f"Built: {output} ({output.stat().st_size} bytes)") def multipart_file(path, content_type="application/zip"): boundary = "----BoooooomCLI" + uuid.uuid4().hex name = pathlib.Path(path).name head = (f"--{boundary}\r\nContent-Disposition: form-data; name=\"file\"; filename=\"{name}\"\r\n" f"Content-Type: {content_type}\r\n\r\n").encode("utf-8") tail = f"\r\n--{boundary}--\r\n".encode("ascii") return boundary, head + pathlib.Path(path).read_bytes() + tail def validate_archive(path, editable=True): try: with zipfile.ZipFile(path) as archive: infos = archive.infolist() files = {} for info in infos: raw = info.filename.replace("\\", "/").rstrip("/") posix = pathlib.PurePosixPath(raw) if not raw or posix.is_absolute() or ".." in posix.parts: raise CliError(f"Archive contains an unsafe path: {raw}") if info.is_dir(): continue if raw in files: raise CliError(f"Archive contains a duplicate path: {raw}") if info.flag_bits & 0x1: raise CliError("Archive contains encrypted files") files[raw] = info if "manifest.json" not in files: raise CliError("Archive root is missing manifest.json") manifest = json.loads(archive.read(files["manifest.json"]).decode("utf-8")) errors = validate_manifest(manifest, editable=editable) for asset in manifest.get("assets", []): if not isinstance(asset, dict): continue try: rel = normalized_asset_path(asset.get("path")) except CliError as exc: errors.append(str(exc)) continue if rel not in files: errors.append(f"asset file is missing: {rel}") continue content_sha = hashlib.sha256(archive.read(files[rel])).hexdigest() if asset.get("sha256") and asset["sha256"] != content_sha: errors.append(f"asset SHA-256 does not match: {rel}") checksum_info = files.get("checksums.sha256") if manifest.get("schema") == SCHEMA_V2 and not checksum_info: errors.append("v2 archives must include checksums.sha256") if checksum_info: declared = {} try: checksum_lines = archive.read(checksum_info).decode("ascii").splitlines() except UnicodeDecodeError: checksum_lines = [] errors.append("checksums.sha256 must use ASCII encoding") for line in checksum_lines: match = re.fullmatch(r"([a-f0-9]{64}) (.+)", line) if not match or match.group(2) in declared: errors.append("checksums.sha256 is malformed or contains duplicate paths") continue declared[match.group(2)] = match.group(1) expected = set(files) - {"checksums.sha256"} if set(declared) != expected: errors.append("checksums.sha256 does not cover every archive file") else: for rel, expected_sha in declared.items(): if hashlib.sha256(archive.read(files[rel])).hexdigest() != expected_sha: errors.append(f"checksum mismatch: {rel}") if errors: raise CliError("Archive validation failed:\n- " + "\n- ".join(errors)) return manifest except (zipfile.BadZipFile, UnicodeDecodeError, json.JSONDecodeError) as exc: raise CliError(f"Invalid asset-pack archive: {exc}") from exc def command_upload(args): path = pathlib.Path(args.archive).resolve() if not path.is_file() or path.suffix.lower() != ".zip": raise CliError("Upload target must be a ZIP file") if path.stat().st_size > 25 * 1024 * 1024: raise CliError("Upload target exceeds 25MB") manifest = validate_archive(path) print(f"Validated upload: {manifest['id']}@{manifest['version']}") creds = load_credentials() boundary, body = multipart_file(path) api = (args.api or creds["api"]).rstrip("/") req = urllib.request.Request(api + "/api/v1/creator/packs/upload", data=body, method="POST", headers={"Authorization": f"Bearer {require_token(args)}", "Content-Type": f"multipart/form-data; boundary={boundary}", "Accept": "application/json"}) try: with urllib.request.urlopen(req, timeout=180) as response: payload = json.loads(response.read().decode("utf-8")) except urllib.error.HTTPError as exc: try: message = json.loads(exc.read().decode("utf-8")).get("error") except Exception: message = f"HTTP {exc.code}" raise CliError(message) from exc print(f"Uploaded {payload['id']}@{payload['version']} to {payload['storage']}") print(f"Member library: {api}{payload.get('libraryUrl', '/school/art-animation.html#library')}") if payload.get("public"): print(f"Public showcase: {api}{payload.get('showcaseUrl', '/school/art-animation.html#showcase')}") def find_pack(args): creds = load_credentials() api = (args.api or creds["api"]).rstrip("/") payload = request_json("GET", "/api/v1/assets/packs", token=require_token(args), api=api) matches = [pack for pack in payload.get("packs", []) if pack.get("id") == args.id and (not args.version or pack.get("version") == args.version)] if not matches: raise CliError(f"Pack not found: {args.id}@{args.version or '*'}") if len(matches) > 1: owned = [pack for pack in matches if pack.get("owned")] candidates = owned or matches versions = {pack.get("version") for pack in candidates} if not args.version and len(versions) == len(candidates): return api, max(candidates, key=lambda pack: semver_key(pack.get("version", "0.0.0"))) if len(candidates) == 1: return api, candidates[0] refs = ", ".join(str(pack.get("ref")) for pack in matches) raise CliError(f"Multiple packs match; use the web library to choose one (refs: {refs})") return api, matches[0] def semver_key(value): match = SEMVER_RE.fullmatch(str(value)) if not match: return (0, 0, 0, 0, "") core, _, metadata = str(value).partition("+") numeric, separator, prerelease = core.partition("-") major, minor, patch = (int(item) for item in numeric.split(".")) return (major, minor, patch, 1 if not separator else 0, prerelease, metadata) def command_packs_list(args): creds = load_credentials() payload = request_json("GET", "/api/v1/assets/packs", token=require_token(args), api=args.api or creds["api"]) for pack in payload.get("packs", []): mode = "official/read-only" if pack.get("official") else "creator" print(f"{pack['id']}@{pack['version']}\t{pack['category']}\t{mode}\t{pack['name']}") print(f" {pack.get('author') or 'Unknown'} · {pack.get('license') or 'Unspecified'} · {pack.get('description') or 'No description'}") def command_packs_info(args): _, pack = find_pack(args) print(f"{pack['name']} ({pack['id']}@{pack['version']})") print(f"Reference: {pack.get('ref')} Category: {pack.get('categoryName') or pack.get('category')}") print(f"Author: {pack.get('author') or 'Unknown'} License: {pack.get('license') or 'Unspecified'}") print(f"Size: {pack.get('size', 0)} bytes SHA-256: {pack.get('sha256') or 'not available'}") print(pack.get("description") or "No description") def command_packs_download(args): api, pack = find_pack(args) download_url = urllib.parse.urljoin(api + "/", str(pack.get("downloadUrl", "")).lstrip("/")) req = urllib.request.Request(download_url, headers={"Authorization": f"Bearer {require_token(args)}"}) output = pathlib.Path(args.output or f"{args.id}-{pack['version']}.zip").resolve() output.parent.mkdir(parents=True, exist_ok=True) temporary = output.with_name(output.name + ".part") try: with urllib.request.urlopen(req, timeout=180) as response, temporary.open("wb") as stream: shutil.copyfileobj(response, stream) actual_sha = hashlib.sha256(temporary.read_bytes()).hexdigest() expected_sha = str(pack.get("sha256") or "") if expected_sha and actual_sha != expected_sha: raise CliError(f"Downloaded SHA-256 mismatch: expected {expected_sha}, got {actual_sha}") manifest = validate_archive(temporary, editable=False) if manifest.get("id") != pack.get("id") or manifest.get("version") != pack.get("version"): raise CliError("Downloaded manifest identity does not match the catalog") os.replace(temporary, output) except urllib.error.HTTPError as exc: raise CliError(f"Download failed: HTTP {exc.code}") from exc finally: if temporary.exists(): temporary.unlink() if args.read_only or pack.get("readOnly"): try: os.chmod(output, 0o444) except OSError: pass print(f"Downloaded: {output}") print(f"Verified SHA-256: {hashlib.sha256(output.read_bytes()).hexdigest()}") def upload_reference(path, token, api): path = pathlib.Path(path).resolve() types = {".png": "image/png", ".jpg": "image/jpeg", ".jpeg": "image/jpeg", ".webp": "image/webp"} content_type = types.get(path.suffix.lower()) if not path.is_file() or not content_type: raise CliError(f"Reference image must be PNG, JPEG or WebP: {path}") if path.stat().st_size > 5 * 1024 * 1024: raise CliError(f"Reference image exceeds 5MB: {path}") boundary, body = multipart_file(path, content_type) req = urllib.request.Request(api.rstrip("/") + "/api/v1/creator/references", data=body, method="POST", headers={"Authorization": f"Bearer {token}", "Content-Type": f"multipart/form-data; boundary={boundary}", "Accept": "application/json"}) try: with urllib.request.urlopen(req, timeout=120) as response: return json.loads(response.read().decode("utf-8"))["url"] except urllib.error.HTTPError as exc: try: message = json.loads(exc.read().decode("utf-8")).get("error") except Exception: message = f"HTTP {exc.code}" raise CliError(f"Reference upload failed: {message}") from exc def command_generate(args): creds = load_credentials() token = require_token(args) api = (args.api or creds["api"]).rstrip("/") if len(args.reference_url or []) + len(args.reference or []) > 3: raise CliError("A generation can use at most three reference images") references = list(args.reference_url or []) for path in args.reference or []: print(f"Uploading reference: {path}") references.append(upload_reference(path, token, api)) category = args.category.replace("-", "_") data = {"category": category, "name": args.name, "prompt": args.prompt, "referenceImages": references, "archetype": args.archetype, "requestId": args.request_id or uuid.uuid4().hex} payload = request_json("POST", "/api/v1/creator/generate", data=data, token=token, api=api, timeout=380) status_url = payload.get("statusUrl") deadline = time.time() + 600 while payload.get("status") in {"queued", "running"}: if time.time() >= deadline: raise CliError("Generation did not finish within 10 minutes") print(f"Generation {payload['id']}: {payload['status']}...") time.sleep(3) payload = request_json("GET", status_url, token=token, api=api, timeout=30) if payload.get("status") == "failed": raise CliError(payload.get("error") or "Generation failed") if payload.get("status") != "ready" or not payload.get("imageUrl"): raise CliError("Generation returned an unexpected status") image_url = payload["imageUrl"] attach_root = pathlib.Path(args.attach).resolve() if args.attach else None if attach_root: read_manifest(attach_root) requested_output = pathlib.Path(args.output).resolve() if args.output else None temporary_parent = requested_output.parent if requested_output else ( attach_root / "assets" if attach_root else pathlib.Path.cwd() ) temporary_parent.mkdir(parents=True, exist_ok=True) temporary = temporary_parent / f".{payload['id']}.part" req = urllib.request.Request(api + image_url, headers={"Authorization": f"Bearer {token}"}) try: with urllib.request.urlopen(req, timeout=180) as response, temporary.open("wb") as stream: shutil.copyfileobj(response, stream) header = temporary.read_bytes()[:16] if header.startswith(b"\x89PNG\r\n\x1a\n"): suffix = ".png" elif header.startswith(b"\xff\xd8\xff"): suffix = ".jpg" elif len(header) >= 12 and header[:4] == b"RIFF" and header[8:12] == b"WEBP": suffix = ".webp" else: raise CliError("Generation download is not a supported image") output = requested_output or ((attach_root / "assets" / f"preview{suffix}") if attach_root else pathlib.Path(f"{payload['id']}{suffix}").resolve()) output.parent.mkdir(parents=True, exist_ok=True) if requested_output and output.suffix.lower() not in {suffix, ".jpeg" if suffix == ".jpg" else suffix}: raise CliError(f"Output extension {output.suffix} does not match generated {suffix} image") os.replace(temporary, output) finally: if temporary.exists(): temporary.unlink() if attach_root: command_asset_add(argparse.Namespace( directory=str(attach_root), file=str(output), path=f"assets/preview{suffix}", id=args.asset_id, role="preview", force=True, )) manifest = read_manifest(attach_root) provenance = manifest.setdefault("provenance", {}) provenance["generation"] = {"id": payload["id"], "model": payload.get("model"), "revisedPrompt": payload.get("revisedPrompt")} (attach_root / "manifest.json").write_text( json.dumps(manifest, ensure_ascii=False, indent=2) + "\n", encoding="utf-8" ) print(f"Generated {category}: {output}") print(f"Model: {payload.get('model')} Generation: {payload.get('id')}") def parser(): root = argparse.ArgumentParser(prog="bomber-assets", description="Bomber Battle creator asset-pack CLI") root.add_argument("--version", action="version", version=f"bomber-assets {CLI_VERSION}") root.add_argument("--api", help=f"API base URL (default: {DEFAULT_API})") root.add_argument("--token", help="Creator Token; prefer auth set-token") commands = root.add_subparsers(dest="command", required=True) auth = commands.add_parser("auth") auth_commands = auth.add_subparsers(dest="auth_command", required=True) set_token = auth_commands.add_parser("set-token") set_token.add_argument("--value", help="Token value; omit for hidden prompt") set_token.set_defaults(func=command_auth_set) whoami = auth_commands.add_parser("whoami") whoami.set_defaults(func=command_auth_whoami) logout = auth_commands.add_parser("logout") logout.set_defaults(func=command_auth_logout) new = commands.add_parser("new") new.add_argument("directory") new.add_argument("--id", required=True) new.add_argument("--name", required=True) new.add_argument("--version", default="0.1.0", help="Semantic version (default: 0.1.0)") new.add_argument("--schema", choices=("v1", "v2"), default="v2") new.add_argument("--package-type", choices=("concept", "runtime", "reference"), default="concept") new.add_argument("--category", choices=CATEGORIES, required=True) new.add_argument("--author") new.add_argument("--description", help="Short member-library description") new.add_argument("--license", default="CC-BY-4.0", help="License identifier or short name") new.set_defaults(func=command_new) asset = commands.add_parser("asset") asset_commands = asset.add_subparsers(dest="asset_command", required=True) asset_add = asset_commands.add_parser("add") asset_add.add_argument("directory") asset_add.add_argument("file") asset_add.add_argument("--id") asset_add.add_argument("--role", choices=ASSET_ROLES, required=True) asset_add.add_argument("--path", help="Destination path inside the pack (default: assets/)") asset_add.add_argument("--force", action="store_true") asset_add.set_defaults(func=command_asset_add) validate = commands.add_parser("validate") validate.add_argument("directory") validate.set_defaults(func=command_validate) build = commands.add_parser("build") build.add_argument("directory") build.add_argument("--output", "-o") build.set_defaults(func=command_build) upload = commands.add_parser("upload") upload.add_argument("archive") upload.set_defaults(func=command_upload) packs = commands.add_parser("packs") pack_commands = packs.add_subparsers(dest="packs_command", required=True) list_command = pack_commands.add_parser("list") list_command.set_defaults(func=command_packs_list) info = pack_commands.add_parser("info") info.add_argument("id") info.add_argument("--version") info.set_defaults(func=command_packs_info) download = pack_commands.add_parser("download") download.add_argument("id") download.add_argument("--version", help="Exact version; omitted selects the newest matching version") download.add_argument("--output", "-o") download.add_argument("--read-only", action="store_true") download.set_defaults(func=command_packs_download) generate = commands.add_parser("generate") generate.add_argument("--category", choices=GENERATABLE, required=True) generate.add_argument("--name", required=True) generate.add_argument("--prompt", required=True) generate.add_argument("--archetype", choices=("ground", "flying", "aquatic", "mechanical", "spectral"), default="ground") generate.add_argument("--reference-url", action="append", default=[]) generate.add_argument("--reference", action="append", default=[], help="Local PNG/JPEG/WebP reference image") generate.add_argument("--output", "-o") generate.add_argument("--attach", help="Attach the generated image as the preview of an existing pack") generate.add_argument("--asset-id", default="preview", help="Asset id used with --attach") generate.add_argument("--request-id", help="Stable idempotency key for safe retries") generate.set_defaults(func=command_generate) return root def main(): args = parser().parse_args() try: args.func(args) except (CliError, OSError, json.JSONDecodeError) as exc: print(f"error: {exc}", file=sys.stderr) return 1 return 0 if __name__ == "__main__": raise SystemExit(main())