Add Aliyun OSS media relay

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:
 

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
 }

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
 

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,

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(