"""Chrome Headless vendor sync — separated from render_video.py. 把 Chrome 下载/解压/共享缓存这块"部署/运维相关"的逻辑收拢到一个文件，让 render_video.py 只剩"调度 + 渲染"。日后如果要把这块搬到 ab-render Dockerfile 或独立 prepare 脚本，只需要换掉这个模块的实现，render_video 不用动。 公共入口只有一个：sync_chrome_headless_vendor()。其它都是辅助函数，给 sync 用。 """ from __future__ import annotations import builtins import json import os import platform import shutil import subprocess # noqa: F401 (kept for parity with original imports) import sys import tempfile import urllib.error import urllib.request import zipfile from datetime import datetime from typing import Optional def _now_iso() -> str: from datetime import timezone return datetime.now(timezone.utc).isoformat() def _log(msg: str) -> None: """Tiny stderr-with-timestamp helper, mirrors render_video.LogPrint format.""" now = datetime.now() stamp = now.strftime("%Y%m%d %H%M%S") + f":{now.microsecond // 1000:03d}" builtins.print(f"[{stamp}] {msg}", file=sys.stderr) def _detect_remotion_chrome_platform() -> Optional[str]: """与 Remotion BrowserFetcher 一致的平台标识；不支持时返回 None。""" system = platform.system() machine = platform.machine().lower() if system == "Darwin": return "mac-arm64" if machine == "arm64" else "mac-x64" if system == "Linux": if machine in ("aarch64", "arm64"): return "linux-arm64" return "linux64" if system == "Windows": return "win64" return None def _load_chrome_vendor_manifest(renderer_dir: str) -> Optional[dict]: override = os.environ.get("REMOTION_CHROME_VENDOR_JSON", "").strip() path = override if override else os.path.join( renderer_dir, "chrome-headless-vendor-template", "vendor-zip-urls.json", ) if not os.path.isfile(path): return None try: with open(path, "r", encoding="utf-8") as f: data = json.load(f) return data if isinstance(data, dict) else None except (OSError, json.JSONDecodeError): return None def _download_chrome_vendor_zip(url: str, dest_path: str) -> None: timeout = int(os.environ.get("REMOTION_CHROME_VENDOR_DOWNLOAD_TIMEOUT", "3600")) req = urllib.request.Request( url, headers={"User-Agent": "Mozilla/5.0 (compatible; render-video/1.0)"}, ) try: with urllib.request.urlopen(req, timeout=timeout) as resp: with open(dest_path, "wb") as out: shutil.copyfileobj(resp, out, 1024 * 256) except urllib.error.HTTPError as e: raise RuntimeError(f"HTTP {e.code}: {e.reason}") from e except urllib.error.URLError as e: raise RuntimeError(f"Download failed: {e.reason}") from e def _vendor_chrome_headless_ready(chrome_root: str, rem_platform: str, expected_version: str) -> bool: version_path = os.path.join(chrome_root, "VERSION") if not os.path.isfile(version_path): return False try: with open(version_path, "r", encoding="utf-8") as f: if f.read().strip() != expected_version: return False except OSError: return False inner = os.path.join(chrome_root, rem_platform, f"chrome-headless-shell-{rem_platform}") if rem_platform == "win64": return os.path.isfile(os.path.join(inner, "chrome-headless-shell.exe")) return os.path.isfile(os.path.join(inner, "chrome-headless-shell")) def _ensure_chrome_headless_shell_executable(chrome_root: str, rem_platform: str) -> None: if rem_platform == "win64": return inner = os.path.join(chrome_root, rem_platform, f"chrome-headless-shell-{rem_platform}") exe = os.path.join(inner, "chrome-headless-shell") if os.path.isfile(exe) and hasattr(os, "chmod"): os.chmod(exe, os.stat(exe).st_mode | 0o111) def _resolve_chrome_share_root() -> str: root = os.environ.get("REMOTION_CHROME_SHARE_DIR", "/tmp/agent-share/chrome-headless").strip() return root if root else "/tmp/agent-share/chrome-headless" def _chrome_headless_share_zip_path(share_root: str, rem_platform: str) -> str: return os.path.join(share_root, f"chrome-headless-shell-{rem_platform}.zip") def _share_chrome_zip_ready(share_root: str, rem_platform: str, expected_version: str) -> bool: """共享目录仅保留 VERSION + 各平台 zip 时，判断是否可用。""" version_path = os.path.join(share_root, "VERSION") if not os.path.isfile(version_path): return False try: with open(version_path, "r", encoding="utf-8") as f: if f.read().strip() != expected_version: return False except OSError: return False zpath = _chrome_headless_share_zip_path(share_root, rem_platform) try: return os.path.isfile(zpath) and os.path.getsize(zpath) > 0 except OSError: return False def _extract_chrome_headless_zip_to_dest(zip_path: str, dest_root: str, rem_platform: str) -> None: """将官方 headless shell zip 解压到 Remotion 缓存根目录下的平台子目录。""" platform_dest = os.path.join(dest_root, rem_platform) if os.path.isdir(platform_dest): shutil.rmtree(platform_dest) os.makedirs(dest_root, exist_ok=True) with zipfile.ZipFile(zip_path, "r") as zf: zf.extractall(platform_dest) def sync_chrome_headless_vendor(renderer_dir: str, render_plan: dict) -> None: """ 共享目录（默认 /tmp/agent-share/chrome-headless）只存 VERSION + chrome-headless-shell-<平台>.zip； 若 zip 与版本匹配则本地解压到 node_modules/.remotion/chrome-headless-shell/，否则从 OSS 下载， 成功后把 zip 写入共享目录（不存解压后的目录树）。 所有 progress / warning 经 stderr 打印；render_plan["logs"] 上记录关键事件。 """ if os.environ.get("REMOTION_SKIP_VENDOR_CHROME", "").strip().lower() in ("1", "true", "yes"): render_plan["logs"].append({ "phase": "render", "message": "Skipped Chrome Headless vendor sync (REMOTION_SKIP_VENDOR_CHROME)", "timestamp": _now_iso(), }) return rem_platform = _detect_remotion_chrome_platform() if not rem_platform: _log("⚠️ 无法识别本机以下载 Chrome Headless，Remotion 可能会尝试在线下载") return manifest = _load_chrome_vendor_manifest(renderer_dir) if not manifest: _log( "⚠️ 未找到 vendor-zip-urls.json（可用 REMOTION_CHROME_VENDOR_JSON 指定路径），" "Remotion 可能会尝试在线下载" ) return zip_urls = manifest.get("zipUrls") if not isinstance(zip_urls, dict): zip_urls = {} url = zip_urls.get(rem_platform) if not url or not isinstance(url, str): _log(f"⚠️ vendor-zip-urls.json 中未配置 {rem_platform} 的下载地址，Remotion 可能会尝试在线下载") render_plan["logs"].append({ "phase": "render", "message": f"No OSS zip URL for platform {rem_platform} in vendor manifest", "timestamp": _now_iso(), }) return expected_version = (manifest.get("chromeVersion") or "").strip() if not expected_version: version_path = os.path.join(renderer_dir, "chrome-headless-vendor-template", "VERSION") try: with open(version_path, "r", encoding="utf-8") as f: expected_version = f.read().strip() except OSError: expected_version = "" if not expected_version: _log("⚠️ 无法确定 Chrome 版本（chromeVersion 或 VERSION），跳过 vendor 同步") return dest_root = os.path.join(renderer_dir, "node_modules", ".remotion", "chrome-headless-shell") force = os.environ.get("REMOTION_FORCE_VENDOR_CHROME", "").strip().lower() in ("1", "true", "yes") force = force or os.environ.get("REMOTION_FORCE_VENDOR_ZIP", "").strip().lower() in ("1", "true", "yes") if not force and _vendor_chrome_headless_ready(dest_root, rem_platform, expected_version): _log(f"ℹ️ 已存在匹配的 Chrome Headless Shell（{expected_version}），跳过下载") return share_root = _resolve_chrome_share_root() platform_dest = os.path.join(dest_root, rem_platform) share_zip = _chrome_headless_share_zip_path(share_root, rem_platform) if not force and _share_chrome_zip_ready(share_root, rem_platform, expected_version): _log(f"ℹ️ 共享目录已有 zip（{rem_platform} {expected_version}），跳过网络下载 → {share_zip}") sys.stderr.flush() try: _extract_chrome_headless_zip_to_dest(share_zip, dest_root, rem_platform) except (OSError, zipfile.BadZipFile) as e: _log(f"❌ 从共享 zip 解压浏览器失败: {e}") render_plan["logs"].append({ "phase": "render", "message": f"Chrome Headless unzip from share zip failed: {e}", "timestamp": _now_iso(), }) return try: with open(os.path.join(dest_root, "VERSION"), "w", encoding="utf-8") as vf: vf.write(expected_version + "\n") except OSError as e: _log(f"⚠️ 写入 VERSION 失败: {e}") _ensure_chrome_headless_shell_executable(dest_root, rem_platform) render_plan["logs"].append({ "phase": "render", "message": f"Installed Chrome Headless from share zip ({share_zip}, {expected_version})", "timestamp": _now_iso(), }) _log(f"✅ Chrome Headless Shell 已就绪: {dest_root}") return if os.path.isdir(platform_dest): shutil.rmtree(platform_dest) os.makedirs(dest_root, exist_ok=True) _log(f"📦 正在从 OSS 下载 Chrome Headless Shell（{rem_platform}）…") sys.stderr.flush() fd, tmp_zip = tempfile.mkstemp(suffix=".zip", prefix="chrome-headless-vendor-") os.close(fd) download_ok = False try: try: _log(f" ⬇️ 开始拉取浏览器包: {url.strip()}") sys.stderr.flush() _download_chrome_vendor_zip(url.strip(), tmp_zip) except RuntimeError as e: _log(f"❌ 下载内置浏览器失败: {e}") render_plan["logs"].append({ "phase": "render", "message": f"Chrome Headless vendor download failed: {e}", "timestamp": _now_iso(), }) return try: _extract_chrome_headless_zip_to_dest(tmp_zip, dest_root, rem_platform) download_ok = True except (OSError, zipfile.BadZipFile) as e: _log(f"❌ 解压内置浏览器失败: {e}") render_plan["logs"].append({ "phase": "render", "message": f"Chrome Headless vendor unzip failed: {e}", "timestamp": _now_iso(), }) return finally: if download_ok: try: os.makedirs(share_root, exist_ok=True) shutil.copy2(tmp_zip, share_zip) with open(os.path.join(share_root, "VERSION"), "w", encoding="utf-8") as vf: vf.write(expected_version + "\n") _log(f" 📁 已缓存 zip 到共享目录: {share_zip}") except OSError as e: _log(f"⚠️ 写入共享 zip 失败（不影响渲染）: {e}") try: os.unlink(tmp_zip) except OSError: pass try: with open(os.path.join(dest_root, "VERSION"), "w", encoding="utf-8") as vf: vf.write(expected_version + "\n") except OSError as e: _log(f"⚠️ 写入 VERSION 失败: {e}") _ensure_chrome_headless_shell_executable(dest_root, rem_platform) render_plan["logs"].append({ "phase": "render", "message": f"Synced Chrome Headless Shell from OSS ({rem_platform}, {expected_version})", "timestamp": _now_iso(), }) _log(f"✅ Chrome Headless Shell 已就绪: {dest_root}")