app.py

"""Blender Render Space 的 Gradio 界面、公开 API 和 MCP 入口。"""

from __future__ import annotations

# ZeroGPU 要求 spaces 在可能加载 CUDA 的库之前导入。
import spaces  # noqa: F401

import json
import logging
import os
from collections.abc import Iterator
from typing import Any, Callable, TypeVar

import gradio as gr

import service
from blender_runtime import ensure_blender
from render_config import RenderConfig, parse_config


LOGGER = logging.getLogger(__name__)
T = TypeVar("T")

CSS = """
.gradio-container { max-width: 1480px !important; margin: 0 auto; }
#render-hero {
    background: linear-gradient(125deg, #102331, #1b4148 62%, #2c6062);
    border: 1px solid #31565b; border-radius: 20px;
    padding: 30px 32px; margin: 8px 0 18px; color: #f3fbfc;
}
#render-hero .eyebrow { font-size: 12px; letter-spacing: .15em; color: #9be3d8; }
#render-hero h1 { color: #f3fbfc; font-size: clamp(27px, 4vw, 39px); margin: 7px 0 9px; }
#render-hero p { color: #c7dcdf; margin: 0; line-height: 1.75; }
#render-hero .pills { display: flex; flex-wrap: wrap; gap: 8px; margin-top: 19px; }
#render-hero .pill {
    font-size: 12px; background: #ffffff12; border: 1px solid #ffffff26;
    border-radius: 30px; padding: 5px 12px; color: #e4f5f4;
}
.workflow-note { border-left: 3px solid #3d9e91; padding-left: 12px; }
.section-caption { margin-top: 8px; margin-bottom: 0; }
#job-actions { align-items: stretch; }
@media (max-width: 640px) {
    #render-hero { padding: 24px 20px; border-radius: 14px; }
}
"""

HERO = """
<section id="render-hero">
  <div class="eyebrow">BLENDER RENDER SPACE</div>
  <h1>把场景变成作品</h1>
  <p>上传 Blender 工程与资源，用 JSON 驱动图片或动画渲染。<br>
  按批次保存帧，通过 API 与 MCP 接入你的创作 Agent。</p>
  <div class="pills">
    <span class="pill">默认 GPU + Eevee</span>
    <span class="pill">CPU / Cycles 可选</span>
    <span class="pill">逐批续渲</span>
    <span class="pill">图片 / 视频</span>
    <span class="pill">API + MCP</span>
  </div>
</section>
"""


def _invoke(function: Callable[..., T], *args: Any) -> T:
    """把服务异常转换成可读的 Gradio / API 错误，保留后台诊断日志。"""
    try:
        return function(*args)
    except gr.Error:
        raise
    except Exception as exc:
        LOGGER.exception("Render service request failed: %s", function.__name__)
        raise gr.Error(str(exc) or "操作失败，请查看 Space 日志。") from exc


def _job_id(value: str) -> str:
    """任务 ID 也是访问任务的凭据，禁止将空字符串传入服务。"""
    value = (value or "").strip()
    if not value:
        raise gr.Error("请先准备任务，或填入之前保存的任务 ID。")
    return value


def environment_info() -> dict[str, Any]:
    """Inspect the runtime, Blender installation and selected rendering backend.

    Returns:
        Environment information without allocating a ZeroGPU lease.
    """
    return _invoke(service.environment_info)


def config_schema() -> dict[str, Any]:
    """Get the JSON Schema for every supported render configuration option.

    Returns:
        The authoritative configuration schema, defaults and validation limits.
    """
    return _invoke(service.config_schema)


def prepare_job(
    files: list[str], config_json: str, config_file: str | None = None
) -> tuple[str, dict[str, Any]]:
    """Upload and validate a Blender project and create a resumable render job.

    Args:
        files: Uploaded .blend files, resource ZIP archives and supporting assets.
        config_json: Render configuration encoded as a JSON object.
        config_file: Optional uploaded JSON configuration file.

    Returns:
        A private job ID and the initial job status. Keep the ID for later calls.
    """
    if not files:
        raise gr.Error("请上传 .blend 工程，或包含工程及资源的 ZIP 文件。")
    return _invoke(service.prepare_job, files, config_json or "{}", config_file)


def inspect_job(job_id: str) -> dict[str, Any]:
    """Inspect a Blender scene, cameras and referenced assets before rendering.

    Args:
        job_id: The private job ID returned by prepare_job.

    Returns:
        Scene inspection results and the current job status.
    """
    return _invoke(service.inspect_job, _job_id(job_id))


def render_chunk(job_id: str) -> dict[str, Any]:
    """Render the next bounded batch and retain completed frames for resuming.

    Args:
        job_id: The private job ID returned by prepare_job.

    Returns:
        Updated job status. Call again while pending frames remain, then finalize.
    """
    return _invoke(service.render_chunk, _job_id(job_id))


def _render_remaining(job_id: str) -> Iterator[dict[str, Any]]:
    """界面专用连续模式：按所选后端逐批渲染，错误或取消后立即结束。"""
    job_id = _job_id(job_id)
    current = _invoke(service.job_status, job_id)
    yield current
    halted = {"cancelled", "cancelling", "canceled", "failed", "error"}
    if current.get("status") in {"cancelled", "cancelling", "canceled"}:
        gr.Warning("任务已停止，请先点击「恢复任务」，再启动连续渲染。")
        return
    if current.get("error") or current.get("status") in halted:
        gr.Warning("任务存在错误。请先检查日志并处理原因，再恢复任务；连续模式不会自动重试。")
        return
    if current.get("status") == "rendering":
        gr.Warning("此任务正在渲染，请等待当前批次完成。")
        return

    while current.get("remaining_frames"):
        # 每轮先检查取消标志；绝不自动清除标志或重试配额错误。
        current = _invoke(service.job_status, job_id)
        if current.get("error") or current.get("status") in halted:
            yield current
            return
        remaining = current.get("remaining_frames", [])
        if not remaining:
            yield current
            return
        previous_count = len(remaining)
        current = _invoke(service.render_chunk, job_id)
        yield current
        if current.get("error") or current.get("status") in halted:
            return
        pending = current.get("remaining_frames", [])
        if pending and len(pending) >= previous_count:
            # 避免异常状态无进展却反复租用 GPU。
            gr.Warning("本批次未减少剩余帧，连续渲染已停止。请检查任务状态与日志。")
            return


def job_status(job_id: str) -> dict[str, Any]:
    """Read render progress without waiting for the Blender execution queue.

    Args:
        job_id: The private job ID returned by prepare_job.

    Returns:
        Current job state and completed-frame progress.
    """
    return _invoke(service.job_status, _job_id(job_id))


def cancel_job(job_id: str) -> dict[str, Any]:
    """Request cancellation while preserving already completed render frames.

    Args:
        job_id: The private job ID returned by prepare_job.

    Returns:
        Cancellation status. Use job_status to observe the final state.
    """
    return _invoke(service.cancel_job, _job_id(job_id))


def resume_job(job_id: str) -> dict[str, Any]:
    """Resume a stopped job while retaining successfully rendered frames.

    Args:
        job_id: The private job ID returned by prepare_job.

    Returns:
        Updated job state. Call render_chunk to render the next batch.
    """
    return _invoke(service.resume_job, _job_id(job_id))


def finalize_job(
    job_id: str,
) -> tuple[dict[str, Any], list[str], str | None, str | None]:
    """Package rendered frames and encode a video when the job is complete.

    Args:
        job_id: The private job ID returned by prepare_job.

    Returns:
        A result summary, downloadable files, optional image and optional video.
    """
    return _invoke(service.finalize_job, _job_id(job_id))


def cpu_self_test() -> tuple[dict[str, Any], list[str], str | None]:
    """Verify default CPU Eevee rendering without allocating a GPU."""
    return _invoke(service.cpu_self_test)


def gpu_self_test() -> tuple[dict[str, Any], list[str], str | None]:
    """Test actual Blender GPU rendering with a small built-in scene.

    Returns:
        Diagnostic results, downloadable logs and an optional test render.
        A failed GPU test does not silently switch the render job to CPU.
    """
    return _invoke(service.gpu_self_test)


def eevee_gpu_self_test(gpu_size: str = "large", graphics_backend: str = "opengl") -> tuple[dict[str, Any], list[str], str | None]:
    """Check GPU Eevee with a real render in the current server environment."""
    return _invoke(service.eevee_gpu_self_test, gpu_size, graphics_backend)


def _preset(mode: str) -> str:
    """从同一配置模型生成示例，避免界面默认值与后端校验发生偏差。"""
    data = RenderConfig().model_dump(mode="json")
    data["mode"] = mode
    return json.dumps(data, indent=2, ensure_ascii=False)


def _render_options(config_json: str, engine: str, device: str, gpu_size: str = "large", graphics_backend: str = "opengl") -> str:
    # Update only render choices; retain camera, animation and quality settings.
    data = json.loads(config_json or "{}")
    data.update(engine=engine, device=device, gpu_size=gpu_size, graphics_backend=graphics_backend)
    return parse_config(json.dumps(data)).model_dump_json(indent=2)


def build_app() -> gr.Blocks:
    """构建界面；导入本模块不会下载 Blender，也不会申请 GPU。"""
    with gr.Blocks(title="Blender Render Space", delete_cache=(3600, 86400)) as demo:
        gr.HTML(HERO)
        gr.Markdown(
            "**默认 GPU + Eevee**，CPU 和 Cycles 仍可选择。"
            "开始正式任务前，请先运行「GPU / Eevee 自检」确认实际出图。"
            "引擎和计算设备可以自由选择。GPU + Eevee 是否可用取决于服务器的图形接口，"
            "可在「环境自检」中检测；不需要用户在自己电脑安装硬件。",
            elem_classes="workflow-note",
        )

        with gr.Tabs():
            with gr.Tab("渲染工作台"):
                with gr.Row():
                    with gr.Column(scale=5, min_width=320):
                        gr.Markdown("### 01 · 上传工程", elem_classes="section-caption")
                        upload = gr.File(
                            label="Blender 工程与资源",
                            file_count="multiple",
                            type="filepath",
                            height=190,
                        )
                        gr.Markdown(
                            "推荐上传**包含 .blend 与纹理目录的 ZIP**，保留相对路径；"
                            "也可以先在 Blender 中打包外部资源，再上传 .blend。"
                            "多个单独上传文件的目录结构不会自动保留。"
                        )
                        config_upload = gr.File(
                            label="渲染配置 JSON（可选）",
                            file_types=[".json"],
                            type="filepath",
                        )
                        gr.Markdown("### 02 · 设置渲染", elem_classes="section-caption")
                        with gr.Row():
                            image_preset = gr.Button("图片配置", size="sm")
                            video_preset = gr.Button("视频配置", size="sm")
                        config_editor = gr.Code(
                            value=_preset("image"),
                            language="json",
                            label="JSON 配置编辑器",
                            lines=19,
                            max_lines=32,
                            interactive=True,
                        )
                        gr.Markdown(
                            "上传 JSON 文件后，将使用文件中的配置。"
                            "使用编辑器时，请先清空上方 JSON 上传框。"
                            "`batch_size` 控制每次提交的帧数；建议先用低分辨率、"
                            "少量采样验证，再提高画质。"
                        )
                        with gr.Row():
                            engine_choice = gr.Dropdown([("Eevee（默认）", "EEVEE"), ("Cycles", "CYCLES")], value="EEVEE", label="渲染引擎")
                            device_choice = gr.Dropdown([("GPU（默认）", "AUTO"), ("CPU", "CPU")], value="AUTO", label="计算设备")
                        with gr.Row():
                            gpu_size_choice = gr.Dropdown([("large · 48 GB", "large"), ("xlarge · 96 GB · 2 倍配额", "xlarge")], value="large", label="ZeroGPU 规格")
                            graphics_backend_choice = gr.Dropdown([("OpenGL", "opengl"), ("Vulkan", "vulkan")], value="opengl", label="GPU Eevee 图形接口")
                        apply_options = gr.Button("将引擎与设备应用到 JSON", size="sm")
                        gr.Markdown("选择引擎或设备会自动更新编辑器中的 JSON；创建任务以 JSON 为准。上传 JSON 文件时，请直接修改文件中的 engine / device。GPU + Eevee 可先运行专用自检。")
                        prepare = gr.Button("准备渲染任务", variant="primary", size="lg")

                    with gr.Column(scale=6, min_width=320):
                        gr.Markdown("### 03 · 检查与渲染", elem_classes="section-caption")
                        job = gr.Textbox(
                            label="任务 ID",
                            placeholder="准备后自动填入，也可粘贴已有任务 ID",
                            interactive=True,
                            buttons=["copy"],
                        )
                        gr.Markdown(
                            "保存任务 ID，便于 API 调用和续渲。"
                            "任务 ID 可访问对应结果，请勿公开。Space 重启可能清除临时任务。"
                        )
                        with gr.Row(elem_id="job-actions"):
                            inspect = gr.Button("检查场景与资源")
                            render = gr.Button("渲染下一批", variant="primary")
                        render_remaining = gr.Button("连续渲染剩余帧", variant="primary")
                        with gr.Row():
                            refresh = gr.Button("刷新进度", size="sm")
                            cancel = gr.Button("停止任务", variant="stop", size="sm")
                            resume = gr.Button("恢复任务", size="sm")
                        status = gr.JSON(label="任务状态 / 场景检查", value={"status": "请先准备任务"})
                        gr.Markdown(
                            "每次点击「渲染下一批」处理一个批次；剩余帧可继续提交。"
                            "「连续渲染剩余帧」会逐批提交，遇到错误或配额不足立即结束，"
                            "不会自动重试。**仅选择 GPU 时消耗 GPU 配额；可随时点击「停止任务」。**"
                            "API / Agent 也可按状态自动循环。"
                            "停止后保留已完成帧；点击「恢复任务」后再提交渲染。"
                        )
                        gr.Markdown("### 04 · 导出作品", elem_classes="section-caption")
                        finalize = gr.Button("整理结果 / 合成视频", variant="primary")
                        result = gr.JSON(label="导出结果")
                        downloads = gr.File(
                            label="下载结果与日志", file_count="multiple", interactive=False
                        )
                        with gr.Tabs():
                            with gr.Tab("图片预览"):
                                image = gr.Image(
                                    label="渲染图片",
                                    type="filepath",
                                    interactive=False,
                                    height=340,
                                    buttons=["download", "fullscreen"],
                                )
                            with gr.Tab("视频预览"):
                                video = gr.Video(label="渲染视频", interactive=False, height=340)

            with gr.Tab("环境自检"):
                gr.Markdown(
                    "### 先验证运行环境\n\n"
                    "「读取环境」不会申请 GPU。"
                    "「运行 GPU 自检」会尝试实际渲染一个小场景，可能消耗 ZeroGPU 配额。"
                    "自检失败时，请下载日志检查原因，或选择常驻 GPU Space。\n\n"
                    "后端由 Space 环境变量 `RENDER_BACKEND` 选择："
                    "`auto`、`zerogpu`、`gpu` 或 `cpu`。"
                    "GPU + Eevee 是默认配置，但可用性必须通过实际出图确认。任务 JSON 的 device=CPU 始终绕过 GPU 租约；GPU 不进行静默 CPU 回退。"
                )
                with gr.Row():
                    env_button = gr.Button("读取环境")
                    cpu_probe_button = gr.Button("运行 CPU / Eevee 自检")
                    probe_button = gr.Button("运行 GPU / Cycles 自检")
                    eevee_probe_button = gr.Button("运行 GPU / Eevee 自检", variant="primary")
                gr.Markdown("GPU / Eevee 自检使用工作台当前选择的 ZeroGPU 规格和图形接口；xlarge 消耗 2 倍配额，规格更大不代表具备图形能力。")
                env_output = gr.JSON(label="环境信息")
                with gr.Row():
                    with gr.Column():
                        probe_output = gr.JSON(label="自检报告")
                        probe_files = gr.File(
                            label="下载自检日志", file_count="multiple", interactive=False
                        )
                    probe_image = gr.Image(
                        label="自检渲染图", type="filepath", interactive=False, height=300
                    )

            with gr.Tab("Agent / API"):
                gr.Markdown(
                    "### 可直接调用的渲染工具\n\n"
                    "API 名称与下表一致。点击页面底部 **Use via API** 查看当前 Space "
                    "地址、参数格式与代码示例。所有接口也作为 MCP 工具提供。\n\n"
                    "| API | 用途 |\n| --- | --- |\n"
                    "| `environment_info` | 检查后端与安装信息 |\n"
                    "| `config_schema` | 获取完整 JSON Schema |\n"
                    "| `cpu_self_test` | 验证可选 CPU Eevee 渲染 |\n"
                    "| `eevee_gpu_self_test` | 检测当前服务器的 GPU + Eevee 实际出图能力 |\n"
                    "| `gpu_self_test` | 验证 Blender 实际 GPU 渲染 |\n"
                    "| `prepare_job` | 上传工程和配置，创建任务 |\n"
                    "| `inspect_job` | 检查场景、相机与资源 |\n"
                    "| `render_chunk` | 渲染下一批，保存已完成帧 |\n"
                    "| `job_status` | 不等待渲染队列，读取进度 |\n"
                    "| `cancel_job` | 请求停止，保留已完成帧 |\n"
                    "| `resume_job` | 将停止的任务恢复为可继续状态 |\n"
                    "| `finalize_job` | 导出结果、日志并合成视频 |\n\n"
                    "**建议调用顺序：** 获取配置 Schema → 准备任务 → 检查场景 → "
                    "循环调用 `render_chunk` → 确认完成 → 调用 `finalize_job`。"
                    "遇到错误先读取状态与日志；不要无条件重复申请 GPU。\n\n"
                    "### MCP 连接\n\n"
                    "Streamable HTTP：`https://<你的-Space-域名>.hf.space/gradio_api/mcp/`\n\n"
                    "兼容 SSE：`https://<你的-Space-域名>.hf.space/gradio_api/mcp/sse`\n\n"
                    "MCP 工具使用上传后的文件引用；Agent 本机路径不会自动出现在 Space 中。"
                    "上传本地大文件优先使用 `gradio_client.handle_file`，"
                    "后续操作只需传任务 ID。私有 Space 的客户端还需要 Hugging Face 令牌。"
                )
                schema_button = gr.Button("读取配置 JSON Schema")
                schema_output = gr.JSON(label="配置 Schema")

        # Blender 操作共用一个执行队列，防止多个进程争用同一个 GPU 或输出目录。
        blender_queue = {"concurrency_id": "blender", "concurrency_limit": 1}
        prepare.click(
            prepare_job,
            inputs=[upload, config_editor, config_upload],
            outputs=[job, status],
            api_name="prepare_job",
            **blender_queue,
        )
        inspect.click(inspect_job, job, status, api_name="inspect_job", **blender_queue)
        render.click(render_chunk, job, status, api_name="render_chunk", **blender_queue)
        render_remaining.click(
            _render_remaining,
            job,
            status,
            api_visibility="private",
            **blender_queue,
        )
        # 进度与停止请求不进入渲染队列，否则长渲染会阻塞取消操作。
        refresh.click(job_status, job, status, api_name="job_status", queue=False)
        cancel.click(cancel_job, job, status, api_name="cancel_job", queue=False)
        resume.click(resume_job, job, status, api_name="resume_job", queue=False)
        finalize.click(
            finalize_job,
            job,
            [result, downloads, image, video],
            api_name="finalize_job",
            **blender_queue,
        )
        env_button.click(environment_info, None, env_output, api_name="environment_info", queue=False)
        schema_button.click(config_schema, None, schema_output, api_name="config_schema", queue=False)
        option_inputs = [config_editor, engine_choice, device_choice, gpu_size_choice, graphics_backend_choice]
        for choice in (engine_choice, device_choice, gpu_size_choice, graphics_backend_choice):
            choice.input(_render_options, option_inputs, config_editor,
                         queue=False, api_visibility="private")
        eevee_probe_button.click(eevee_gpu_self_test, [gpu_size_choice, graphics_backend_choice], [probe_output, probe_files, probe_image],
                                 api_name="eevee_gpu_self_test", **blender_queue)
        apply_options.click(_render_options, option_inputs, config_editor,
                            queue=False, api_visibility="private")
        cpu_probe_button.click(cpu_self_test, None, [probe_output, probe_files, probe_image],
                               api_name="cpu_self_test", **blender_queue)
        probe_button.click(
            gpu_self_test,
            None,
            [probe_output, probe_files, probe_image],
            api_name="gpu_self_test",
            **blender_queue,
        )
        # 纯界面快捷操作不作为 Agent 工具暴露。
        image_preset.click(
            lambda: _preset("image"), None, config_editor, queue=False, api_visibility="private"
        )
        video_preset.click(
            lambda: _preset("video"), None, config_editor, queue=False, api_visibility="private"
        )

    return demo.queue(default_concurrency_limit=1, max_size=32)


demo = build_app()


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
    # 安装与解压在申请 GPU 之前完成，避免把下载时间算进 ZeroGPU 租约。
    LOGGER.info("正在准备 Blender 运行时（首次运行需要下载并校验安装包）")
    binary = ensure_blender()
    LOGGER.info("Blender 运行时已就绪：%s", binary)
    service.PUBLIC_ROOT.mkdir(parents=True, exist_ok=True)
    demo.launch(
        server_name=os.environ.get("GRADIO_SERVER_NAME", "0.0.0.0"),
        server_port=int(os.environ.get("PORT", os.environ.get("GRADIO_SERVER_PORT", "7860"))),
        mcp_server=True,
        max_file_size="1gb",
        theme=gr.themes.Soft(primary_hue="teal", neutral_hue="slate"),
        css=CSS,
        # 仅发布服务显式导出的文件；不允许浏览整个任务目录或上传源文件。
        allowed_paths=[str(service.PUBLIC_ROOT.resolve())],
        footer_links=["api"],
        show_error=False,
    )