diff --git a/README.md b/README.md index 5cb143f..ce69c5b 100644 --- a/README.md +++ b/README.md @@ -106,7 +106,13 @@ Python dependencies are listed in [requirements.txt](./requirements.txt). "secret_key": "", "poll_interval": 1.0, "request_timeout": 30, - "area": "global" + "area": "global", + "oss_endpoint": "", + "oss_access_key_id": "", + "oss_access_key_secret": "", + "oss_bucket": "", + "oss_prefix": "kling_uploads", + "oss_signed_url_expires": 86400 } ``` @@ -146,10 +152,9 @@ Python dependencies are listed in [requirements.txt](./requirements.txt). - `reference_video_input`: connect a node that outputs `VIDEO` - `reference_video_frames` + `reference_video_info`: connect an uploaded video loader that outputs decoded frames plus timing metadata +- `reference_video`: pass an existing public http(s) video URL directly -Direct reference-video URLs are no longer supported in this node. - -Reference images and videos are uploaded to a public temporary media host before Kling reads them. The wrapper tries `tmpfiles.org` first and automatically falls back to `catbox.moe` if that upload service resets the connection. +Reference images and local reference videos must be reachable by Kling before generation starts. If OSS settings are present in `config.local.json` or environment variables (`OSS_ENDPOINT`, `OSS_ACCESS_KEY_ID`, `OSS_ACCESS_KEY_SECRET`, `OSS_BUCKET`, `OSS_PREFIX`), the wrapper uploads media to Aliyun OSS first and sends Kling a signed temporary download URL. Without OSS, it falls back to public temporary hosts: `tmpfiles.org`, then `catbox.moe`. The visible `model_name` dropdown for motion control is: diff --git a/config.example.json b/config.example.json index 10f3fb2..a05455a 100644 --- a/config.example.json +++ b/config.example.json @@ -3,5 +3,11 @@ "secret_key": "", "poll_interval": 1.0, "request_timeout": 30, - "area": "china" + "area": "china", + "oss_endpoint": "", + "oss_access_key_id": "", + "oss_access_key_secret": "", + "oss_bucket": "", + "oss_prefix": "kling_uploads", + "oss_signed_url_expires": 86400 } diff --git a/examples/README.md b/examples/README.md index c9b27c9..15af5aa 100644 --- a/examples/README.md +++ b/examples/README.md @@ -15,7 +15,7 @@ This directory contains importable ComfyUI workflow JSON examples and small Pyth - `05_comfyui_kling_wrapper_text2video_v26_sound.json` is the native audio example and is built around `kling-v2-6`. - `06_comfyui_kling_wrapper_advanced_element_subject_to_image2video.json` requires 1 frontal portrait plus 1-3 additional photos of the same subject. Background or scene images do not count as advanced-element reference images. - If you need both subject binding and speech, generate the bound video first and then add speech with the lip-sync or audio nodes. -- `Motion Control` now expects uploaded/local video workflow inputs rather than a direct reference-video URL. Use a video loader node that provides either `VIDEO` or `reference_video_frames` plus `reference_video_info`. Set `duration` to `auto` if you want it to match the uploaded motion reference length. +- `Motion Control` accepts either a direct http(s) reference-video URL or uploaded/local video workflow inputs. For local media, use a video loader node that provides either `VIDEO` or `reference_video_frames` plus `reference_video_info`. Set `duration` to `auto` if you want it to match the uploaded motion reference length. ## Included files diff --git a/py/nodes.py b/py/nodes.py index a11f3e2..b95d010 100644 --- a/py/nodes.py +++ b/py/nodes.py @@ -10,8 +10,12 @@ from .api.exceptions import KLingAPIError import base64 import configparser from contextlib import contextmanager +import email.utils +import hashlib +import hmac import io import json +import mimetypes import os import re import shutil @@ -26,6 +30,7 @@ import folder_paths from comfy_extras.nodes_audio import LoadAudio import time import urllib.parse +import uuid from pathlib import Path ROOT_DIR = Path(__file__).resolve().parents[1] @@ -48,6 +53,7 @@ CATBOX_UPLOAD_API_URL = "https://catbox.moe/user/api.php" TMPFILES_MAX_SIZE_BYTES = 100 * 1024 * 1024 TMPFILES_UPLOAD_RETRY_COUNT = 3 TMPFILES_UPLOAD_RETRY_DELAY = 1.0 +LOCAL_MEDIA_UPLOAD_SUBFOLDER = "kling_uploads" DEFAULT_REFERENCE_VIDEO_FPS = 24.0 LIPSYNC_AUDIO_TYPES = { "阳光少年": "genshin_vindi2", @@ -763,6 +769,209 @@ def _normalize_catbox_upload_url(raw_url): return normalized +def _is_http_url(value): + parsed = urllib.parse.urlparse(str(value or "").strip()) + return parsed.scheme in {"http", "https"} and bool(parsed.netloc) + + +def _config_or_env_value(config_data, json_key, env_key, default=""): + if _json_value_present(config_data, json_key): + return str(config_data[json_key]).strip() + + env_value = _load_env_value(env_key) + if env_value: + return env_value + + return default + + +def _normalize_oss_endpoint(endpoint): + raw_endpoint = str(endpoint or "").strip().rstrip("/") + if not raw_endpoint: + raise ValueError("oss_endpoint is required.") + + if "://" not in raw_endpoint: + raw_endpoint = f"https://{raw_endpoint}" + + parsed = urllib.parse.urlparse(raw_endpoint) + if parsed.scheme not in {"http", "https"} or not parsed.netloc: + raise ValueError("oss_endpoint must be a valid OSS endpoint host or URL.") + if parsed.path not in {"", "/"}: + raise ValueError("oss_endpoint must not include a path.") + + return parsed.scheme, parsed.netloc + + +def _normalize_oss_prefix(prefix): + normalized = str(prefix or "").strip().strip("/") + return normalized or LOCAL_MEDIA_UPLOAD_SUBFOLDER + + +def _parse_oss_signed_url_expires(value): + if value in (None, ""): + return 24 * 60 * 60 + + try: + expires = int(str(value).strip()) + except (TypeError, ValueError) as exc: + raise ValueError("oss_signed_url_expires must be an integer number of seconds.") from exc + + if expires <= 0: + raise ValueError("oss_signed_url_expires must be greater than 0.") + + return expires + + +def _resolve_oss_upload_config(): + config_data = _load_json_config() + endpoint = _config_or_env_value(config_data, "oss_endpoint", "OSS_ENDPOINT") + access_key_id = _config_or_env_value(config_data, "oss_access_key_id", "OSS_ACCESS_KEY_ID") + access_key_secret = _config_or_env_value(config_data, "oss_access_key_secret", "OSS_ACCESS_KEY_SECRET") + bucket = _config_or_env_value(config_data, "oss_bucket", "OSS_BUCKET") + prefix = _config_or_env_value(config_data, "oss_prefix", "OSS_PREFIX", LOCAL_MEDIA_UPLOAD_SUBFOLDER) + signed_url_expires = _config_or_env_value( + config_data, + "oss_signed_url_expires", + "OSS_SIGNED_URL_EXPIRES", + 24 * 60 * 60, + ) + + required_values = { + "oss_endpoint": endpoint, + "oss_access_key_id": access_key_id, + "oss_access_key_secret": access_key_secret, + "oss_bucket": bucket, + } + present_count = sum(1 for value in required_values.values() if value) + if present_count == 0: + return None + + missing_keys = [key for key, value in required_values.items() if not value] + if missing_keys: + raise ValueError( + "OSS upload config is incomplete; set " + + ", ".join(missing_keys) + + " in config.local.json or environment variables." + ) + + scheme, endpoint_host = _normalize_oss_endpoint(endpoint) + return { + "scheme": scheme, + "endpoint_host": endpoint_host, + "access_key_id": access_key_id, + "access_key_secret": access_key_secret, + "bucket": str(bucket).strip(), + "prefix": _normalize_oss_prefix(prefix), + "signed_url_expires": _parse_oss_signed_url_expires(signed_url_expires), + } + + +def _oss_signature(access_key_secret, string_to_sign): + digest = hmac.new( + access_key_secret.encode("utf-8"), + string_to_sign.encode("utf-8"), + hashlib.sha1, + ).digest() + return base64.b64encode(digest).decode("utf-8") + + +def _build_oss_object_key(file_path, prefix): + filename = os.path.basename(os.fspath(file_path)) + safe_filename = re.sub(r"[^A-Za-z0-9._-]+", "_", filename).strip("._") + if not safe_filename: + safe_filename = "media" + + date_folder = time.strftime("%Y%m%d", time.gmtime()) + return f"{prefix}/{date_folder}/{uuid.uuid4().hex}-{safe_filename}" + + +def _build_oss_url(oss_config, object_key, query=None): + quoted_key = urllib.parse.quote(object_key, safe="/") + base_url = ( + f"{oss_config['scheme']}://{oss_config['bucket']}." + f"{oss_config['endpoint_host']}/{quoted_key}" + ) + if not query: + return base_url + + return base_url + "?" + urllib.parse.urlencode(query) + + +def _build_oss_authorization_header(oss_config, method, object_key, content_type, date): + string_to_sign = ( + f"{method}\n" + f"\n" + f"{content_type}\n" + f"{date}\n" + f"/{oss_config['bucket']}/{object_key}" + ) + signature = _oss_signature(oss_config["access_key_secret"], string_to_sign) + return f"OSS {oss_config['access_key_id']}:{signature}" + + +def _build_oss_signed_download_url(oss_config, object_key): + expires = int(time.time()) + int(oss_config["signed_url_expires"]) + string_to_sign = f"GET\n\n\n{expires}\n/{oss_config['bucket']}/{object_key}" + signature = _oss_signature(oss_config["access_key_secret"], string_to_sign) + return _build_oss_url( + oss_config, + object_key, + { + "OSSAccessKeyId": oss_config["access_key_id"], + "Expires": str(expires), + "Signature": signature, + }, + ) + + +def _upload_file_to_oss(file_path, timeout=DEFAULT_UPLOAD_TIMEOUT, oss_config=None): + normalized_path = _resolve_upload_file_path(file_path) + oss_config = oss_config or _resolve_oss_upload_config() + if not oss_config: + raise ValueError("OSS upload config is not configured.") + + object_key = _build_oss_object_key(normalized_path, oss_config["prefix"]) + content_type = mimetypes.guess_type(normalized_path)[0] or "application/octet-stream" + upload_url = _build_oss_url(oss_config, object_key) + + for attempt in range(1, TMPFILES_UPLOAD_RETRY_COUNT + 1): + try: + date = email.utils.formatdate(usegmt=True) + authorization = _build_oss_authorization_header( + oss_config, + "PUT", + object_key, + content_type, + date, + ) + with open(normalized_path, "rb") as handle: + response = requests.put( + upload_url, + data=handle, + headers={ + "Authorization": authorization, + "Content-Type": content_type, + "Date": date, + }, + timeout=float(timeout), + ) + response.raise_for_status() + return _build_oss_signed_download_url(oss_config, object_key) + except requests.RequestException as exc: + if attempt >= TMPFILES_UPLOAD_RETRY_COUNT: + raise ConnectionError( + "OSS media upload failed after multiple attempts. " + f"Check OSS endpoint, bucket permissions, and network access: {exc}" + ) from exc + print( + f"[{NODE_PREFIX}] OSS upload retry {attempt}/{TMPFILES_UPLOAD_RETRY_COUNT} " + f"after transient error: {exc}" + ) + time.sleep(TMPFILES_UPLOAD_RETRY_DELAY) + + raise ConnectionError("OSS media upload failed.") + + def _upload_file_to_tmpfiles(file_path, timeout=DEFAULT_UPLOAD_TIMEOUT): normalized_path = _resolve_upload_file_path(file_path) @@ -849,6 +1058,24 @@ def _upload_file_to_temporary_media_host(file_path, timeout=DEFAULT_UPLOAD_TIMEO failures = [] uploaders = [] + try: + oss_config = _resolve_oss_upload_config() + except Exception as exc: + oss_config = None + failures.append(f"aliyun OSS: {exc}") + + if oss_config is not None: + uploaders.append( + ( + "aliyun OSS", + lambda path, timeout=DEFAULT_UPLOAD_TIMEOUT, _oss_config=oss_config: _upload_file_to_oss( + path, + timeout=timeout, + oss_config=_oss_config, + ), + ) + ) + if file_size <= TMPFILES_MAX_SIZE_BYTES: uploaders.append(("tmpfiles.org", _upload_file_to_tmpfiles)) else: @@ -985,7 +1212,8 @@ def _resolve_motion_control_duration( return _format_duration_value(frame_count / fps) raise ValueError( - "duration='auto' requires reference_video_input or reference_video_frames with usable duration metadata." + "duration='auto' requires reference_video_input or reference_video_frames with usable duration metadata. " + "Set duration explicitly when using a direct reference_video URL." ) @@ -1079,10 +1307,10 @@ def _resolve_motion_control_reference_video( return _upload_reference_video_frames(reference_video_frames, reference_video_info) if isinstance(reference_video, str) and reference_video.strip(): - raise ValueError( - "Direct reference_video URLs are no longer supported for motion control. " - "Use reference_video_input or reference_video_frames instead." - ) + normalized = reference_video.strip() + if _is_http_url(normalized): + return normalized + raise ValueError("reference_video must be an http(s) URL when provided.") return "" @@ -2401,6 +2629,7 @@ class MotionControlNode: "reference_image": ("IMAGE",), }, "optional": { + "reference_video": ("STRING", {"multiline": False, "default": ""}), "reference_video_input": ("VIDEO",), "reference_video_frames": ("IMAGE",), "reference_video_info": ("VHS_VIDEOINFO",), @@ -2447,7 +2676,7 @@ class MotionControlNode: ) if not reference_video: raise ValueError( - "reference_video_input or reference_video_frames is required for motion control." + "reference_video, reference_video_input, or reference_video_frames is required for motion control." ) duration = _resolve_motion_control_duration( duration, diff --git a/tests/test_backend_config.py b/tests/test_backend_config.py index 91ab310..35da71f 100644 --- a/tests/test_backend_config.py +++ b/tests/test_backend_config.py @@ -4,6 +4,7 @@ import os import sys import tempfile import unittest +import urllib.parse from contextlib import contextmanager from pathlib import Path from types import ModuleType, SimpleNamespace @@ -33,6 +34,12 @@ ENV_KEYS = { "KLINGAI_AREA": "", "KLINGAI_POLL_INTERVAL": "", "KLINGAI_REQUEST_TIMEOUT": "", + "OSS_ENDPOINT": "", + "OSS_ACCESS_KEY_ID": "", + "OSS_ACCESS_KEY_SECRET": "", + "OSS_BUCKET": "", + "OSS_PREFIX": "", + "OSS_SIGNED_URL_EXPIRES": "", } @@ -271,6 +278,70 @@ class BackendConfigTests(unittest.TestCase): self.assertEqual(url, "https://tmpfiles.org/dl/123/sample.mp4") self.assertEqual(post_mock.call_count, 2) + def test_upload_file_to_oss_returns_signed_download_url(self): + with tempfile.TemporaryDirectory() as tmpdir: + file_path = Path(tmpdir) / "sample.mp4" + file_path.write_bytes(b"video-bytes") + + ok_response = mock.Mock() + ok_response.raise_for_status.return_value = None + oss_config = { + "scheme": "https", + "endpoint_host": "oss-cn-hangzhou.aliyuncs.com", + "access_key_id": "access-key-id", + "access_key_secret": "access-key-secret", + "bucket": "example-bucket", + "prefix": "GouMei-Video-Cut", + "signed_url_expires": 3600, + } + + with mock.patch.object(kling_nodes, "_build_oss_object_key", return_value="GouMei-Video-Cut/sample.mp4"): + with mock.patch.object(kling_nodes.time, "time", return_value=1000): + with mock.patch.object(kling_nodes.requests, "put", return_value=ok_response) as put_mock: + url = kling_nodes._upload_file_to_oss(file_path, oss_config=oss_config) + + put_mock.assert_called_once() + self.assertEqual( + put_mock.call_args.args[0], + "https://example-bucket.oss-cn-hangzhou.aliyuncs.com/GouMei-Video-Cut/sample.mp4", + ) + self.assertIn("Authorization", put_mock.call_args.kwargs["headers"]) + self.assertTrue(put_mock.call_args.kwargs["headers"]["Authorization"].startswith("OSS access-key-id:")) + + parsed_url = urllib.parse.urlparse(url) + query = urllib.parse.parse_qs(parsed_url.query) + self.assertEqual(parsed_url.scheme, "https") + self.assertEqual(parsed_url.netloc, "example-bucket.oss-cn-hangzhou.aliyuncs.com") + self.assertEqual(parsed_url.path, "/GouMei-Video-Cut/sample.mp4") + self.assertEqual(query["OSSAccessKeyId"], ["access-key-id"]) + self.assertEqual(query["Expires"], ["4600"]) + self.assertIn("Signature", query) + + def test_oss_upload_config_reads_environment_variables(self): + with tempfile.TemporaryDirectory() as tmpdir: + json_path = Path(tmpdir) / "missing-config.local.json" + env_values = { + **ENV_KEYS, + "OSS_ENDPOINT": "oss-cn-hangzhou.aliyuncs.com", + "OSS_ACCESS_KEY_ID": "access-key-id", + "OSS_ACCESS_KEY_SECRET": "access-key-secret", + "OSS_BUCKET": "example-bucket", + "OSS_PREFIX": "GouMei-Video-Cut", + "OSS_SIGNED_URL_EXPIRES": "7200", + } + + with mock.patch.dict(os.environ, env_values, clear=False): + with mock.patch.object(kling_nodes, "CONFIG_JSON_PATH", json_path): + config = kling_nodes._resolve_oss_upload_config() + + self.assertEqual(config["scheme"], "https") + self.assertEqual(config["endpoint_host"], "oss-cn-hangzhou.aliyuncs.com") + self.assertEqual(config["access_key_id"], "access-key-id") + self.assertEqual(config["access_key_secret"], "access-key-secret") + self.assertEqual(config["bucket"], "example-bucket") + self.assertEqual(config["prefix"], "GouMei-Video-Cut") + self.assertEqual(config["signed_url_expires"], 7200) + def test_temporary_media_upload_falls_back_to_catbox(self): with tempfile.TemporaryDirectory() as tmpdir: file_path = Path(tmpdir) / "sample.mp4" @@ -281,17 +352,18 @@ class BackendConfigTests(unittest.TestCase): catbox_response.text = "https://files.catbox.moe/sample.mp4" with mock.patch.object(kling_nodes.time, "sleep"): - with mock.patch.object( - kling_nodes.requests, - "post", - side_effect=[ - requests.exceptions.ConnectionError("reset"), - requests.exceptions.ConnectionError("reset"), - requests.exceptions.ConnectionError("reset"), - catbox_response, - ], - ) as post_mock: - url = kling_nodes._upload_file_to_temporary_media_host(file_path) + with mock.patch.object(kling_nodes, "_resolve_oss_upload_config", return_value=None): + with mock.patch.object( + kling_nodes.requests, + "post", + side_effect=[ + requests.exceptions.ConnectionError("reset"), + requests.exceptions.ConnectionError("reset"), + requests.exceptions.ConnectionError("reset"), + catbox_response, + ], + ) as post_mock: + url = kling_nodes._upload_file_to_temporary_media_host(file_path) self.assertEqual(url, "https://files.catbox.moe/sample.mp4") self.assertEqual(post_mock.call_count, 4) @@ -315,7 +387,7 @@ class BackendConfigTests(unittest.TestCase): self.assertEqual(metadata["default"], "kling-v2-6") self.assertEqual(duration_options, kling_nodes.MOTION_CONTROL_DURATIONS) self.assertEqual(duration_metadata["default"], "auto") - self.assertNotIn("reference_video", input_types["optional"]) + self.assertIn("reference_video", input_types["optional"]) def test_motion_control_uploads_reference_video_frames(self): sentinel_client = object() @@ -571,16 +643,47 @@ class BackendConfigTests(unittest.TestCase): self.assertEqual(first_client.request.call_count, 1) self.assertEqual(second_client.request.call_count, 1) - def test_motion_control_rejects_direct_reference_video_url(self): - with self.assertRaisesRegex( - ValueError, - "Direct reference_video URLs are no longer supported", - ): - kling_nodes.MotionControlNode().generate( - model_name="kling-v2-6", - reference_image=object(), - reference_video="https://example.com/reference-motion.mp4", - ) + def test_motion_control_accepts_direct_reference_video_url(self): + sentinel_client = object() + + @contextmanager + def fake_runtime_client(): + yield sentinel_client + + class FakeMotionControl: + last_instance = None + + def __init__(self): + FakeMotionControl.last_instance = self + + def run(self, client): + self.seen_client = client + return SimpleNamespace( + final_unit_deduction=1.0, + task_result=SimpleNamespace( + videos=[SimpleNamespace(url="https://example.com/video.mp4", id="video-123")] + ), + ) + + reference_video_url = "https://example.com/reference-motion.mp4" + with mock.patch.object(kling_nodes, "_runtime_client", fake_runtime_client): + with mock.patch.object(kling_nodes, "MotionControl", FakeMotionControl): + with mock.patch.object( + kling_nodes, + "_upload_image_reference", + return_value="https://tmpfiles.org/dl/img/ref.png", + ): + url, video_id = kling_nodes.MotionControlNode().generate( + model_name="kling-v2-6", + reference_image=object(), + reference_video=reference_video_url, + duration="5", + ) + + self.assertEqual(url, "https://example.com/video.mp4") + self.assertEqual(video_id, "video-123") + self.assertEqual(FakeMotionControl.last_instance.video_url, reference_video_url) + self.assertIs(FakeMotionControl.last_instance.seen_client, sentinel_client) def test_motion_control_auto_duration_requires_video_timing(self): with self.assertRaisesRegex(