#!/usr/bin/env python3 """Create an SVG with MomoVector. Reads MOMOVECTOR_API_KEY; requires Python 3.10+.""" import argparse import base64 import json import mimetypes import os import re from pathlib import Path import sys import time import urllib.error import urllib.parse import urllib.request import uuid class NoRedirect(urllib.request.HTTPRedirectHandler): def redirect_request(self, req, fp, code, msg, headers, newurl): # Never forward a Bearer key to a redirect destination. return None def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--prompt', default='') parser.add_argument('--image', type=Path) parser.add_argument('--reference', action='append', default=[], help='Style image file or public HTTPS image URL; at most two') parser.add_argument('--style', choices=['playful', 'flat', 'minimal', 'retro', 'business', 'tech', '3d'], default='playful') parser.add_argument('--purpose', choices=['illustration', 'logo'], default='illustration') parser.add_argument('--primary-color', default=None) parser.add_argument('--output', type=Path, default=Path('artwork.svg')) parser.add_argument('--request-id', default=None) parser.add_argument('--timeout', type=int, default=300, help='Overall wait limit in seconds') args = parser.parse_args() token = os.environ.get('MOMOVECTOR_API_KEY') if not token: parser.error('Set MOMOVECTOR_API_KEY in your environment first.') base_url = os.environ.get('MOMOVECTOR_BASE_URL', 'https://momovector.com').rstrip('/') parsed = urllib.parse.urlsplit(base_url) if parsed.username or parsed.password or parsed.query or parsed.fragment or parsed.path not in ('', '/'): parser.error('MOMOVECTOR_BASE_URL must be a bare origin, without credentials or a path.') if not parsed.hostname or not (parsed.scheme == 'https' or parsed.scheme == 'http' and parsed.hostname in ('localhost', '127.0.0.1', '::1')): parser.error('Use HTTPS, or HTTP only for a local demonstration.') if args.timeout <= 0: parser.error('--timeout must be positive.') payload = {'mode': 'image' if args.image else 'text', 'prompt': args.prompt, 'style': args.style} if args.purpose == 'logo': if not args.image: parser.error('Logo generation requires --image with a sketch.') payload.update(purpose='logo', style='minimal') if args.primary_color is not None: if not re.fullmatch(r'#[0-9a-fA-F]{6}', args.primary_color): parser.error('--primary-color must use #RRGGBB.') payload['primaryColor'] = args.primary_color.upper() elif args.primary_color is not None: parser.error('--primary-color requires --purpose logo.') if args.image: mime = mimetypes.guess_type(args.image)[0] if mime not in ('image/png', 'image/jpeg', 'image/webp'): parser.error('Choose PNG, JPEG or WebP.') if args.image.stat().st_size > 8 * 1024 * 1024: parser.error('The image exceeds 8 MiB.') payload['image'] = f'data:{mime};base64,' + base64.b64encode(args.image.read_bytes()).decode('ascii') elif len(args.prompt.strip()) < 3: parser.error('Provide a prompt with at least 3 characters, or --image.') if len(args.reference) > 2: parser.error('Use at most two --reference values.') references = [] for value in args.reference: if value.startswith('https://'): link = urllib.parse.urlsplit(value) if not link.hostname or link.username or link.password or len(value) > 4096: parser.error('References require public HTTPS image URLs without credentials.') references.append(value) else: path = Path(value) mime = mimetypes.guess_type(path)[0] if mime not in ('image/png', 'image/jpeg', 'image/webp') or not path.is_file() or path.stat().st_size > 2 * 1024 * 1024: parser.error('Reference files must be PNG, JPEG or WebP, at most 2 MiB each.') references.append(f'data:{mime};base64,' + base64.b64encode(path.read_bytes()).decode('ascii')) if references: payload['styleReferences'] = references request_id = args.request_id or str(uuid.uuid4()) print(f'Idempotency-Key: {request_id}', file=sys.stderr) deadline = time.monotonic() + args.timeout opener = urllib.request.build_opener(NoRedirect()) def call(path, body=None, binary=False): headers = {'Authorization': f'Bearer {token}'} data = None if body is not None: headers.update({'Content-Type': 'application/json', 'Idempotency-Key': request_id}) data = json.dumps(body, ensure_ascii=False).encode('utf-8') for attempt in range(4): remaining = deadline - time.monotonic() if remaining <= 0: raise RuntimeError('Wait limit reached. Query the existing task later; do not submit a new task.') request = urllib.request.Request(base_url + path, data=data, headers=headers) try: with opener.open(request, timeout=min(30, remaining)) as response: raw = response.read(2_000_001) if len(raw) > 2_000_000: raise RuntimeError('Response exceeds the size limit.') return raw if binary else json.loads(raw) except urllib.error.HTTPError as error: message = error.read(16000).decode('utf-8', errors='replace') if error.code not in (429, 500, 502, 503, 504) or attempt == 3: raise RuntimeError(f'API error {error.code}: {message}') from None try: delay = max(1, min(60, int(error.headers.get('Retry-After', 2 ** attempt)))) except ValueError: delay = 2 ** attempt except (urllib.error.URLError, TimeoutError): if attempt == 3: raise RuntimeError('Network error. Retry with the same --request-id.') from None delay = 2 ** attempt time.sleep(min(delay, max(0, deadline - time.monotonic()))) raise RuntimeError('Retry limit reached.') job = call('/api/v1/generations', payload)['job'] job_id = urllib.parse.quote(job['id'], safe='') print(f"Job: {job['id']}" + (' (demo sample)' if job['demo'] else ''), file=sys.stderr) path = f'/api/v1/generations/{job_id}' while job['status'] not in ('completed', 'failed'): time.sleep(min(3, max(0, deadline - time.monotonic()))) job = call(path)['job'] if job['status'] == 'failed': raise RuntimeError(job['error'] or 'Generation failed; credits were returned.') args.output.write_bytes(call(path + '/file', binary=True)) print(f'Saved SVG to {args.output}') if __name__ == '__main__': try: main() except (RuntimeError, OSError, ValueError) as error: print(str(error), file=sys.stderr) sys.exit(1)