#!/usr/bin/env python3 """ FLUX Image Generator for Newsletter Hero Images Generates a topic-relevant hero image using the FLUX model via: 1. Black Forest Labs (BFL) API — if BFL_API_KEY is set 2. OpenRouter API — if OPENROUTER_API_KEY is set Usage: python3 flux_image.py --prompt "Professional editorial photograph of..." \ --aspect landscape \ --output ./output/hero-image.png """ import argparse import json import os import sys import time import urllib.request import urllib.error import base64 from pathlib import Path # --------------------------------------------------------------------------- # Aspect-ratio presets (width x height) # --------------------------------------------------------------------------- ASPECT_PRESETS = { "landscape": (1344, 768), # ~16:9, good for newsletter hero "wide": (1536, 640), # ultra-wide banner "square": (1024, 1024), "portrait": (768, 1344), } # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _find_api_key(): """Find a FLUX-compatible API key from environment or .env files.""" for var in ("BFL_API_KEY", "OPENROUTER_API_KEY"): val = os.environ.get(var) if val: return var, val # Walk up directories looking for .env search = Path.cwd() for _ in range(6): env_file = search / ".env" if env_file.exists(): for line in env_file.read_text().splitlines(): line = line.strip() if line.startswith("#") or "=" not in line: continue key, _, value = line.partition("=") key = key.strip() value = value.strip().strip("'\"") if key in ("BFL_API_KEY", "OPENROUTER_API_KEY") and value: return key, value search = search.parent return None, None def _http_json(url, data=None, headers=None, method="GET"): """Simple HTTPS JSON request.""" headers = headers or {} headers.setdefault("Content-Type", "application/json") headers.setdefault("User-Agent", "newsletter-designer/1.0") body = json.dumps(data).encode() if data else None req = urllib.request.Request(url, data=body, headers=headers, method=method) with urllib.request.urlopen(req, timeout=60) as resp: return json.loads(resp.read().decode()) def _download(url, dest): """Download a file from URL to local path.""" req = urllib.request.Request(url, headers={"User-Agent": "newsletter-designer/1.0"}) with urllib.request.urlopen(req, timeout=120) as resp: Path(dest).parent.mkdir(parents=True, exist_ok=True) with open(dest, "wb") as f: f.write(resp.read()) print(f"Downloaded: {dest} ({os.path.getsize(dest)} bytes)") # --------------------------------------------------------------------------- # BFL (Black Forest Labs) Provider # --------------------------------------------------------------------------- def generate_bfl(prompt, width, height, output, api_key): """Generate image via BFL FLUX API (async task + polling).""" print("Using BFL FLUX API...") # Submit task result = _http_json( "https://api.bfl.ml/v1/flux-pro-1.1", data={ "prompt": prompt, "width": width, "height": height, }, headers={"X-Key": api_key}, method="POST", ) task_id = result.get("id") if not task_id: print(f"Error: No task ID returned. Response: {result}") sys.exit(1) print(f"Task submitted: {task_id}") # Poll for result for attempt in range(120): time.sleep(2) status = _http_json( f"https://api.bfl.ml/v1/get_result?id={task_id}", headers={"X-Key": api_key}, ) state = status.get("status", "") if state == "Ready": image_url = status.get("result", {}).get("sample") if image_url: _download(image_url, output) return output else: print(f"Error: Ready but no image URL. Response: {status}") sys.exit(1) elif state in ("Error", "Failed"): print(f"Task failed: {status}") sys.exit(1) else: if attempt % 10 == 0: print(f" Polling... ({attempt * 2}s, status: {state})") print("Error: Timed out waiting for image generation.") sys.exit(1) # --------------------------------------------------------------------------- # OpenRouter Provider # --------------------------------------------------------------------------- def generate_openrouter(prompt, width, height, output, api_key): """Generate image via OpenRouter FLUX endpoint.""" print("Using OpenRouter FLUX API...") result = _http_json( "https://openrouter.ai/api/v1/images/generations", data={ "model": "black-forest-labs/flux-1.1-pro", "prompt": prompt, "n": 1, "size": f"{width}x{height}", }, headers={"Authorization": f"Bearer {api_key}"}, method="POST", ) images = result.get("data", []) if not images: print(f"Error: No images returned. Response: {result}") sys.exit(1) img = images[0] if img.get("url"): _download(img["url"], output) elif img.get("b64_json"): Path(output).parent.mkdir(parents=True, exist_ok=True) with open(output, "wb") as f: f.write(base64.b64decode(img["b64_json"])) print(f"Saved from base64: {output}") else: print(f"Error: Unexpected image format. Response: {result}") sys.exit(1) return output # --------------------------------------------------------------------------- # Main # --------------------------------------------------------------------------- def main(): parser = argparse.ArgumentParser(description="Generate FLUX hero image for newsletter") parser.add_argument("--prompt", required=True, help="Image generation prompt") parser.add_argument("--aspect", default="landscape", choices=ASPECT_PRESETS.keys(), help="Aspect ratio preset (default: landscape)") parser.add_argument("--width", type=int, help="Custom width (overrides aspect)") parser.add_argument("--height", type=int, help="Custom height (overrides aspect)") parser.add_argument("--output", default="./output/hero-image.png", help="Output file path") args = parser.parse_args() width, height = ASPECT_PRESETS[args.aspect] if args.width: width = args.width if args.height: height = args.height key_name, api_key = _find_api_key() if not api_key: print("Error: No FLUX API key found.") print("Set BFL_API_KEY or OPENROUTER_API_KEY in environment or .env file.") sys.exit(1) print(f"Prompt: {args.prompt[:100]}...") print(f"Size: {width}x{height}") print(f"Output: {args.output}") print(f"Provider: {key_name}") if key_name == "OPENROUTER_API_KEY" or (api_key and api_key.startswith("sk-or-")): generate_openrouter(args.prompt, width, height, args.output, api_key) else: generate_bfl(args.prompt, width, height, args.output, api_key) print("Done!") if __name__ == "__main__": main()