my-project-public

repository

loading code, commits, and activity

repositories

loading repo index

my-project-public — gitlawb

#1	"""MiniMax API Client for multimodal generation (image, music, video, TTS)"""
#2
#3	import aiohttp
#4	import asyncio
#5	import base64
#6	from typing import Optional, Dict, Any, List
#7	from dataclasses import dataclass
#8	from enum import Enum
#9
#10
#11	class ImageRatio(str, Enum):
#12	"""Image aspect ratios"""
#13	SQUARE = "1:1"
#14	PORTRAIT = "3:4"
#15	LANDSCAPE = "4:3"
#16	WIDESCREEN = "16:9"
#17
#18
#19	class VideoRatio(str, Enum):
#20	"""Video aspect ratios"""
#21	SQUARE = "1:1"
#22	PORTRAIT = "9:16"
#23	LANDSCAPE = "16:9"
#24
#25
#26	@dataclass
#27	class GenerationResult:
#28	"""Result from a generation task"""
#29	task_id: str
#30	status: str
#31	file_id: Optional[str] = None
#32	download_url: Optional[str] = None
#33	error: Optional[str] = None
#34
#35
#36	class MinimaxClient:
#37	"""Client for MiniMax multimodal generation APIs"""
#38
#39	BASE_URL = "https://api.minimaxi.chat/v1"
#40
#41	def __init__(self, api_key: str):
#42	"""Initialize MiniMax client.
#43
#44	Args:
#45	api_key: MiniMax API key
#46	"""
#47	self.api_key = api_key
#48	self.session: Optional[aiohttp.ClientSession] = None
#49
#50	async def _ensure_session(self):
#51	"""Ensure aiohttp session exists."""
#52	if self.session is None or self.session.closed:
#53	self.session = aiohttp.ClientSession(
#54	headers={
#55	"Authorization": f"Bearer {self.api_key}",
#56	"Content-Type": "application/json"
#57	}
#58	)
#59
#60	async def close(self):
#61	"""Close the aiohttp session."""
#62	if self.session and not self.session.closed:
#63	await self.session.close()
#64
#65	async def _post(self, endpoint: str, json_data: Dict[str, Any]) -> Dict[str, Any]:
#66	"""Make POST request to MiniMax API."""
#67	await self._ensure_session()
#68
#69	url = f"{self.BASE_URL}{endpoint}"
#70
#71	async with self.session.post(url, json=json_data) as response:
#72	response.raise_for_status()
#73	return await response.json()
#74
#75	async def _get(self, endpoint: str) -> Dict[str, Any]:
#76	"""Make GET request to MiniMax API."""
#77	await self._ensure_session()
#78
#79	url = f"{self.BASE_URL}{endpoint}"
#80
#81	async with self.session.get(url) as response:
#82	response.raise_for_status()
#83	return await response.json()
#84
#85	async def _poll_task(
#86	self,
#87	task_id: str,
#88	max_wait: int = 300,
#89	poll_interval: int = 3
#90	) -> GenerationResult:
#91	"""Poll a task until completion or timeout.
#92
#93	Args:
#94	task_id: Task ID to poll
#95	max_wait: Maximum time to wait in seconds
#96	poll_interval: Time between polls in seconds
#97
#98	Returns:
#99	GenerationResult with task status and file info
#100	"""
#101	start_time = asyncio.get_event_loop().time()
#102
#103	while True:
#104	elapsed = asyncio.get_event_loop().time() - start_time
#105	if elapsed > max_wait:
#106	return GenerationResult(
#107	task_id=task_id,
#108	status="timeout",
#109	error=f"Task timed out after {max_wait} seconds"
#110	)
#111
#112	result = await self._get(f"/query/video_generation?task_id={task_id}")
#113
#114	status = result.get("status")
#115
#116	if status == "Success":
#117	file_id = result.get("file_id")
#118	return GenerationResult(
#119	task_id=task_id,
#120	status="success",
#121	file_id=file_id,
#122	download_url=f"{self.BASE_URL}/files/retrieve?file_id={file_id}"
#123	)
#124	elif status == "Failed":
#125	return GenerationResult(
#126	task_id=task_id,
#127	status="failed",
#128	error=result.get("error", "Unknown error")
#129	)
#130
#131	# Still processing, wait and retry
#132	await asyncio.sleep(poll_interval)
#133
#134	async def generate_image(
#135	self,
#136	prompt: str,
#137	model: str = "text-to-image-v3",
#138	ratio: ImageRatio = ImageRatio.SQUARE,
#139	num_images: int = 1
#140	) -> Dict[str, Any]:
#141	"""Generate image from text prompt.
#142
#143	Args:
#144	prompt: Text description of desired image
#145	model: Model to use (text-to-image-v3, image-to-image-v2)
#146	ratio: Aspect ratio for the image
#147	num_images: Number of images to generate (1-4)
#148
#149	Returns:
#150	Response with image URLs
#151	"""
#152	data = {
#153	"model": model,
#154	"prompt": prompt,
#155	"ratio": ratio.value,
#156	"n": min(max(num_images, 1), 4)
#157	}
#158
#159	return await self._post("/image/generation", data)
#160
#161	async def generate_image_from_image(
#162	self,
#163	prompt: str,
#164	image_url: str,
#165	ratio: ImageRatio = ImageRatio.SQUARE
#166	) -> Dict[str, Any]:
#167	"""Generate image from text prompt and reference image.
#168
#169	Args:
#170	prompt: Text description of desired modifications
#171	image_url: URL or base64 of reference image
#172	ratio: Aspect ratio for the image
#173
#174	Returns:
#175	Response with generated image URL
#176	"""
#177	data = {
#178	"model": "image-to-image-v2",
#179	"prompt": prompt,
#180	"image_url": image_url,
#181	"ratio": ratio.value
#182	}
#183
#184	return await self._post("/image/generation", data)
#185
#186	async def generate_music(
#187	self,
#188	prompt: str,
#189	lyrics: Optional[str] = None,
#190	duration: int = 30,
#191	instrumental: bool = False
#192	) -> Dict[str, Any]:
#193	"""Generate music from text prompt.
#194
#195	Args:
#196	prompt: Description of desired music style/mood
#197	lyrics: Optional lyrics to sing
#198	duration: Duration in seconds (max 120)
#199	instrumental: Generate instrumental only
#200
#201	Returns:
#202	Response with audio file URL
#203	"""
#204	data = {
#205	"model": "music-generation-v1",
#206	"prompt": prompt,
#207	"duration": min(duration, 120)
#208	}
#209
#210	if lyrics and not instrumental:
#211	data["lyrics"] = lyrics
#212
#213	if instrumental:
#214	data["type"] = "instrumental"
#215
#216	return await self._post("/music/generation", data)
#217
#218	async def generate_video(
#219	self,
#220	prompt: str,
#221	first_frame_image: Optional[str] = None,
#222	last_frame_image: Optional[str] = None,
#223	ratio: VideoRatio = VideoRatio.LANDSCAPE,
#224	duration: int = 5,
#225	wait_for_completion: bool = True,
#226	max_wait: int = 300
#227	) -> GenerationResult:
#228	"""Generate video from text prompt.
#229
#230	Args:
#231	prompt: Description of desired video
#232	first_frame_image: Optional URL/base64 of first frame
#233	last_frame_image: Optional URL/base64 of last frame
#234	ratio: Aspect ratio for video
#235	duration: Duration in seconds (2-6)
#236	wait_for_completion: Whether to poll until complete
#237	max_wait: Maximum time to wait in seconds
#238
#239	Returns:
#240	GenerationResult with video file info
#241	"""
#242	data = {
#243	"model": "video-generation-v1",
#244	"prompt": prompt,
#245	"ratio": ratio.value,
#246	"duration": max(2, min(duration, 6))
#247	}
#248
#249	if first_frame_image:
#250	data["first_frame_image"] = first_frame_image
#251
#252	if last_frame_image:
#253	data["last_frame_image"] = last_frame_image
#254
#255	response = await self._post("/video/generation", data)
#256	task_id = response.get("task_id")
#257
#258	if not task_id:
#259	return GenerationResult(
#260	task_id="",
#261	status="failed",
#262	error="No task_id returned"
#263	)
#264
#265	if wait_for_completion:
#266	return await self._poll_task(task_id, max_wait=max_wait)
#267	else:
#268	return GenerationResult(
#269	task_id=task_id,
#270	status="processing"
#271	)
#272
#273	async def text_to_speech(
#274	self,
#275	text: str,
#276	voice_id: str = "default",
#277	speed: float = 1.0,
#278	pitch: float = 1.0
#279	) -> Dict[str, Any]:
#280	"""Convert text to speech.
#281
#282	Args:
#283	text: Text to convert to speech
#284	voice_id: Voice ID to use
#285	speed: Speech speed (0.5-2.0)
#286	pitch: Voice pitch (0.5-2.0)
#287
#288	Returns:
#289	Response with audio file URL
#290	"""
#291	data = {
#292	"model": "speech-generation-v1",
#293	"text": text,
#294	"voice_id": voice_id,
#295	"speed": max(0.5, min(speed, 2.0)),
#296	"pitch": max(0.5, min(pitch, 2.0))
#297	}
#298
#299	return await self._post("/speech/generation", data)
#300
#301	async def download_file(self, file_id: str) -> bytes:
#302	"""Download a generated file.
#303
#304	Args:
#305	file_id: File ID from generation result
#306
#307	Returns:
#308	File content as bytes
#309	"""
#310	await self._ensure_session()
#311
#312	url = f"{self.BASE_URL}/files/retrieve?file_id={file_id}"
#313
#314	async with self.session.get(url) as response:
#315	response.raise_for_status()
#316	return await response.read()
#317
#318	async def save_file(self, file_id: str, output_path: str):
#319	"""Download and save a generated file.
#320
#321	Args:
#322	file_id: File ID from generation result
#323	output_path: Path to save file
#324	"""
#325	content = await self.download_file(file_id)
#326
#327	with open(output_path, "wb") as f:
#328	f.write(content)
#329

z6Mkq5mY3JWtxoxUobWcfNHm7AkRubgSWEZTkBVqZXJviFZ5/my-project-public