fix: worker 对接平台 watermark-task API
Docker Build and Push / build-and-push (cuda, 12.8) (push) Has been skipped
Docker Build and Push / check-secrets (push) Failing after 4s
Docker Build and Push / build-and-push (cpu, latest) (push) Has been skipped
Docker Build and Push / build-and-push (cuda, 11.8) (push) Has been skipped
Docker Build and Push / build-and-push (cuda, 12.6) (push) Has been skipped
Docker Build and Push / build-and-push (directml, latest) (push) Has been skipped
Build Windows CPU / build (push) Has been cancelled
Build Windows CUDA 11.8 / build (push) Has been cancelled
Build Windows CUDA 12.6 / build (push) Has been cancelled
Build Windows CUDA 12.8 / build (push) Has been cancelled
Build Windows DirectML / build (push) Has been cancelled

- claim/progress/result 路径改为 /openapi/watermark-task/*
- 认领字段 video_url → media_url(兼容旧字段)
- result status: passed→done,字段 output_path→output_url,error→error_message
- 新增 _upload_result_video:处理完后上传结果到平台(no_transcode 直传),获取 play_url 回传

Co-Authored-By: Claude Sonnet 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
Claude
2026-06-26 23:55:56 -07:00
co-authored by Claude Sonnet 4.6
parent 76f5d1c857
commit 073a2a4fba
+60 -10
View File
@@ -670,7 +670,7 @@ class VSRTaskWorker:
async def _claim(self) -> dict[str, Any] | None:
resp = await self._signed_post(
"/openapi/vsr-task/claim",
"/openapi/watermark-task/claim",
{"slot_id": self._slot_id, "slot_total": self._slot_total},
)
if resp is None:
@@ -682,7 +682,7 @@ class VSRTaskWorker:
async def _run_task(self, ptask: dict[str, Any]) -> None:
platform_task_id: int = ptask["id"]
video_url: str = ptask.get("video_url", "")
video_url: str = ptask.get("media_url", "") or ptask.get("video_url", "")
logger.info(f"[vsr_worker] 认领任务 #{platform_task_id} url={video_url[:60]}")
task_id = uuid.uuid4().hex[:8]
@@ -744,29 +744,79 @@ class VSRTaskWorker:
final = tasks.get(task_id, {})
if final.get("status") == "done" and final.get("output_path"):
await self._report_result(platform_task_id, status="passed",
task_id=task_id, output_path=final["output_path"])
output_url = await self._upload_result_video(final["output_path"])
await self._report_result(platform_task_id, status="done", output_url=output_url)
else:
await self._report_result(platform_task_id, status="failed",
task_id=task_id, error=final.get("error", "unknown"))
error_message=final.get("error", "unknown"))
async def _report_progress(self, platform_task_id: int, message: str) -> None:
await self._signed_post(
f"/openapi/vsr-task/{platform_task_id}/progress",
f"/openapi/watermark-task/{platform_task_id}/progress",
{"message": message},
headers={"X-Worker-Slot": str(self._slot_id)},
)
async def _report_result(self, platform_task_id: int, *, status: str,
task_id: str | None = None, output_path: str | None = None,
error: str = "") -> None:
output_url: str | None = None,
error_message: str = "") -> None:
await self._signed_post(
f"/openapi/vsr-task/{platform_task_id}/result",
{"status": status, "task_id": task_id, "output_path": output_path, "error": error},
f"/openapi/watermark-task/{platform_task_id}/result",
{"status": status, "output_url": output_url, "error_message": error_message},
headers={"X-Worker-Slot": str(self._slot_id)},
)
logger.info(f"[vsr_worker] 任务 #{platform_task_id} 上报结果 status={status}")
async def _upload_result_video(self, local_path: str) -> str | None:
"""处理完成后将输出视频上传到平台,返回 play_url(S3 地址)。"""
import hashlib as _hl
try:
data = Path(local_path).read_bytes()
sha256 = _hl.sha256(data).hexdigest()
filename = Path(local_path).name
sign_resp = await self._signed_post(
"/openapi/file/upload/get_upload_sign",
{
"sha256": sha256,
"size": len(data),
"filename": filename,
"content_type": "video/mp4",
"file_type": "video",
"biz_ref": "vsr-output",
"no_transcode": True,
},
)
if sign_resp is None:
return None
action = sign_resp.get("action", "")
if action == "skip":
return sign_resp.get("play_url")
presigned_url = sign_resp.get("presigned_url", "")
object_key = sign_resp.get("object_key", "")
if not presigned_url or not object_key:
logger.warning(f"[vsr_worker] upload sign 未返回 presigned_url/object_key: {sign_resp}")
return None
put_resp = await self._client.put(presigned_url, content=data,
headers={"Content-Type": "video/mp4"})
put_resp.raise_for_status()
reg_resp = await self._signed_post(
"/openapi/file/assets/register",
{
"sha256": sha256,
"object_key": object_key,
"size": len(data),
"filename": filename,
"content_type": "video/mp4",
"file_type": "video",
"biz_ref": "vsr-output",
"no_transcode": True,
},
)
return reg_resp.get("play_url") if reg_resp else None
except Exception as e:
logger.warning(f"[vsr_worker] 上传结果视频失败: {e}")
return None
async def _signed_post(self, path: str, body: dict[str, Any],
headers: dict[str, str] | None = None) -> dict[str, Any] | None:
body_str = json.dumps(body, separators=(",", ":"), ensure_ascii=False)