feat: 试用过程中的细节优化

server/services/generator.py

@@ -10,6 +10,7 @@ from __future__ import annotations
 
 import asyncio
 import base64
+import io
 import logging
 import mimetypes
 import re
@@ -17,6 +18,7 @@ from dataclasses import dataclass
 from typing import Awaitable, Callable
 
 import httpx
+from PIL import Image, ImageOps
 
 from config import get_settings
 from services.prompts import build_context, build_prompt, type_name
@@ -84,8 +86,8 @@ def _image_size(provider: str, ratio: str, is_wan: bool = True, model: str = "")
         # gpt-image 自定义尺寸约束：16 的倍数、长短边比 ≤ 3:1（1536x2048 合法）
         return "1536x2048" if ratio == "3:4" else "2048x2048"
     if provider == "pcore":
-        # pcore 的 size 是枚举（默认 1024x1024），不接受自定义尺寸；用 gpt-image 枚举：竖版 1024x1536（近似 3:4）
-        return "1024x1536" if ratio == "3:4" else "1024x1024"
+        # 竖图必须与 Ozon 的 3:4 规格一致。1024x1536 实际是 2:3，会使成图明显偏瘦高。
+        return "1152x1536" if ratio == "3:4" else "1024x1024"
     # tongyi：万象与千问的 size 语法相同（* 分隔），档位不同
     # wan2.6 系列总像素限制在 [1280², 1440²]，wan2.7 的 1536*2048/2048*2048 会超限
     if model.startswith("wan2.6"):
@@ -141,6 +143,75 @@ async def _resolve_ref(url: str) -> str:
     return _bytes_to_data_uri(data, mime)
 
 
+@dataclass(frozen=True)
+class OutpaintCanvas:
+    """用于受限扩图的预处理结果。
+
+    ``image_data_url`` 与 ``mask_data_url`` 尺寸严格一致：蒙版不透明的区域是原图，
+    透明区域才允许供应商生成。最终仍会把 ``original`` 覆盖回结果，防止供应商在
+    蒙版区域有轻微重绘。
+    """
+
+    image_data_url: str
+    mask_data_url: str
+    original: Image.Image
+    box: tuple[int, int, int, int]
+    size: tuple[int, int]
+
+    def restore_original(self, generated: bytes) -> bytes:
+        """恢复原图区域，确保输出中原图像素不受模型影响。"""
+        with Image.open(io.BytesIO(generated)) as result:
+            result = ImageOps.exif_transpose(result).convert("RGBA")
+            if result.size != self.size:
+                # 供应商偶发不遵从 size 时仍输出固定 3:4；随后会精确覆盖原图区域。
+                result = ImageOps.fit(result, self.size, method=Image.Resampling.LANCZOS)
+            result.alpha_composite(self.original, self.box[:2])
+            out = io.BytesIO()
+            result.save(out, format="PNG")
+            return out.getvalue()
+
+
+def _png_data_url(image: Image.Image) -> str:
+    out = io.BytesIO()
+    image.save(out, format="PNG")
+    return _bytes_to_data_uri(out.getvalue(), "image/png")
+
+
+async def prepare_outpaint_canvas(url: str, size: str) -> OutpaintCanvas:
+    """把单张参考图置于目标画布，并生成仅开放新增边缘的 PNG alpha 蒙版。"""
+    try:
+        target_width, target_height = (int(value) for value in size.lower().split("x", 1))
+    except (TypeError, ValueError) as exc:
+        raise ValueError(f"仅扩图需要合法的宽高尺寸，收到: {size}") from exc
+    if target_width <= 0 or target_height <= 0:
+        raise ValueError(f"仅扩图需要正数宽高尺寸，收到: {size}")
+
+    data, _ = await _resolve_ref_bytes(url)
+    with Image.open(io.BytesIO(data)) as opened:
+        source = ImageOps.exif_transpose(opened).convert("RGBA")
+        if source.width <= 0 or source.height <= 0:
+            raise ValueError("参考底图尺寸无效")
+        scale = min(target_width / source.width, target_height / source.height)
+        source_width = max(1, round(source.width * scale))
+        source_height = max(1, round(source.height * scale))
+        original = source.resize((source_width, source_height), Image.Resampling.LANCZOS)
+
+    x = (target_width - original.width) // 2
+    y = (target_height - original.height) // 2
+    canvas = Image.new("RGBA", (target_width, target_height), (0, 0, 0, 0))
+    canvas.alpha_composite(original, (x, y))
+    # OpenAI/GPT Image 兼容蒙版约定：透明像素是可编辑区域；原图区域不透明。
+    mask = Image.new("RGBA", (target_width, target_height), (0, 0, 0, 0))
+    mask.paste((255, 255, 255, 255), (x, y, x + original.width, y + original.height))
+    return OutpaintCanvas(
+        image_data_url=_png_data_url(canvas),
+        mask_data_url=_png_data_url(mask),
+        original=original,
+        box=(x, y, x + original.width, y + original.height),
+        size=(target_width, target_height),
+    )
+
+
 # ── Provider：豆包 Seedream（火山方舟）────────────────────────────────────
 
 async def generate_doubao(prompt: str, ref_images: list[str], size: str = "2048x2048", model: str | None = None) -> bytes:
@@ -562,6 +633,7 @@ async def advance_generation(
     model: str,
     provider_task_id: str | None = None,
     provider_task_kind: str | None = None,
+    edit_mask: str | None = None,
 ) -> GenerationAdvance:
     """推进一次图片生成，不在本函数内无限轮询。
 
@@ -569,6 +641,8 @@ async def advance_generation(
     能继续外部任务，且在等待期间可以领取其他任务。豆包和同步千问仍在一次调用内完成。
     """
     settings = get_settings()
+    if edit_mask and provider != "pcore":
+        raise RuntimeError("带蒙版的仅扩图当前仅支持 Pcore gpt-image-2")
     if provider == "doubao":
         return GenerationAdvance(data=await generate_doubao(prompt, ref_images, size=size, model=model))
 
@@ -650,6 +724,10 @@ async def advance_generation(
             body: dict = {"model": model, "prompt": prompt, "n": 1, "size": size, "async": True}
             if ref_images:
                 body["images"] = [{"image_url": await _resolve_ref(item)} for item in ref_images]
+            if edit_mask:
+                # Pcore 的图片参数使用 JSON {image_url}；蒙版沿用相同的 OpenAI 兼容对象形式。
+                # 透明区域允许编辑，不透明区域保留。若供应商轻微偏移，Worker 会二次覆盖原图。
+                body["mask"] = {"image_url": edit_mask}
             response = await client.post(
                 f"{base}/v1/images/{kind}", headers={**headers, "Content-Type": "application/json"}, json=body,
             )

server/services/job_worker.py

@@ -10,7 +10,7 @@ from config import get_settings
 from db import get_session_factory
 from models import ProductAsset
 from services.image_assets import get_image_asset_service
-from services.generator import _image_size, _refs_for_job, advance_generation
+from services.generator import OutpaintCanvas, _image_size, _refs_for_job, advance_generation, prepare_outpaint_canvas
 from services.jobs import (
     ClaimedItem,
     claim_next_item,
@@ -22,6 +22,7 @@ from services.jobs import (
     recover_expired_leases,
 )
 from services.prompts import build_context, build_prompt
+from services.prompts.common import requirements_block
 from services.storage import get_storage
 from services.suite_service import append_generated_asset
 from services.tasks import TaskImage
@@ -152,6 +153,22 @@ async def _process_slideshow_video_item(claim: ClaimedItem) -> None:
         await mark_item_failed(claim.item_id, str(exc))
 
 
+def _outpaint_prompt(requirements: str | None) -> str:
+    """仅扩图不复用套图提示词，避免默认风格、营销构图或文字污染边缘生成。"""
+    requested = (requirements or "").strip()
+    if not requested:
+        requested = "Continue the existing surroundings naturally and seamlessly."
+    return "\n\n".join([
+        requirements_block(requested),
+        "OUTPAINT ONLY: Image 1 contains an original photo placed on a 3:4 canvas. Generate pixels only "
+        "in the transparent masked margins outside the original photo. Do not crop, move, resize, redraw, "
+        "retouch, recolor, relight, or otherwise alter any pixel of the original photo.",
+        "Do not add marketing layouts, product captions, overlays, props, or visible text unless the STRICT "
+        "REQUIREMENTS explicitly ask for them. Make the new margins blend naturally with the original edges.",
+        "FINAL CHECK: output exactly 3:4 and change only the newly expanded margins.",
+    ])
+
+
 async def process_item(claim: ClaimedItem) -> None:
     """推进一次子任务。外部模型未完成时只写 provider id 并退出，让出 Worker。"""
     if claim.job_kind == "watermark_batch":
@@ -174,18 +191,34 @@ async def process_item(claim: ClaimedItem) -> None:
         "pcore": "gpt-image-2",
     }.get(provider, settings.ark_image_model)
     try:
-        context = build_context(payload.get("context") or {}, fallback_name="")
-        prompt_extra = job_item.get("prompt_extra") or job_item
-        prompt = build_prompt(
-            provider, model, job_item.get("kind", "custom"), context, claim.style_set, claim.lang,
-            extra=prompt_extra, style_prompt=payload.get("style_prompt"), requirements=payload.get("requirements"),
-        )
         refs = _refs_for_job(list(payload.get("ref_images") or []), job_item)
+        operation = payload.get("operation", "edit")
+        target_size = _image_size(provider, claim.ratio, is_wan=provider == "tongyi" and model.lower().startswith("wan"), model=model)
+        edit_mask: str | None = None
+        outpaint: OutpaintCanvas | None = None
+        if operation == "outpaint":
+            if len(refs) != 1:
+                raise RuntimeError("仅扩图必须且只能使用一张参考底图")
+            prompt = _outpaint_prompt(payload.get("requirements"))
+            # 首次提交时才构造图片和蒙版；之后轮询不再向供应商重复上传。
+            if not claim.provider_task_id:
+                outpaint = await prepare_outpaint_canvas(refs[0], target_size)
+                refs = [outpaint.image_data_url]
+                edit_mask = outpaint.mask_data_url
+        else:
+            context = build_context(payload.get("context") or {}, fallback_name="")
+            prompt_extra = job_item.get("prompt_extra") or job_item
+            prompt = build_prompt(
+                provider, model, job_item.get("kind", "custom"), context, claim.style_set, claim.lang,
+                extra=prompt_extra, style_prompt=payload.get("style_prompt"), requirements=payload.get("requirements"),
+                requirements_exclusive=bool(payload.get("requirements_exclusive")),
+            )
         is_wan = provider == "tongyi" and model.lower().startswith("wan")
         result = await advance_generation(
             provider, prompt, refs,
-            size=_image_size(provider, claim.ratio, is_wan=is_wan, model=model), model=model,
+            size=target_size, model=model,
             provider_task_id=claim.provider_task_id, provider_task_kind=claim.provider_task_kind,
+            edit_mask=edit_mask,
         )
         if result.data is None:
             if not result.provider_task_id or not result.provider_task_kind:
@@ -203,6 +236,11 @@ async def process_item(claim: ClaimedItem) -> None:
             await mark_item_cancelled(claim.item_id)
             return
 
+        if operation == "outpaint":
+            # 异步任务完成时本轮未必创建过画布，按同一源图与尺寸确定性重建后覆盖原图区域。
+            outpaint = outpaint or await prepare_outpaint_canvas(refs[0], target_size)
+            result = type(result)(data=outpaint.restore_original(result.data))
+
         watermark = payload.get("watermark") or {}
         position = int(job_item.get("position", 0))
         if watermark.get("enabled"):

server/services/planner.py

@@ -56,11 +56,11 @@ SYSTEM_PROMPT = """你是一名资深电商视觉策划。根据商品信息规
 5. variant_name 只能从「SKU规格」列表原样照抄；没有绑定就输出 null。
 
 ## 规划规则
-1. 套图必须均衡覆盖：白底主图、场景图、细节图；参数含可靠的长宽高/尺寸时还必须有尺寸标注图。按这个顺序先预留张数，再安排其他类型。
+1. 套图必须均衡覆盖：白底主图、场景图、细节图；参数含可靠的任一方向尺寸时还必须有尺寸标注图。按这个顺序先预留张数，再安排其他类型。
 2. 白底主图（kind=white_bg）通常只出 1 张；有多个视觉差异明显的 SKU 时最多选 2 个代表款各出 1 张，variant_name 填对应规格名。**禁止因为 SKU 多而为每个 SKU 都规划主图**。仅当「生图要求」明确要求每个/全部 SKU 分别出图时才可例外。
 3. 场景图（kind=lifestyle）：至少 1 张，按核心使用场景出 1-2 张，每张聚焦一个场景。
 4. 细节图（kind=material / key_features / selling_pt / ecommerce_detail）：至少 1 张，展示关键细节、材质或结构；通常 1-2 张，每张聚焦一个卖点。
-5. 尺寸标注图（kind=size_chart）：参数含长宽高/尺寸时必须出 1 张；无可靠尺寸参数时不要虚构尺寸。
+5. 尺寸标注图（kind=size_chart）：参数含任一可靠尺寸时必须出 1 张；只标注已知方向，无可靠尺寸参数时不要虚构尺寸。
 6. SKU 合集图（kind=sku_collection）：仅在 SKU >1 且确有必要展示差异时出 1 张，优先级低于上述四类。
 7. kind 枚举：white_bg / key_features / selling_pt / material / lifestyle / multi_scene / ecommerce_detail / size_chart / sku_collection / custom。
 8. title 用中文短语（如「主图·粉色」「浴室壁挂」）；detail 中文说明这张图展示什么；prompt_hint 用英文描述构图要点。
@@ -124,9 +124,9 @@ def _normalize_items(raw_items: list, sku_variants: list[str]) -> list[dict]:
 
 
 def _has_dimensions(product_info: dict) -> bool:
-    """仅在已采集到尺寸信息时规划尺寸图，避免模型编造数值。"""
+    """有任一可靠尺寸即可规划尺寸图；未给出的方向不允许模型编造。"""
     dimensions = product_info.get("dimensions") or {}
-    if isinstance(dimensions, dict) and all(dimensions.get(key) is not None for key in ("length", "width", "height")):
+    if isinstance(dimensions, dict) and any(dimensions.get(key) is not None for key in ("length", "width", "height")):
         return True
     for param in product_info.get("params") or []:
         if not isinstance(param, dict):

server/services/prompts/__init__.py

@@ -36,10 +36,11 @@ def prompt_family(provider: str, model: str | None) -> str:
 
 def build_prompt(provider: str, model: str | None, type_id: str, ctx: dict, style_set: int,
                  lang: str, extra: dict | None = None, style_prompt: str | None = None,
-                 requirements: str | None = None) -> str:
+                 requirements: str | None = None, requirements_exclusive: bool = False) -> str:
     """按模型家族构造指定图类型的完整生图 prompt。参数含义见各家族 build_prompt。"""
     module = _MODULE_BY_FAMILY[prompt_family(provider, model)]
     return module.build_prompt(
         type_id, ctx, style_set, lang,
         extra=extra, style_prompt=style_prompt, requirements=requirements,
+        requirements_exclusive=requirements_exclusive,
     )

server/services/prompts/alibaba.py

@@ -96,12 +96,13 @@ def _prompt_ecommerce_detail(ctx: dict, style: dict, lang: str) -> str:
 
 def _prompt_size_chart(ctx: dict, style: dict, lang: str) -> str:
     dims = params_line_for_output(ctx, lang)
+    annotation = ctx.get("dimension_annotation_instruction") or "Do not invent measurement directions or values."
     return (
         f"Product size chart infographic for \"{ctx['title']}\": product shown in clean front and side views "
-        f"on light background, with thin measurement annotation lines (arrows) marking length, width and height, "
+        f"on light background, with thin measurement annotation lines (arrows) only for supplied dimensions, "
         f"measurement values rendered next to each line"
         + (f" (known specs: {dims})" if dims else "")
-        + f", small caption row, precise technical drawing aesthetic. {style['tone']}. "
+        + f". {annotation} Small caption row, precise technical drawing aesthetic. {style['tone']}. "
         f"{TEXT_RENDER[lang]} {QUALITY} {PRODUCT_REF_LOCK}"
     )
 
@@ -136,7 +137,8 @@ _PROMPT_BUILDERS = {
 
 
 def build_prompt(type_id: str, ctx: dict, style_set: int, lang: str, extra: dict | None = None,
-                 style_prompt: str | None = None, requirements: str | None = None) -> str:
+                 style_prompt: str | None = None, requirements: str | None = None,
+                 requirements_exclusive: bool = False) -> str:
     """构造指定图类型的完整生图 prompt（主体参考语义）。
 
     extra: 方案项信息 {title, detail, prompt_hint}——custom 类型必需，
@@ -145,6 +147,15 @@ def build_prompt(type_id: str, ctx: dict, style_set: int, lang: str, extra: dict
     requirements: 生图要求（最高优先级，强制约束），置于 prompt 最前面，
     声明覆盖一切冲突指令，用户可在此输入强制要求。
     """
+    if requirements_exclusive:
+        return " ".join([
+            requirements_block(requirements),
+            "Execute only the strict requirements above. Do not add marketing layouts, captions, overlays, "
+            "props, default studio styling, or visible text unless explicitly requested.",
+            "Preserve the reference image exactly unless the strict requirements explicitly request a "
+            "specific change. When they do, make only that requested change and preserve everything else.",
+            "Do not make any change that was not explicitly required.",
+        ])
     style = resolve_style(style_set, style_prompt)
     extra = extra or {}
     if type_id == "custom":

server/services/prompts/common.py

@@ -134,11 +134,28 @@ def build_context(raw: dict, fallback_name: str = "", fallback_desc: str = "") -
                 break
 
     dimensions = raw.get("dimensions") or {}
-    dimension_values = (dimensions.get("length"), dimensions.get("width"), dimensions.get("height")) if isinstance(dimensions, dict) else ()
-    dimensions_line = ""
-    if len(dimension_values) == 3 and all(value is not None for value in dimension_values):
-        unit = dimensions.get("unit") or "mm"
-        dimensions_line = "L × W × H: " + " × ".join(str(value) for value in dimension_values) + f" {unit}"
+    known_dimensions: list[tuple[str, object]] = []
+    if isinstance(dimensions, dict):
+        known_dimensions = [
+            (label, dimensions[key]) for key, label in (("length", "length"), ("width", "width"), ("height", "height"))
+            if dimensions.get(key) is not None
+        ]
+    unit = (dimensions.get("unit") or "mm") if isinstance(dimensions, dict) else "mm"
+    dimensions_line = "; ".join(f"{label}: {value} {unit}" for label, value in known_dimensions)
+    known_labels = [label for label, _ in known_dimensions]
+    missing_labels = [label for label in ("length", "width", "height") if label not in known_labels]
+    dimension_annotation_instruction = (
+        "STRICT MEASUREMENT OUTPUT RULE: render exactly "
+        + str(len(known_labels))
+        + " measurement arrows and labels, only for: "
+        + ", ".join(known_labels)
+        + ". The only permitted dimension values are: "
+        + dimensions_line
+        + ". Do not infer product measurements. Do not draw, label, or show an arrow for: "
+        + ", ".join(missing_labels or ["any additional direction"])
+        + ", depth, thickness, diameter, or any other unlisted dimension."
+        if known_dimensions else ""
+    )
     params_line = "; ".join(
         f"{p.get('key')}: {p.get('value')}" for p in (raw.get("params") or [])[:8]
     )
@@ -158,6 +175,7 @@ def build_context(raw: dict, fallback_name: str = "", fallback_desc: str = "") -
             dimensions_line,
         ))),
         "price": raw.get("price") or "",
+        "dimension_annotation_instruction": dimension_annotation_instruction,
     }
 
 def selling_point_lines(ctx: dict, lang: str, max_n: int = 3) -> str:

server/services/prompts/google.py

@@ -98,12 +98,13 @@ def _task_ecommerce_detail(ctx: dict, lang: str) -> str:
 
 def _task_size_chart(ctx: dict, lang: str) -> str:
     dims = params_line_for_output(ctx, lang)
+    annotation = ctx.get("dimension_annotation_instruction") or "Do not invent measurement directions or values."
     return (
         "Create a size chart: the unchanged product in clean front and side views on a light background, "
-        "thin measurement annotation lines (arrows) marking length, width and height with values beside "
+        "thin measurement annotation lines (arrows) only for supplied dimensions with values beside "
         "each line"
         + (f" (known specs: {dims})" if dims else "")
-        + ", small caption row, precise technical-drawing aesthetic."
+        + f". {annotation} Small caption row, precise technical-drawing aesthetic."
     )
 
 def _task_sku_collection(ctx: dict, lang: str) -> str:
@@ -128,8 +129,19 @@ _TASK_BUILDERS = {
 
 
 def build_prompt(type_id: str, ctx: dict, style_set: int, lang: str, extra: dict | None = None,
-                 style_prompt: str | None = None, requirements: str | None = None) -> str:
+                 style_prompt: str | None = None, requirements: str | None = None,
+                 requirements_exclusive: bool = False) -> str:
     """构造指定图类型的 prompt：要求块 → 指令 → 主体锁 → 锚定 → 风格 → 文案 → 画质 → 提醒。"""
+    if requirements_exclusive:
+        return "\n\n".join([
+            requirements_block(requirements),
+            "Edit the first image only as required above. Do not add marketing layouts, captions, overlays, "
+            "props, default studio styling, or visible text unless explicitly requested above.",
+            "PROTECTION: preserve the first image exactly unless the strict requirements explicitly request "
+            "a specific change. When they do, make only that requested change and preserve everything else.",
+            "Final check: follow the strict requirements over every default behavior; do not make changes "
+            "that were not explicitly required.",
+        ])
     style = resolve_style(style_set, style_prompt)
     extra = extra or {}
     hint = (extra.get("prompt_hint") or "").strip()

server/services/prompts/gpt.py

@@ -129,12 +129,13 @@ def _task_ecommerce_detail(ctx: dict, lang: str) -> str:
 
 def _task_size_chart(ctx: dict, lang: str) -> str:
     dims = params_line_for_output(ctx, lang)
+    annotation = ctx.get("dimension_annotation_instruction") or "Do not invent measurement directions or values."
     return (
         "TASK: Measurement chart. Show the unchanged product in clean front and side views on a light "
-        "background; overlay thin technical annotation lines (arrows) marking length, width and height "
-        "with measurement values rendered beside each line"
+        "background; overlay thin technical annotation lines (arrows) only for the supplied dimensions, "
+        "with each measurement value rendered beside its line"
         + (f" (known specs: {dims})" if dims else "")
-        + "; precise technical-drawing aesthetic, small caption row."
+        + f". {annotation}; precise technical-drawing aesthetic, small caption row."
     )
 
 def _task_sku_collection(ctx: dict, lang: str) -> str:
@@ -160,8 +161,21 @@ _TASK_BUILDERS = {
 
 
 def build_prompt(type_id: str, ctx: dict, style_set: int, lang: str, extra: dict | None = None,
-                 style_prompt: str | None = None, requirements: str | None = None) -> str:
+                 style_prompt: str | None = None, requirements: str | None = None,
+                 requirements_exclusive: bool = False) -> str:
     """构造指定图类型的 edits 语义 prompt：要求块 → 编辑指令 → 保真锁 → 锚定 → 风格 → 文案 → 画质 → 终检。"""
+    if requirements_exclusive:
+        # 单张 AI 编辑：用户要求是唯一的画面意图，绝不能被套图的营销图/棚拍/俄文文案覆盖。
+        return "\n\n".join([
+            requirements_block(requirements),
+            "TASK: Edit Image 1 only according to the STRICT REQUIREMENTS above. Do not add marketing "
+            "layouts, captions, overlays, props, default studio styling, or any visible text unless the "
+            "STRICT REQUIREMENTS explicitly ask for them.",
+            "PROTECTION: preserve Image 1 exactly unless the STRICT REQUIREMENTS explicitly request a "
+            "specific change. When they do, make only that requested change and preserve everything else.",
+            "FINAL CHECK: follow the STRICT REQUIREMENTS over every default behavior. Preserve the product "
+            "and do not make any change that is not explicitly required.",
+        ])
     style = resolve_style(style_set, style_prompt)
     extra = extra or {}
     hint = (extra.get("prompt_hint") or "").strip()

server/services/video_renderer.py

@@ -58,11 +58,16 @@ async def _run_process(
     should_cancel: Callable[[], Awaitable[bool]],
     timeout_seconds: int,
 ) -> None:
-    process = await asyncio.create_subprocess_exec(
-        *args,
-        stdout=asyncio.subprocess.DEVNULL,
-        stderr=asyncio.subprocess.PIPE,
-    )
+    try:
+        process = await asyncio.create_subprocess_exec(
+            *args,
+            stdout=asyncio.subprocess.DEVNULL,
+            stderr=asyncio.subprocess.PIPE,
+        )
+    except FileNotFoundError as exc:
+        raise VideoRenderError(
+            f"未找到视频合成工具「{args[0]}」。请安装 FFmpeg，并配置 VIDEO_FFMPEG_PATH 为其可执行文件的绝对路径。"
+        ) from exc
     try:
         waited = 0.0
         while process.returncode is None:
@@ -91,13 +96,18 @@ async def _run_process(
 
 async def _probe(path: Path) -> dict:
     settings = get_settings()
-    process = await asyncio.create_subprocess_exec(
-        settings.video_ffprobe_path,
-        "-v", "error", "-show_entries", "format=duration,size:stream=codec_name,width,height,pix_fmt",
-        "-of", "json", str(path),
-        stdout=asyncio.subprocess.PIPE,
-        stderr=asyncio.subprocess.PIPE,
-    )
+    try:
+        process = await asyncio.create_subprocess_exec(
+            settings.video_ffprobe_path,
+            "-v", "error", "-show_entries", "format=duration,size:stream=codec_name,width,height,pix_fmt",
+            "-of", "json", str(path),
+            stdout=asyncio.subprocess.PIPE,
+            stderr=asyncio.subprocess.PIPE,
+        )
+    except FileNotFoundError as exc:
+        raise VideoRenderError(
+            f"未找到视频校验工具「{settings.video_ffprobe_path}」。请安装 FFmpeg，并配置 VIDEO_FFPROBE_PATH 为其可执行文件的绝对路径。"
+        ) from exc
     stdout, stderr = await process.communicate()
     if process.returncode != 0:
         raise VideoRenderError(f"无法校验视频输出：{stderr.decode('utf-8', errors='replace')[-400:]}")