Skip to content

Commit a3ff7f5

Browse files
timerringtimerring
committed
feat: kling image to video
1 parent 1c51b38 commit a3ff7f5

File tree

2 files changed

+395
-0
lines changed

2 files changed

+395
-0
lines changed

‎docs/zh/models/kling/kling-image-to-video.md‎

Lines changed: 95 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,95 @@
1+
---
2+
title: 可灵 Kling 图生视频案例
3+
gitChangelog: false
4+
updatedAt: 2025-05-11
5+
---
6+
7+
# 可灵 Kling 图生视频
8+
9+
这是一个图生视频的示例，使用 kling 生成视频。
10+
11+
> [!TIP]
12+
> 视频生成是计算密集型任务，特别是高质量、高分辨率视频可能需要数十秒甚至数分钟处理时间，为了让用户发送请求后可以立即收到响应（任务ID），而不必等待整个生成过程，因此用户可以同时提交多个生成任务，然后异步查询结果。
13+
> 同时这样的队列系统允许服务提供商根据可用GPU/TPU资源智能调度任务。
14+
15+
通常来说，视频生成的常见流程是：
16+
17+
1. `POST`: 调用 `生成视频api` 提交视频生成任务，返回获取 `task_id`。
18+
2. `GET`: 根据 `task_id` 调用 `查询视频api` 查看视频生成任务是否完成。
19+
20+
本示例实现了每隔一秒轮询任务状态，直到任务完成，然后返回视频 url 列表。
21+
22+
## 代码示例
23+
24+
> 深色背景为可以修改的参数，非必选参数已经注释，可以按照自己的需求启用。
25+
26+
27+
<<< @/zh/snippets/kling-image-to-video.py{254-255,261-272,276-296}
28+
29+
30+
## 返回结果
31+
32+
返回结果为视频的 url 和 id，视频的有效期一般为 30 天，推荐尽快下载或者转存。
33+
34+
```
35+
36+
```
37+
38+
## 流程图
39+
40+
```mermaid
41+
flowchart TD
42+
A[开始] --> B[初始化 KlingImageToVideo 实例]
43+
B --> C[调用 generate_video 方法]
44+
45+
C --> D1{检查起始图片类型}
46+
D1 -->|URL| E1[直接使用URL]
47+
D1 -->|本地文件| F1[转换为base64]
48+
49+
C --> D2{检查结束图片类型}
50+
D2 -->|URL| E2[直接使用URL]
51+
D2 -->|本地文件| F2[转换为base64]
52+
D2 -->|未提供| G2[跳过]
53+
54+
C --> D3{检查静态遮罩类型}
55+
D3 -->|URL| E3[直接使用URL]
56+
D3 -->|本地文件| F3[转换为base64]
57+
D3 -->|未提供| G3[跳过]
58+
59+
C --> D4{检查动态遮罩列表}
60+
D4 -->|存在| E4[处理每个遮罩项]
61+
D4 -->|未提供| G4[跳过]
62+
63+
E4 --> F4{检查遮罩图像类型}
64+
F4 -->|URL| G4[直接使用URL]
65+
F4 -->|本地文件| H4[转换为base64]
66+
67+
E1 --> I[准备API请求参数]
68+
F1 --> I
69+
E2 --> I
70+
F2 --> I
71+
G2 --> I
72+
E3 --> I
73+
F3 --> I
74+
G3 --> I
75+
G4 --> I
76+
H4 --> I
77+
78+
I --> J[调用_kling_generate_video提交任务]
79+
J --> K[获取task_id]
80+
81+
K --> L[开始轮询查询结果]
82+
L --> M[调用_query_video_result]
83+
84+
M --> N{任务是否完成?}
85+
N -->|否| O{是否超时?}
86+
O -->|否| P[等待3秒]
87+
P --> M
88+
O -->|是| Q[返回超时信息]
89+
90+
N -->|是| R[获取video_url和video_id]
91+
R --> S[返回结果]
92+
93+
Q --> T[结束]
94+
S --> T
95+
```

‎docs/zh/snippets/kling-image-to-video.py‎

Lines changed: 300 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,300 @@
1+
import http.client
2+
import json
3+
import base64
4+
import time
5+
6+
class KlingImageToVideo:
7+
def __init__(self, api_token, api_url):
8+
"""初始化 Kling 图像生成视频转换器
9+
10+
参数:
11+
api_token: API 密钥
12+
api_url: API 节点地址
13+
"""
14+
self.api_url = api_url
15+
self.api_token = api_token
16+
# 初始化 HTTP 连接
17+
self.conn = http.client.HTTPSConnection(self.api_url)
18+
self.endpoint = "/kling/v1/videos/image2video"
19+
# 设置请求头
20+
self.headers = {
21+
'Authorization': f'Bearer {self.api_token}',
22+
'Content-Type': 'application/json'
23+
}
24+
25+
@staticmethod
26+
def get_image_base64(image_path):
27+
"""将图片转换为 base64 编码形式
28+
29+
参数:
30+
image_path: 图片路径
31+
返回:
32+
base64 编码后的图片字符串
33+
"""
34+
with open(image_path, "rb") as image_file:
35+
return base64.b64encode(image_file.read()).decode("utf-8")
36+
37+
def _kling_generate_video(self, model_name, image, prompt,
38+
image_tail=None, negative_prompt="",
39+
cfg_scale=0.5, mode="std", duration="5",
40+
camera_control=None, static_mask=None, dynamic_masks=None,
41+
callback_url="", external_task_id=""):
42+
"""使用 kling 进行图像转视频
43+
44+
参数:
45+
model_name: str, 模型版本 kling-v1, kling-v1-5, kling-v1-6
46+
image: str, 起始图片，base64编码或URL
47+
prompt: str, 正向提示词
48+
image_tail: str, 结束图片，base64编码或URL
49+
negative_prompt: str, 负向提示词
50+
cfg_scale: float, 生成视频的自由度，取值范围：[0, 1]
51+
mode: str, 生成模式：std(标准模式) 或 pro(专家模式)
52+
duration: str, 视频时长(秒)：5 或 10
53+
camera_control: dict, 摄像机控制参数
54+
static_mask: str, 静态区域遮罩，base64编码或URL
55+
dynamic_masks: list, 动态区域遮罩列表
56+
callback_url: str, 回调地址
57+
external_task_id: str, 自定义任务ID
58+
返回:
59+
task_id: 生成任务的 id
60+
"""
61+
# 构建请求体，请求的核心参数
62+
payload = {
63+
"model_name": model_name,
64+
"image": image,
65+
"prompt": prompt,
66+
"negative_prompt": negative_prompt,
67+
"cfg_scale": cfg_scale,
68+
"mode": mode,
69+
"duration": duration,
70+
"callback_url": callback_url
71+
}
72+
73+
# 如果提供了结束图片
74+
if image_tail:
75+
payload["image_tail"] = image_tail
76+
77+
# 如果提供了摄像机控制参数
78+
if camera_control:
79+
payload["camera_control"] = camera_control
80+
81+
# 如果提供了静态遮罩
82+
if static_mask:
83+
payload["static_mask"] = static_mask
84+
85+
# 如果提供了动态遮罩
86+
if dynamic_masks:
87+
payload["dynamic_masks"] = dynamic_masks
88+
89+
# 如果提供了自定义任务ID
90+
if external_task_id:
91+
payload["external_task_id"] = external_task_id
92+
93+
# 发送 POST 请求，提交视频生成任务
94+
self.conn.request("POST", self.endpoint, json.dumps(payload), self.headers)
95+
# 获取响应
96+
res = self.conn.getresponse()
97+
# 读取响应内容并解析为 JSON
98+
json_data = json.loads(res.read().decode("utf-8"))
99+
100+
if 'code' in json_data and json_data['code'] == 0:
101+
# 成功则返回提交的任务 id
102+
return json_data['data']['task_id']
103+
else:
104+
# 失败则返回错误信息
105+
raise Exception(f"API调用失败：{json_data['message']}")
106+
107+
def _query_video_result(self, task_id):
108+
"""使用查询接口获取生成视频结果
109+
110+
参数:
111+
task_id: 生成任务的 id
112+
返回:
113+
video_url: 视频 url，任务未完成时返回 None
114+
video_id: 视频 id，任务未完成时返回 None
115+
"""
116+
# 构建查询路径
117+
query_path = f"/kling/v1/videos/generations/{task_id}"
118+
119+
# 发送 GET 请求，查询视频生成任务状态
120+
self.conn.request("GET", query_path, None, self.headers)
121+
# 获取响应
122+
res = self.conn.getresponse()
123+
# 读取响应内容并解析为 JSON
124+
json_data = json.loads(res.read().decode("utf-8"))
125+
126+
# 检查响应是否成功
127+
if json_data['code'] == 0:
128+
# 如果任务状态为成功，则返回视频 url 和 id
129+
if json_data['data']['task_status'] == "succeed":
130+
video_url = json_data['data']['task_result']['videos'][0]['url']
131+
video_id = json_data['data']['task_result']['videos'][0]['id']
132+
return video_url, video_id
133+
else:
134+
return None, None
135+
else:
136+
# 如果查询失败，抛出异常
137+
raise Exception(f"查询失败: {json_data['message']}")
138+
139+
def generate_video(self, model_name, image, prompt,
140+
image_tail=None, negative_prompt="",
141+
cfg_scale=0.5, mode="std", duration="5",
142+
camera_control=None, static_mask=None, dynamic_masks=None,
143+
callback_url="", external_task_id="", timeout=600):
144+
"""实现功能，根据图片生成视频并返回结果
145+
146+
参数:
147+
model_name: str, 模型版本 kling-v1, kling-v1-5, kling-v1-6
148+
image: str, 起始图片URL或本地文件路径
149+
prompt: str, 正向提示词
150+
image_tail: str, 结束图片URL或本地文件路径
151+
negative_prompt: str, 负向提示词
152+
cfg_scale: float, 生成视频的自由度，取值范围：[0, 1]
153+
mode: str, 生成模式：std(标准模式) 或 pro(专家模式)
154+
duration: str, 视频时长(秒)：5 或 10
155+
camera_control: dict, 摄像机控制参数
156+
static_mask: str, 静态区域遮罩URL或本地文件路径
157+
dynamic_masks: list, 动态区域遮罩列表
158+
callback_url: str, 回调地址
159+
external_task_id: str, 自定义任务ID
160+
timeout: int, 超时时间（秒）
161+
返回:
162+
video_url: 视频URL
163+
video_id: 视频ID
164+
"""
165+
# 处理起始图片输入
166+
if image.startswith(('http://', 'https://', 'ftp://')):
167+
# 如果是URL，直接使用
168+
image_data = image
169+
else:
170+
# 否则当作本地文件路径处理，转换为base64
171+
try:
172+
image_data = KlingImageToVideo.get_image_base64(image)
173+
except Exception as e:
174+
raise ValueError(f"无法读取起始图像文件: {str(e)}")
175+
176+
# 处理结束图片输入(如果有)
177+
image_tail_data = None
178+
if image_tail:
179+
if image_tail.startswith(('http://', 'https://', 'ftp://')):
180+
# 如果是URL，直接使用
181+
image_tail_data = image_tail
182+
else:
183+
# 否则当作本地文件路径处理，转换为base64
184+
try:
185+
image_tail_data = KlingImageToVideo.get_image_base64(image_tail)
186+
except Exception as e:
187+
raise ValueError(f"无法读取结束图像文件: {str(e)}")
188+
189+
# 处理静态遮罩(如果有)
190+
static_mask_data = None
191+
if static_mask:
192+
if static_mask.startswith(('http://', 'https://', 'ftp://')):
193+
# 如果是URL，直接使用
194+
static_mask_data = static_mask
195+
else:
196+
# 否则当作本地文件路径处理，转换为base64
197+
try:
198+
static_mask_data = KlingImageToVideo.get_image_base64(static_mask)
199+
except Exception as e:
200+
raise ValueError(f"无法读取静态遮罩文件: {str(e)}")
201+
202+
# 处理动态遮罩(如果有)
203+
if dynamic_masks:
204+
processed_masks = []
205+
for mask_item in dynamic_masks:
206+
processed_item = mask_item.copy()
207+
208+
# 处理遮罩图像
209+
if mask_item.get('mask'):
210+
mask_image = mask_item['mask']
211+
if mask_image.startswith(('http://', 'https://', 'ftp://')):
212+
# 如果是URL，直接使用
213+
processed_item['mask'] = mask_image
214+
else:
215+
# 否则当作本地文件路径处理，转换为base64
216+
try:
217+
processed_item['mask'] = KlingImageToVideo.get_image_base64(mask_image)
218+
except Exception as e:
219+
raise ValueError(f"无法读取动态遮罩文件: {str(e)}")
220+
221+
processed_masks.append(processed_item)
222+
223+
dynamic_masks = processed_masks
224+
225+
# 调用生成视频 API 提交任务
226+
task_id = self._kling_generate_video(
227+
model_name, image_data, prompt,
228+
image_tail_data, negative_prompt,
229+
cfg_scale, mode, duration,
230+
camera_control, static_mask_data, dynamic_masks,
231+
callback_url, external_task_id
232+
)
233+
234+
start_time = time.time()
235+
236+
# 轮询等待生成完成
237+
while True:
238+
# 查询任务状态
239+
video_url, video_id = self._query_video_result(task_id)
240+
# 如果任务完成，返回结果
241+
if video_url is not None:
242+
return video_url, video_id
243+
# 如果超时，返回 None
244+
if time.time() - start_time > timeout:
245+
print(f"请求达到 {timeout} 秒超时")
246+
return None, None
247+
# 轮询间隔 3 秒
248+
time.sleep(3)
249+
print(f"等待视频生成，{int(time.time() - start_time)} 秒", flush=True)
250+
251+
252+
# 使用示例
253+
if __name__ == "__main__":
254+
API_URL = "www.dmxapi.cn" # API 节点地址
255+
DMX_API_TOKEN = "sk-XXXXXXXXXXXXXX" # API 密钥
256+
257+
# 创建图像转视频生成器实例
258+
kling_image_to_video = KlingImageToVideo(api_token=DMX_API_TOKEN, api_url=API_URL)
259+
260+
# 示例摄像机控制
261+
# camera_control = {
262+
# "type": "forward_up", # 预定义运镜类型 可选 “simple”, “down_back”, “forward_up”, “right_turn_forward”, “left_turn_forward”
263+
# # 如果使用 simple 类型，需要配置以下参数（六选一）
264+
# # "config": {
265+
# # "horizontal": 0, # 水平运镜 [-10, 10]
266+
# # "vertical": 0, # 垂直运镜 [-10, 10]
267+
# # "pan": 5, # 水平摇镜 [-10, 10]
268+
# # "tilt": 0, # 垂直摇镜 [-10, 10]
269+
# # "roll": 0, # 旋转运镜 [-10, 10]
270+
# # "zoom": 0 # 变焦 [-10, 10]
271+
# # }
272+
# }
273+
274+
# 生成视频
275+
video_url, video_id = kling_image_to_video.generate_video(
276+
model_name="kling-v1-6", # [必选] 模型版本 kling-v1, kling-v1-5, kling-v1-6
277+
image="/Users/dmxapi/Desktop/dmx.png", # [必选] 起始图片，可以是 URL 或 本地文件 路径
278+
prompt="生成图中的几只动物走路的场景", # [必选] 正向提示词
279+
# image_tail="end.jpg", # 结束图片，可以是 URL 或 本地文件 路径
280+
# negative_prompt="模糊, 扭曲", # 负向提示词
281+
# cfg_scale=0.5, # 生成视频的自由度，取值范围：[0, 1]
282+
# mode="std", # 生成模式：std(标准模式) 或 pro(专家模式)
283+
# duration="5", # 视频时长(秒)：5 或 10
284+
# camera_control=camera_control, # 摄像机控制参数
285+
# static_mask="mask.png", # 静态区域遮罩，可以是URL或本地文件路径
286+
# dynamic_masks=[{ # 动态区域遮罩列表
287+
# "mask": "mask.png", # 动态区域遮罩，可以是URL或本地文件路径
288+
# "trajectories": [
289+
# {"x": 100, "y": 100}, # 起始点
290+
# {"x": 150, "y": 200}, # 中间点
291+
# {"x": 200, "y": 300} # 结束点
292+
# ]
293+
# }],
294+
# callback_url="", # 回调地址
295+
# external_task_id="", # 自定义任务ID
296+
# timeout=600 # 等待超时时间（秒）
297+
)
298+
299+
print("生成的视频URL:", video_url)
300+
print("生成的视频ID:", video_id)

0 commit comments

Comments
 (0)