""" 七牛云 Kling 图像生成后端. 使用七牛云 AI 大模型推理 API 的 Kling 系列模型进行文生图。 接口采用异步模式: 提交任务 → 轮询状态 → 下载图片。 支持的模型: - kling-v1: 文生图 V1.0 - kling-v1-5: 文生图 V1.5 (支持角色/人脸参考) - kling-v2: 文生图 V2.0 (推荐) 图片以 URL 形式返回 (有效期 7 天), 本插件自动下载到本地缓存。 """ from __future__ import annotations import json import logging import os import time import urllib.request import urllib.error from typing import Any, Dict, List, Optional from agent.image_gen_provider import ( DEFAULT_ASPECT_RATIO, ImageGenProvider, error_response, resolve_aspect_ratio, save_url_image, success_response, ) logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # Model catalog # --------------------------------------------------------------------------- _MODELS: Dict[str, Dict[str, Any]] = { "kling-v2": { "display": "Kling v2", "speed": "~60s", "strengths": "最新文生图, 画面质量最高", "price": "按量计费", }, "kling-v1-5": { "display": "Kling v1.5", "speed": "~45s", "strengths": "支持角色特征参考/人脸参考", "price": "按量计费", }, "kling-v1": { "display": "Kling v1", "speed": "~30s", "strengths": "基础文生图, 速度快", "price": "按量计费", }, } DEFAULT_MODEL = "kling-v2" # 七牛云 Kling 支持的画面比例 _ASPECT_MAP = { "landscape": "16:9", "portrait": "9:16", "square": "1:1", } # 轮询配置 _POLL_INTERVAL = 5 # 秒 _POLL_TIMEOUT = 300 # 最大等待 5 分钟 _API_BASE = "https://api.qnaigc.com/v1" def _get_api_key() -> Optional[str]: """获取七牛云 API Key.""" # 优先从环境变量 for env_var in ("QINIU_API_KEY", "OPENROUTER_API_KEY"): key = os.environ.get(env_var, "").strip() if key: return key # 从 config 读取 try: from hermes_cli.config import load_config cfg = load_config() providers = cfg.get("providers", {}) qiniu_cfg = providers.get("qiniu", {}) if isinstance(qiniu_cfg, dict): key = qiniu_cfg.get("api_key", "").strip() if key: return key except Exception: pass return None def _resolve_model() -> tuple: """从配置解析当前使用的模型.""" env_override = os.environ.get("QINIU_IMAGE_MODEL", "").strip() if env_override and env_override in _MODELS: return env_override, _MODELS[env_override] try: from hermes_cli.config import load_config cfg = load_config() img_cfg = cfg.get("image_gen") if isinstance(cfg, dict) else {} qiniu_cfg = img_cfg.get("qiniu") if isinstance(img_cfg, dict) else {} if isinstance(qiniu_cfg, dict): model_id = qiniu_cfg.get("model", "").strip() if model_id and model_id in _MODELS: return model_id, _MODELS[model_id] top_model = img_cfg.get("model", "").strip() if isinstance(img_cfg, dict) else "" if top_model and top_model in _MODELS: return top_model, _MODELS[top_model] except Exception as exc: logger.debug("Could not load image_gen config: %s", exc) return DEFAULT_MODEL, _MODELS[DEFAULT_MODEL] def _kling_aspect_ratio(hermes_aspect: str) -> str: """将 Hermes 抽象比例映射为 Kling 原生比例.""" return _ASPECT_MAP.get(hermes_aspect, "16:9") # --------------------------------------------------------------------------- # Provider # --------------------------------------------------------------------------- class QiniuImageGenProvider(ImageGenProvider): """七牛云 Kling 文生图后端.""" @property def name(self) -> str: return "qiniu" @property def display_name(self) -> str: return "七牛云 Kling" def is_available(self) -> bool: return bool(_get_api_key()) def list_models(self) -> List[Dict[str, Any]]: return [ { "id": model_id, "display": meta["display"], "speed": meta["speed"], "strengths": meta["strengths"], "price": meta["price"], } for model_id, meta in _MODELS.items() ] def default_model(self) -> Optional[str]: return DEFAULT_MODEL def get_setup_schema(self) -> Dict[str, Any]: return { "name": "七牛云 Kling", "badge": "paid", "tag": "七牛云 Kling 系列文生图 (kling-v1/v1-5/v2), 异步接口", "env_vars": [ { "key": "QINIU_API_KEY", "prompt": "七牛云 AI API Key", "url": "https://console.qnaigc.com/", }, ], } def generate( self, prompt: str, aspect_ratio: str = DEFAULT_ASPECT_RATIO, **kwargs: Any, ) -> Dict[str, Any]: prompt = (prompt or "").strip() aspect = resolve_aspect_ratio(aspect_ratio) if not prompt: return error_response( error="Prompt is required", error_type="invalid_argument", provider="qiniu", aspect_ratio=aspect, ) api_key = _get_api_key() if not api_key: return error_response( error=( "七牛云 API Key 未设置. 请设置环境变量 QINIU_API_KEY " "或通过 hermes auth add qiniu 添加." ), error_type="auth_required", provider="qiniu", aspect_ratio=aspect, ) model_id, meta = _resolve_model() kling_ratio = _kling_aspect_ratio(aspect) # ---- Step 1: 提交生成任务 ---- payload = json.dumps({ "model": model_id, "prompt": prompt[:2500], # Kling 限制 2500 字符 "aspect_ratio": kling_ratio, "n": 1, }).encode("utf-8") req = urllib.request.Request( f"{_API_BASE}/images/generations", data=payload, headers={ "Authorization": f"Bearer {api_key}", "Content-Type": "application/json", }, ) try: resp = urllib.request.urlopen(req, timeout=30) result = json.loads(resp.read()) task_id = result.get("task_id") except urllib.error.HTTPError as e: body = e.read().decode()[:500] logger.debug("Kling submit failed: %s", body) return error_response( error=f"Kling 提交失败 (HTTP {e.code}): {body}", error_type="api_error", provider="qiniu", model=model_id, prompt=prompt, aspect_ratio=aspect, ) except Exception as exc: return error_response( error=f"Kling 提交异常: {exc}", error_type="api_error", provider="qiniu", model=model_id, prompt=prompt, aspect_ratio=aspect, ) if not task_id: return error_response( error="Kling 返回了空的 task_id", error_type="empty_response", provider="qiniu", model=model_id, prompt=prompt, aspect_ratio=aspect, ) logger.info("Kling task submitted: %s (model=%s)", task_id, model_id) # ---- Step 2: 轮询任务状态 ---- deadline = time.time() + _POLL_TIMEOUT image_url: Optional[str] = None while time.time() < deadline: time.sleep(_POLL_INTERVAL) req2 = urllib.request.Request( f"{_API_BASE}/images/tasks/{task_id}", headers={"Authorization": f"Bearer {api_key}"}, ) try: resp2 = urllib.request.urlopen(req2, timeout=10) status = json.loads(resp2.read()) except Exception as exc: logger.warning("Kling poll error for %s: %s", task_id, exc) continue s = status.get("status", "") if s == "succeed": data = status.get("data", []) if data and isinstance(data, list): first = data[0] if isinstance(first, dict): image_url = first.get("url") break elif s == "failed": return error_response( error=f"Kling 生成失败: {status.get('status_message', '未知错误')}", error_type="generation_failed", provider="qiniu", model=model_id, prompt=prompt, aspect_ratio=aspect, ) if image_url is None: return error_response( error=f"Kling 任务超时 (>{_POLL_TIMEOUT}s): {task_id}", error_type="timeout", provider="qiniu", model=model_id, prompt=prompt, aspect_ratio=aspect, ) # ---- Step 3: 下载图片到本地 ---- try: saved_path = save_url_image( image_url, prefix=f"kling_{model_id}", timeout=60, ) except Exception as exc: # 下载失败，返回原始 URL（用户可在7天内手动下载） logger.warning("Failed to cache Kling image: %s", exc) return success_response( image=image_url, model=model_id, prompt=prompt, aspect_ratio=aspect, provider="qiniu", extra={ "task_id": task_id, "note": "图片缓存失败，URL有效期7天", }, ) return success_response( image=str(saved_path), model=model_id, prompt=prompt, aspect_ratio=aspect, provider="qiniu", extra={"task_id": task_id}, ) # --------------------------------------------------------------------------- # Plugin entry point # --------------------------------------------------------------------------- def register(ctx) -> None: """注册七牛云 Kling 图像生成 provider.""" ctx.register_image_gen_provider(QiniuImageGenProvider())