""" title: Freepik AI (Mystic & Kling) author: You description: Generate images with Mystic and make videos from images with Kling via Freepik API version: 0.1.0 requirements: httpx, pydantic """ from typing import Optional from pydantic import BaseModel, Field import httpx import asyncio FREEPIK_BASE = "https://api.freepik.com" class Tools: def __init__(self): # اختیار‌ی: توضیح برای UI self.description = "Freepik image/video via Mystic & Kling" self.valves = self.Valves() class Valves(BaseModel): api_key: str = Field("", description="Freepik API key (x-freepik-api-key)") async def _poll(self, client: httpx.AsyncClient, url: str, api_key: str, timeout_s: int = 120) -> dict: deadline = asyncio.get_event_loop().time() + timeout_s while asyncio.get_event_loop().time() < deadline: r = await client.get(url, headers={"x-freepik-api-key": api_key}) r.raise_for_status() data = r.json().get("data", {}) status = data.get("status", "") generated = data.get("generated") or data.get("video") or data.get("result") if status in ("COMPLETED","FINISHED","SUCCEEDED") or (isinstance(generated, list) and generated): return data if status in ("FAILED","ERROR"): raise RuntimeError(f"Task failed: {data}") await asyncio.sleep(2) raise TimeoutError("Polling timeout") async def mystic_generate( self, prompt: str, aspect_ratio: str = "square_1_1", resolution: str = "2k", model: str = "realism", timeout_s: int = 120 ) -> str: """ Generate an image with Freepik Mystic and return the first result URL. :param prompt: Text prompt :param aspect_ratio: e.g. square_1_1, widescreen_16_9 :param resolution: 1k | 2k | 4k :param model: realism | fluid | zen (see Freepik docs) :param timeout_s: polling timeout in seconds """ api_key = self.valves.api_key if not api_key: return "Set your Freepik API key in Tool valves first." async with httpx.AsyncClient(base_url=FREEPIK_BASE, timeout=60) as client: r = await client.post( "/v1/ai/mystic", headers={"x-freepik-api-key": api_key, "Content-Type": "application/json"}, json={"prompt": prompt, "aspect_ratio": aspect_ratio, "resolution": resolution, "model": model}, ) r.raise_for_status() task_id = r.json().get("data", {}).get("task_id") if not task_id: return f"Unexpected response: {r.text}" data = await self._poll(client, f"/v1/ai/mystic/{task_id}", api_key, timeout_s) urls = [] gen = data.get("generated") or [] if isinstance(gen, list): for g in gen: urls.append(g.get("url") or g.get("image") or g.get("src")) return urls[0] if urls else f"No URL in response: {data}" async def kling_image_to_video( self, image: str, duration: str = "5", timeout_s: int = 180 ) -> str: """ Make a short video from an image with Freepik's Kling model. Returns the video URL. :param image: Public URL or Base64 of the image :param duration: "5" or "10" :param timeout_s: polling timeout in seconds """ api_key = self.valves.api_key if not api_key: return "Set your Freepik API key in Tool valves first." async with httpx.AsyncClient(base_url=FREEPIK_BASE, timeout=90) as client: r = await client.post( "/v1/ai/image-to-video/kling-v2", headers={"x-freepik-api-key": api_key, "Content-Type": "application/json"}, json={"image": image, "duration": duration}, ) r.raise_for_status() task_id = r.json().get("data", {}).get("task_id") if not task_id: return f"Unexpected response: {r.text}" data = await self._poll(client, f"/v1/ai/image-to-video/kling-v2/{task_id}", api_key, timeout_s) # Some variants return {"video":{"url":...}}, others via generated[] if isinstance(data.get("video"), dict): return data["video"].get("url") or data["video"].get("src") or str(data) gen = data.get("generated") if isinstance(gen, list) and gen: return gen[0].get("url") or str(data) return f"No video URL in response: {data}"