refactor: rename project from Freepik to Magnific

Rename all identifiers, strings, file names, env vars, CLI entry point,
ASCII banner, and API endpoint to reflect the company rebrand.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-05-16 13:06:16 +02:00
parent e013db9065
commit 941fd14ccf
24 changed files with 294 additions and 294 deletions
+3
View File
@@ -0,0 +1,3 @@
"""Magnific AI CLI — generate images, videos, and more."""
__version__ = "0.1.0"
View File
+116
View File
@@ -0,0 +1,116 @@
"""Magnific HTTP client with authentication, error handling, and download support."""
from __future__ import annotations
from typing import Any, Optional
import httpx
from magnific_cli import __version__
BASE_URL = "https://api.magnific.ai"
DEFAULT_TIMEOUT = 60.0
class MagnificAPIError(Exception):
"""Raised when the Magnific API returns an error response."""
def __init__(self, message: str, status_code: Optional[int] = None, raw: Optional[dict] = None):
super().__init__(message)
self.status_code = status_code
self.raw = raw or {}
@classmethod
def from_response(cls, response: httpx.Response) -> "MagnificAPIError":
try:
body = response.json()
except Exception:
body = {}
message = (
body.get("message")
or body.get("error", {}).get("message")
or body.get("errors", [{}])[0].get("message")
or f"HTTP {response.status_code}"
)
# Append individual field validation errors when present
invalid = body.get("invalid_params", [])
if invalid:
details = "\n".join(
f"{p.get('field', '?')}: {p.get('reason', '')}" for p in invalid
)
message = f"{message}\n\n{details}"
hints = {
401: "Check your API key — set MAGNIFIC_API_KEY or use --api-key.",
403: "Your plan may not support this feature. Check your Magnific subscription.",
429: "Rate limit exceeded. Please wait before retrying.",
}
hint = hints.get(response.status_code)
if hint:
message = f"{message}\n\nHint: {hint}"
return cls(message, status_code=response.status_code, raw=body)
class MagnificClient:
"""Thin synchronous HTTP wrapper around the Magnific API."""
def __init__(
self,
api_key: str,
base_url: str = BASE_URL,
timeout: float = DEFAULT_TIMEOUT,
) -> None:
self._client = httpx.Client(
base_url=base_url,
headers={
"x-magnific-api-key": api_key,
"Content-Type": "application/json",
"Accept": "application/json",
"User-Agent": f"magnific-cli/{__version__}",
},
timeout=httpx.Timeout(timeout),
)
def post(self, path: str, json: dict[str, Any]) -> dict[str, Any]:
try:
response = self._client.post(path, json=json)
response.raise_for_status()
return response.json()
except httpx.HTTPStatusError as exc:
raise MagnificAPIError.from_response(exc.response) from exc
except httpx.RequestError as exc:
raise MagnificAPIError(f"Network error: {exc}") from exc
def post_multipart(self, path: str, data: dict[str, Any], files: dict[str, Any]) -> dict[str, Any]:
"""POST with multipart/form-data (for file uploads)."""
headers = {k: v for k, v in self._client.headers.items() if k.lower() != "content-type"}
try:
response = self._client.post(path, data=data, files=files, headers=headers)
response.raise_for_status()
return response.json()
except httpx.HTTPStatusError as exc:
raise MagnificAPIError.from_response(exc.response) from exc
except httpx.RequestError as exc:
raise MagnificAPIError(f"Network error: {exc}") from exc
def get(self, path: str, params: dict[str, Any] | None = None) -> dict[str, Any]:
try:
response = self._client.get(path, params=params)
response.raise_for_status()
return response.json()
except httpx.HTTPStatusError as exc:
raise MagnificAPIError.from_response(exc.response) from exc
except httpx.RequestError as exc:
raise MagnificAPIError(f"Network error: {exc}") from exc
def __enter__(self) -> "MagnificClient":
return self
def __exit__(self, *args: Any) -> None:
self._client.close()
def close(self) -> None:
self._client.close()
+98
View File
@@ -0,0 +1,98 @@
"""Image editing API methods: expand, relight, style-transfer, icons."""
from __future__ import annotations
from typing import Any, Optional, Tuple
from magnific_cli.api.client import MagnificClient
from magnific_cli.api.models import IconStyle, get_output_urls, get_status, get_task_id
class EditAPI:
def __init__(self, client: MagnificClient) -> None:
self._client = client
# ------------------------------------------------------------------
# Icon generation
# ------------------------------------------------------------------
def generate_icon(
self,
prompt: str,
style: IconStyle = IconStyle.COLOR,
num_inference_steps: int = 30,
guidance_scale: float = 7.5,
seed: Optional[int] = None,
) -> str:
"""Submit a text-to-icon task. Returns task_id."""
payload: dict[str, Any] = {
"prompt": prompt,
"style": style.value,
"num_inference_steps": num_inference_steps,
"guidance_scale": guidance_scale,
}
if seed is not None:
payload["seed"] = seed
raw = self._client.post("/v1/ai/text-to-icon", json=payload)
return get_task_id(raw)
def icon_status(self, task_id: str) -> Tuple[str, dict[str, Any]]:
raw = self._client.get(f"/v1/ai/text-to-icon/{task_id}")
return get_status(raw), raw
def render_icon(self, task_id: str, fmt: str = "png") -> str:
"""Get the download URL for a completed icon in PNG or SVG format."""
raw = self._client.post(f"/v1/ai/text-to-icon/{task_id}/render/{fmt}", json={})
data = raw.get("data", raw)
return data.get("url") or data.get("download_url") or ""
# ------------------------------------------------------------------
# Relight
# ------------------------------------------------------------------
def relight_submit(
self,
image_b64: str,
prompt: Optional[str] = None,
style: Optional[str] = None,
) -> str:
payload: dict[str, Any] = {"image": image_b64}
if prompt:
payload["prompt"] = prompt
if style:
payload["style"] = style
raw = self._client.post("/v1/ai/image-relight", json=payload)
return get_task_id(raw)
def relight_status(self, task_id: str) -> Tuple[str, dict[str, Any]]:
raw = self._client.get(f"/v1/ai/image-relight/{task_id}")
return get_status(raw), raw
# ------------------------------------------------------------------
# Style Transfer
# ------------------------------------------------------------------
def style_transfer_submit(
self,
content_image_b64: str,
style_image_b64: str,
strength: Optional[float] = None,
) -> str:
payload: dict[str, Any] = {
"content_image": content_image_b64,
"style_image": style_image_b64,
}
if strength is not None:
payload["strength"] = strength
raw = self._client.post("/v1/ai/image-style-transfer", json=payload)
return get_task_id(raw)
def style_transfer_status(self, task_id: str) -> Tuple[str, dict[str, Any]]:
raw = self._client.get(f"/v1/ai/image-style-transfer/{task_id}")
return get_status(raw), raw
def get_output_urls(self, raw: dict[str, Any]) -> list[str]:
return get_output_urls(raw)
+105
View File
@@ -0,0 +1,105 @@
"""Image generation and analysis API methods."""
from __future__ import annotations
from typing import Any, Optional, Tuple
from magnific_cli.api.client import MagnificClient
from magnific_cli.api.models import (
IMAGE_POST_ENDPOINTS,
IMAGE_STATUS_ENDPOINTS,
ImageModel,
get_output_urls,
get_status,
get_task_id,
)
class ImageAPI:
def __init__(self, client: MagnificClient) -> None:
self._client = client
def generate(self, model: ImageModel, payload: dict[str, Any]) -> str:
"""Submit a generation task. Returns task_id."""
endpoint = IMAGE_POST_ENDPOINTS[model]
raw = self._client.post(endpoint, json=payload)
return get_task_id(raw)
def get_status(self, model: ImageModel, task_id: str) -> Tuple[str, dict[str, Any]]:
"""Poll status. Returns (status_str, raw_response)."""
endpoint = IMAGE_STATUS_ENDPOINTS[model].format(task_id=task_id)
raw = self._client.get(endpoint)
return get_status(raw), raw
def get_output_urls(self, raw: dict[str, Any]) -> list[str]:
return get_output_urls(raw)
# ------------------------------------------------------------------
# Image-to-prompt (describe)
# ------------------------------------------------------------------
def describe_submit(self, image_b64: str) -> str:
"""Submit image-to-prompt task. Returns task_id."""
raw = self._client.post("/v1/ai/image-to-prompt", json={"image": image_b64})
return get_task_id(raw)
def describe_status(self, task_id: str) -> Tuple[str, dict[str, Any]]:
raw = self._client.get(f"/v1/ai/image-to-prompt/{task_id}")
return get_status(raw), raw
def get_prompt_text(self, raw: dict[str, Any]) -> str:
"""Extract generated prompt text from a completed describe response."""
data = raw.get("data", raw)
return (
data.get("prompt")
or data.get("description")
or data.get("text")
or data.get("result", {}).get("prompt", "")
or ""
)
# ------------------------------------------------------------------
# Image expansion (outpainting)
# ------------------------------------------------------------------
def expand_submit(
self,
model: str,
image_b64: str,
left: int = 0,
right: int = 0,
top: int = 0,
bottom: int = 0,
prompt: Optional[str] = None,
seed: Optional[int] = None,
) -> str:
payload: dict[str, Any] = {
"image": image_b64,
"left": left,
"right": right,
"top": top,
"bottom": bottom,
}
if prompt:
payload["prompt"] = prompt
if seed is not None:
payload["seed"] = seed
endpoint_map = {
"flux-pro": "/v1/ai/image-expand/flux-pro",
"ideogram": "/v1/ai/image-expand/ideogram",
"seedream-v4-5": "/v1/ai/image-expand/seedream-v4-5",
}
endpoint = endpoint_map.get(model, "/v1/ai/image-expand/flux-pro")
raw = self._client.post(endpoint, json=payload)
return get_task_id(raw)
def expand_status(self, model: str, task_id: str) -> Tuple[str, dict[str, Any]]:
endpoint_map = {
"flux-pro": "/v1/ai/image-expand/flux-pro",
"ideogram": "/v1/ai/image-expand/ideogram",
"seedream-v4-5": "/v1/ai/image-expand/seedream-v4-5",
}
base = endpoint_map.get(model, "/v1/ai/image-expand/flux-pro")
raw = self._client.get(f"{base}/{task_id}")
return get_status(raw), raw
+246
View File
@@ -0,0 +1,246 @@
"""Enums, endpoint maps, and response normalization utilities."""
from __future__ import annotations
from enum import Enum
from typing import Any
# ---------------------------------------------------------------------------
# Model enums
# ---------------------------------------------------------------------------
class ImageModel(str, Enum):
MYSTIC = "mystic"
FLUX_KONTEXT_PRO = "flux-kontext-pro"
FLUX_2_PRO = "flux-2-pro"
FLUX_2_TURBO = "flux-2-turbo"
FLUX_PRO_1_1 = "flux-pro-1.1"
SEEDREAM_V4 = "seedream-v4"
SEEDREAM_V4_5 = "seedream-v4-5"
IDEOGRAM_V2 = "ideogram-v2"
class VideoModel(str, Enum):
KLING_O1_PRO = "kling-o1-pro"
KLING_O1_STD = "kling-o1-std"
KLING_ELEMENTS_PRO = "kling-elements-pro"
KLING_ELEMENTS_STD = "kling-elements-std"
MINIMAX_HAILUO = "minimax-hailuo"
class UpscaleMode(str, Enum):
CREATIVE = "creative"
PRECISION = "precision"
PRECISION_V2 = "precision-v2"
class VideoUpscaleMode(str, Enum):
STANDARD = "standard"
TURBO = "turbo"
class AspectRatio(str, Enum):
LANDSCAPE = "16:9"
PORTRAIT = "9:16"
SQUARE = "1:1"
CLASSIC = "4:3"
WIDE = "21:9"
class IconStyle(str, Enum):
SOLID = "solid"
OUTLINE = "outline"
COLOR = "color"
FLAT = "flat"
STICKER = "sticker"
# ---------------------------------------------------------------------------
# Endpoint maps
# ---------------------------------------------------------------------------
IMAGE_POST_ENDPOINTS: dict[ImageModel, str] = {
ImageModel.MYSTIC: "/v1/ai/mystic",
ImageModel.FLUX_KONTEXT_PRO: "/v1/ai/text-to-image/flux-kontext-pro",
ImageModel.FLUX_2_PRO: "/v1/ai/text-to-image/flux-2-pro",
ImageModel.FLUX_2_TURBO: "/v1/ai/text-to-image/flux-2-turbo",
ImageModel.FLUX_PRO_1_1: "/v1/ai/text-to-image/flux-pro-v1-1",
ImageModel.SEEDREAM_V4: "/v1/ai/text-to-image/seedream-v4",
ImageModel.SEEDREAM_V4_5: "/v1/ai/text-to-image/seedream-v4-5",
ImageModel.IDEOGRAM_V2: "/v1/ai/text-to-image/ideogram-v2",
}
IMAGE_STATUS_ENDPOINTS: dict[ImageModel, str] = {
ImageModel.MYSTIC: "/v1/ai/mystic/{task_id}",
ImageModel.FLUX_KONTEXT_PRO: "/v1/ai/text-to-image/flux-kontext-pro/{task_id}",
ImageModel.FLUX_2_PRO: "/v1/ai/text-to-image/flux-2-pro/{task_id}",
ImageModel.FLUX_2_TURBO: "/v1/ai/text-to-image/flux-2-turbo/{task_id}",
ImageModel.FLUX_PRO_1_1: "/v1/ai/text-to-image/flux-pro-v1-1/{task_id}",
ImageModel.SEEDREAM_V4: "/v1/ai/text-to-image/seedream-v4/{task_id}",
ImageModel.SEEDREAM_V4_5: "/v1/ai/text-to-image/seedream-v4-5/{task_id}",
ImageModel.IDEOGRAM_V2: "/v1/ai/text-to-image/ideogram-v2/{task_id}",
}
VIDEO_POST_ENDPOINTS: dict[VideoModel, str] = {
VideoModel.KLING_O1_PRO: "/v1/ai/image-to-video/kling-o1-pro",
VideoModel.KLING_O1_STD: "/v1/ai/image-to-video/kling-o1-std",
VideoModel.KLING_ELEMENTS_PRO: "/v1/ai/image-to-video/kling-elements-pro",
VideoModel.KLING_ELEMENTS_STD: "/v1/ai/image-to-video/kling-elements-std",
VideoModel.MINIMAX_HAILUO: "/v1/ai/image-to-video/minimax-hailuo-02-1080p",
}
VIDEO_STATUS_ENDPOINTS: dict[VideoModel, str] = {
VideoModel.KLING_O1_PRO: "/v1/ai/image-to-video/kling-o1/{task_id}",
VideoModel.KLING_O1_STD: "/v1/ai/image-to-video/kling-o1/{task_id}",
VideoModel.KLING_ELEMENTS_PRO: "/v1/ai/image-to-video/kling-elements-pro/{task_id}",
VideoModel.KLING_ELEMENTS_STD: "/v1/ai/image-to-video/kling-elements-std/{task_id}",
VideoModel.MINIMAX_HAILUO: "/v1/ai/image-to-video/minimax-hailuo-02-1080p/{task_id}",
}
# Per-model image input field name
VIDEO_IMAGE_FIELDS: dict[VideoModel, str] = {
VideoModel.KLING_O1_PRO: "first_frame",
VideoModel.KLING_O1_STD: "first_frame",
VideoModel.KLING_ELEMENTS_PRO: "images", # expects an array
VideoModel.KLING_ELEMENTS_STD: "images", # expects an array
VideoModel.MINIMAX_HAILUO: "image",
}
# Models that support an aspect_ratio parameter at all
VIDEO_ASPECT_RATIO_MODELS: set[VideoModel] = {
VideoModel.KLING_ELEMENTS_PRO,
VideoModel.KLING_ELEMENTS_STD,
VideoModel.MINIMAX_HAILUO,
}
# Of those, these require named slug aspect ratios instead of "W:H" strings
VIDEO_SLUG_ASPECT_RATIO_MODELS: set[VideoModel] = {
VideoModel.KLING_ELEMENTS_PRO,
VideoModel.KLING_ELEMENTS_STD,
}
_VIDEO_RATIO_TO_SLUG: dict[str, str] = {
"1:1": "square_1_1",
"16:9": "widescreen_16_9",
"9:16": "social_story_9_16",
}
def normalize_aspect_ratio_video(ratio: str, model: VideoModel) -> str:
"""Convert a user-facing aspect ratio to the format required by the video model."""
if model not in VIDEO_SLUG_ASPECT_RATIO_MODELS:
return ratio
return _VIDEO_RATIO_TO_SLUG.get(ratio, ratio)
UPSCALE_POST_ENDPOINTS: dict[UpscaleMode, str] = {
UpscaleMode.CREATIVE: "/v1/ai/image-upscaler",
UpscaleMode.PRECISION: "/v1/ai/image-upscaler-precision",
UpscaleMode.PRECISION_V2: "/v1/ai/image-upscaler-precision-v2",
}
UPSCALE_STATUS_ENDPOINTS: dict[UpscaleMode, str] = {
UpscaleMode.CREATIVE: "/v1/ai/image-upscaler/{task_id}",
UpscaleMode.PRECISION: "/v1/ai/image-upscaler-precision/{task_id}",
UpscaleMode.PRECISION_V2: "/v1/ai/image-upscaler-precision-v2/{task_id}",
}
VIDEO_UPSCALE_POST_ENDPOINTS: dict[VideoUpscaleMode, str] = {
VideoUpscaleMode.STANDARD: "/v1/ai/video-upscaler",
VideoUpscaleMode.TURBO: "/v1/ai/video-upscaler/turbo",
}
VIDEO_UPSCALE_STATUS_ENDPOINT = "/v1/ai/video-upscaler/{task_id}"
# ---------------------------------------------------------------------------
# Aspect ratio normalization
# ---------------------------------------------------------------------------
# Models that require named slug aspect ratios instead of "W:H" strings
SLUG_ASPECT_RATIO_MODELS: set[ImageModel] = {
ImageModel.MYSTIC,
ImageModel.FLUX_PRO_1_1,
ImageModel.SEEDREAM_V4,
ImageModel.SEEDREAM_V4_5,
}
# User-friendly "W:H" → API slug mapping
_RATIO_TO_SLUG: dict[str, str] = {
"1:1": "square_1_1",
"16:9": "widescreen_16_9",
"9:16": "social_story_9_16",
"4:3": "classic_4_3",
"3:4": "traditional_3_4",
"3:2": "standard_3_2",
"2:3": "portrait_2_3",
"2:1": "horizontal_2_1",
"1:2": "vertical_1_2",
"4:5": "social_post_4_5",
"21:9": "widescreen_16_9", # closest match
}
def normalize_aspect_ratio(ratio: str, model: ImageModel) -> str:
"""Convert a user-facing aspect ratio to the format required by the model."""
if model not in SLUG_ASPECT_RATIO_MODELS:
return ratio # free-form models accept "1:1" directly
slug = _RATIO_TO_SLUG.get(ratio)
if slug:
return slug
# Already a slug (user passed "square_1_1" directly) — pass through
return ratio
# ---------------------------------------------------------------------------
# Response normalization helpers
# ---------------------------------------------------------------------------
def get_task_id(raw: dict[str, Any]) -> str:
"""Extract task_id from any response shape."""
data = raw.get("data", raw)
task_id = data.get("task_id") or data.get("id") or raw.get("task_id") or raw.get("id")
if not task_id:
raise ValueError(f"No task_id found in response: {raw}")
return str(task_id)
def get_status(raw: dict[str, Any]) -> str:
"""Extract normalized status string from any response shape."""
data = raw.get("data", raw)
status = data.get("status") or raw.get("status") or "PENDING"
return status.upper().replace(" ", "_")
def get_output_urls(raw: dict[str, Any]) -> list[str]:
"""Extract all output file URLs from a completed task response."""
data = raw.get("data", raw)
# Try common key names in order of likelihood
for key in ("generated", "output", "outputs", "result", "results", "images", "videos"):
items = data.get(key)
if items is None:
continue
if isinstance(items, list) and items:
urls = []
for item in items:
if isinstance(item, dict):
url = item.get("url") or item.get("download_url") or item.get("src")
if url:
urls.append(url)
elif isinstance(item, str):
urls.append(item)
if urls:
return urls
elif isinstance(items, dict):
url = items.get("url") or items.get("download_url") or items.get("src")
if url:
return [url]
elif isinstance(items, str):
return [items]
# Fallback: top-level url field
if "url" in data:
return [data["url"]]
return []
+88
View File
@@ -0,0 +1,88 @@
"""Image and video upscaling API methods."""
from __future__ import annotations
from typing import Any, Optional, Tuple
from magnific_cli.api.client import MagnificClient
from magnific_cli.api.models import (
UPSCALE_POST_ENDPOINTS,
UPSCALE_STATUS_ENDPOINTS,
VIDEO_UPSCALE_POST_ENDPOINTS,
VIDEO_UPSCALE_STATUS_ENDPOINT,
UpscaleMode,
VideoUpscaleMode,
get_output_urls,
get_status,
get_task_id,
)
class UpscaleAPI:
def __init__(self, client: MagnificClient) -> None:
self._client = client
# ------------------------------------------------------------------
# Image upscaling
# ------------------------------------------------------------------
def upscale_image(
self,
mode: UpscaleMode,
image_b64: str,
scale_factor: Optional[str] = None,
creativity: Optional[int] = None,
prompt: Optional[str] = None,
seed: Optional[int] = None,
) -> str:
"""Submit an image upscale task. Returns task_id."""
payload: dict[str, Any] = {"image": image_b64}
if scale_factor:
# Convert "2x" → 2, "4x" → 4
factor = scale_factor.rstrip("xX")
try:
payload["scale_factor"] = int(factor)
except ValueError:
payload["scale_factor"] = 2
if mode == UpscaleMode.CREATIVE:
if creativity is not None:
payload["creativity"] = creativity
if prompt:
payload["prompt"] = prompt
if seed is not None:
payload["seed"] = seed
endpoint = UPSCALE_POST_ENDPOINTS[mode]
raw = self._client.post(endpoint, json=payload)
return get_task_id(raw)
def upscale_image_status(self, mode: UpscaleMode, task_id: str) -> Tuple[str, dict[str, Any]]:
endpoint = UPSCALE_STATUS_ENDPOINTS[mode].format(task_id=task_id)
raw = self._client.get(endpoint)
return get_status(raw), raw
# ------------------------------------------------------------------
# Video upscaling
# ------------------------------------------------------------------
def upscale_video(
self,
mode: VideoUpscaleMode,
video_b64: str,
) -> str:
"""Submit a video upscale task. Returns task_id."""
payload: dict[str, Any] = {"video": video_b64}
endpoint = VIDEO_UPSCALE_POST_ENDPOINTS[mode]
raw = self._client.post(endpoint, json=payload)
return get_task_id(raw)
def upscale_video_status(self, task_id: str) -> Tuple[str, dict[str, Any]]:
endpoint = VIDEO_UPSCALE_STATUS_ENDPOINT.format(task_id=task_id)
raw = self._client.get(endpoint)
return get_status(raw), raw
def get_output_urls(self, raw: dict[str, Any]) -> list[str]:
return get_output_urls(raw)
+73
View File
@@ -0,0 +1,73 @@
"""Video generation API methods."""
from __future__ import annotations
from typing import Any, Optional, Tuple
from magnific_cli.api.client import MagnificClient
from magnific_cli.api.models import (
VIDEO_POST_ENDPOINTS,
VIDEO_STATUS_ENDPOINTS,
VIDEO_IMAGE_FIELDS,
VIDEO_ASPECT_RATIO_MODELS,
VideoModel,
get_output_urls,
get_status,
get_task_id,
normalize_aspect_ratio_video,
)
class VideoAPI:
def __init__(self, client: MagnificClient) -> None:
self._client = client
def generate(
self,
model: VideoModel,
image_b64: str,
prompt: Optional[str] = None,
duration: int = 5,
aspect_ratio: str = "16:9",
seed: Optional[int] = None,
) -> str:
"""Submit an image-to-video task. Returns task_id."""
# Video API expects raw base64, not a data URI
if ";base64," in image_b64:
image_b64 = image_b64.split(";base64,", 1)[1]
image_field = VIDEO_IMAGE_FIELDS[model]
# kling-elements uses an array; all others use a scalar
if image_field == "images":
payload: dict[str, Any] = {"images": [image_b64]}
else:
payload = {image_field: image_b64}
if prompt:
payload["prompt"] = prompt
# minimax only supports duration=6; clamp silently
effective_duration = duration
if model == VideoModel.MINIMAX_HAILUO:
effective_duration = 6
payload["duration"] = str(effective_duration)
if aspect_ratio and model in VIDEO_ASPECT_RATIO_MODELS:
payload["aspect_ratio"] = normalize_aspect_ratio_video(aspect_ratio, model)
if seed is not None:
payload["seed"] = seed
endpoint = VIDEO_POST_ENDPOINTS[model]
raw = self._client.post(endpoint, json=payload)
return get_task_id(raw)
def get_status(self, model: VideoModel, task_id: str) -> Tuple[str, dict[str, Any]]:
"""Poll status. Returns (status_str, raw_response)."""
endpoint = VIDEO_STATUS_ENDPOINTS[model].format(task_id=task_id)
raw = self._client.get(endpoint)
return get_status(raw), raw
def get_output_urls(self, raw: dict[str, Any]) -> list[str]:
return get_output_urls(raw)
View File
+97
View File
@@ -0,0 +1,97 @@
"""describe-image command."""
from __future__ import annotations
from pathlib import Path
from typing import Annotated, Optional
import typer
from magnific_cli.api.client import MagnificAPIError, MagnificClient
from magnific_cli.api.images import ImageAPI
from magnific_cli.utils.config import MagnificConfig
from magnific_cli.utils.console import console, print_describe_result, print_error, print_no_wait
from magnific_cli.utils.files import image_to_base64
from magnific_cli.utils.polling import MagnificTaskError, MagnificTimeoutError, PollConfig, poll_task
def _get_api_key(api_key: Optional[str], config: MagnificConfig) -> str:
key = api_key or config.api_key
if not key:
print_error("No API key found.", hint="Set [cyan]MAGNIFIC_API_KEY[/cyan] or pass [cyan]--api-key[/cyan].")
raise typer.Exit(1)
return key
def describe_image(
image: Annotated[
Path,
typer.Argument(
help="Image to analyze and describe",
exists=True, file_okay=True, dir_okay=False,
),
],
output: Annotated[
Optional[Path],
typer.Option("--output", "-o", help="Save the generated prompt to a text file"),
] = None,
wait: Annotated[
bool,
typer.Option("--wait/--no-wait"),
] = True,
api_key: Annotated[
Optional[str],
typer.Option("--api-key", envvar="MAGNIFIC_API_KEY"),
] = None,
) -> None:
"""
[bold]Describe an image[/bold] and generate a text prompt for it.
Reverse-engineers the image into an AI-ready prompt you can use with
[cyan]generate-image[/cyan].
[dim]Examples:[/dim]
magnific describe-image photo.jpg
magnific describe-image scene.png --output prompt.txt
"""
config = MagnificConfig.load()
key = _get_api_key(api_key, config)
image_b64 = image_to_base64(image)
with MagnificClient(key, base_url=config.base_url) as client:
api = ImageAPI(client)
with console.status("[info]Submitting image analysis…[/info]"):
try:
task_id = api.describe_submit(image_b64)
except MagnificAPIError as exc:
print_error(str(exc))
raise typer.Exit(1)
if not wait:
print_no_wait(task_id, "describe-image", "image-to-prompt")
return
poll_config = PollConfig(task_type="describe", max_wait=config.poll_timeout)
try:
result = poll_task(
check_fn=lambda tid: api.describe_status(tid),
task_id=task_id,
config=poll_config,
console=console,
)
except (MagnificTaskError, MagnificTimeoutError) as exc:
print_error(str(exc))
raise typer.Exit(1)
prompt_text = api.get_prompt_text(result)
if not prompt_text:
print_error("Analysis completed but no prompt text found.")
raise typer.Exit(1)
saved_path: Optional[Path] = None
if output:
output.write_text(prompt_text, encoding="utf-8")
saved_path = output
print_describe_result(task_id, prompt_text, saved_path)
+120
View File
@@ -0,0 +1,120 @@
"""config set/get/show/reset commands."""
from __future__ import annotations
from typing import Annotated, Optional
import typer
from rich.prompt import Confirm
from magnific_cli.utils.config import CONFIG_FILE, MagnificConfig
from magnific_cli.utils.console import console, print_config_table, print_config_toml, print_error, print_warning
app = typer.Typer(
name="config",
help="[bold]Manage[/bold] Magnific CLI configuration.",
rich_markup_mode="rich",
no_args_is_help=True,
)
@app.command("show")
def config_show(
toml: Annotated[
bool,
typer.Option("--toml", help="Output as TOML syntax instead of a table"),
] = False,
) -> None:
"""[bold]Show[/bold] all current configuration values."""
config = MagnificConfig.load()
d = config.to_display_dict()
if toml:
print_config_toml(d)
else:
print_config_table(d)
console.print(f"\n[dim]Config file:[/dim] {CONFIG_FILE}")
@app.command("get")
def config_get(
key: Annotated[str, typer.Argument(help="Config key to retrieve")],
) -> None:
"""[bold]Get[/bold] the value of a single configuration key."""
config = MagnificConfig.load()
d = config.to_display_dict()
if key not in d:
print_error(
f"Unknown config key: '{key}'",
hint=f"Run [cyan]magnific config show[/cyan] to see all available keys.",
)
raise typer.Exit(1)
value = d[key]
if key == "api_key" and value:
masked = f"{'*' * 8}{str(value)[-4:]}"
console.print(f"[dim]{key}[/dim] = [bold]{masked}[/bold] [dim](masked)[/dim]")
elif value is None:
console.print(f"[dim]{key}[/dim] = [dim]not set[/dim]")
else:
console.print(f"[dim]{key}[/dim] = [bold]{value}[/bold]")
@app.command("set")
def config_set(
key: Annotated[str, typer.Argument(help="Config key to update")],
value: Annotated[str, typer.Argument(help="New value")],
) -> None:
"""
[bold]Set[/bold] a configuration value.
[dim]Examples:[/dim]
magnific config set default_image_model mystic
magnific config set default_output_dir ~/images
magnific config set poll_timeout 300
[dim]Note:[/dim] The API key is never saved to disk. Use the
[cyan]MAGNIFIC_API_KEY[/cyan] environment variable instead.
"""
config = MagnificConfig.load()
try:
config.set_value(key, value)
console.print(f"[success]✓[/success] Set [cyan]{key}[/cyan] = [bold]{value}[/bold]")
console.print(f"[dim]Saved to:[/dim] {CONFIG_FILE}")
except ValueError as exc:
print_error(str(exc))
raise typer.Exit(1)
@app.command("reset")
def config_reset(
yes: Annotated[
bool,
typer.Option("--yes", "-y", help="Skip confirmation prompt"),
] = False,
) -> None:
"""[bold]Reset[/bold] all configuration to defaults."""
if not yes:
confirmed = Confirm.ask(
"[warning]Reset all configuration to defaults?[/warning]",
console=console,
default=False,
)
if not confirmed:
console.print("[dim]Aborted.[/dim]")
return
if CONFIG_FILE.exists():
CONFIG_FILE.unlink()
console.print(f"[success]✓[/success] Configuration reset. Deleted: [path]{CONFIG_FILE}[/path]")
else:
print_warning("No config file found — already at defaults.")
@app.command("path")
def config_path() -> None:
"""Show the path to the configuration file."""
console.print(f"[path]{CONFIG_FILE}[/path]")
if CONFIG_FILE.exists():
console.print("[dim](file exists)[/dim]")
else:
console.print("[dim](file does not exist — using defaults)[/dim]")
+215
View File
@@ -0,0 +1,215 @@
"""expand-image, relight, style-transfer commands."""
from __future__ import annotations
from pathlib import Path
from typing import Annotated, Optional
import typer
from magnific_cli.api.client import MagnificAPIError, MagnificClient
from magnific_cli.api.edit import EditAPI
from magnific_cli.utils.config import MagnificConfig
from magnific_cli.utils.console import GenerationResult, console, print_error, print_no_wait, print_result
from magnific_cli.utils.files import auto_output_path, get_image_dimensions, image_to_base64, save_from_url
from magnific_cli.utils.polling import MagnificTaskError, MagnificTimeoutError, PollConfig, poll_task
_EXPAND_MODELS = ["flux-pro", "ideogram", "seedream-v4-5"]
def _get_api_key(api_key: Optional[str], config: MagnificConfig) -> str:
key = api_key or config.api_key
if not key:
print_error("No API key found.", hint="Set [cyan]MAGNIFIC_API_KEY[/cyan] or pass [cyan]--api-key[/cyan].")
raise typer.Exit(1)
return key
def expand_image(
image: Annotated[
Path,
typer.Argument(help="Image to expand (outpaint)", exists=True),
],
left: Annotated[int, typer.Option("--left", help="Pixels to add on the left (02048)", min=0, max=2048)] = 0,
right: Annotated[int, typer.Option("--right", help="Pixels to add on the right (02048)", min=0, max=2048)] = 0,
top: Annotated[int, typer.Option("--top", help="Pixels to add on top (02048)", min=0, max=2048)] = 0,
bottom: Annotated[int, typer.Option("--bottom", help="Pixels to add on bottom (02048)", min=0, max=2048)] = 0,
prompt: Annotated[Optional[str], typer.Option("--prompt", "-p", help="Optional prompt to guide the expansion")] = None,
model: Annotated[str, typer.Option("--model", "-m", help=f"Expansion model: {', '.join(_EXPAND_MODELS)}")] = "flux-pro",
seed: Annotated[Optional[int], typer.Option("--seed")] = None,
output: Annotated[Optional[Path], typer.Option("--output", "-o")] = None,
wait: Annotated[bool, typer.Option("--wait/--no-wait")] = True,
api_key: Annotated[Optional[str], typer.Option("--api-key", envvar="MAGNIFIC_API_KEY")] = None,
) -> None:
"""
[bold]Expand an image[/bold] by adding new content around its edges (outpainting).
[dim]Examples:[/dim]
magnific expand-image photo.jpg --left 512 --right 512 --prompt "lush forest"
magnific expand-image banner.png --bottom 256 --model seedream-v4-5
"""
if not any([left, right, top, bottom]):
print_error("At least one of --left, --right, --top, --bottom must be > 0.")
raise typer.Exit(1)
if model not in _EXPAND_MODELS:
print_error(f"Unknown model '{model}'.", hint=f"Choose from: {', '.join(_EXPAND_MODELS)}")
raise typer.Exit(1)
config = MagnificConfig.load()
key = _get_api_key(api_key, config)
image_b64 = image_to_base64(image)
with MagnificClient(key, base_url=config.base_url) as client:
api = EditAPI(client)
with console.status("[info]Submitting image expansion…[/info]"):
try:
task_id = api.expand_submit(model, image_b64, left, right, top, bottom, prompt, seed)
except MagnificAPIError as exc:
print_error(str(exc))
raise typer.Exit(1)
if not wait:
print_no_wait(task_id, "expand-image", model)
return
poll_config = PollConfig(task_type="expand", max_wait=config.poll_timeout)
from magnific_cli.api.images import ImageAPI
img_api = ImageAPI(client)
try:
result = poll_task(
check_fn=lambda tid: img_api.expand_status(model, tid),
task_id=task_id,
config=poll_config,
console=console,
extra_info={"Model": model, "Expand": f"L{left} R{right} T{top} B{bottom}"},
)
except (MagnificTaskError, MagnificTimeoutError) as exc:
print_error(str(exc))
raise typer.Exit(1)
from magnific_cli.api.models import get_output_urls
urls = get_output_urls(result)
if not urls:
print_error("Expansion completed but no output URL found.")
raise typer.Exit(1)
out = output or auto_output_path("expanded", model, image.suffix.lstrip(".") or "jpg", config.default_output_dir)
save_from_url(urls[0], out, console)
w, h = get_image_dimensions(out)
print_result(GenerationResult(task_id=task_id, model=model, output_path=out, width=w, height=h, task_type="expand"))
def relight_image(
image: Annotated[Path, typer.Argument(help="Image to relight", exists=True)],
prompt: Annotated[Optional[str], typer.Option("--prompt", "-p", help="Lighting description e.g. 'golden hour sunlight'")] = None,
style: Annotated[Optional[str], typer.Option("--style", "-s", help="Lighting style preset")] = None,
output: Annotated[Optional[Path], typer.Option("--output", "-o")] = None,
wait: Annotated[bool, typer.Option("--wait/--no-wait")] = True,
api_key: Annotated[Optional[str], typer.Option("--api-key", envvar="MAGNIFIC_API_KEY")] = None,
) -> None:
"""
[bold]Relight an image[/bold] using AI-controlled lighting. [dim](Premium feature)[/dim]
[dim]Examples:[/dim]
magnific relight portrait.jpg --prompt "dramatic studio lighting"
magnific relight scene.png --prompt "warm golden hour" --output relit.jpg
"""
config = MagnificConfig.load()
key = _get_api_key(api_key, config)
image_b64 = image_to_base64(image)
with MagnificClient(key, base_url=config.base_url) as client:
api = EditAPI(client)
with console.status("[info]Submitting image relight…[/info]"):
try:
task_id = api.relight_submit(image_b64, prompt, style)
except MagnificAPIError as exc:
print_error(str(exc))
raise typer.Exit(1)
if not wait:
print_no_wait(task_id, "relight", "image-relight")
return
poll_config = PollConfig(task_type="relight", max_wait=config.poll_timeout)
try:
result = poll_task(
check_fn=lambda tid: api.relight_status(tid),
task_id=task_id,
config=poll_config,
console=console,
)
except (MagnificTaskError, MagnificTimeoutError) as exc:
print_error(str(exc))
raise typer.Exit(1)
urls = api.get_output_urls(result)
if not urls:
print_error("Relight completed but no output URL found.")
raise typer.Exit(1)
out = output or auto_output_path("relit", "relight", image.suffix.lstrip(".") or "jpg", config.default_output_dir)
save_from_url(urls[0], out, console)
w, h = get_image_dimensions(out)
print_result(GenerationResult(task_id=task_id, model="image-relight", output_path=out, width=w, height=h, task_type="relight"))
def style_transfer(
content: Annotated[Path, typer.Argument(help="Content image (what to keep)", exists=True)],
style_image: Annotated[Path, typer.Argument(help="Style image (the artistic style to apply)", exists=True)],
strength: Annotated[Optional[float], typer.Option("--strength", help="Style strength 0.01.0", min=0.0, max=1.0)] = None,
output: Annotated[Optional[Path], typer.Option("--output", "-o")] = None,
wait: Annotated[bool, typer.Option("--wait/--no-wait")] = True,
api_key: Annotated[Optional[str], typer.Option("--api-key", envvar="MAGNIFIC_API_KEY")] = None,
) -> None:
"""
[bold]Apply an artistic style[/bold] from one image onto another. [dim](Premium feature)[/dim]
[dim]Examples:[/dim]
magnific style-transfer photo.jpg painting.jpg --strength 0.8
magnific style-transfer portrait.png van_gogh.jpg --output styled.jpg
"""
config = MagnificConfig.load()
key = _get_api_key(api_key, config)
content_b64 = image_to_base64(content)
style_b64 = image_to_base64(style_image)
with MagnificClient(key, base_url=config.base_url) as client:
api = EditAPI(client)
with console.status("[info]Submitting style transfer…[/info]"):
try:
task_id = api.style_transfer_submit(content_b64, style_b64, strength)
except MagnificAPIError as exc:
print_error(str(exc))
raise typer.Exit(1)
if not wait:
print_no_wait(task_id, "style-transfer", "image-style-transfer")
return
poll_config = PollConfig(task_type="style-transfer", max_wait=config.poll_timeout)
try:
result = poll_task(
check_fn=lambda tid: api.style_transfer_status(tid),
task_id=task_id,
config=poll_config,
console=console,
)
except (MagnificTaskError, MagnificTimeoutError) as exc:
print_error(str(exc))
raise typer.Exit(1)
urls = api.get_output_urls(result)
if not urls:
print_error("Style transfer completed but no output URL found.")
raise typer.Exit(1)
out = output or auto_output_path("styled", "style-transfer", "jpg", config.default_output_dir)
save_from_url(urls[0], out, console)
w, h = get_image_dimensions(out)
print_result(GenerationResult(task_id=task_id, model="image-style-transfer", output_path=out, width=w, height=h, task_type="style-transfer"))
+359
View File
@@ -0,0 +1,359 @@
"""generate-image, generate-video, generate-icon commands."""
from __future__ import annotations
from pathlib import Path
from typing import Annotated, Optional
import typer
from magnific_cli.api.client import MagnificAPIError, MagnificClient
from magnific_cli.api.edit import EditAPI
from magnific_cli.api.images import ImageAPI
from magnific_cli.api.models import IconStyle, ImageModel, VideoModel, normalize_aspect_ratio
from magnific_cli.api.videos import VideoAPI
from magnific_cli.utils.config import MagnificConfig
from magnific_cli.utils.console import GenerationResult, console, print_error, print_no_wait, print_result
from magnific_cli.utils.files import auto_output_path, get_image_dimensions, image_to_base64, save_from_url
from magnific_cli.utils.polling import MagnificTaskError, MagnificTimeoutError, PollConfig, poll_task
def _get_api_key(api_key: Optional[str], config: MagnificConfig) -> str:
key = api_key or config.api_key
if not key:
print_error(
"No API key found.",
hint="Set the [cyan]MAGNIFIC_API_KEY[/cyan] environment variable, "
"or pass [cyan]--api-key YOUR_KEY[/cyan].",
)
raise typer.Exit(1)
return key
def generate_image(
prompt: Annotated[str, typer.Argument(help="Text prompt describing the image to generate")],
model: Annotated[
ImageModel,
typer.Option("--model", "-m", help="AI model to use for generation", show_default=True),
] = ImageModel.FLUX_2_PRO,
output: Annotated[
Optional[Path],
typer.Option("--output", "-o", help="Output file path (auto-generated if omitted)"),
] = None,
aspect_ratio: Annotated[
Optional[str],
typer.Option(
"--aspect-ratio", "-a",
help="Aspect ratio e.g. [cyan]16:9[/cyan], [cyan]1:1[/cyan], [cyan]9:16[/cyan], [cyan]4:3[/cyan]",
),
] = None,
negative_prompt: Annotated[
Optional[str],
typer.Option("--negative-prompt", "-n", help="Concepts to exclude from the image"),
] = None,
seed: Annotated[
Optional[int],
typer.Option("--seed", help="Random seed for reproducibility"),
] = None,
input_image: Annotated[
Optional[Path],
typer.Option(
"--input-image", "-i",
help="Reference image for img2img / editing (flux-kontext-pro)",
exists=True, file_okay=True, dir_okay=False,
),
] = None,
wait: Annotated[
bool,
typer.Option("--wait/--no-wait", help="Wait for completion or return task ID immediately"),
] = True,
api_key: Annotated[
Optional[str],
typer.Option("--api-key", envvar="MAGNIFIC_API_KEY", help="Magnific API key"),
] = None,
) -> None:
"""
[bold]Generate an image[/bold] using Magnific AI models.
[dim]Examples:[/dim]
magnific generate-image "a cat on the moon" --model flux-2-pro
magnific generate-image "cyberpunk city at night" --model mystic --aspect-ratio 16:9
magnific generate-image "make the sky orange" --model flux-kontext-pro --input-image photo.jpg
"""
config = MagnificConfig.load()
key = _get_api_key(api_key, config)
# Build request payload
payload: dict = {"prompt": prompt}
if aspect_ratio:
payload["aspect_ratio"] = normalize_aspect_ratio(aspect_ratio, model)
if negative_prompt:
payload["negative_prompt"] = negative_prompt
if seed is not None:
payload["seed"] = seed
if input_image:
payload["image"] = image_to_base64(input_image)
with MagnificClient(key, base_url=config.base_url) as client:
api = ImageAPI(client)
with console.status(f"[info]Submitting {model.value} generation…[/info]"):
try:
task_id = api.generate(model, payload)
except MagnificAPIError as exc:
print_error(str(exc))
raise typer.Exit(1)
if not wait:
print_no_wait(task_id, "image", model.value)
return
poll_config = PollConfig(task_type="image", max_wait=config.poll_timeout)
try:
result = poll_task(
check_fn=lambda tid: api.get_status(model, tid),
task_id=task_id,
config=poll_config,
console=console,
extra_info={"Model": f"[magenta]{model.value}[/magenta]"},
)
except (MagnificTaskError, MagnificTimeoutError) as exc:
print_error(str(exc))
raise typer.Exit(1)
urls = api.get_output_urls(result)
if not urls:
print_error("Generation completed but no output URLs found.", hint="Check the Magnific dashboard.")
raise typer.Exit(1)
out = output or auto_output_path("image", model.value, "jpg", config.default_output_dir)
save_from_url(urls[0], out, console)
w, h = get_image_dimensions(out)
print_result(
GenerationResult(
task_id=task_id,
model=model.value,
output_path=out,
width=w,
height=h,
seed=seed,
task_type="image",
)
)
def generate_video(
image: Annotated[
Path,
typer.Argument(
help="Source image path to animate",
exists=True, file_okay=True, dir_okay=False,
),
],
model: Annotated[
VideoModel,
typer.Option("--model", "-m", help="Video AI model", show_default=True),
] = VideoModel.KLING_O1_PRO,
prompt: Annotated[
Optional[str],
typer.Option("--prompt", "-p", help="Motion/style guidance prompt"),
] = None,
duration: Annotated[
int,
typer.Option(
"--duration", "-d",
help="Duration in seconds: [cyan]5[/cyan] or [cyan]10[/cyan] (minimax-hailuo is fixed at 6s)",
min=5, max=10,
),
] = 5,
aspect_ratio: Annotated[
str,
typer.Option("--aspect-ratio", "-a", help="Output aspect ratio: [cyan]16:9[/cyan] | [cyan]9:16[/cyan] | [cyan]1:1[/cyan]"),
] = "16:9",
seed: Annotated[
Optional[int],
typer.Option("--seed", help="Random seed for reproducibility"),
] = None,
output: Annotated[
Optional[Path],
typer.Option("--output", "-o", help="Output video file path"),
] = None,
wait: Annotated[
bool,
typer.Option("--wait/--no-wait", help="Wait for completion or return task ID immediately"),
] = True,
api_key: Annotated[
Optional[str],
typer.Option("--api-key", envvar="MAGNIFIC_API_KEY", help="Magnific API key"),
] = None,
) -> None:
"""
[bold]Generate a video[/bold] from a source image using AI.
[dim]Examples:[/dim]
magnific generate-video photo.jpg --prompt "gentle ocean waves" --model kling-o1-pro
magnific generate-video portrait.png --model kling-elements-pro --aspect-ratio 9:16
magnific generate-video photo.jpg --model minimax-hailuo --aspect-ratio 16:9
"""
config = MagnificConfig.load()
key = _get_api_key(api_key, config)
image_b64 = image_to_base64(image)
with MagnificClient(key, base_url=config.base_url) as client:
api = VideoAPI(client)
with console.status(f"[info]Submitting {model.value} video generation…[/info]"):
try:
task_id = api.generate(
model=model,
image_b64=image_b64,
prompt=prompt,
duration=duration,
aspect_ratio=aspect_ratio,
seed=seed,
)
except MagnificAPIError as exc:
print_error(str(exc))
raise typer.Exit(1)
if not wait:
print_no_wait(task_id, "video", model.value)
return
poll_config = PollConfig(
task_type="video",
initial_delay=5.0,
max_wait=config.poll_timeout,
)
try:
result = poll_task(
check_fn=lambda tid: api.get_status(model, tid),
task_id=task_id,
config=poll_config,
console=console,
extra_info={
"Model": f"[magenta]{model.value}[/magenta]",
"Duration": f"{duration}s",
"Ratio": aspect_ratio,
},
)
except (MagnificTaskError, MagnificTimeoutError) as exc:
print_error(str(exc))
raise typer.Exit(1)
urls = api.get_output_urls(result)
if not urls:
print_error("Generation completed but no output URLs found.")
raise typer.Exit(1)
out = output or auto_output_path("video", model.value, "mp4", config.default_output_dir)
save_from_url(urls[0], out, console)
print_result(
GenerationResult(
task_id=task_id,
model=model.value,
output_path=out,
duration=str(duration),
task_type="video",
)
)
def generate_icon(
prompt: Annotated[str, typer.Argument(help="Text prompt for the icon")],
style: Annotated[
IconStyle,
typer.Option("--style", "-s", help="Icon style", show_default=True),
] = IconStyle.COLOR,
steps: Annotated[
int,
typer.Option("--steps", help="Inference steps (1050)", min=10, max=50),
] = 30,
guidance: Annotated[
float,
typer.Option("--guidance", help="Guidance scale (010)", min=0.0, max=10.0),
] = 7.5,
seed: Annotated[
Optional[int],
typer.Option("--seed", help="Random seed"),
] = None,
fmt: Annotated[
str,
typer.Option("--format", "-f", help="Output format: [cyan]png[/cyan] | [cyan]svg[/cyan]"),
] = "png",
output: Annotated[
Optional[Path],
typer.Option("--output", "-o", help="Output file path"),
] = None,
wait: Annotated[
bool,
typer.Option("--wait/--no-wait"),
] = True,
api_key: Annotated[
Optional[str],
typer.Option("--api-key", envvar="MAGNIFIC_API_KEY"),
] = None,
) -> None:
"""
[bold]Generate an icon[/bold] from a text prompt.
[dim]Examples:[/dim]
magnific generate-icon "shopping cart" --style solid --format svg
magnific generate-icon "rocket ship" --style color --format png
"""
config = MagnificConfig.load()
key = _get_api_key(api_key, config)
with MagnificClient(key, base_url=config.base_url) as client:
api = EditAPI(client)
with console.status("[info]Submitting icon generation…[/info]"):
try:
task_id = api.generate_icon(prompt, style, steps, guidance, seed)
except MagnificAPIError as exc:
print_error(str(exc))
raise typer.Exit(1)
if not wait:
print_no_wait(task_id, "icon", f"text-to-icon/{style.value}")
return
poll_config = PollConfig(task_type="icon", max_wait=config.poll_timeout)
try:
poll_task(
check_fn=lambda tid: api.icon_status(tid),
task_id=task_id,
config=poll_config,
console=console,
extra_info={"Style": style.value, "Format": fmt},
)
except (MagnificTaskError, MagnificTimeoutError) as exc:
print_error(str(exc))
raise typer.Exit(1)
# Render to get download URL
with console.status(f"[info]Rendering icon as {fmt.upper()}…[/info]"):
try:
url = api.render_icon(task_id, fmt)
except MagnificAPIError as exc:
print_error(str(exc))
raise typer.Exit(1)
if not url:
print_error("Icon generated but render URL not found.")
raise typer.Exit(1)
out = output or auto_output_path("icon", style.value, fmt, config.default_output_dir)
save_from_url(url, out, console)
print_result(
GenerationResult(
task_id=task_id,
model=f"text-to-icon/{style.value}",
output_path=out,
task_type="icon",
)
)
+243
View File
@@ -0,0 +1,243 @@
"""upscale-image and upscale-video commands."""
from __future__ import annotations
from pathlib import Path
from typing import Annotated, Optional
import typer
from magnific_cli.api.client import MagnificAPIError, MagnificClient
from magnific_cli.api.models import UpscaleMode, VideoUpscaleMode
from magnific_cli.api.upscale import UpscaleAPI
from magnific_cli.utils.config import MagnificConfig
from magnific_cli.utils.console import GenerationResult, console, print_error, print_no_wait, print_result
from magnific_cli.utils.files import auto_output_path, get_image_dimensions, image_to_base64, save_from_url, video_to_base64
from magnific_cli.utils.polling import MagnificTaskError, MagnificTimeoutError, PollConfig, poll_task
def _get_api_key(api_key: Optional[str], config: MagnificConfig) -> str:
key = api_key or config.api_key
if not key:
print_error(
"No API key found.",
hint="Set [cyan]MAGNIFIC_API_KEY[/cyan] or pass [cyan]--api-key YOUR_KEY[/cyan].",
)
raise typer.Exit(1)
return key
def upscale_image(
image: Annotated[
Path,
typer.Argument(
help="Image file to upscale",
exists=True, file_okay=True, dir_okay=False,
),
],
mode: Annotated[
UpscaleMode,
typer.Option(
"--mode",
help="Upscaling mode:\n"
" [cyan]precision-v2[/cyan] — faithful, detail-preserving (recommended)\n"
" [cyan]precision[/cyan] — faithful upscaling\n"
" [cyan]creative[/cyan] — AI-enhanced creative reinterpretation",
show_default=True,
),
] = UpscaleMode.PRECISION_V2,
scale: Annotated[
str,
typer.Option("--scale", help="Scale factor: [cyan]2x[/cyan] or [cyan]4x[/cyan]"),
] = "2x",
creativity: Annotated[
Optional[int],
typer.Option(
"--creativity",
help="Creative enhancement level 010 ([cyan]creative[/cyan] mode only)",
min=0, max=10,
),
] = None,
prompt: Annotated[
Optional[str],
typer.Option("--prompt", "-p", help="Enhancement guidance ([cyan]creative[/cyan] mode only)"),
] = None,
seed: Annotated[
Optional[int],
typer.Option("--seed", help="Random seed for reproducibility"),
] = None,
output: Annotated[
Optional[Path],
typer.Option("--output", "-o", help="Output file path"),
] = None,
wait: Annotated[
bool,
typer.Option("--wait/--no-wait", help="Wait for completion or return task ID immediately"),
] = True,
api_key: Annotated[
Optional[str],
typer.Option("--api-key", envvar="MAGNIFIC_API_KEY", help="Magnific API key"),
] = None,
) -> None:
"""
[bold]Upscale and enhance an image[/bold] using AI.
[dim]Examples:[/dim]
magnific upscale-image photo.jpg --mode precision-v2 --scale 4x
magnific upscale-image portrait.png --mode creative --creativity 7 --prompt "sharp cinematic"
"""
config = MagnificConfig.load()
key = _get_api_key(api_key, config)
if mode != UpscaleMode.CREATIVE and (creativity is not None or prompt):
print_error(
"--creativity and --prompt are only supported with --mode creative.",
hint="Switch to [cyan]--mode creative[/cyan] or remove those options.",
)
raise typer.Exit(1)
image_b64 = image_to_base64(image)
with MagnificClient(key, base_url=config.base_url) as client:
api = UpscaleAPI(client)
with console.status(f"[info]Submitting {mode.value} upscale ({scale})…[/info]"):
try:
task_id = api.upscale_image(
mode=mode,
image_b64=image_b64,
scale_factor=scale,
creativity=creativity,
prompt=prompt,
seed=seed,
)
except MagnificAPIError as exc:
print_error(str(exc))
raise typer.Exit(1)
if not wait:
print_no_wait(task_id, "upscale-image", mode.value)
return
poll_config = PollConfig(task_type="upscale-image", max_wait=config.poll_timeout)
try:
result = poll_task(
check_fn=lambda tid: api.upscale_image_status(mode, tid),
task_id=task_id,
config=poll_config,
console=console,
extra_info={"Mode": mode.value, "Scale": scale},
)
except (MagnificTaskError, MagnificTimeoutError) as exc:
print_error(str(exc))
raise typer.Exit(1)
urls = api.get_output_urls(result)
if not urls:
print_error("Upscaling completed but no output URL found.")
raise typer.Exit(1)
out = output or auto_output_path("upscaled", mode.value, image.suffix.lstrip(".") or "jpg", config.default_output_dir)
save_from_url(urls[0], out, console)
w, h = get_image_dimensions(out)
print_result(
GenerationResult(
task_id=task_id,
model=mode.value,
output_path=out,
width=w,
height=h,
task_type="upscale-image",
)
)
def upscale_video(
video: Annotated[
Path,
typer.Argument(
help="Video file to upscale",
exists=True, file_okay=True, dir_okay=False,
),
],
mode: Annotated[
VideoUpscaleMode,
typer.Option(
"--mode",
help="Upscaling mode: [cyan]standard[/cyan] | [cyan]turbo[/cyan] (faster)",
show_default=True,
),
] = VideoUpscaleMode.STANDARD,
output: Annotated[
Optional[Path],
typer.Option("--output", "-o", help="Output video file path"),
] = None,
wait: Annotated[
bool,
typer.Option("--wait/--no-wait"),
] = True,
api_key: Annotated[
Optional[str],
typer.Option("--api-key", envvar="MAGNIFIC_API_KEY"),
] = None,
) -> None:
"""
[bold]Upscale a video[/bold] to higher resolution using AI.
[dim]Examples:[/dim]
magnific upscale-video clip.mp4 --mode standard
magnific upscale-video clip.mp4 --mode turbo --output clip_4k.mp4
"""
config = MagnificConfig.load()
key = _get_api_key(api_key, config)
video_b64 = video_to_base64(video)
with MagnificClient(key, base_url=config.base_url) as client:
api = UpscaleAPI(client)
with console.status(f"[info]Submitting video upscale ({mode.value})…[/info]"):
try:
task_id = api.upscale_video(mode=mode, video_b64=video_b64)
except MagnificAPIError as exc:
print_error(str(exc))
raise typer.Exit(1)
if not wait:
print_no_wait(task_id, "upscale-video", mode.value)
return
poll_config = PollConfig(
task_type="upscale-video",
initial_delay=5.0,
max_wait=config.poll_timeout,
)
try:
result = poll_task(
check_fn=lambda tid: api.upscale_video_status(tid),
task_id=task_id,
config=poll_config,
console=console,
extra_info={"Mode": mode.value},
)
except (MagnificTaskError, MagnificTimeoutError) as exc:
print_error(str(exc))
raise typer.Exit(1)
urls = api.get_output_urls(result)
if not urls:
print_error("Upscaling completed but no output URL found.")
raise typer.Exit(1)
out = output or auto_output_path("upscaled_video", mode.value, "mp4", config.default_output_dir)
save_from_url(urls[0], out, console)
print_result(
GenerationResult(
task_id=task_id,
model=mode.value,
output_path=out,
task_type="upscale-video",
)
)
+123
View File
@@ -0,0 +1,123 @@
"""Root typer application — command registration and banner."""
from __future__ import annotations
from typing import Annotated, Optional
import typer
from magnific_cli import __version__
from magnific_cli.commands import config as config_cmd
from magnific_cli.commands.analyze import describe_image
from magnific_cli.commands.edit import expand_image, relight_image, style_transfer
from magnific_cli.commands.generate import generate_icon, generate_image, generate_video
from magnific_cli.commands.upscale import upscale_image, upscale_video
from magnific_cli.utils.console import console, print_banner
from magnific_cli.utils.config import MagnificConfig
app = typer.Typer(
name="magnific",
help=(
"[bold magenta]Magnific AI[/bold magenta] — generate images, videos, and more "
"from the command line.\n\n"
"[dim]Set your API key:[/dim] [cyan]export MAGNIFIC_API_KEY=your_key[/cyan]\n"
"[dim]Get an API key:[/dim] https://magnific.ai"
),
rich_markup_mode="rich",
no_args_is_help=True,
pretty_exceptions_enable=True,
pretty_exceptions_show_locals=False,
pretty_exceptions_short=True,
add_completion=True,
)
# ── Flat commands ──────────────────────────────────────────────────────────────
app.command(
"generate-image",
help="[bold]Generate an image[/bold] from a text prompt using AI models.",
rich_help_panel="Generation",
)(generate_image)
app.command(
"generate-video",
help="[bold]Generate a video[/bold] by animating a source image using AI.",
rich_help_panel="Generation",
)(generate_video)
app.command(
"generate-icon",
help="[bold]Generate an icon[/bold] from a text prompt in various styles.",
rich_help_panel="Generation",
)(generate_icon)
app.command(
"upscale-image",
help="[bold]Upscale and enhance[/bold] an image using AI.",
rich_help_panel="Enhancement",
)(upscale_image)
app.command(
"upscale-video",
help="[bold]Upscale a video[/bold] to higher resolution using AI.",
rich_help_panel="Enhancement",
)(upscale_video)
app.command(
"expand-image",
help="[bold]Expand an image[/bold] by adding new content around its edges (outpainting).",
rich_help_panel="Editing",
)(expand_image)
app.command(
"relight",
help="[bold]Relight an image[/bold] using AI-controlled lighting. [dim](Premium)[/dim]",
rich_help_panel="Editing",
)(relight_image)
app.command(
"style-transfer",
help="[bold]Apply an artistic style[/bold] from one image onto another. [dim](Premium)[/dim]",
rich_help_panel="Editing",
)(style_transfer)
app.command(
"describe-image",
help="[bold]Describe an image[/bold] and generate a text prompt from it.",
rich_help_panel="Analysis",
)(describe_image)
# ── Nested config sub-app ──────────────────────────────────────────────────────
app.add_typer(
config_cmd.app,
rich_help_panel="Settings",
)
# ── Root callback ──────────────────────────────────────────────────────────────
@app.callback(invoke_without_command=True)
def main(
ctx: typer.Context,
version: Annotated[
bool,
typer.Option("--version", "-v", help="Show version and exit", is_eager=True),
] = False,
) -> None:
"""
[bold magenta]Magnific AI[/bold magenta] — generate images, videos, icons, and more.
"""
cfg = MagnificConfig.load()
if version:
print_banner()
console.print(f"[dim]Version:[/dim] [bold]{__version__}[/bold]\n")
raise typer.Exit()
if ctx.invoked_subcommand is None:
if cfg.show_banner:
print_banner()
console.print(ctx.get_help())
elif cfg.show_banner and ctx.invoked_subcommand not in ("config",):
print_banner()
View File
+96
View File
@@ -0,0 +1,96 @@
"""Configuration management — env vars, config file, and defaults."""
from __future__ import annotations
from pathlib import Path
from typing import Optional
import toml
from platformdirs import user_config_dir
from pydantic import field_validator
from pydantic_settings import BaseSettings, SettingsConfigDict
CONFIG_DIR = Path(user_config_dir("magnific-cli"))
CONFIG_FILE = CONFIG_DIR / "config.toml"
class MagnificConfig(BaseSettings):
"""
Configuration with priority (highest to lowest):
1. CLI --api-key flag (handled in commands directly)
2. MAGNIFIC_* environment variables
3. ~/.config/magnific-cli/config.toml
4. Defaults below
"""
model_config = SettingsConfigDict(
env_prefix="MAGNIFIC_",
env_file=".env",
env_file_encoding="utf-8",
extra="ignore",
)
api_key: Optional[str] = None
base_url: str = "https://api.magnific.ai"
default_output_dir: str = "."
default_image_model: str = "flux-2-pro"
default_video_model: str = "kling-o1-pro"
default_upscale_mode: str = "precision-v2"
poll_timeout: int = 600
poll_max_interval: int = 15
show_banner: bool = True
@field_validator("api_key", mode="before")
@classmethod
def strip_api_key(cls, v: Optional[str]) -> Optional[str]:
return v.strip() if isinstance(v, str) else v
@classmethod
def load(cls) -> "MagnificConfig":
"""Load from config file, then overlay environment variables."""
file_data: dict = {}
if CONFIG_FILE.exists():
try:
file_data = toml.load(CONFIG_FILE)
except Exception:
pass
return cls(**file_data)
def save(self, exclude_keys: set[str] | None = None) -> None:
"""Persist non-sensitive config to disk."""
exclude_keys = (exclude_keys or set()) | {"api_key"}
CONFIG_DIR.mkdir(parents=True, exist_ok=True)
data = self.model_dump(exclude=exclude_keys, exclude_none=True)
# Stringify paths
for k, v in data.items():
if isinstance(v, Path):
data[k] = str(v)
with open(CONFIG_FILE, "w") as f:
toml.dump(data, f)
def to_display_dict(self) -> dict:
"""Return all settings as a displayable dict (keeps api_key for masking)."""
d = self.model_dump()
return {k: v for k, v in d.items()}
def set_value(self, key: str, value: str) -> None:
"""Update a single config key and save."""
allowed = {
"base_url", "default_output_dir", "default_image_model",
"default_video_model", "default_upscale_mode", "poll_timeout",
"poll_max_interval", "show_banner",
}
if key not in allowed:
raise ValueError(
f"Key '{key}' is not configurable via this command. "
f"Use the MAGNIFIC_API_KEY environment variable to set the API key."
)
current = self.model_dump()
if key in ("poll_timeout", "poll_max_interval"):
current[key] = int(value)
elif key == "show_banner":
current[key] = value.lower() in ("true", "1", "yes")
else:
current[key] = value
updated = MagnificConfig(**{k: v for k, v in current.items() if v is not None})
updated.save()
+213
View File
@@ -0,0 +1,213 @@
"""Rich console singleton, theme, banner, and all display helpers."""
from __future__ import annotations
from dataclasses import dataclass
from pathlib import Path
from typing import Optional
import rich.box
from rich.align import Align
from rich.columns import Columns
from rich.console import Console
from rich.panel import Panel
from rich.rule import Rule
from rich.syntax import Syntax
from rich.table import Table
from rich.text import Text
from rich.theme import Theme
MAGNIFIC_THEME = Theme(
{
"info": "bold cyan",
"success": "bold green",
"warning": "bold yellow",
"error": "bold red",
"model": "bold magenta",
"taskid": "bold blue",
"path": "bold white underline",
"dim.label": "dim white",
"highlight": "bold magenta",
"brand": "bold magenta",
}
)
console = Console(theme=MAGNIFIC_THEME, highlight=True)
err_console = Console(stderr=True, theme=MAGNIFIC_THEME)
BANNER = """\
[bold magenta]███╗ ███╗ █████╗ ██████╗ ███╗ ██╗██╗███████╗██╗ ██████╗ [/bold magenta]
[bold magenta]████╗ ████║██╔══██╗██╔════╝ ████╗ ██║██║██╔════╝██║██╔════╝ [/bold magenta]
[bold magenta]██╔████╔██║███████║██║ ███╗██╔██╗ ██║██║█████╗ ██║██║ [/bold magenta]
[bold magenta]██║╚██╔╝██║██╔══██║██║ ██║██║╚██╗██║██║██╔══╝ ██║██║ [/bold magenta]
[bold magenta]██║ ╚═╝ ██║██║ ██║╚██████╔╝██║ ╚████║██║██║ ██║╚██████╗ [/bold magenta]
[bold magenta]╚═╝ ╚═╝╚═╝ ╚═╝ ╚═════╝ ╚═╝ ╚═══╝╚═╝╚═╝ ╚═╝ ╚═════╝[/bold magenta]
[dim] AI Media Generation CLI • v0.1.0[/dim]"""
def print_banner() -> None:
console.print()
console.print(Align.center(Text.from_markup(BANNER)))
console.print(Rule(style="magenta dim"))
console.print()
@dataclass
class GenerationResult:
task_id: str
model: str
output_path: Path
width: Optional[int] = None
height: Optional[int] = None
seed: Optional[int] = None
duration: Optional[str] = None
task_type: str = "image"
def print_result(result: GenerationResult) -> None:
table = Table(
show_header=False,
box=rich.box.SIMPLE,
padding=(0, 1),
show_edge=False,
)
table.add_column(style="dim.label", width=16, no_wrap=True)
table.add_column(style="bold", overflow="fold")
table.add_row("Model", f"[model]{result.model}[/model]")
table.add_row("Task ID", f"[taskid]{result.task_id[:16]}…[/taskid]")
if result.width and result.height:
table.add_row("Dimensions", f"{result.width} × {result.height} px")
if result.duration:
table.add_row("Duration", f"{result.duration}s")
if result.seed is not None:
table.add_row("Seed", str(result.seed))
else:
table.add_row("Seed", "[dim]random[/dim]")
table.add_row("Saved to", f"[path]{result.output_path}[/path]")
title_map = {
"image": "[success] Image Generated [/success]",
"video": "[success] Video Generated [/success]",
"upscale-image": "[success] Image Upscaled [/success]",
"upscale-video": "[success] Video Upscaled [/success]",
"icon": "[success] Icon Generated [/success]",
"expand": "[success] Image Expanded [/success]",
"describe": "[success] Image Described [/success]",
}
title = title_map.get(result.task_type, "[success] Done [/success]")
console.print(Panel(table, title=title, border_style="green", padding=(1, 2)))
def print_describe_result(task_id: str, prompt_text: str, output_path: Optional[Path] = None) -> None:
content = Text(prompt_text, style="italic")
footer = ""
if output_path:
footer = f"\n\n[dim]Saved to:[/dim] [path]{output_path}[/path]"
console.print(
Panel(
Text.from_markup(f"{prompt_text}{footer}"),
title="[success] Image Description [/success]",
subtitle=f"[dim]Task: {task_id[:16]}…[/dim]",
border_style="green",
padding=(1, 2),
)
)
def print_no_wait(task_id: str, task_type: str, model: str) -> None:
console.print(
Panel(
Text.from_markup(
f"[info]Task submitted successfully.[/info]\n\n"
f"[dim.label]Task ID:[/dim.label] [taskid]{task_id}[/taskid]\n"
f"[dim.label]Type:[/dim.label] {task_type}\n"
f"[dim.label]Model:[/dim.label] [model]{model}[/model]\n\n"
f"[dim]The task is processing asynchronously. Results will be available\n"
f"on the Magnific dashboard when complete.[/dim]"
),
title="⏳ Task Queued",
border_style="cyan",
padding=(1, 2),
)
)
def print_error(message: str, hint: Optional[str] = None) -> None:
body = f"[error]{message}[/error]"
if hint:
body += f"\n\n[dim]Hint:[/dim] {hint}"
err_console.print(
Panel(
Text.from_markup(body),
title="[error] Error [/error]",
border_style="red",
padding=(1, 2),
)
)
def print_warning(message: str) -> None:
console.print(
Panel(
Text.from_markup(f"[warning]{message}[/warning]"),
title="[warning] Warning [/warning]",
border_style="yellow",
padding=(0, 2),
)
)
def print_config_table(config_dict: dict, masked_keys: set[str] | None = None) -> None:
masked_keys = masked_keys or {"api_key"}
table = Table(
title="[brand]Magnific CLI Configuration[/brand]",
show_header=True,
header_style="bold magenta",
border_style="magenta",
box=rich.box.ROUNDED,
padding=(0, 1),
)
table.add_column("Key", style="dim.label", width=28)
table.add_column("Value", style="bold", overflow="fold")
for key, value in config_dict.items():
if key in masked_keys and value:
display = f"[dim]{'*' * 8}{str(value)[-4:]}[/dim]" if len(str(value)) > 4 else "[dim]****[/dim]"
elif value is None or value == "":
display = "[dim]not set[/dim]"
else:
display = str(value)
table.add_row(key, display)
console.print(table)
def print_config_toml(config_dict: dict, masked_keys: set[str] | None = None) -> None:
masked_keys = masked_keys or {"api_key"}
lines = []
for key, value in config_dict.items():
if key in masked_keys:
continue
if value is None:
continue
if isinstance(value, str):
lines.append(f'{key} = "{value}"')
elif isinstance(value, bool):
lines.append(f"{key} = {str(value).lower()}")
else:
lines.append(f"{key} = {value}")
toml_str = "\n".join(lines)
console.print(
Panel(
Syntax(toml_str, "toml", theme="monokai", background_color="default"),
title="[brand]~/.config/magnific-cli/config.toml[/brand]",
border_style="magenta",
padding=(1, 2),
)
)
+105
View File
@@ -0,0 +1,105 @@
"""File I/O utilities: base64 encoding, output path generation, download."""
from __future__ import annotations
import base64
from datetime import datetime
from pathlib import Path
from typing import Optional
import httpx
from rich.console import Console
from rich.progress import (
BarColumn,
DownloadColumn,
Progress,
SpinnerColumn,
TimeElapsedColumn,
TransferSpeedColumn,
)
def image_to_base64(path: Path) -> str:
"""Read an image file and return a base64-encoded string.
Uses Pillow to detect the actual format rather than trusting the file
extension — mismatched extensions (e.g. a JPEG saved as .png) would
produce an incorrect MIME type that causes silent failures with some models.
"""
from PIL import Image
_pillow_to_mime = {
"JPEG": "image/jpeg",
"PNG": "image/png",
"GIF": "image/gif",
"WEBP": "image/webp",
}
with Image.open(path) as img:
fmt = img.format or "JPEG"
mime = _pillow_to_mime.get(fmt, "image/jpeg")
with open(path, "rb") as f:
encoded = base64.b64encode(f.read()).decode()
return f"data:{mime};base64,{encoded}"
def video_to_base64(path: Path) -> str:
"""Read a video file and return a base64-encoded string."""
suffix = path.suffix.lower().lstrip(".")
mime_map = {
"mp4": "video/mp4",
"mov": "video/quicktime",
"avi": "video/x-msvideo",
"webm": "video/webm",
}
mime = mime_map.get(suffix, "video/mp4")
with open(path, "rb") as f:
encoded = base64.b64encode(f.read()).decode()
return f"data:{mime};base64,{encoded}"
def auto_output_path(task_type: str, model: str, ext: str = "jpg", output_dir: str = ".") -> Path:
"""Generate a timestamped output filename."""
ts = datetime.now().strftime("%Y%m%d_%H%M%S")
safe_model = model.replace("/", "-").replace(" ", "-")
filename = f"magnific_{task_type}_{safe_model}_{ts}.{ext}"
return Path(output_dir) / filename
def save_from_url(url: str, dest: Path, console: Console) -> None:
"""Stream-download a URL to a file, displaying a Rich progress bar."""
dest.parent.mkdir(parents=True, exist_ok=True)
with Progress(
SpinnerColumn(style="bold magenta"),
"[progress.description]{task.description}",
BarColumn(bar_width=30, style="magenta", complete_style="green"),
"[progress.percentage]{task.percentage:>3.0f}%",
DownloadColumn(),
TransferSpeedColumn(),
TimeElapsedColumn(),
console=console,
transient=True,
) as progress:
with httpx.stream("GET", url, follow_redirects=True, timeout=120) as r:
r.raise_for_status()
total = int(r.headers.get("content-length", 0)) or None
task = progress.add_task(
f"[dim]Saving[/dim] [bold]{dest.name}[/bold]",
total=total,
)
with open(dest, "wb") as f:
for chunk in r.iter_bytes(chunk_size=65536):
f.write(chunk)
progress.advance(task, len(chunk))
def get_image_dimensions(path: Path) -> tuple[Optional[int], Optional[int]]:
"""Return (width, height) of an image using Pillow."""
try:
from PIL import Image
with Image.open(path) as img:
return img.width, img.height
except Exception:
return None, None
+184
View File
@@ -0,0 +1,184 @@
"""Async task polling with beautiful Rich Live display."""
from __future__ import annotations
import time
from dataclasses import dataclass, field
from typing import Any, Callable, Literal, Tuple
from rich.console import Console
from rich.live import Live
from rich.panel import Panel
from rich.spinner import Spinner
from rich.table import Table
from rich.text import Text
TaskStatus = Literal["PENDING", "IN_PROGRESS", "COMPLETED", "FAILED", "CANCELLED", "CREATED"]
STATUS_COLORS: dict[str, str] = {
"CREATED": "yellow",
"PENDING": "yellow",
"IN_PROGRESS": "cyan",
"COMPLETED": "green",
"FAILED": "red",
"CANCELLED": "dim red",
}
STATUS_ICONS: dict[str, str] = {
"CREATED": "",
"PENDING": "",
"IN_PROGRESS": "",
"COMPLETED": "",
"FAILED": "",
"CANCELLED": "",
}
TASK_TYPE_LABELS: dict[str, str] = {
"image": "Image Generation",
"video": "Video Generation",
"upscale-image": "Image Upscaling",
"upscale-video": "Video Upscaling",
"icon": "Icon Generation",
"expand": "Image Expansion",
"describe": "Image Analysis",
"relight": "Image Relighting",
"style-transfer": "Style Transfer",
}
class MagnificTaskError(Exception):
pass
class MagnificTimeoutError(Exception):
pass
@dataclass
class PollConfig:
initial_delay: float = 2.0
min_interval: float = 2.0
max_interval: float = 15.0
backoff_factor: float = 1.5
max_wait: float = 600.0
task_type: str = "image"
def _render_panel(
task_id: str,
status: str,
elapsed: float,
task_type: str,
extra_info: dict[str, str] | None = None,
) -> Panel:
color = STATUS_COLORS.get(status.upper(), "white")
icon = STATUS_ICONS.get(status.upper(), "?")
label = TASK_TYPE_LABELS.get(task_type, task_type.replace("-", " ").title())
grid = Table.grid(padding=(0, 2))
grid.add_column(style="dim", width=14, no_wrap=True)
grid.add_column(overflow="fold")
spinner = Spinner("dots", style="bold magenta")
# Header row with spinner
header = Text()
header.append(f"{label}", style="bold white")
grid.add_row(spinner, header)
grid.add_row("", "") # spacer
short_id = task_id[:12] + "" if len(task_id) > 12 else task_id
grid.add_row("Task ID", f"[bold blue]{short_id}[/bold blue]")
grid.add_row(
"Status",
f"[{color}]{icon} {status.replace('_', ' ')}[/{color}]",
)
mins, secs = divmod(int(elapsed), 60)
time_str = f"{mins}m {secs:02d}s" if mins else f"{secs}s"
grid.add_row("Elapsed", f"[dim]{time_str}[/dim]")
if extra_info:
for k, v in extra_info.items():
grid.add_row(k, v)
return Panel(
grid,
title="[bold magenta]~ Magnific AI ~[/bold magenta]",
border_style="magenta",
padding=(1, 2),
width=52,
)
def poll_task(
check_fn: Callable[[str], Tuple[str, dict[str, Any]]],
task_id: str,
config: PollConfig,
console: Console,
extra_info: dict[str, str] | None = None,
) -> dict[str, Any]:
"""
Poll until COMPLETED or FAILED, displaying a live status panel.
Args:
check_fn: Callable(task_id) → (status_str, raw_response_dict)
task_id: The task ID to poll
config: Polling configuration
console: Rich console instance
extra_info: Extra rows to display in the status panel
Returns:
The raw response dict when status is COMPLETED
"""
start = time.monotonic()
interval = config.initial_delay
current_status = "PENDING"
result: dict[str, Any] = {}
with Live(
_render_panel(task_id, current_status, 0, config.task_type, extra_info),
console=console,
refresh_per_second=8,
transient=False,
) as live:
time.sleep(config.initial_delay)
while True:
elapsed = time.monotonic() - start
if elapsed > config.max_wait:
live.stop()
raise MagnificTimeoutError(
f"Task {task_id} timed out after {config.max_wait:.0f}s"
)
try:
current_status, result = check_fn(task_id)
except Exception as exc:
live.stop()
raise exc
live.update(
_render_panel(task_id, current_status, elapsed, config.task_type, extra_info)
)
upper = current_status.upper()
if upper == "COMPLETED":
live.update(
_render_panel(task_id, "COMPLETED", elapsed, config.task_type, extra_info)
)
return result
if upper in ("FAILED", "CANCELLED"):
live.stop()
data = result.get("data", result)
error_msg = (
data.get("error", {}).get("message")
or data.get("message")
or f"Task ended with status {upper}"
)
raise MagnificTaskError(f"{upper}: {error_msg}")
time.sleep(interval)
interval = min(interval * config.backoff_factor, config.max_interval)