diff --git a/.env.example b/.env.example index 73e7122b7..a15bea607 100644 --- a/.env.example +++ b/.env.example @@ -32,6 +32,8 @@ SUNO_API_KEY= # Suno AI music generation (full songs, instrumenta # --- Video Generation --- HEYGEN_API_KEY= # HeyGen API (VEO, Sora, Runway, Kling, Seedance via single key) RUNWAY_API_KEY= # Runway Gen-4 (direct API, alternative to fal.ai routing) +VOLC_ACCESSKEY= # Volcengine Jimeng (即梦 AI) video generation via official API (HMAC-SHA256 V4 signing) +VOLC_SECRETKEY= # Secret Access Key paired with VOLC_ACCESSKEY. Get both at https://console.volcengine.com/iam/keymanage VIDEO_GEN_LOCAL_ENABLED= # Set to "true" for local video gen (needs GPU + diffusers) VIDEO_GEN_LOCAL_MODEL= # Local model: wan2.1-1.3b, wan2.1-14b, hunyuan-1.5, ltx2-local, cogvideo-5b MODAL_LTX2_ENDPOINT_URL= # Modal self-hosted LTX-2 endpoint (optional) diff --git a/docs/PROVIDERS.md b/docs/PROVIDERS.md index f5e34f7e6..0dc38b185 100644 --- a/docs/PROVIDERS.md +++ b/docs/PROVIDERS.md @@ -95,6 +95,44 @@ OpenMontage now uses those published rates in the Grok tool estimators. --- +### Volcengine Jimeng — 即梦 AI Video Generation + +> **Direct ByteDance API via V4 signing.** Calls the Volcengine visual API (visual.volcengineapi.com) with HMAC-SHA256 request signing using IAM AK/SK credentials. Supports text-to-video and image-to-video via Jimeng 3.0 Pro. + +**Tools unlocked:** `jimeng_video` +**Env vars:** `VOLC_ACCESSKEY` (Access Key ID) + `VOLC_SECRETKEY` (Secret Access Key) + +#### Setup + +1. Go to [console.volcengine.com/iam/keymanage](https://console.volcengine.com/iam/keymanage) +2. Create a Volcengine account if you don't have one +3. Create an Access Key pair (AK + SK) +4. Ensure your account has access to Jimeng AI (即梦) video generation service +5. Add to `.env`: `VOLC_ACCESSKEY=...` and `VOLC_SECRETKEY=...` + +#### What it's best for + +- Direct ByteDance/Volcengine API quota usage +- Jimeng 3.0 Pro text-to-video and image-to-video +- Chinese-language prompt understanding +- Configurable frame count (121=5s, 241=10s) and aspect ratio + +#### API notes + +Authentication uses Volcengine IAM V4 signing (HMAC-SHA256), not a Bearer token. The signing process builds a canonical request, derives a signing key from SK → date → region → service, and signs the request. + +API flow: `POST ?Action=CVSync2AsyncSubmitTask` → poll `POST ?Action=CVSync2AsyncGetResult` → download `video_url`. + +The `req_key` for video is `jimeng_ti2v_v30_pro`. Success code is `10000`. Task statuses: `in_queue`, `generating`, `done`, `not_found`, `expired`. + +#### Pricing + +| Model | Price | +|------|-------| +| Jimeng 3.0 Pro (video) | ~$0.05/sec (check Volcengine console for actual rate) | + +--- + ### Alibaba DashScope — Qwen Image + TTS + ASR > **Best for Chinese-language production.** One key unlocks Qwen-Image generation, Qwen-TTS Mandarin narration, and Qwen-ASR with word-level timestamps — the only DashScope path that provides word-level granularity for subtitle alignment. diff --git a/tests/contracts/test_jimeng_video.py b/tests/contracts/test_jimeng_video.py new file mode 100644 index 000000000..a5fd0f74d --- /dev/null +++ b/tests/contracts/test_jimeng_video.py @@ -0,0 +1,320 @@ +"""Contract tests for the Volcengine Jimeng video provider tool. + +These tests verify that the tool satisfies the BaseTool contract without +requiring real Volcengine AK/SK credentials or making any API calls. + +Run: pytest tests/contracts/test_jimeng_video.py -v +""" + +import pytest + +from tools.base_tool import ( + BaseTool, + Determinism, + ExecutionMode, + ToolRuntime, + ToolStability, + ToolStatus, + ToolTier, +) +from tools.video.jimeng_video import JimengVideo + + +# ------------------------------------------------------------------ +# Contract compliance +# ------------------------------------------------------------------ + +class TestContract: + + def test_inherits_base_tool(self): + assert issubclass(JimengVideo, BaseTool) + + def test_has_required_identity(self): + tool = JimengVideo() + assert tool.name == "jimeng_video" + assert tool.version + assert tool.provider == "volcengine" + assert tool.capability == "video_generation" + assert tool.tier == ToolTier.GENERATE + assert tool.stability == ToolStability.EXPERIMENTAL + assert tool.runtime == ToolRuntime.API + + def test_execution_mode_is_async(self): + assert JimengVideo().execution_mode == ExecutionMode.ASYNC + + def test_has_input_schema(self): + schema = JimengVideo().input_schema + assert schema.get("type") == "object" + props = schema.get("properties", {}) + required = schema.get("required", []) + assert required == ["prompt"] + for field in required: + assert field in props + + def test_has_capabilities(self): + tool = JimengVideo() + assert "text_to_video" in tool.capabilities + assert "image_to_video" in tool.capabilities + + def test_has_agent_skills(self): + assert "ai-video-gen" in JimengVideo().agent_skills + + def test_has_fallbacks(self): + tool = JimengVideo() + assert "minimax_video" in tool.fallback_tools + assert "kling_video" in tool.fallback_tools + + def test_has_install_instructions(self): + tool = JimengVideo() + assert "VOLC_ACCESSKEY" in tool.install_instructions + assert "VOLC_SECRETKEY" in tool.install_instructions + + def test_get_info_returns_dict(self): + info = JimengVideo().get_info() + assert isinstance(info, dict) + assert info["name"] == "jimeng_video" + assert info["provider"] == "volcengine" + assert info["runtime"] == "api" + + def test_status_unavailable_without_keys(self, monkeypatch): + monkeypatch.delenv("VOLC_ACCESSKEY", raising=False) + monkeypatch.delenv("VOLC_SECRETKEY", raising=False) + assert JimengVideo().get_status() == ToolStatus.UNAVAILABLE + + def test_status_available_with_keys(self, monkeypatch): + monkeypatch.setenv("VOLC_ACCESSKEY", "fake-ak") + monkeypatch.setenv("VOLC_SECRETKEY", "fake-sk") + assert JimengVideo().get_status() == ToolStatus.AVAILABLE + + def test_status_unavailable_with_only_ak(self, monkeypatch): + monkeypatch.setenv("VOLC_ACCESSKEY", "fake-ak") + monkeypatch.delenv("VOLC_SECRETKEY", raising=False) + assert JimengVideo().get_status() == ToolStatus.UNAVAILABLE + + def test_has_resource_profile(self): + rp = JimengVideo().resource_profile + assert rp.network_required is True + assert rp.vram_mb == 0 + + def test_has_retry_policy(self): + assert JimengVideo().retry_policy.max_retries >= 0 + + def test_has_side_effects(self): + side = JimengVideo().side_effects + assert len(side) > 0 + assert any("API" in s for s in side) + + def test_has_user_visible_verification(self): + assert len(JimengVideo().user_visible_verification) > 0 + + def test_lazy_imports_requests(self): + import importlib + import sys + mod_name = "tools.video.jimeng_video" + if "requests" in sys.modules: + del sys.modules["requests"] + importlib.reload(sys.modules[mod_name]) + + def test_estimate_cost_returns_float(self): + cost = JimengVideo().estimate_cost({"prompt": "x", "frames": 121}) + assert isinstance(cost, float) + assert cost > 0.0 + + def test_dry_run_returns_dict(self): + result = JimengVideo().dry_run({"prompt": "test"}) + assert isinstance(result, dict) + assert result["tool"] == "jimeng_video" + + +# ------------------------------------------------------------------ +# Idempotency keys +# ------------------------------------------------------------------ + +class TestIdempotencyKeys: + + def test_includes_all_output_affecting_fields(self): + fields = JimengVideo().idempotency_key_fields + for field in ("prompt", "operation", "image_url", "frames", "aspect_ratio", "seed"): + assert field in fields, f"missing idempotency field: {field}" + + def test_excludes_execution_only_fields(self): + fields = JimengVideo().idempotency_key_fields + for field in ("output_path", "poll_interval_seconds", "timeout_seconds"): + assert field not in fields + + def test_differs_on_frames(self): + tool = JimengVideo() + base = {"prompt": "x"} + assert tool.idempotency_key(base) != tool.idempotency_key({**base, "frames": 241}) + + def test_differs_on_aspect_ratio(self): + tool = JimengVideo() + base = {"prompt": "x"} + assert tool.idempotency_key({**base, "aspect_ratio": "16:9"}) != tool.idempotency_key( + {**base, "aspect_ratio": "9:16"} + ) + + def test_differs_on_seed(self): + tool = JimengVideo() + base = {"prompt": "x"} + assert tool.idempotency_key({**base, "seed": -1}) != tool.idempotency_key( + {**base, "seed": 42} + ) + + def test_differs_on_image_url(self): + tool = JimengVideo() + base = {"prompt": "x", "operation": "image_to_video"} + assert tool.idempotency_key(base) != tool.idempotency_key( + {**base, "image_url": "https://example.com/img.png"} + ) + + +# ------------------------------------------------------------------ +# Tool-specific behavior +# ------------------------------------------------------------------ + +class TestToolSpecific: + + def test_default_frames_is_121(self): + tool = JimengVideo() + assert tool.input_schema["properties"]["frames"]["default"] == 121 + + def test_default_aspect_ratio_is_16_9(self): + tool = JimengVideo() + assert tool.input_schema["properties"]["aspect_ratio"]["default"] == "16:9" + + def test_default_seed_is_negative_one(self): + tool = JimengVideo() + assert tool.input_schema["properties"]["seed"]["default"] == -1 + + def test_cost_scales_with_frames(self): + tool = JimengVideo() + cost_5s = tool.estimate_cost({"prompt": "x", "frames": 121}) + cost_10s = tool.estimate_cost({"prompt": "x", "frames": 241}) + assert cost_10s > cost_5s + + def test_build_payload_t2v(self): + tool = JimengVideo() + payload = tool._build_payload({"prompt": "a cat"}) + assert payload["req_key"] == "jimeng_ti2v_v30_pro" + assert payload["prompt"] == "a cat" + assert payload["frames"] == 121 + assert payload["aspect_ratio"] == "16:9" + assert payload["seed"] == -1 + assert "image_urls" not in payload + + def test_build_payload_i2v_includes_image(self): + tool = JimengVideo() + payload = tool._build_payload({ + "prompt": "motion", + "operation": "image_to_video", + "image_url": "https://example.com/img.png", + }) + assert payload["image_urls"] == ["https://example.com/img.png"] + + def test_build_payload_t2v_omits_image(self): + tool = JimengVideo() + payload = tool._build_payload({"prompt": "a cat", "operation": "text_to_video"}) + assert "image_urls" not in payload + + def test_i2v_without_image_fails(self, monkeypatch): + monkeypatch.setenv("VOLC_ACCESSKEY", "fake-ak") + monkeypatch.setenv("VOLC_SECRETKEY", "fake-sk") + result = JimengVideo().execute({"prompt": "test", "operation": "image_to_video"}) + assert result.success is False + assert "image_url" in result.error + + def test_no_keys_returns_error(self, monkeypatch): + monkeypatch.delenv("VOLC_ACCESSKEY", raising=False) + monkeypatch.delenv("VOLC_SECRETKEY", raising=False) + result = JimengVideo().execute({"prompt": "test"}) + assert result.success is False + assert "VOLC_ACCESSKEY" in result.error + assert "VOLC_SECRETKEY" in result.error + + def test_safe_error_redacts_keys(self, monkeypatch): + monkeypatch.setenv("VOLC_ACCESSKEY", "my-ak-secret") + monkeypatch.setenv("VOLC_SECRETKEY", "my-sk-secret") + redacted = JimengVideo._safe_error( + Exception("failed with ak=my-ak-secret sk=my-sk-secret") + ) + assert "my-ak-secret" not in redacted + assert "my-sk-secret" not in redacted + assert "[redacted]" in redacted + + def test_safe_error_no_empty_string_bug(self, monkeypatch): + """Regression: when no keys are set, _safe_error must not mangle.""" + monkeypatch.delenv("VOLC_ACCESSKEY", raising=False) + monkeypatch.delenv("VOLC_SECRETKEY", raising=False) + msg = JimengVideo._safe_error(Exception("abc")) + assert msg == "abc" + + def test_sign_returns_authorization_header(self): + headers = JimengVideo._sign( + "POST", "/", + {"Action": "CVSync2AsyncSubmitTask", "Version": "2022-08-31"}, + {}, b'{"prompt":"test"}', + "fake-ak", "fake-sk", + ) + assert "Authorization" in headers + assert "HMAC-SHA256" in headers["Authorization"] + assert "fake-ak" in headers["Authorization"] + assert "Host" in headers + assert "X-Date" in headers + assert "X-Content-Sha256" in headers + + def test_sign_includes_content_type(self): + headers = JimengVideo._sign("POST", "/", {}, {}, b"{}", "ak", "sk") + assert headers["Content-Type"] == "application/json" + + def test_json_or_raise_returns_dict(self): + class FakeResp: + status_code = 200 + def json(self): + return {"code": 10000, "data": {"task_id": "123"}} + assert JimengVideo._json_or_raise(FakeResp()) == {"code": 10000, "data": {"task_id": "123"}} + + def test_json_or_raise_raises_on_non_json(self): + class FakeResp: + status_code = 500 + def json(self): + raise ValueError("not JSON") + with pytest.raises(RuntimeError, match="Non-JSON"): + JimengVideo._json_or_raise(FakeResp()) + + def test_check_code_passes_on_success(self): + JimengVideo._check_code(200, {"code": 10000, "message": "Success"}) + + def test_check_code_raises_on_api_error(self): + with pytest.raises(RuntimeError, match="code=10008"): + JimengVideo._check_code(200, {"code": 10008, "message": "Insufficient balance"}) + + def test_check_code_raises_on_http_error(self): + with pytest.raises(RuntimeError, match="HTTP 401"): + JimengVideo._check_code(401, {"code": 10004, "message": "Auth failed"}) + + def test_check_code_defaults_to_success_when_code_missing(self): + """If code field is absent on HTTP 2xx, default to 10000 (success).""" + JimengVideo._check_code(200, {"data": {"task_id": "123"}}) + + +# ------------------------------------------------------------------ +# Registry discovery +# ------------------------------------------------------------------ + +class TestRegistryDiscovery: + + def test_discoverable(self): + from tools.tool_registry import ToolRegistry + registry = ToolRegistry() + registry.discover() + names = {t.name for t in registry._tools.values()} + assert "jimeng_video" in names + + def test_distinct_from_other_minimax_tools(self): + from tools.tool_registry import ToolRegistry + registry = ToolRegistry() + registry.discover() + jimeng = [t for t in registry._tools.values() if t.name == "jimeng_video"] + assert len(jimeng) == 1 + assert jimeng[0].provider == "volcengine" diff --git a/tools/video/jimeng_video.py b/tools/video/jimeng_video.py new file mode 100644 index 000000000..3208a05f2 --- /dev/null +++ b/tools/video/jimeng_video.py @@ -0,0 +1,394 @@ +"""Volcengine Jimeng (即梦 AI) video generation via the official API. + +Calls the Volcengine visual API directly (visual.volcengineapi.com) using +HMAC-SHA256 V4 request signing with AK/SK credentials. Supports text-to-video +and image-to-video via the Hailuo/Jimeng 3.0 Pro model. + +API flow: POST CVSync2AsyncSubmitTask -> poll CVSync2AsyncGetResult -> +download video_url. + +Authentication uses Volcengine IAM V4 signing (not Bearer token), which +requires an Access Key ID (AK) and Secret Access Key (SK) pair from +console.volcengine.com/iam/keymanage. +""" + +from __future__ import annotations + +import hashlib +import hmac +import json +import os +import time +import urllib.parse +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + +from tools.base_tool import ( + BaseTool, + Determinism, + ExecutionMode, + ResourceProfile, + RetryPolicy, + ToolResult, + ToolRuntime, + ToolStability, + ToolStatus, + ToolTier, +) + + +_HOST = "visual.volcengineapi.com" +_REGION = "cn-north-1" +_SERVICE = "cv" +_ALGORITHM = "HMAC-SHA256" +_API_VERSION = "2022-08-31" +_REQ_KEY_VIDEO = "jimeng_ti2v_v30_pro" + + +class JimengVideo(BaseTool): + name = "jimeng_video" + version = "0.1.0" + tier = ToolTier.GENERATE + capability = "video_generation" + provider = "volcengine" + stability = ToolStability.EXPERIMENTAL + execution_mode = ExecutionMode.ASYNC + determinism = Determinism.STOCHASTIC + runtime = ToolRuntime.API + + dependencies = [] + install_instructions = ( + "Set VOLC_ACCESSKEY and VOLC_SECRETKEY to your Volcengine IAM credentials.\n" + " Get them at https://console.volcengine.com/iam/keymanage\n" + " Ensure your account has access to Jimeng AI (即梦) video generation." + ) + agent_skills = ["ai-video-gen"] + + capabilities = ["text_to_video", "image_to_video"] + supports = { + "text_to_video": True, + "image_to_video": True, + "native_audio": False, + "seed": True, + } + best_for = [ + "Jimeng 3.0 Pro text-to-video and image-to-video via Volcengine", + "direct ByteDance API quota usage (not through a gateway)", + "Chinese-language prompt understanding", + ] + not_good_for = ["offline generation", "users without Volcengine AK/SK"] + fallback_tools = ["minimax_video", "kling_video", "veo_video"] + + input_schema = { + "type": "object", + "required": ["prompt"], + "properties": { + "prompt": { + "type": "string", + "description": "Video description. Max 2000 chars. Supports Chinese.", + }, + "operation": { + "type": "string", + "enum": ["text_to_video", "image_to_video"], + "default": "text_to_video", + }, + "image_url": { + "type": "string", + "description": ( + "First frame image URL for image-to-video. " + "Must be publicly accessible." + ), + }, + "frames": { + "type": "integer", + "minimum": 1, + "default": 121, + "description": "Total frames. 121=5s, 241=10s at 24fps.", + }, + "aspect_ratio": { + "type": "string", + "enum": ["16:9", "9:16", "1:1", "4:3", "3:4", "21:9"], + "default": "16:9", + }, + "seed": { + "type": "integer", + "default": -1, + "description": "Random seed. -1 for random.", + }, + "output_path": {"type": "string"}, + "poll_interval_seconds": { + "type": "number", + "minimum": 2, + "default": 5.0, + }, + "timeout_seconds": { + "type": "integer", + "minimum": 60, + "default": 600, + }, + }, + } + + resource_profile = ResourceProfile( + cpu_cores=1, ram_mb=512, vram_mb=0, disk_mb=500, network_required=True + ) + retry_policy = RetryPolicy( + max_retries=2, + backoff_seconds=2.0, + retryable_errors=["rate_limit", "timeout"], + ) + idempotency_key_fields = [ + "prompt", + "operation", + "image_url", + "frames", + "aspect_ratio", + "seed", + ] + side_effects = [ + "writes video file to output_path", + "calls Volcengine Jimeng API (V4-signed submit + poll + download)", + ] + user_visible_verification = [ + "Watch generated clip for motion coherence and prompt adherence", + ] + + def _ak(self) -> str | None: + val = os.environ.get("VOLC_ACCESSKEY", "") + if val and not val.strip().startswith("#"): + return val.strip() + return None + + def _sk(self) -> str | None: + val = os.environ.get("VOLC_SECRETKEY", "") + if val and not val.strip().startswith("#"): + return val.strip() + return None + + def get_status(self) -> ToolStatus: + if self._ak() and self._sk(): + return ToolStatus.AVAILABLE + return ToolStatus.UNAVAILABLE + + def estimate_cost(self, inputs: dict[str, Any]) -> float: + frames = int(inputs.get("frames", 121)) + seconds = frames / 24.0 + return round(0.05 * seconds, 2) + + def estimate_runtime(self, inputs: dict[str, Any]) -> float: + return 120.0 + int(inputs.get("frames", 121)) * 1.0 + + def execute(self, inputs: dict[str, Any]) -> ToolResult: + ak = self._ak() + sk = self._sk() + if not ak or not sk: + return ToolResult( + success=False, + error="VOLC_ACCESSKEY or VOLC_SECRETKEY not set. " + self.install_instructions, + ) + + operation = inputs.get("operation", "text_to_video") + if operation == "image_to_video" and not inputs.get("image_url"): + return ToolResult( + success=False, + error="image_to_video requires image_url (public URL).", + ) + + start = time.time() + try: + result = self._generate(inputs, ak=ak, sk=sk) + except Exception as exc: + return ToolResult( + success=False, + error=f"Jimeng video generation failed: {self._safe_error(exc)}", + ) + + result.duration_seconds = round(time.time() - start, 2) + return result + + def _generate(self, inputs: dict[str, Any], *, ak: str, sk: str) -> ToolResult: + import requests + + from tools.video._shared import probe_output + + payload = self._build_payload(inputs) + task_id = self._submit_task(payload, ak=ak, sk=sk) + video_url = self._poll_task( + task_id, ak=ak, sk=sk, + poll_interval=float(inputs.get("poll_interval_seconds", 5.0)), + timeout_seconds=int(inputs.get("timeout_seconds", 600)), + ) + + download = requests.get(video_url, timeout=120) + download.raise_for_status() + + output_path = Path(inputs.get("output_path", "jimeng_video.mp4")) + output_path.parent.mkdir(parents=True, exist_ok=True) + output_path.write_bytes(download.content) + + probed = probe_output(output_path) + return ToolResult( + success=True, + data={ + "provider": "volcengine", + "route": "jimeng_direct", + "model": _REQ_KEY_VIDEO, + "prompt": inputs["prompt"], + "operation": inputs.get("operation", "text_to_video"), + "frames": payload.get("frames", 121), + "aspect_ratio": payload.get("aspect_ratio", "16:9"), + "seed": payload.get("seed", -1), + "task_id": task_id, + "video_url": video_url, + "output": str(output_path), + "format": "mp4", + **probed, + }, + artifacts=[str(output_path)], + cost_usd=self.estimate_cost(inputs), + model=_REQ_KEY_VIDEO, + ) + + @staticmethod + def _build_payload(inputs: dict[str, Any]) -> dict[str, Any]: + operation = inputs.get("operation", "text_to_video") + payload: dict[str, Any] = { + "req_key": _REQ_KEY_VIDEO, + "prompt": inputs["prompt"], + "frames": int(inputs.get("frames", 121)), + "aspect_ratio": inputs.get("aspect_ratio", "16:9"), + "seed": int(inputs.get("seed", -1)), + } + if operation == "image_to_video" and inputs.get("image_url"): + payload["image_urls"] = [inputs["image_url"]] + return payload + + def _submit_task(self, payload: dict[str, Any], *, ak: str, sk: str) -> str: + import requests + + query = {"Action": "CVSync2AsyncSubmitTask", "Version": _API_VERSION} + body = json.dumps(payload, ensure_ascii=False).encode("utf-8") + headers = self._sign("POST", "/", query, {}, body, ak, sk) + url = f"https://{_HOST}/?{urllib.parse.urlencode(sorted(query.items()))}" + resp = requests.post(url, data=body, headers=headers, timeout=30) + data = self._json_or_raise(resp) + self._check_code(resp.status_code, data) + task_id = data.get("data", {}).get("task_id") + if not task_id: + raise RuntimeError(f"Jimeng submit returned no task_id: {data}") + return task_id + + def _poll_task( + self, task_id: str, *, ak: str, sk: str, + poll_interval: float, timeout_seconds: int, + ) -> str: + import requests + + query = {"Action": "CVSync2AsyncGetResult", "Version": _API_VERSION} + body = json.dumps({ + "req_key": _REQ_KEY_VIDEO, + "task_id": task_id, + "req_json": json.dumps({"return_url": True}), + }, ensure_ascii=False).encode("utf-8") + + deadline = time.time() + timeout_seconds + while time.time() < deadline: + time.sleep(poll_interval) + headers = self._sign("POST", "/", query, {}, body, ak, sk) + url = f"https://{_HOST}/?{urllib.parse.urlencode(sorted(query.items()))}" + resp = requests.post(url, data=body, headers=headers, timeout=30) + data = self._json_or_raise(resp) + self._check_code(resp.status_code, data) + status = (data.get("data") or {}).get("status", "") + if status == "done": + video_url = (data.get("data") or {}).get("video_url") + if not video_url: + raise RuntimeError(f"Jimeng task done but no video_url: {data}") + return video_url + if status in ("not_found", "expired"): + raise RuntimeError(f"Jimeng task invalid: status={status}") + raise TimeoutError(f"Jimeng task {task_id} did not finish within {timeout_seconds}s") + + @staticmethod + def _sign( + method: str, path: str, query_params: dict, + headers: dict, body: bytes, ak: str, sk: str, + ) -> dict: + now = datetime.now(timezone.utc) + x_date = now.strftime("%Y%m%dT%H%M%SZ") + short_date = x_date[:8] + + body_hash = hashlib.sha256(body).hexdigest() + headers = dict(headers) + headers["Host"] = _HOST + headers["X-Date"] = x_date + headers["X-Content-Sha256"] = body_hash + headers["Content-Type"] = "application/json" + + lower_headers = {k.lower(): v.strip() for k, v in headers.items()} + signed_names = sorted(lower_headers) + canonical_headers = "".join( + f"{k}:{lower_headers[k]}\n" + for k in signed_names + ) + signed_str = ";".join(signed_names) + + canonical_query = "&".join( + f"{urllib.parse.quote(str(k), safe='')}={urllib.parse.quote(str(v), safe='')}" + for k, v in sorted(query_params.items()) + ) + + canonical_request = "\n".join([ + method.upper(), path, canonical_query, + canonical_headers, signed_str, body_hash, + ]) + + credential_scope = f"{short_date}/{_REGION}/{_SERVICE}/request" + string_to_sign = "\n".join([ + _ALGORITHM, x_date, credential_scope, + hashlib.sha256(canonical_request.encode("utf-8")).hexdigest(), + ]) + + k_date = hmac.new(sk.encode("utf-8"), short_date.encode("utf-8"), hashlib.sha256).digest() + k_region = hmac.new(k_date, _REGION.encode("utf-8"), hashlib.sha256).digest() + k_service = hmac.new(k_region, _SERVICE.encode("utf-8"), hashlib.sha256).digest() + k_signing = hmac.new(k_service, b"request", hashlib.sha256).digest() + + signature = hmac.new(k_signing, string_to_sign.encode("utf-8"), hashlib.sha256).hexdigest() + + headers["Authorization"] = ( + f"{_ALGORITHM} Credential={ak}/{credential_scope}, " + f"SignedHeaders={signed_str}, Signature={signature}" + ) + return headers + + @staticmethod + def _safe_error(exc: Exception) -> str: + msg = str(exc) + for var in ("VOLC_SECRETKEY", "VOLC_ACCESSKEY"): + val = os.environ.get(var, "") + if val: + msg = msg.replace(val, "[redacted]") + return msg + + @staticmethod + def _json_or_raise(response: Any) -> dict[str, Any]: + try: + return response.json() + except ValueError as exc: + raise RuntimeError( + f"Non-JSON response from Jimeng API: HTTP {response.status_code}" + ) from exc + + @staticmethod + def _check_code(http_status: int, payload: dict[str, Any]) -> None: + if http_status < 400: + code = payload.get("code", 10000) + if code == 10000: + return + msg = payload.get("message", "unknown error") + raise RuntimeError(f"Jimeng API error: code={code}, msg={msg}") + code = payload.get("code", "unknown") + msg = payload.get("message", "unknown error") + raise RuntimeError(f"Jimeng API error: HTTP {http_status}, code={code}, msg={msg}")