diff --git a/apps/models_provider/impl/minimax_model_provider/model/ttv.py b/apps/models_provider/impl/minimax_model_provider/model/ttv.py index 5691fe1d8b2..8522d8c238f 100644 --- a/apps/models_provider/impl/minimax_model_provider/model/ttv.py +++ b/apps/models_provider/impl/minimax_model_provider/model/ttv.py @@ -1,5 +1,7 @@ +import json import time from typing import Dict, ClassVar +from urllib.parse import urlparse import requests @@ -18,6 +20,7 @@ class GenerationVideoModel(MaxKBBaseModel, BaseGenerationVideo): v2_models: ClassVar[tuple] = ("MiniMax-H3", "MiniMax-H3-Max") v2_extra_fields: ClassVar[tuple] = ("resolution", "duration", "ratio", "callback_url") + official_hosts: ClassVar[tuple] = ("api.minimaxi.com", "api.minimaxi.com.cn", "api.minimaxi.io") v2_success_status: ClassVar[frozenset] = frozenset({"succeeded", "Success"}) v2_fail_status: ClassVar[frozenset] = frozenset({"failed", "Fail", "cancelled", "Cancel"}) @@ -75,6 +78,160 @@ def _v2(self) -> bool: return True return self._detect_api_version() == "v2" + def _is_private_deploy(self) -> bool: + """私有部署：api_base 主机不是官方 MiniMax 云端域名时，走本地 OpenAI 兼容的 /v1/videos 接口。""" + host = (urlparse(self.api_base).hostname or "").lower() + return not any(host == h or host.endswith("." + h) for h in self.official_hosts) + + def _generate_video_private(self, prompt, first_frame_url=None, last_frame_url=None, **kwargs): + """私有部署：api_base 即完整地址，直接发 multipart/form-data。 + + 私有部署可能直接返回视频二进制（此时原样返回），也可能返回 JSON（视频 URL 或异步任务 id）。 + """ + url = self.api_base + fields = {"model": self.model_name, "prompt": prompt} + # 透传 model_params_setting 里的参数（width/height/fps/seed/extra_params 等），不从 UI 表单约束 + for key, value in (self.params or {}).items(): + if value is None: + continue + if isinstance(value, (dict, list, bool)): + value = json.dumps(value, ensure_ascii=False) + fields[str(key)] = value + + headers = {"Authorization": f"Bearer {self.api_key}"} + files = {key: (None, str(value)) for key, value in fields.items()} + maxkb_logger.info(f"提交私有部署视频生成任务，模型: {self.model_name} -> {url}") + + body = self._post_private(url, headers, files) + # 私有部署可能直接返回视频二进制 + if isinstance(body, (bytes, bytearray)): + maxkb_logger.info("私有部署直接返回视频二进制") + return body + # 返回 JSON：先找同步生成的视频 URL + video_url = self._find_video_url(body) + if video_url: + maxkb_logger.info(f"私有部署视频生成成功，视频 URL: {video_url}") + return video_url + # 否则按异步任务处理，用 video id 轮询 + video_id = self._find_video_id(body) + if video_id: + return self._poll_private_task(video_id) + raise RuntimeError(f"私有部署返回内容既非视频二进制也无可识别的视频地址/任务 id: {body}") + + def _post_private(self, url, headers, files): + """私有部署 POST 提交，带重试；能解析为 JSON 则返回 JSON，否则返回视频二进制。""" + for attempt in range(self.max_retries): + try: + response = requests.post(url, headers=headers, files=files, timeout=600) + response.raise_for_status() + return self._parse_private_body(response) + except (requests.exceptions.ProxyError, requests.exceptions.ConnectionError, + requests.exceptions.Timeout) as e: + maxkb_logger.error(f"⚠️ 网络错误: {e}，正在重试 {attempt + 1}/{self.max_retries}...") + time.sleep(self.retry_delay) + except requests.exceptions.HTTPError as e: + raise RuntimeError( + f"私有部署 HTTP 请求失败: {e.response.text if hasattr(e, 'response') else str(e)}") + raise RuntimeError("多次重试后仍无法连接到私有部署服务") + + @staticmethod + def _parse_private_body(response): + """私有部署可能直接返回视频二进制：能解析成 JSON 才当 JSON，否则原样返回二进制。""" + try: + return response.json() + except ValueError: + return response.content + + def _poll_private_task(self, video_id: str) -> str: + """异步任务：轮询 GET {api_base}/{video_id} 直至成功/失败。""" + query_url = f"{self.api_base.rstrip('/')}/{video_id}" + headers = {"Authorization": f"Bearer {self.api_key}"} + max_attempts = 60 + for attempt in range(max_attempts): + try: + response = requests.get(query_url, headers=headers, timeout=60) + response.raise_for_status() + body = self._parse_private_body(response) + except (requests.exceptions.ProxyError, requests.exceptions.ConnectionError, + requests.exceptions.Timeout) as e: + maxkb_logger.error(f"⚠️ 网络错误: {e}，正在重试 {attempt + 1}/{max_attempts}...") + time.sleep(self.retry_delay) + continue + except requests.exceptions.HTTPError as e: + raise RuntimeError( + f"私有部署查询任务失败: {e.response.text if hasattr(e, 'response') else str(e)}") + if isinstance(body, (bytes, bytearray)): + maxkb_logger.info("私有部署查询直接返回视频二进制") + return body + status = self._find_status(body) + maxkb_logger.info(f"私有部署任务状态 (尝试 {attempt + 1}/{max_attempts}): {status}") + if status in self.v2_success_status: + video_url = self._find_video_url(body) + if video_url: + maxkb_logger.info(f"私有部署任务成功，视频 URL: {video_url}") + return video_url + raise RuntimeError(f"私有部署任务成功但未找到视频地址: {body}") + if status in self.v2_fail_status: + raise RuntimeError(f"视频生成失败，状态: {status}, 返回: {body}") + time.sleep(self.retry_delay) + raise RuntimeError(f"任务超时：经过 {max_attempts} 次轮询后仍未完成") + + @staticmethod + def _find_video_url(obj): + """递归查找返回体里第一个以 http 开头的视频地址。""" + if isinstance(obj, dict): + for value in obj.values(): + if isinstance(value, str) and value.startswith("http"): + return value + if isinstance(value, (dict, list)): + found = GenerationVideoModel._find_video_url(value) + if found: + return found + elif isinstance(obj, list): + for value in obj: + found = GenerationVideoModel._find_video_url(value) + if found: + return found + return None + + @staticmethod + def _find_video_id(obj): + """递归查找返回体里第一个视频/任务 id。""" + id_keys = {"video_id", "videoId", "task_id", "taskId", "job_id", "jobId", "id"} + if isinstance(obj, dict): + for key, value in obj.items(): + if key in id_keys and isinstance(value, str) and value: + return value + if isinstance(value, (dict, list)): + found = GenerationVideoModel._find_video_id(value) + if found: + return found + elif isinstance(obj, list): + for value in obj: + found = GenerationVideoModel._find_video_id(value) + if found: + return found + return None + + @staticmethod + def _find_status(obj): + """递归查找返回体里的状态字段值。""" + status_keys = {"status", "state", "status_message", "message"} + if isinstance(obj, dict): + for key, value in obj.items(): + if key in status_keys and isinstance(value, str) and value: + return value + if isinstance(value, (dict, list)): + found = GenerationVideoModel._find_status(value) + if found: + return found + elif isinstance(obj, list): + for value in obj: + found = GenerationVideoModel._find_status(value) + if found: + return found + return None + def _safe_call(self, method, url, **kwargs): """带重试的请求封装""" headers = {"Authorization": f"Bearer {self.api_key}"} @@ -113,6 +270,9 @@ def generate_video(self, prompt, negative_prompt=None, first_frame_url=None, las 返回: 视频下载 URL """ + # 私有部署（本地 /v1/videos，OpenAI 兼容 multipart 接口） + if self._is_private_deploy(): + return self._generate_video_private(prompt, first_frame_url, last_frame_url, **kwargs) # 自动兼容 V1 / V2 (MiniMax-H3) 两套参数逻辑 if self._v2(): return self._generate_video_v2(prompt, first_frame_url, last_frame_url, **kwargs)