#!/usr/bin/env python3 """ render_job_client.py — ab-api renderJob 接口 Python 客户端 封装对 ab-api /api/renderJob/* 的调用，供 render_video.py 和 gen_jianying_draft.py 使用。 认证方式: X-Priv-Token header（与其他 skill 脚本一致） 接口地址: 优先读 MM_API_BASE_URL（executor 显式转发、全链路标准名）， 回退 MM_BACKEND_API_URL，再回退默认 https://api.remixmate.ai/api。 两个变量语义相同，保留回退是为兼容只配了其中一个的部署环境。 """ from log_diagnostics import trace_headers import json import os import sys import urllib.error import urllib.request def _api_base() -> str: return os.environ.get( "MM_API_BASE_URL", os.environ.get("MM_BACKEND_API_URL", "https://api.remixmate.ai/api"), ).rstrip("/") def _make_headers(priv_token: str) -> dict: headers = { "Content-Type": "application/json", "X-Priv-Token": priv_token, "x-invoke-skill": "render-video", } agent_name = os.environ.get("AGENT_NAME", "") if agent_name: headers["x-invoke-agent"] = agent_name headers.update(trace_headers()) return headers # ─── 扣费回传 ──────────────────────────────────────────────────────────────── # ab-api 会在扣了积分的成功响应上挂 billing 字段（见 core.Success）。渲染积分是在 # /renderJob/saveManifest 里扣的，所以这里是渲染花费唯一的可见入口——累加起来交给 # render_video.py 在收尾时统一告知用户。 _BILLING: dict = {"credits": 0, "balance": None} def _record_billing(result: dict) -> None: billing = result.get("billing") if not isinstance(billing, dict): return credits = billing.get("credits") if not isinstance(credits, (int, float)) or credits <= 0: return _BILLING["credits"] += int(credits) balance = billing.get("balance") if isinstance(balance, (int, float)): _BILLING["balance"] = int(balance) def billing_summary() -> dict: """本进程通过本模块产生的累计扣费：{"credits": int, "balance": int|None}。""" return dict(_BILLING) def _post(path: str, payload: dict, priv_token: str, timeout: int = 30) -> dict: """发送 POST 请求并返回响应 data 字段；失败时抛出 RuntimeError。""" url = f"{_api_base()}{path}" body = json.dumps(payload).encode("utf-8") req = urllib.request.Request(url, data=body, headers=_make_headers(priv_token), method="POST") try: with urllib.request.urlopen(req, timeout=timeout) as resp: result = json.loads(resp.read().decode("utf-8")) except urllib.error.HTTPError as e: raise RuntimeError(f"HTTP {e.code} request failed: {path}") from e except urllib.error.URLError as e: raise RuntimeError(f"network request failed ({path}): {e.reason}") from e _record_billing(result) if result.get("code") != 0: msg = result.get("msg") or result.get("message") or "unknown error" raise RuntimeError(f"API returned an error [{path}]: {msg}") return result.get("data") or {} # ───────────────────────────────────────────────────────────────────────────── def create_job(priv_token: str, dsl: str = "", dsl_meta: dict = None) -> int: """ 创建渲染任务。 会话归属：本函数会自动带上环境变量 CONVERSATION_ID（若已设置），让服务端在 创建时就把 job 挂到该会话的 take 上。 为什么要在这里传：渲染任务过去出生就是孤儿——归属全靠事后一次独立的补绑调用， 而那条路上有四个静默盲区（CLI 的 take 绑定在 job 出生前就跑完了；云端因为 CONVERSATION_ID 已被注入而整段跳过它；兜底补绑的 jobId 末位来源是从 stdout 正则抠的）。归属断了，成片在旁白面板里就查不到已生成的音频。 而 CONVERSATION_ID 在这一刻本来就已经在环境里了——信息在场，以前只是没传。 见 ab-platform/docs/render-job-binding-fix.md。 服务端对该字段是软失败：绑不上只记日志，不影响任务创建。 Args: priv_token: PrivToken dsl: 完整 DSL JSON 字符串（可选） dsl_meta: DSL 摘要字典（可选，将被序列化为 JSON） Returns: jobId (int) """ payload: dict = {} if dsl: payload["dsl"] = dsl if dsl_meta: payload["dslMeta"] = dsl_meta conversation_id = os.environ.get("CONVERSATION_ID", "").strip() if conversation_id: payload["conversationId"] = conversation_id data = _post("/renderJob/create", payload, priv_token) job_id = data.get("jobId") if not job_id: raise RuntimeError("create render job failed: response has no jobId") return int(job_id) def save_plan(job_id: int, render_plan: str, priv_token: str) -> None: """ 保存 RenderPlan 到数据库（Phase 1 完成后调用）。 Args: job_id: 渲染任务 ID render_plan: 完整 RenderPlan JSON 字符串 priv_token: PrivToken """ _post("/renderJob/savePlan", {"jobId": job_id, "renderPlan": render_plan}, priv_token) def get_plan(job_id: int, priv_token: str) -> str: """ 从数据库加载 RenderPlan（Phase 2 开始前调用）。 Args: job_id: 渲染任务 ID priv_token: PrivToken Returns: RenderPlan JSON 字符串 """ data = _post("/renderJob/getPlan", {"jobId": job_id}, priv_token) render_plan = data.get("renderPlan") if not render_plan: raise RuntimeError(f"job {job_id} has no saved RenderPlan; run --resolve-only --save-job first") return render_plan def save_manifest(job_id: int, manifest: dict, priv_token: str) -> None: """ 保存渲染结果到数据库（渲染完成后调用）。 Args: job_id: 渲染任务 ID manifest: 渲染结果字典 priv_token: PrivToken """ _post("/renderJob/saveManifest", {"jobId": job_id, "manifest": manifest}, priv_token) def get_job(job_id: int, priv_token: str) -> dict: """ 获取渲染任务完整详情。 Args: job_id: 渲染任务 ID priv_token: PrivToken Returns: 任务详情字典（含 status / renderPlan / manifest 等） """ return _post("/renderJob/get", {"jobId": job_id}, priv_token) def presign_template_source(template_key: str, priv_token: str) -> dict: """ 为已注册私有模板的源码包申请临时 GET 下载 URL（渲染取包用）。 调 ab-api /remotionTemplate/presignSource，返回 { tarballUrl, sourceOssKey, version }。 ab-render 用 tarballUrl 走 /renderDraft 动态 bundle 渲染。 注意：presignSource 端点不在 /renderJob 下，路径以 / 开头直接拼到 _api_base。 """ return _post("/remotionTemplate/presignSource", {"templateKey": template_key}, priv_token) # ─── CLI self-test ──────────────────────────────────────────────────────────── if __name__ == "__main__": import argparse parser = argparse.ArgumentParser(description="render_job_client self-test tool") parser.add_argument("--priv-token", default=os.environ.get("PRIV_TOKEN", ""), help="PrivToken") parser.add_argument("--get", type=int, metavar="JOB_ID", help="Fetch job details") parser.add_argument("--get-plan", type=int, metavar="JOB_ID", help="Fetch RenderPlan") args = parser.parse_args() token = args.priv_token if not token: print("❌ PRIV_TOKEN not configured (or pass --priv-token)", file=sys.stderr) sys.exit(1) if args.get: result = get_job(args.get, token) print(json.dumps(result, ensure_ascii=False, indent=2)) elif args.get_plan: plan = get_plan(args.get_plan, token) print(plan) else: parser.print_help()