feat: kling video extend

timerring · timerring · commit cd6d882d1437 · 2025-05-11T22:39:09.000+08:00
diff --git a/docs/zh/models/kling/api/video-extend.md b/docs/zh/models/kling/api/video-extend.md
@@ -47,7 +47,7 @@ updatedAt: 2025-05-09
 
 <<< @/zh/snippets/video-extend-api.py{5-6,19-24}
 
-## 响应参数示例
+## 生成响应参数
 
 > 业务码的含义请参考 [业务码](/zh/models/kling/api/business-code.md)。
 
@@ -65,5 +65,36 @@ updatedAt: 2025-05-09
 }
 ```
 
+## 查询响应参数
 
+```
+{
+  "code": 0, //错误码；具体定义见1.1错误码
+  "message": "string", //错误信息；具体定义见1.1错误码
+  "request_id": "string", //请求ID，系统生成，用于跟踪请求、排查问题；全局唯一
+  "data":{
+  	"task_id": "string", //任务ID，系统生成；全局唯一
+    "task_status": "string", //任务状态，枚举值：submitted（已提交）、processing（处理中）、succeed（成功）、failed（失败）
+    "task_status_msg": "string", //任务状态信息，当任务失败时展示失败原因（如触发平台的内容风控等）
+    "task_info":{ //任务创建时的参数信息
+       "parent_video": {
+         	"id": "string", //续写前的视频ID；全局唯一
+      		"url": "string", //续写前视频的URL（请注意，为保障信息安全，生成的图片/视频会在30天后被清理，请及时转存）
+      		"duration": "string" //续写前的视频总时长，单位s
+       }
+    }, //任务创建时用户填写的详细信息
+    "task_result":{
+      "videos":[  //数组是为了保留扩展性，以防未来要支持n
+        {
+          "id": "string", //续写后的完整视频ID；全局唯一
+          "url": "string", //续写后视频的URL
+          "duration": "string" //视频总时长，单位s
+        }
+      ]
+    }
+    "created_at": 1722769557708, //任务创建时间，Unix时间戳、单位ms
+    "updated_at": 1722769557708, //任务更新时间，Unix时间戳、单位ms
+  }
+}
+```
 
diff --git a/docs/zh/models/kling/kling-video-extend.md b/docs/zh/models/kling/kling-video-extend.md
@@ -0,0 +1,63 @@
+---
+title: 可灵 Kling 视频延长案例
+gitChangelog: false
+updatedAt: 2025-05-11
+---
+
+# 可灵 Kling 视频延长
+
+这是一个视频延长的示例，使用 kling 延长视频。
+
+> [!TIP]
+> 视频延长是计算密集型任务，特别是高质量、高分辨率视频可能需要数十秒甚至数分钟处理时间，为了让用户发送请求后可以立即收到响应（任务ID），而不必等待整个生成过程，因此用户可以同时提交多个生成任务，然后异步查询结果。
+> 同时这样的队列系统允许服务提供商根据可用GPU/TPU资源智能调度任务。
+
+通常来说，视频延长的常见流程是：
+
+1. `POST`: 调用 `视频延长api` 提交视频延长任务，返回获取 `task_id`。
+2. `GET`: 根据 `task_id` 调用 `查询视频api` 查看视频延长任务是否完成。
+
+本示例实现了每隔一秒轮询任务状态，直到任务完成，然后返回视频 url 和 id。
+
+## 代码示例
+
+> 深色背景为可以修改的参数，非必选参数已经注释，可以按照自己的需求启用。
+
+
+<<< @/zh/snippets/kling-video-extend.py{124-125,132-138}
+
+
+## 返回结果
+
+返回结果为视频的 url 和 id，视频的有效期一般为 30 天，推荐尽快下载或者转存。
+
+```
+
+```
+
+## 流程图
+
+```mermaid
+flowchart TD
+    A[开始] --> B[初始化 KlingVideoExtend 实例]
+    B --> C[调用 extend_video 方法]
+    
+    C --> D[准备参数: task_id, video_id, prompt等]
+    D --> E[调用 _kling_extend_video 提交任务]
+    E --> F[获取task_id]
+    
+    F --> G[开始循环查询结果]
+    G --> H[调用 _query_video_extend_result 查询]
+    
+    H --> I{任务是否完成?}
+    I -->|否| J{是否超时?}
+    J -->|否| K[等待1秒]
+    K --> H
+    J -->|是| L[返回超时信息]
+    
+    I -->|是| M[获取结果: video_url, video_id]
+    M --> N[返回视频URL和视频ID]
+    
+    L --> O[结束]
+    N --> O
+```
diff --git a/docs/zh/snippets/kling-video-extend.py b/docs/zh/snippets/kling-video-extend.py
@@ -0,0 +1,142 @@
+import http.client
+import json
+import time
+
+class KlingVideoExtend:
+    def __init__(self, api_token, api_url):
+        """初始化 Kling 视频延长生成器
+        
+        参数:
+            api_token: API 密钥
+            api_url: API 节点地址
+        """
+        self.api_url = api_url
+        self.api_token = api_token
+        # 初始化 HTTP 连接
+        self.conn = http.client.HTTPSConnection(self.api_url)
+        self.endpoint = "/kling/v1/videos/video-extend"
+        # 设置请求头
+        self.headers = {
+            'Authorization': f'Bearer {self.api_token}',
+            'Content-Type': 'application/json'
+        }
+
+    def _kling_extend_video(self, task_id, video_id, prompt, negative_prompt="", cfg_scale=0.5, callback_url=""):
+        """使用 kling 提交视频延长任务
+        
+        参数: 
+            task_id: str, 任务ID
+            video_id: str, 要延长的视频ID
+            prompt: str, 文本提示词，指导延长方向
+            negative_prompt: str, 负向文本提示词
+            cfg_scale: float, 提示词参考强度
+            callback_url: str, 回调地址，可以用于 webhook 等通知场景
+        返回:
+            task_id: 生成任务的 id
+        """
+        # 构建请求体，请求的核心参数
+        payload = {
+            "task_id": task_id,
+            "video_id": video_id,
+            "prompt": prompt,
+            "negative_prompt": negative_prompt,
+            "cfg_scale": cfg_scale,
+            "callback_url": callback_url
+        }
+        
+        # 发送 POST 请求，提交视频延长任务
+        self.conn.request("POST", self.endpoint, json.dumps(payload), self.headers)
+        # 获取响应
+        res = self.conn.getresponse()
+        # 读取响应内容并解析为 JSON
+        json_data = json.loads(res.read().decode("utf-8"))
+        
+        if 'code' in json_data and json_data['code'] == 0:
+            # 成功则返回提交的任务 id
+            return json_data['data']['task_id']
+        else:
+            # 失败则返回错误信息
+            raise Exception(f"API调用失败：{json_data['message']}")
+    
+    def _query_video_extend_result(self, task_id):
+        """使用查询接口获取视频延长结果
+        
+        参数:
+            task_id: 生成任务的 id
+        返回:
+            video_url: 延长后的视频 url，任务未完成时返回 None
+        """
+        # 构建查询路径
+        query_path = f"{self.endpoint}/{task_id}"
+
+        # 发送 GET 请求，查询视频延长任务状态
+        self.conn.request("GET", query_path, None, self.headers)
+        # 获取响应
+        res = self.conn.getresponse()
+        # 读取响应内容并解析为 JSON
+        json_data = json.loads(res.read().decode("utf-8"))
+        
+        # 如果任务状态为成功，则返回视频 url
+        if json_data['data']['task_status'] == "succeed":
+            video_url = json_data['data']['task_result']['videos'][0]['url']
+            video_id = json_data['data']['task_result']['videos'][0]['id']
+            return video_url, video_id
+        else: 
+            return None
+    
+    def extend_video(self, task_id, video_id, prompt, negative_prompt="", cfg_scale=0.5, callback_url="", timeout=300):
+        """实现功能，根据预设的参数延长视频并返回延长后的视频 url
+        
+        参数:
+            task_id: str, 任务ID
+            video_id: str, 要延长的视频ID
+            prompt: str, 文本提示词，指导延长方向
+            negative_prompt: str, 负向文本提示词
+            cfg_scale: float, 提示词参考强度
+            callback_url: str, 回调地址，可以用于 webhook 等通知场景
+            timeout: int, 超时时间（秒）
+        返回:
+            video_url: 延长后的视频 url
+        """
+        # 调用视频延长 API 提交任务，返回获取 task_id
+        task_id = self._kling_extend_video(task_id, video_id, prompt, negative_prompt, cfg_scale, callback_url)
+        
+        start_time = time.time()
+        
+        # 轮询等待生成完成
+        while True:
+            # 根据 task_id 调用查询 API 查看任务是否完成
+            video_url, video_id = self._query_video_extend_result(task_id)
+            # 如果任务完成，则返回视频 url
+            if video_url is not None:
+                return video_url, video_id
+            # 如果轮询超时，则返回 None
+            if time.time() - start_time > timeout:
+                print(f"请求达到 {timeout} 秒超时")
+                return None
+            # 轮询间隔 1 秒
+            time.sleep(1)
+            print(f"等待视频延长结果生成，{int(time.time() - start_time)} 秒", flush=True)
+
+
+# 使用示例
+if __name__ == "__main__":
+    API_URL = "www.dmxapi.cn"  # API 节点地址
+    DMX_API_TOKEN = "sk-XXXXXXXXXXXXXX"  # API 密钥
+    
+    # 创建视频延长生成器实例
+    kling_video_extend = KlingVideoExtend(api_token=DMX_API_TOKEN, api_url=API_URL)
+    
+    # 延长视频
+    video_url, video_id = kling_video_extend.extend_video(
+        task_id="CjhDaWgU7GAAAAAAAb1QfA",  # [必选] 任务ID
+        video_id="bc17f1e8-6c7b-440c-bd30-e47fb1c82932",  # [必选] 要延长的视频ID
+        prompt="继续展示动物的走动，保持相同的风格和氛围",  # [必选] 文本提示词
+        # negative_prompt="突然的场景转换, 光线变化, 画面抖动",  # 负向提示词
+        # cfg_scale=0.5,  # 提示词参考强度，设置较高以保持连贯性
+        # callback_url="",  # 回调地址
+        # timeout=300  # 等待超时时间，视频处理通常需要更长时间
+    )
+    
+    print(video_url)
+    print(video_id)
