feat(cache): 增强缓存锁机制支持进程存活检测

- 添加了锁元数据写入和读取功能，记录进程ID和启动时间
- 实现了进程存活检查机制，防止PID复用导致的死锁
- 引入了过期锁检测和自动清理机制
- 集成了psutil库进行系统进程监控
- 优化了缓存清理逻辑，支持跳过活跃锁文件
- 使用JSON格式存储锁元数据信息

.env.example

@@ -1,13 +1,65 @@
-TEMPLATE_DIR=template/
-API_ENDPOINT=https://zhentuai.com/task/v1
+# ===================
+# API 配置
+# ===================
+API_ENDPOINT=http://127.0.0.1:18084/api
 ACCESS_KEY=TEST_ACCESS_KEY
+WORKER_ID=1
+
+# ===================
+# 目录配置
+# ===================
 TEMP_DIR=tmp/
-#REDIRECT_TO_URL=https://renderworker-deuvulkhes.cn-shanghai.fcapp.run/
-# QSV
-ENCODER_ARGS="-c:v h264_qsv -global_quality 28 -look_ahead 1"
-# NVENC
-#ENCODER_ARGS="-c:v h264_nvenc -cq:v 24 -preset:v p7 -tune:v hq -profile:v high"
-# HEVC
-#VIDEO_ARGS="-profile:v main
-UPLOAD_METHOD="rclone"
-RCLONE_REPLACE_MAP="https://oss.zhentuai.com|alioss://frametour-assets,https://frametour-assets.oss-cn-shanghai.aliyuncs.com|alioss://frametour-assets"
+
+# ===================
+# 并发与调度
+# ===================
+#MAX_CONCURRENCY=4                    # 最大并发任务数
+#HEARTBEAT_INTERVAL=5                 # 心跳间隔（秒）
+#LEASE_EXTENSION_THRESHOLD=60         # 租约续期阈值（秒），提前多久续期
+#LEASE_EXTENSION_DURATION=300         # 租约续期时长（秒）
+
+# ===================
+# 能力配置
+# ===================
+# 支持的任务类型，逗号分隔，默认全部支持
+#CAPABILITIES=RENDER_SEGMENT_VIDEO,PREPARE_JOB_AUDIO,PACKAGE_SEGMENT_TS,FINALIZE_MP4
+
+# ===================
+# 超时配置
+# ===================
+#FFMPEG_TIMEOUT=3600                  # FFmpeg 执行超时（秒）
+#DOWNLOAD_TIMEOUT=300                 # 下载超时（秒）
+#UPLOAD_TIMEOUT=600                   # 上传超时（秒）
+
+# ===================
+# 硬件加速与多显卡
+# ===================
+# 硬件加速类型: none, qsv, cuda
+HW_ACCEL=none
+
+# GPU 设备列表（逗号分隔的设备索引）
+# 不配置时：自动检测所有设备
+# 单设备示例：GPU_DEVICES=0
+# 多设备示例：GPU_DEVICES=0,1,2
+#GPU_DEVICES=0,1
+
+# ===================
+# 素材缓存
+# ===================
+#CACHE_ENABLED=true                   # 是否启用素材缓存
+#CACHE_DIR=                           # 缓存目录，默认 TEMP_DIR/cache
+#CACHE_MAX_SIZE_GB=0                  # 最大缓存大小（GB），0 表示不限制
+
+# ===================
+# URL 映射（内网下载加速）
+# ===================
+# 格式: src1|dst1,src2|dst2
+#HTTP_REPLACE_MAP="https://cdcdn.zhentuai.com|http://192.168.10.254:9000"
+
+# ===================
+# 上传配置
+# ===================
+# 上传方式: 默认 HTTP，可选 rclone
+#UPLOAD_METHOD=rclone
+#RCLONE_CONFIG_FILE=                  # rclone 配置文件路径
+#RCLONE_REPLACE_MAP="https://oss.example.com|alioss://bucket"

constant/__init__.py

@@ -34,7 +34,21 @@ TRANSITION_TYPES = (
     'slidedown',  # 向下滑动
 )
 
-# 统一视频编码参数（来自集成文档）
+# 支持的特效类型
+EFFECT_TYPES = (
+    'cameraShot',  # 相机定格效果：在指定时间点冻结画面
+    'zoom',        # 缩放效果（预留）
+    'blur',        # 模糊效果（预留）
+)
+
+# 硬件加速类型
+HW_ACCEL_NONE = 'none'      # 纯软件编解码
+HW_ACCEL_QSV = 'qsv'        # Intel Quick Sync Video (核显/独显)
+HW_ACCEL_CUDA = 'cuda'      # NVIDIA NVENC/NVDEC
+
+HW_ACCEL_TYPES = (HW_ACCEL_NONE, HW_ACCEL_QSV, HW_ACCEL_CUDA)
+
+# 统一视频编码参数（软件编码，来自集成文档）
 VIDEO_ENCODE_PARAMS = {
     'codec': 'libx264',
     'preset': 'medium',
@@ -44,6 +58,28 @@ VIDEO_ENCODE_PARAMS = {
     'pix_fmt': 'yuv420p',
 }
 
+# QSV 硬件加速视频编码参数（Intel Quick Sync）
+VIDEO_ENCODE_PARAMS_QSV = {
+    'codec': 'h264_qsv',
+    'preset': 'medium',          # QSV 支持: veryfast, faster, fast, medium, slow, slower, veryslow
+    'profile': 'main',
+    'level': '4.0',
+    'global_quality': '23',      # QSV 使用 global_quality 代替 crf（1-51，值越低质量越高）
+    'look_ahead': '1',           # 启用前瞻分析提升质量
+    'pix_fmt': 'nv12',           # QSV 硬件表面格式
+}
+
+# CUDA 硬件加速视频编码参数（NVIDIA NVENC）
+VIDEO_ENCODE_PARAMS_CUDA = {
+    'codec': 'h264_nvenc',
+    'preset': 'p4',              # NVENC 预设 p1-p7（p1 最快，p7 最慢/质量最高），p4 ≈ medium
+    'profile': 'main',
+    'level': '4.0',
+    'rc': 'vbr',                 # 码率控制模式：vbr 可变码率
+    'cq': '23',                  # 恒定质量模式的质量值（0-51）
+    'pix_fmt': 'yuv420p',        # NVENC 输入格式（会自动转换）
+}
+
 # 统一音频编码参数
 AUDIO_ENCODE_PARAMS = {
     'codec': 'aac',

domain/config.py

@@ -5,10 +5,15 @@ Worker 配置模型
 定义 Worker 运行时的配置参数。
 """
 
+import logging
 import os
 from dataclasses import dataclass, field
 from typing import List, Optional
 
+from constant import HW_ACCEL_NONE, HW_ACCEL_QSV, HW_ACCEL_CUDA, HW_ACCEL_TYPES
+
+logger = logging.getLogger(__name__)
+
 
 # 默认支持的任务类型
 DEFAULT_CAPABILITIES = [
@@ -54,6 +59,17 @@ class WorkerConfig:
     download_timeout: int = 300  # 秒，下载超时
     upload_timeout: int = 600  # 秒，上传超时
 
+    # 硬件加速配置
+    hw_accel: str = HW_ACCEL_NONE  # 硬件加速类型: none, qsv, cuda
+
+    # GPU 设备配置（多显卡调度）
+    gpu_devices: List[int] = field(default_factory=list)  # 空列表表示使用默认设备
+
+    # 素材缓存配置
+    cache_enabled: bool = True  # 是否启用素材缓存
+    cache_dir: str = ""  # 缓存目录，默认为 temp_dir/cache
+    cache_max_size_gb: float = 0  # 最大缓存大小（GB），0 表示不限制
+
     @classmethod
     def from_env(cls) -> 'WorkerConfig':
         """从环境变量创建配置"""
@@ -98,6 +114,26 @@ class WorkerConfig:
         download_timeout = int(os.getenv('DOWNLOAD_TIMEOUT', '300'))
         upload_timeout = int(os.getenv('UPLOAD_TIMEOUT', '600'))
 
+        # 硬件加速配置
+        hw_accel = os.getenv('HW_ACCEL', HW_ACCEL_NONE).lower()
+        if hw_accel not in HW_ACCEL_TYPES:
+            hw_accel = HW_ACCEL_NONE
+
+        # GPU 设备列表（用于多显卡调度）
+        gpu_devices_str = os.getenv('GPU_DEVICES', '')
+        gpu_devices: List[int] = []
+        if gpu_devices_str:
+            try:
+                gpu_devices = [int(d.strip()) for d in gpu_devices_str.split(',') if d.strip()]
+            except ValueError:
+                logger.warning(f"Invalid GPU_DEVICES value: {gpu_devices_str}, using auto-detect")
+                gpu_devices = []
+
+        # 素材缓存配置
+        cache_enabled = os.getenv('CACHE_ENABLED', 'true').lower() in ('true', '1', 'yes')
+        cache_dir = os.getenv('CACHE_DIR', '')  # 空字符串表示使用默认路径
+        cache_max_size_gb = float(os.getenv('CACHE_MAX_SIZE_GB', '0'))
+
         return cls(
             api_endpoint=api_endpoint,
             access_key=access_key,
@@ -110,7 +146,12 @@ class WorkerConfig:
             capabilities=capabilities,
             ffmpeg_timeout=ffmpeg_timeout,
             download_timeout=download_timeout,
-            upload_timeout=upload_timeout
+            upload_timeout=upload_timeout,
+            hw_accel=hw_accel,
+            gpu_devices=gpu_devices,
+            cache_enabled=cache_enabled,
+            cache_dir=cache_dir if cache_dir else os.path.join(temp_dir, 'cache'),
+            cache_max_size_gb=cache_max_size_gb
         )
 
     def get_work_dir_path(self, task_id: str) -> str:
@@ -120,3 +161,23 @@ class WorkerConfig:
     def ensure_temp_dir(self) -> None:
         """确保临时目录存在"""
         os.makedirs(self.temp_dir, exist_ok=True)
+
+    def is_hw_accel_enabled(self) -> bool:
+        """是否启用了硬件加速"""
+        return self.hw_accel != HW_ACCEL_NONE
+
+    def is_qsv(self) -> bool:
+        """是否使用 QSV 硬件加速"""
+        return self.hw_accel == HW_ACCEL_QSV
+
+    def is_cuda(self) -> bool:
+        """是否使用 CUDA 硬件加速"""
+        return self.hw_accel == HW_ACCEL_CUDA
+
+    def has_multi_gpu(self) -> bool:
+        """是否配置了多 GPU"""
+        return len(self.gpu_devices) > 1
+
+    def get_gpu_devices(self) -> List[int]:
+        """获取 GPU 设备列表"""
+        return self.gpu_devices.copy()

domain/gpu.py

@@ -0,0 +1,31 @@
+# -*- coding: utf-8 -*-
+"""
+GPU 设备模型
+
+定义 GPU 设备的数据结构。
+"""
+
+from dataclasses import dataclass
+from typing import Optional
+
+
+@dataclass
+class GPUDevice:
+    """
+    GPU 设备信息
+
+    Attributes:
+        index: 设备索引（对应 nvidia-smi 中的 GPU ID）
+        name: 设备名称（如 "NVIDIA GeForce RTX 3090"）
+        memory_total: 显存总量（MB），可选
+        available: 设备是否可用
+    """
+    index: int
+    name: str
+    memory_total: Optional[int] = None
+    available: bool = True
+
+    def __str__(self) -> str:
+        status = "available" if self.available else "unavailable"
+        mem_info = f", {self.memory_total}MB" if self.memory_total else ""
+        return f"GPU[{self.index}]: {self.name}{mem_info} ({status})"

domain/task.py

@@ -9,6 +9,12 @@ from enum import Enum
 from dataclasses import dataclass, field
 from typing import Dict, Any, Optional, List
 from datetime import datetime
+from urllib.parse import urlparse, unquote
+import os
+
+
+# 支持的图片扩展名
+IMAGE_EXTENSIONS = {'.jpg', '.jpeg', '.png', '.webp', '.bmp', '.gif'}
 
 
 class TaskType(Enum):
@@ -34,6 +40,13 @@ TRANSITION_TYPES = {
     'slidedown': 'slidedown', # 向下滑动
 }
 
+# 支持的特效类型
+EFFECT_TYPES = {
+    'cameraShot',  # 相机定格效果
+    'zoom',        # 缩放效果（预留）
+    'blur',        # 模糊效果（预留）
+}
+
 
 class TaskStatus(Enum):
     """任务状态枚举"""
@@ -76,6 +89,70 @@ class TransitionConfig:
         return TRANSITION_TYPES.get(self.type, 'fade')
 
 
+@dataclass
+class Effect:
+    """
+    特效配置
+
+    格式：type:params
+    例如：cameraShot:3,1 表示在第3秒定格1秒
+    """
+    effect_type: str           # 效果类型
+    params: str = ""           # 参数字符串
+
+    @classmethod
+    def from_string(cls, effect_str: str) -> Optional['Effect']:
+        """
+        从字符串解析 Effect
+
+        格式：type:params 或 type（无参数时）
+        """
+        if not effect_str:
+            return None
+        parts = effect_str.split(':', 1)
+        effect_type = parts[0].strip()
+        if effect_type not in EFFECT_TYPES:
+            return None
+        params = parts[1].strip() if len(parts) > 1 else ""
+        return cls(effect_type=effect_type, params=params)
+
+    @classmethod
+    def parse_effects(cls, effects_str: Optional[str]) -> List['Effect']:
+        """
+        解析效果字符串
+
+        格式：effect1|effect2|effect3
+        例如：cameraShot:3,1|blur:5
+        """
+        if not effects_str:
+            return []
+        effects = []
+        for part in effects_str.split('|'):
+            effect = cls.from_string(part.strip())
+            if effect:
+                effects.append(effect)
+        return effects
+
+    def get_camera_shot_params(self) -> tuple:
+        """
+        获取 cameraShot 效果参数
+
+        Returns:
+            (start_sec, duration_sec): 开始时间和持续时间（秒）
+        """
+        if self.effect_type != 'cameraShot':
+            return (0, 0)
+        if not self.params:
+            return (3, 1)  # 默认值
+        parts = self.params.split(',')
+        try:
+            start = int(parts[0]) if len(parts) >= 1 else 3
+            duration = int(parts[1]) if len(parts) >= 2 else 1
+            return (start, duration)
+        except ValueError:
+            return (3, 1)
+
+
 @dataclass
 class RenderSpec:
     """
@@ -137,6 +214,10 @@ class RenderSpec:
             return self.transition_out.get_overlap_ms()
         return 0
 
+    def get_effects(self) -> List['Effect']:
+        """获取解析后的特效列表"""
+        return Effect.parse_effects(self.effects)
+
 
 @dataclass
 class OutputSpec:
@@ -275,9 +356,56 @@ class Task:
         return int(self.payload.get('durationMs', 5000))
 
     def get_material_url(self) -> Optional[str]:
-        """获取素材 URL"""
+        """
+        获取素材 URL
+
+        优先使用 boundMaterialUrl（实际可下载的 HTTP URL），
+        如果不存在则回退到 sourceRef（可能是 slot 引用）。
+
+        Returns:
+            素材 URL，如果都不存在返回 None
+        """
         return self.payload.get('boundMaterialUrl') or self.payload.get('sourceRef')
 
+    def get_source_ref(self) -> Optional[str]:
+        """获取素材源引用（slot 标识符，如 device:xxx）"""
+        return self.payload.get('sourceRef')
+
+    def get_bound_material_url(self) -> Optional[str]:
+        """获取绑定的素材 URL（实际可下载的 HTTP URL）"""
+        return self.payload.get('boundMaterialUrl')
+
+    def get_material_type(self) -> str:
+        """
+        获取素材类型
+
+        优先使用服务端下发的 materialType 字段，
+        如果不存在则根据 URL 后缀自动推断。
+
+        Returns:
+            素材类型："video" 或 "image"
+        """
+        # 优先使用服务端下发的类型
+        material_type = self.payload.get('materialType')
+        if material_type in ('video', 'image'):
+            return material_type
+
+        # 降级：根据 URL 后缀推断
+        material_url = self.get_material_url()
+        if material_url:
+            parsed = urlparse(material_url)
+            path = unquote(parsed.path)
+            _, ext = os.path.splitext(path)
+            if ext.lower() in IMAGE_EXTENSIONS:
+                return 'image'
+
+        # 默认视频类型
+        return 'video'
+
+    def is_image_material(self) -> bool:
+        """判断素材是否为图片类型"""
+        return self.get_material_type() == 'image'
+
     def get_render_spec(self) -> RenderSpec:
         """获取渲染规格"""
         return RenderSpec.from_dict(self.payload.get('renderSpec'))
@@ -286,6 +414,69 @@ class Task:
         """获取输出规格"""
         return OutputSpec.from_dict(self.payload.get('output'))
 
+    def get_transition_type(self) -> Optional[str]:
+        """获取转场类型（来自 TaskPayload 顶层）"""
+        return self.payload.get('transitionType')
+
+    def get_transition_ms(self) -> int:
+        """获取转场时长（毫秒，来自 TaskPayload 顶层）"""
+        return int(self.payload.get('transitionMs', 0))
+
+    def has_transition(self) -> bool:
+        """是否有转场效果"""
+        return self.get_transition_ms() > 0
+
+    def get_overlap_tail_ms(self) -> int:
+        """
+        获取尾部 overlap 时长（毫秒）
+
+        转场发生在当前片段与下一片段之间，当前片段需要在尾部多渲染 overlap 帧。
+        overlap = transitionMs / 2
+        """
+        return self.get_transition_ms() // 2
+
+    def get_transition_in_type(self) -> Optional[str]:
+        """获取入场转场类型（来自前一片段的出场转场）"""
+        return self.payload.get('transitionInType')
+
+    def get_transition_in_ms(self) -> int:
+        """获取入场转场时长（毫秒）"""
+        return int(self.payload.get('transitionInMs', 0))
+
+    def get_transition_out_type(self) -> Optional[str]:
+        """获取出场转场类型（当前片段的转场配置）"""
+        return self.payload.get('transitionOutType')
+
+    def get_transition_out_ms(self) -> int:
+        """获取出场转场时长（毫秒）"""
+        return int(self.payload.get('transitionOutMs', 0))
+
+    def has_transition_in(self) -> bool:
+        """是否有入场转场"""
+        return self.get_transition_in_ms() > 0
+
+    def has_transition_out(self) -> bool:
+        """是否有出场转场"""
+        return self.get_transition_out_ms() > 0
+
+    def get_overlap_head_ms(self) -> int:
+        """
+        获取头部 overlap 时长（毫秒）
+
+        入场转场来自前一个片段，当前片段需要在头部多渲染 overlap 帧。
+        overlap = transitionInMs / 2
+        """
+        return self.get_transition_in_ms() // 2
+
+    def get_overlap_tail_ms_v2(self) -> int:
+        """
+        获取尾部 overlap 时长（毫秒）- 使用新的字段名
+
+        出场转场用于当前片段与下一片段之间，当前片段需要在尾部多渲染 overlap 帧。
+        overlap = transitionOutMs / 2
+        """
+        return self.get_transition_out_ms() // 2
+
     def get_bgm_url(self) -> Optional[str]:
         """获取 BGM URL"""
         return self.payload.get('bgmUrl')

handlers/base.py

@@ -11,6 +11,7 @@ import logging
 import shutil
 import tempfile
 import subprocess
+import threading
 from abc import ABC
 from typing import Optional, List, Dict, Any, Tuple, TYPE_CHECKING
 
@@ -19,6 +20,11 @@ from domain.task import Task
 from domain.result import TaskResult, ErrorCode
 from domain.config import WorkerConfig
 from services import storage
+from services.cache import MaterialCache
+from constant import (
+    HW_ACCEL_NONE, HW_ACCEL_QSV, HW_ACCEL_CUDA,
+    VIDEO_ENCODE_PARAMS, VIDEO_ENCODE_PARAMS_QSV, VIDEO_ENCODE_PARAMS_CUDA
+)
 
 if TYPE_CHECKING:
     from services.api_client import APIClientV2
@@ -26,15 +32,108 @@ if TYPE_CHECKING:
 logger = logging.getLogger(__name__)
 
 
-# v2 统一视频编码参数（来自集成文档）
-VIDEO_ENCODE_ARGS = [
-    '-c:v', 'libx264',
-    '-preset', 'medium',
-    '-profile:v', 'main',
-    '-level', '4.0',
-    '-crf', '23',
-    '-pix_fmt', 'yuv420p',
-]
+def get_video_encode_args(hw_accel: str = HW_ACCEL_NONE) -> List[str]:
+    """
+    根据硬件加速配置获取视频编码参数
+
+    Args:
+        hw_accel: 硬件加速类型 (none, qsv, cuda)
+
+    Returns:
+        FFmpeg 视频编码参数列表
+    """
+    if hw_accel == HW_ACCEL_QSV:
+        params = VIDEO_ENCODE_PARAMS_QSV
+        return [
+            '-c:v', params['codec'],
+            '-preset', params['preset'],
+            '-profile:v', params['profile'],
+            '-level', params['level'],
+            '-global_quality', params['global_quality'],
+            '-look_ahead', params['look_ahead'],
+        ]
+    elif hw_accel == HW_ACCEL_CUDA:
+        params = VIDEO_ENCODE_PARAMS_CUDA
+        return [
+            '-c:v', params['codec'],
+            '-preset', params['preset'],
+            '-profile:v', params['profile'],
+            '-level', params['level'],
+            '-rc', params['rc'],
+            '-cq', params['cq'],
+            '-b:v', '0',  # 配合 vbr 模式使用 cq
+        ]
+    else:
+        # 软件编码（默认）
+        params = VIDEO_ENCODE_PARAMS
+        return [
+            '-c:v', params['codec'],
+            '-preset', params['preset'],
+            '-profile:v', params['profile'],
+            '-level', params['level'],
+            '-crf', params['crf'],
+            '-pix_fmt', params['pix_fmt'],
+        ]
+
+
+def get_hwaccel_decode_args(hw_accel: str = HW_ACCEL_NONE, device_index: Optional[int] = None) -> List[str]:
+    """
+    获取硬件加速解码参数（输入文件之前使用）
+
+    Args:
+        hw_accel: 硬件加速类型 (none, qsv, cuda)
+        device_index: GPU 设备索引，用于多显卡调度
+
+    Returns:
+        FFmpeg 硬件加速解码参数列表
+    """
+    if hw_accel == HW_ACCEL_CUDA:
+        # CUDA 硬件加速解码
+        args = ['-hwaccel', 'cuda']
+        # 多显卡模式下指定设备
+        if device_index is not None:
+            args.extend(['-hwaccel_device', str(device_index)])
+        args.extend(['-hwaccel_output_format', 'cuda'])
+        return args
+    elif hw_accel == HW_ACCEL_QSV:
+        # QSV 硬件加速解码
+        args = ['-hwaccel', 'qsv']
+        # QSV 在 Windows 上使用 -qsv_device
+        if device_index is not None:
+            args.extend(['-qsv_device', str(device_index)])
+        args.extend(['-hwaccel_output_format', 'qsv'])
+        return args
+    else:
+        return []
+
+
+def get_hwaccel_filter_prefix(hw_accel: str = HW_ACCEL_NONE) -> str:
+    """
+    获取硬件加速滤镜前缀（用于 hwdownload 从 GPU 到 CPU）
+
+    注意：由于大多数复杂滤镜（如 lut3d, overlay, crop 等）不支持硬件表面，
+    我们需要在滤镜链开始时将硬件表面下载到系统内存。
+
+    CUDA/QSV hwdownload 只支持 nv12 格式输出，因此需要两步转换：
+    1. hwdownload,format=nv12 - 从 GPU 下载到 CPU
+    2. format=yuv420p - 转换为标准格式（确保与 RGBA/YUVA overlay 混合时颜色正确）
+
+    Args:
+        hw_accel: 硬件加速类型
+
+    Returns:
+        需要添加到滤镜链开头的 hwdownload 滤镜字符串
+    """
+    if hw_accel == HW_ACCEL_CUDA:
+        return 'hwdownload,format=nv12,format=yuv420p,'
+    elif hw_accel == HW_ACCEL_QSV:
+        return 'hwdownload,format=nv12,format=yuv420p,'
+    else:
+        return ''
+
+
+# v2 统一视频编码参数（兼容旧代码，使用软件编码）
+VIDEO_ENCODE_ARGS = get_video_encode_args(HW_ACCEL_NONE)
 
 # v2 统一音频编码参数
 AUDIO_ENCODE_ARGS = [
@@ -44,6 +143,8 @@ AUDIO_ENCODE_ARGS = [
     '-ac', '2',
 ]
 
+FFMPEG_LOGLEVEL = 'error'
+
 
 def subprocess_args(include_stdout: bool = True) -> Dict[str, Any]:
     """
@@ -164,9 +265,13 @@ class BaseHandler(TaskHandler, ABC):
     - 临时目录管理
     - 文件下载/上传
     - FFmpeg 命令执行
+    - GPU 设备管理（多显卡调度）
     - 日志记录
     """
 
+    # 线程本地存储：用于存储当前线程的 GPU 设备索引
+    _thread_local = threading.local()
+
     def __init__(self, config: WorkerConfig, api_client: 'APIClientV2'):
         """
         初始化处理器
@@ -177,6 +282,72 @@ class BaseHandler(TaskHandler, ABC):
         """
         self.config = config
         self.api_client = api_client
+        self.material_cache = MaterialCache(
+            cache_dir=config.cache_dir,
+            enabled=config.cache_enabled,
+            max_size_gb=config.cache_max_size_gb
+        )
+
+    # ========== GPU 设备管理 ==========
+
+    def set_gpu_device(self, device_index: int) -> None:
+        """
+        设置当前线程的 GPU 设备索引
+
+        由 TaskExecutor 在任务执行前调用。
+
+        Args:
+            device_index: GPU 设备索引
+        """
+        self._thread_local.gpu_device = device_index
+
+    def get_gpu_device(self) -> Optional[int]:
+        """
+        获取当前线程的 GPU 设备索引
+
+        Returns:
+            GPU 设备索引，未设置则返回 None
+        """
+        return getattr(self._thread_local, 'gpu_device', None)
+
+    def clear_gpu_device(self) -> None:
+        """
+        清除当前线程的 GPU 设备索引
+
+        由 TaskExecutor 在任务执行后调用。
+        """
+        if hasattr(self._thread_local, 'gpu_device'):
+            del self._thread_local.gpu_device
+
+    # ========== FFmpeg 参数生成 ==========
+
+    def get_video_encode_args(self) -> List[str]:
+        """
+        获取当前配置的视频编码参数
+
+        Returns:
+            FFmpeg 视频编码参数列表
+        """
+        return get_video_encode_args(self.config.hw_accel)
+
+    def get_hwaccel_decode_args(self) -> List[str]:
+        """
+        获取硬件加速解码参数（支持设备指定）
+
+        Returns:
+            FFmpeg 硬件加速解码参数列表
+        """
+        device_index = self.get_gpu_device()
+        return get_hwaccel_decode_args(self.config.hw_accel, device_index)
+
+    def get_hwaccel_filter_prefix(self) -> str:
+        """
+        获取硬件加速滤镜前缀
+
+        Returns:
+            需要添加到滤镜链开头的 hwdownload 滤镜字符串
+        """
+        return get_hwaccel_filter_prefix(self.config.hw_accel)
 
     def before_handle(self, task: Task) -> None:
         """处理前钩子"""
@@ -223,14 +394,15 @@ class BaseHandler(TaskHandler, ABC):
         except Exception as e:
             logger.warning(f"Failed to cleanup work directory {work_dir}: {e}")
 
-    def download_file(self, url: str, dest: str, timeout: int = None) -> bool:
+    def download_file(self, url: str, dest: str, timeout: int = None, use_cache: bool = True) -> bool:
         """
-        下载文件
+        下载文件（支持缓存）
 
         Args:
             url: 文件 URL
             dest: 目标路径
             timeout: 超时时间（秒）
+            use_cache: 是否使用缓存（默认 True）
 
         Returns:
             是否成功
@@ -239,7 +411,13 @@ class BaseHandler(TaskHandler, ABC):
             timeout = self.config.download_timeout
 
         try:
-            result = storage.download_file(url, dest, timeout=timeout)
+            if use_cache:
+                # 使用缓存下载
+                result = self.material_cache.get_or_download(url, dest, timeout=timeout)
+            else:
+                # 直接下载（不走缓存）
+                result = storage.download_file(url, dest, timeout=timeout)
+
             if result:
                 file_size = os.path.getsize(dest) if os.path.exists(dest) else 0
                 logger.debug(f"Downloaded: {url} -> {dest} ({file_size} bytes)")
@@ -286,6 +464,11 @@ class BaseHandler(TaskHandler, ABC):
             if result:
                 file_size = os.path.getsize(file_path)
                 logger.info(f"[task:{task_id}] Uploaded: {file_path} ({file_size} bytes)")
+
+                # 将上传成功的文件加入缓存
+                if access_url:
+                    self.material_cache.add_to_cache(access_url, file_path)
+
                 return access_url
             else:
                 logger.error(f"[task:{task_id}] Upload failed: {file_path}")
@@ -314,22 +497,28 @@ class BaseHandler(TaskHandler, ABC):
         if timeout is None:
             timeout = self.config.ffmpeg_timeout
 
+        cmd_to_run = list(cmd)
+        if cmd_to_run and cmd_to_run[0] == 'ffmpeg' and '-loglevel' not in cmd_to_run:
+            cmd_to_run[1:1] = ['-loglevel', FFMPEG_LOGLEVEL]
+
         # 日志记录命令（限制长度）
-        cmd_str = ' '.join(cmd)
+        cmd_str = ' '.join(cmd_to_run)
         if len(cmd_str) > 500:
             cmd_str = cmd_str[:500] + '...'
         logger.info(f"[task:{task_id}] FFmpeg: {cmd_str}")
 
         try:
+            run_args = subprocess_args(False)
+            run_args['stdout'] = subprocess.DEVNULL
+            run_args['stderr'] = subprocess.PIPE
             result = subprocess.run(
-                cmd,
-                capture_output=True,
+                cmd_to_run,
                 timeout=timeout,
-                **subprocess_args(False)
+                **run_args
             )
 
             if result.returncode != 0:
-                stderr = result.stderr.decode('utf-8', errors='replace')[:1000]
+                stderr = (result.stderr or b'').decode('utf-8', errors='replace')[:1000]
                 logger.error(f"[task:{task_id}] FFmpeg failed (code={result.returncode}): {stderr}")
                 return False
 

handlers/compose_transition.py

@@ -10,7 +10,7 @@ import os
 import logging
 from typing import List, Optional
 
-from handlers.base import BaseHandler, VIDEO_ENCODE_ARGS
+from handlers.base import BaseHandler
 from domain.task import Task, TaskType, TransitionConfig, TRANSITION_TYPES
 from domain.result import TaskResult, ErrorCode
 
@@ -235,8 +235,8 @@ class ComposeTransitionHandler(BaseHandler):
             '-map', '[outv]',
         ]
 
-        # 编码参数（与片段视频一致）
-        cmd.extend(VIDEO_ENCODE_ARGS)
+        # 编码参数（根据硬件加速配置动态获取）
+        cmd.extend(self.get_video_encode_args())
 
         # 帧率
         fps = output_spec.fps

handlers/finalize_mp4.py

@@ -127,9 +127,9 @@ class FinalizeMp4Handler(BaseHandler):
         concat_file = os.path.join(work_dir, 'concat.txt')
         with open(concat_file, 'w', encoding='utf-8') as f:
             for ts_file in ts_files:
-                # 路径中的反斜杠需要转义或使用正斜杠
-                ts_path = ts_file.replace('\\', '/')
-                f.write(f"file '{ts_path}'\n")
+                # FFmpeg concat 路径相对于 concat.txt 所在目录，只需写文件名
+                ts_filename = os.path.basename(ts_file)
+                f.write(f"file '{ts_filename}'\n")
 
         # 3. 构建合并命令（remux，不重编码）
         cmd = [

handlers/render_video.py

@@ -9,14 +9,23 @@
 import os
 import logging
 from typing import List, Optional, Tuple
+from urllib.parse import urlparse, unquote
 
-from handlers.base import BaseHandler, VIDEO_ENCODE_ARGS
-from domain.task import Task, TaskType, RenderSpec, OutputSpec
+from handlers.base import BaseHandler
+from domain.task import Task, TaskType, RenderSpec, OutputSpec, Effect, IMAGE_EXTENSIONS
 from domain.result import TaskResult, ErrorCode
 
 logger = logging.getLogger(__name__)
 
 
+def _get_extension_from_url(url: str) -> str:
+    """从 URL 提取文件扩展名"""
+    parsed = urlparse(url)
+    path = unquote(parsed.path)
+    _, ext = os.path.splitext(path)
+    return ext.lower() if ext else ''
+
+
 class RenderSegmentVideoHandler(BaseHandler):
     """
     视频片段渲染处理器
@@ -46,19 +55,63 @@ class RenderSegmentVideoHandler(BaseHandler):
                     "Missing material URL (boundMaterialUrl or sourceRef)"
                 )
 
+            # 检查 URL 格式：必须是 HTTP 或 HTTPS 协议
+            if not material_url.startswith(('http://', 'https://')):
+                source_ref = task.get_source_ref()
+                bound_url = task.get_bound_material_url()
+                logger.error(
+                    f"[task:{task.task_id}] Invalid material URL format: '{material_url}'. "
+                    f"boundMaterialUrl={bound_url}, sourceRef={source_ref}. "
+                    f"Server should provide boundMaterialUrl with HTTP/HTTPS URL."
+                )
+                return TaskResult.fail(
+                    ErrorCode.E_SPEC_INVALID,
+                    f"Invalid material URL: '{material_url}' is not a valid HTTP/HTTPS URL. "
+                    f"Server must provide boundMaterialUrl."
+                )
+
             render_spec = task.get_render_spec()
             output_spec = task.get_output_spec()
             duration_ms = task.get_duration_ms()
 
-            # 1. 下载素材
-            input_file = os.path.join(work_dir, 'input.mp4')
+            # 1. 检测素材类型并确定输入文件扩展名
+            is_image = task.is_image_material()
+            if is_image:
+                # 图片素材：根据 URL 确定扩展名
+                ext = _get_extension_from_url(material_url)
+                if not ext or ext not in IMAGE_EXTENSIONS:
+                    ext = '.jpg'  # 默认扩展名
+                input_file = os.path.join(work_dir, f'input{ext}')
+            else:
+                input_file = os.path.join(work_dir, 'input.mp4')
+
+            # 2. 下载素材
             if not self.download_file(material_url, input_file):
                 return TaskResult.fail(
                     ErrorCode.E_INPUT_UNAVAILABLE,
                     f"Failed to download material: {material_url}"
                 )
 
-            # 2. 下载 LUT（如有）
+            # 3. 图片素材转换为视频
+            if is_image:
+                video_input_file = os.path.join(work_dir, 'input_video.mp4')
+                if not self._convert_image_to_video(
+                    image_file=input_file,
+                    output_file=video_input_file,
+                    duration_ms=duration_ms,
+                    output_spec=output_spec,
+                    render_spec=render_spec,
+                    task_id=task.task_id
+                ):
+                    return TaskResult.fail(
+                        ErrorCode.E_FFMPEG_FAILED,
+                        "Failed to convert image to video"
+                    )
+                # 使用转换后的视频作为输入
+                input_file = video_input_file
+                logger.info(f"[task:{task.task_id}] Image converted to video successfully")
+
+            # 4. 下载 LUT（如有）
             lut_file = None
             if render_spec.lut_url:
                 lut_file = os.path.join(work_dir, 'lut.cube')
@@ -66,23 +119,29 @@ class RenderSegmentVideoHandler(BaseHandler):
                     logger.warning(f"[task:{task.task_id}] Failed to download LUT, continuing without it")
                     lut_file = None
 
-            # 3. 下载叠加层（如有）
+            # 5. 下载叠加层（如有）
             overlay_file = None
             if render_spec.overlay_url:
                 # 根据 URL 后缀确定文件扩展名
-                ext = '.png'
-                if render_spec.overlay_url.lower().endswith('.jpg') or render_spec.overlay_url.lower().endswith('.jpeg'):
+                url_lower = render_spec.overlay_url.lower()
+                if url_lower.endswith('.jpg') or url_lower.endswith('.jpeg'):
                     ext = '.jpg'
+                elif url_lower.endswith('.mov'):
+                    ext = '.mov'
+                else:
+                    ext = '.png'  # 默认
                 overlay_file = os.path.join(work_dir, f'overlay{ext}')
                 if not self.download_file(render_spec.overlay_url, overlay_file):
                     logger.warning(f"[task:{task.task_id}] Failed to download overlay, continuing without it")
                     overlay_file = None
 
-            # 4. 计算 overlap 时长
-            overlap_head_ms = render_spec.get_overlap_head_ms()
-            overlap_tail_ms = render_spec.get_overlap_tail_ms()
+            # 6. 计算 overlap 时长（用于转场帧冻结）
+            # 头部 overlap: 来自前一片段的出场转场
+            overlap_head_ms = task.get_overlap_head_ms()
+            # 尾部 overlap: 当前片段的出场转场
+            overlap_tail_ms = task.get_overlap_tail_ms_v2()
 
-            # 5. 构建 FFmpeg 命令
+            # 7. 构建 FFmpeg 命令
             output_file = os.path.join(work_dir, 'output.mp4')
             cmd = self._build_command(
                 input_file=input_file,
@@ -96,25 +155,25 @@ class RenderSegmentVideoHandler(BaseHandler):
                 overlap_tail_ms=overlap_tail_ms
             )
 
-            # 6. 执行 FFmpeg
+            # 8. 执行 FFmpeg
             if not self.run_ffmpeg(cmd, task.task_id):
                 return TaskResult.fail(
                     ErrorCode.E_FFMPEG_FAILED,
                     "FFmpeg rendering failed"
                 )
 
-            # 7. 验证输出文件
+            # 9. 验证输出文件
             if not self.ensure_file_exists(output_file, min_size=4096):
                 return TaskResult.fail(
                     ErrorCode.E_FFMPEG_FAILED,
                     "Output file is missing or too small"
                 )
 
-            # 8. 获取实际时长
+            # 10. 获取实际时长
             actual_duration = self.probe_duration(output_file)
             actual_duration_ms = int(actual_duration * 1000) if actual_duration else duration_ms
 
-            # 9. 上传产物
+            # 11. 上传产物
             video_url = self.upload_file(task.task_id, 'video', output_file)
             if not video_url:
                 return TaskResult.fail(
@@ -122,7 +181,7 @@ class RenderSegmentVideoHandler(BaseHandler):
                     "Failed to upload video"
                 )
 
-            # 10. 构建结果（包含 overlap 信息）
+            # 12. 构建结果（包含 overlap 信息）
             result_data = {
                 'videoUrl': video_url,
                 'actualDurationMs': actual_duration_ms,
@@ -139,6 +198,96 @@ class RenderSegmentVideoHandler(BaseHandler):
         finally:
             self.cleanup_work_dir(work_dir)
 
+    def _convert_image_to_video(
+        self,
+        image_file: str,
+        output_file: str,
+        duration_ms: int,
+        output_spec: OutputSpec,
+        render_spec: RenderSpec,
+        task_id: str
+    ) -> bool:
+        """
+        将图片转换为视频
+
+        使用 FFmpeg 将静态图片转换为指定时长的视频，
+        同时应用缩放填充和变速处理。
+
+        Args:
+            image_file: 输入图片文件路径
+            output_file: 输出视频文件路径
+            duration_ms: 目标时长（毫秒）
+            output_spec: 输出规格
+            render_spec: 渲染规格
+            task_id: 任务 ID（用于日志）
+
+        Returns:
+            是否成功
+        """
+        width = output_spec.width
+        height = output_spec.height
+        fps = output_spec.fps
+
+        # 计算实际时长（考虑变速）
+        speed = float(render_spec.speed) if render_spec.speed else 1.0
+        if speed <= 0:
+            speed = 1.0
+        # 变速后的实际播放时长
+        actual_duration_sec = (duration_ms / 1000.0) / speed
+
+        # 构建 FFmpeg 命令
+        cmd = [
+            'ffmpeg', '-y', '-hide_banner',
+            '-loop', '1',  # 循环输入图片
+            '-i', image_file,
+            '-t', str(actual_duration_sec),  # 输出时长
+        ]
+
+        # 构建滤镜：缩放填充到目标尺寸
+        filters = []
+
+        # 裁切处理（与视频相同逻辑）
+        if render_spec.crop_enable and render_spec.face_pos:
+            try:
+                fx, fy = map(float, render_spec.face_pos.split(','))
+                target_ratio = width / height
+                filters.append(
+                    f"crop='min(iw,ih*{target_ratio})':'min(ih,iw/{target_ratio})':"
+                    f"'(iw-min(iw,ih*{target_ratio}))*{fx}':"
+                    f"'(ih-min(ih,iw/{target_ratio}))*{fy}'"
+                )
+            except (ValueError, ZeroDivisionError):
+                logger.warning(f"[task:{task_id}] Invalid face position: {render_spec.face_pos}")
+        elif render_spec.zoom_cut:
+            target_ratio = width / height
+            filters.append(
+                f"crop='min(iw,ih*{target_ratio})':'min(ih,iw/{target_ratio})'"
+            )
+
+        # 缩放填充
+        filters.append(
+            f"scale={width}:{height}:force_original_aspect_ratio=decrease,"
+            f"pad={width}:{height}:(ow-iw)/2:(oh-ih)/2:black"
+        )
+
+        # 格式转换（确保兼容性）
+        filters.append("format=yuv420p")
+
+        cmd.extend(['-vf', ','.join(filters)])
+
+        # 编码参数
+        cmd.extend([
+            '-c:v', 'libx264',
+            '-preset', 'fast',
+            '-crf', '18',
+            '-r', str(fps),
+            '-an',  # 无音频
+            output_file
+        ])
+
+        logger.info(f"[task:{task_id}] Converting image to video: {actual_duration_sec:.2f}s at {fps}fps")
+        return self.run_ffmpeg(cmd, task_id)
+
     def _build_command(
         self,
         input_file: str,
@@ -170,6 +319,11 @@ class RenderSegmentVideoHandler(BaseHandler):
         """
         cmd = ['ffmpeg', '-y', '-hide_banner']
 
+        # 硬件加速解码参数（在输入文件之前）
+        hwaccel_args = self.get_hwaccel_decode_args()
+        if hwaccel_args:
+            cmd.extend(hwaccel_args)
+
         # 输入文件
         cmd.extend(['-i', input_file])
 
@@ -182,20 +336,22 @@ class RenderSegmentVideoHandler(BaseHandler):
             render_spec=render_spec,
             output_spec=output_spec,
             lut_file=lut_file,
-            has_overlay=overlay_file is not None,
+            overlay_file=overlay_file,
             overlap_head_ms=overlap_head_ms,
             overlap_tail_ms=overlap_tail_ms
         )
 
         # 应用滤镜
-        if overlay_file:
-            # 使用 filter_complex 处理叠加
+        # 检测是否为 filter_complex 格式（包含分号或方括号标签）
+        is_filter_complex = ';' in filters or (filters.startswith('[') and ']' in filters)
+        if is_filter_complex or overlay_file:
+            # 使用 filter_complex 处理
             cmd.extend(['-filter_complex', filters])
         elif filters:
             cmd.extend(['-vf', filters])
 
-        # 编码参数（v2 统一参数）
-        cmd.extend(VIDEO_ENCODE_ARGS)
+        # 编码参数（根据硬件加速配置动态获取）
+        cmd.extend(self.get_video_encode_args())
 
         # 帧率
         fps = output_spec.fps
@@ -224,7 +380,7 @@ class RenderSegmentVideoHandler(BaseHandler):
         render_spec: RenderSpec,
         output_spec: OutputSpec,
         lut_file: Optional[str] = None,
-        has_overlay: bool = False,
+        overlay_file: Optional[str] = None,
         overlap_head_ms: int = 0,
         overlap_tail_ms: int = 0
     ) -> str:
@@ -235,7 +391,7 @@ class RenderSegmentVideoHandler(BaseHandler):
             render_spec: 渲染规格
             output_spec: 输出规格
             lut_file: LUT 文件路径
-            has_overlay: 是否有叠加层
+            overlay_file: 叠加层文件路径（支持图片 png/jpg 和视频 mov）
             overlap_head_ms: 头部 overlap 时长（毫秒）
             overlap_tail_ms: 尾部 overlap 时长（毫秒）
 
@@ -245,6 +401,21 @@ class RenderSegmentVideoHandler(BaseHandler):
         filters = []
         width = output_spec.width
         height = output_spec.height
+        fps = output_spec.fps
+
+        # 判断 overlay 类型
+        has_overlay = overlay_file is not None
+        is_video_overlay = has_overlay and overlay_file.lower().endswith('.mov')
+
+        # 解析 effects
+        effects = render_spec.get_effects()
+        has_camera_shot = any(e.effect_type == 'cameraShot' for e in effects)
+
+        # 硬件加速时需要先 hwdownload（将 GPU 表面下载到系统内存）
+        hwaccel_prefix = self.get_hwaccel_filter_prefix()
+        if hwaccel_prefix:
+            # 去掉末尾的逗号，作为第一个滤镜
+            filters.append(hwaccel_prefix.rstrip(','))
 
         # 1. 变速处理
         speed = float(render_spec.speed) if render_spec.speed else 1.0
@@ -255,8 +426,8 @@ class RenderSegmentVideoHandler(BaseHandler):
 
         # 2. LUT 调色
         if lut_file:
-            # 路径中的反斜杠需要转义
-            lut_path = lut_file.replace('\\', '/')
+            # 路径中的反斜杠需要转换，冒号需要转义（FFmpeg filter语法中冒号是特殊字符）
+            lut_path = lut_file.replace('\\', '/').replace(':', r'\:')
             filters.append(f"lut3d='{lut_path}'")
 
         # 3. 裁切处理
@@ -288,7 +459,23 @@ class RenderSegmentVideoHandler(BaseHandler):
         )
         filters.append(scale_filter)
 
-        # 5. 帧冻结（tpad）- 用于转场 overlap 区域
+        # 5. 特效处理（cameraShot 需要特殊处理）
+        if has_camera_shot:
+            # cameraShot 需要使用 filter_complex 格式
+            return self._build_filter_complex_with_effects(
+                base_filters=filters,
+                effects=effects,
+                fps=fps,
+                width=width,
+                height=height,
+                has_overlay=has_overlay,
+                is_video_overlay=is_video_overlay,
+                overlap_head_ms=overlap_head_ms,
+                overlap_tail_ms=overlap_tail_ms,
+                use_hwdownload=bool(hwaccel_prefix)
+            )
+
+        # 6. 帧冻结（tpad）- 用于转场 overlap 区域
         # 注意：tpad 必须在缩放之后应用
         tpad_parts = []
         if overlap_head_ms > 0:
@@ -303,10 +490,140 @@ class RenderSegmentVideoHandler(BaseHandler):
         if tpad_parts:
             filters.append(f"tpad={':'.join(tpad_parts)}")
 
-        # 6. 构建最终滤镜
+        # 7. 构建最终滤镜
         if has_overlay:
             # 使用 filter_complex 格式
             base_filters = ','.join(filters) if filters else 'copy'
-            return f"[0:v]{base_filters}[base];[base][1:v]overlay=0:0"
+            overlay_scale = f"scale={width}:{height}"
+            # 视频 overlay 使用 eof_action=pass（结束后消失），图片 overlay 使用默认行为（保持显示）
+            overlay_params = 'eof_action=pass' if is_video_overlay else ''
+            overlay_filter = f"overlay=0:0:{overlay_params}" if overlay_params else 'overlay=0:0'
+            # 视频 overlay 需要在末尾统一颜色范围，避免 overlay 结束后 range 从 tv 变为 pc
+            range_fix = ',format=yuv420p,setrange=tv' if is_video_overlay else ''
+            return f"[0:v]{base_filters}[base];[1:v]{overlay_scale}[overlay];[base][overlay]{overlay_filter}{range_fix}"
         else:
             return ','.join(filters) if filters else ''
+
+    def _build_filter_complex_with_effects(
+        self,
+        base_filters: List[str],
+        effects: List[Effect],
+        fps: int,
+        width: int,
+        height: int,
+        has_overlay: bool = False,
+        is_video_overlay: bool = False,
+        overlap_head_ms: int = 0,
+        overlap_tail_ms: int = 0,
+        use_hwdownload: bool = False
+    ) -> str:
+        """
+        构建包含特效的 filter_complex 滤镜图
+
+        cameraShot 效果需要使用 split/freezeframes/concat 滤镜组合。
+
+        Args:
+            base_filters: 基础滤镜列表
+            effects: 特效列表
+            fps: 帧率
+            width: 输出宽度
+            height: 输出高度
+            has_overlay: 是否有叠加层
+            is_video_overlay: 叠加层是否为视频格式（如 .mov）
+            overlap_head_ms: 头部 overlap 时长
+            overlap_tail_ms: 尾部 overlap 时长
+            use_hwdownload: 是否使用了硬件加速解码（已在 base_filters 中包含 hwdownload）
+
+        Returns:
+            filter_complex 格式的滤镜字符串
+        """
+        filter_parts = []
+
+        # 基础滤镜链
+        base_chain = ','.join(base_filters) if base_filters else 'copy'
+
+        # 当前输出标签
+        current_output = '[v_base]'
+        filter_parts.append(f"[0:v]{base_chain}{current_output}")
+
+        # 处理每个特效
+        effect_idx = 0
+        for effect in effects:
+            if effect.effect_type == 'cameraShot':
+                start_sec, duration_sec = effect.get_camera_shot_params()
+                if start_sec <= 0 or duration_sec <= 0:
+                    continue
+
+                # cameraShot 实现（定格效果）：
+                # 1. fps + split 分割
+                # 2. 第一路：trim(0, start) + tpad冻结duration秒
+                # 3. 第二路：trim(start, end)
+                # 4. concat 拼接
+
+                split_out_a = f'[eff{effect_idx}_a]'
+                split_out_b = f'[eff{effect_idx}_b]'
+                frozen_out = f'[eff{effect_idx}_frozen]'
+                rest_out = f'[eff{effect_idx}_rest]'
+                effect_output = f'[v_eff{effect_idx}]'
+
+                # fps + split
+                filter_parts.append(
+                    f"{current_output}fps=fps={fps},split{split_out_a}{split_out_b}"
+                )
+
+                # 第一路：trim(0, start) + tpad冻结
+                # tpad=stop_mode=clone 将最后一帧冻结指定时长
+                filter_parts.append(
+                    f"{split_out_a}trim=start=0:end={start_sec},setpts=PTS-STARTPTS,"
+                    f"tpad=stop_mode=clone:stop_duration={duration_sec}{frozen_out}"
+                )
+
+                # 第二路：trim 从 start 开始
+                filter_parts.append(
+                    f"{split_out_b}trim=start={start_sec},setpts=PTS-STARTPTS{rest_out}"
+                )
+
+                # concat 拼接
+                filter_parts.append(
+                    f"{frozen_out}{rest_out}concat=n=2:v=1:a=0{effect_output}"
+                )
+
+                current_output = effect_output
+                effect_idx += 1
+
+        # 帧冻结（tpad）- 用于转场 overlap 区域
+        tpad_parts = []
+        if overlap_head_ms > 0:
+            head_duration_sec = overlap_head_ms / 1000.0
+            tpad_parts.append(f"start_mode=clone:start_duration={head_duration_sec}")
+        if overlap_tail_ms > 0:
+            tail_duration_sec = overlap_tail_ms / 1000.0
+            tpad_parts.append(f"stop_mode=clone:stop_duration={tail_duration_sec}")
+
+        if tpad_parts:
+            tpad_output = '[v_tpad]'
+            filter_parts.append(f"{current_output}tpad={':'.join(tpad_parts)}{tpad_output}")
+            current_output = tpad_output
+
+        # 最终输出
+        if has_overlay:
+            # 叠加层处理
+            # 视频 overlay 使用 eof_action=pass（结束后消失），图片 overlay 使用默认行为（保持显示）
+            overlay_params = 'eof_action=pass' if is_video_overlay else ''
+            overlay_filter = f"overlay=0:0:{overlay_params}" if overlay_params else 'overlay=0:0'
+            overlay_scale = f"scale={width}:{height}"
+            overlay_output = '[v_overlay]'
+            # 视频 overlay 需要在末尾统一颜色范围，避免 overlay 结束后 range 从 tv 变为 pc
+            range_fix = ',format=yuv420p,setrange=tv' if is_video_overlay else ''
+            filter_parts.append(f"[1:v]{overlay_scale}{overlay_output}")
+            filter_parts.append(f"{current_output}{overlay_output}{overlay_filter}{range_fix}")
+        else:
+            # 移除最后一个标签，直接输出
+            # 将最后一个滤镜的输出标签替换为空（直接输出）
+            if filter_parts:
+                last_filter = filter_parts[-1]
+                # 移除末尾的输出标签
+                if last_filter.endswith(current_output):
+                    filter_parts[-1] = last_filter[:-len(current_output)]
+
+        return ';'.join(filter_parts)

index.py

@@ -26,6 +26,8 @@ import time
 import signal
 import logging
 
+from dotenv import load_dotenv
+
 from domain.config import WorkerConfig
 from services.api_client import APIClientV2
 from services.task_executor import TaskExecutor
@@ -166,6 +168,9 @@ class WorkerV2:
 
 def main():
     """主函数"""
+    # 加载 .env 文件（如果存在）
+    load_dotenv()
+
     logger.info(f"RenderWorker v{SOFTWARE_VERSION}")
 
     # 创建并运行 Worker

services/api_client.py

@@ -7,11 +7,13 @@ v2 API 客户端
 
 import logging
 import subprocess
+import time
 import requests
 from typing import Dict, List, Optional, Any
 
 from domain.task import Task
 from domain.config import WorkerConfig
+from util.system import get_hw_accel_info_str
 
 logger = logging.getLogger(__name__)
 
@@ -23,6 +25,8 @@ class APIClientV2:
     负责与渲染服务端的所有 HTTP 通信。
     """
 
+    SYSTEM_INFO_TTL_SECONDS = 30
+
     def __init__(self, config: WorkerConfig):
         """
         初始化 API 客户端
@@ -36,6 +40,15 @@ class APIClientV2:
         self.worker_id = config.worker_id
         self.session = requests.Session()
 
+        self._ffmpeg_version: Optional[str] = None
+        self._codec_info: Optional[str] = None
+        self._hw_accel_info: Optional[str] = None
+        self._gpu_info: Optional[str] = None
+        self._gpu_info_checked = False
+        self._static_system_info: Optional[Dict[str, Any]] = None
+        self._system_info_cache: Optional[Dict[str, Any]] = None
+        self._system_info_cache_ts = 0.0
+
         # 设置默认请求头
         self.session.headers.update({
             'Content-Type': 'application/json',
@@ -80,7 +93,7 @@ class APIClientV2:
 
             # 解析任务列表
             tasks = []
-            for task_data in data.get('data', {}).get('tasks', []):
+            for task_data in data.get('data', {}).get('tasks') or []:
                 try:
                     task = Task.from_dict(task_data)
                     tasks.append(task)
@@ -286,6 +299,8 @@ class APIClientV2:
 
     def _get_ffmpeg_version(self) -> str:
         """获取 FFmpeg 版本"""
+        if self._ffmpeg_version is not None:
+            return self._ffmpeg_version
         try:
             result = subprocess.run(
                 ['ffmpeg', '-version'],
@@ -298,13 +313,18 @@ class APIClientV2:
                 parts = first_line.split()
                 for i, part in enumerate(parts):
                     if part == 'version' and i + 1 < len(parts):
-                        return parts[i + 1]
-            return 'unknown'
+                        self._ffmpeg_version = parts[i + 1]
+                        return self._ffmpeg_version
+            self._ffmpeg_version = 'unknown'
+            return self._ffmpeg_version
         except Exception:
-            return 'unknown'
+            self._ffmpeg_version = 'unknown'
+            return self._ffmpeg_version
 
     def _get_codec_info(self) -> str:
         """获取支持的编解码器信息"""
+        if self._codec_info is not None:
+            return self._codec_info
         try:
             result = subprocess.run(
                 ['ffmpeg', '-codecs'],
@@ -323,35 +343,60 @@ class APIClientV2:
                 codecs.append('aac')
             if 'libfdk_aac' in output:
                 codecs.append('libfdk_aac')
-            return ', '.join(codecs) if codecs else 'unknown'
+            self._codec_info = ', '.join(codecs) if codecs else 'unknown'
+            return self._codec_info
         except Exception:
-            return 'unknown'
+            self._codec_info = 'unknown'
+            return self._codec_info
 
     def _get_system_info(self) -> Dict[str, Any]:
         """获取系统信息"""
         try:
+            now = time.monotonic()
+            if (
+                self._system_info_cache
+                and now - self._system_info_cache_ts < self.SYSTEM_INFO_TTL_SECONDS
+            ):
+                return self._system_info_cache
+
             import platform
             import psutil
 
-            info = {
-                'os': platform.system(),
-                'cpu': f"{psutil.cpu_count()} cores",
-                'memory': f"{psutil.virtual_memory().total // (1024**3)}GB",
+            if self._hw_accel_info is None:
+                self._hw_accel_info = get_hw_accel_info_str()
+
+            if self._static_system_info is None:
+                self._static_system_info = {
+                    'os': platform.system(),
+                    'cpu': f"{psutil.cpu_count()} cores",
+                    'memory': f"{psutil.virtual_memory().total // (1024**3)}GB",
+                    'hwAccelConfig': self.config.hw_accel,  # 当前配置的硬件加速
+                    'hwAccelSupport': self._hw_accel_info,  # 系统支持的硬件加速
+                }
+
+            info = dict(self._static_system_info)
+            info.update({
                 'cpuUsage': f"{psutil.cpu_percent()}%",
-                'memoryAvailable': f"{psutil.virtual_memory().available // (1024**3)}GB"
-            }
+                'memoryAvailable': f"{psutil.virtual_memory().available // (1024**3)}GB",
+            })
 
             # 尝试获取 GPU 信息
             gpu_info = self._get_gpu_info()
             if gpu_info:
                 info['gpu'] = gpu_info
 
+            self._system_info_cache = info
+            self._system_info_cache_ts = now
             return info
         except Exception:
             return {}
 
     def _get_gpu_info(self) -> Optional[str]:
         """获取 GPU 信息"""
+        if self._gpu_info_checked:
+            return self._gpu_info
+
+        self._gpu_info_checked = True
         try:
             result = subprocess.run(
                 ['nvidia-smi', '--query-gpu=name', '--format=csv,noheader'],
@@ -361,10 +406,11 @@ class APIClientV2:
             )
             if result.returncode == 0:
                 gpu_name = result.stdout.strip().split('\n')[0]
-                return gpu_name
+                self._gpu_info = gpu_name
         except Exception:
-            pass
-        return None
+            self._gpu_info = None
+
+        return self._gpu_info
 
     def close(self):
         """关闭会话"""

services/cache.py

@@ -0,0 +1,513 @@
+# -*- coding: utf-8 -*-
+"""
+素材缓存服务
+
+提供素材下载缓存功能，避免相同素材重复下载。
+"""
+
+import json
+import os
+import hashlib
+import logging
+import shutil
+import time
+import uuid
+from typing import Optional, Tuple
+from urllib.parse import urlparse, unquote
+
+import psutil
+
+from services import storage
+
+logger = logging.getLogger(__name__)
+
+
+def _extract_cache_key(url: str) -> str:
+    """
+    从 URL 提取缓存键
+
+    去除签名等查询参数，保留路径作为唯一标识。
+
+    Args:
+        url: 完整的素材 URL
+
+    Returns:
+        缓存键（URL 路径的 MD5 哈希）
+    """
+    parsed = urlparse(url)
+    # 使用 scheme + host + path 作为唯一标识（忽略签名等查询参数）
+    cache_key_source = f"{parsed.scheme}://{parsed.netloc}{unquote(parsed.path)}"
+    return hashlib.md5(cache_key_source.encode('utf-8')).hexdigest()
+
+
+def _get_file_extension(url: str) -> str:
+    """
+    从 URL 提取文件扩展名
+
+    Args:
+        url: 素材 URL
+
+    Returns:
+        文件扩展名（如 .mp4, .png），无法识别时返回空字符串
+    """
+    parsed = urlparse(url)
+    path = unquote(parsed.path)
+    _, ext = os.path.splitext(path)
+    return ext.lower() if ext else ''
+
+
+class MaterialCache:
+    """
+    素材缓存管理器
+
+    负责素材文件的缓存存储和检索。
+    """
+
+    LOCK_TIMEOUT_SEC = 30.0
+    LOCK_POLL_INTERVAL_SEC = 0.1
+    LOCK_STALE_SECONDS = 24 * 60 * 60
+
+    def __init__(self, cache_dir: str, enabled: bool = True, max_size_gb: float = 0):
+        """
+        初始化缓存管理器
+
+        Args:
+            cache_dir: 缓存目录路径
+            enabled: 是否启用缓存
+            max_size_gb: 最大缓存大小（GB），0 表示不限制
+        """
+        self.cache_dir = cache_dir
+        self.enabled = enabled
+        self.max_size_bytes = int(max_size_gb * 1024 * 1024 * 1024) if max_size_gb > 0 else 0
+
+        if self.enabled:
+            os.makedirs(self.cache_dir, exist_ok=True)
+            logger.info(f"Material cache initialized: {cache_dir}")
+
+    def get_cache_path(self, url: str) -> str:
+        """
+        获取素材的缓存文件路径
+
+        Args:
+            url: 素材 URL
+
+        Returns:
+            缓存文件的完整路径
+        """
+        cache_key = _extract_cache_key(url)
+        ext = _get_file_extension(url)
+        filename = f"{cache_key}{ext}"
+        return os.path.join(self.cache_dir, filename)
+
+    def _get_lock_path(self, cache_key: str) -> str:
+        """获取缓存锁文件路径"""
+        assert self.cache_dir
+        return os.path.join(self.cache_dir, f"{cache_key}.lock")
+
+    def _write_lock_metadata(self, lock_fd: int, lock_path: str) -> bool:
+        """写入锁元数据，失败则清理锁文件"""
+        try:
+            try:
+                process_start_time = psutil.Process(os.getpid()).create_time()
+            except Exception as e:
+                process_start_time = None
+                logger.warning(f"Cache lock process start time error: {e}")
+            metadata = {
+                'pid': os.getpid(),
+                'process_start_time': process_start_time,
+                'created_at': time.time()
+            }
+            with os.fdopen(lock_fd, 'w', encoding='utf-8') as lock_file:
+                json.dump(metadata, lock_file)
+            return True
+        except Exception as e:
+            try:
+                os.close(lock_fd)
+            except Exception:
+                pass
+            self._remove_lock_file(lock_path, f"write metadata failed: {e}")
+            return False
+
+    def _read_lock_metadata(self, lock_path: str) -> Optional[dict]:
+        """读取锁元数据，失败返回 None（兼容历史空锁文件）"""
+        try:
+            with open(lock_path, 'r', encoding='utf-8') as lock_file:
+                data = json.load(lock_file)
+            return data if isinstance(data, dict) else None
+        except Exception:
+            return None
+
+    def _is_process_alive(self, pid: int, expected_start_time: Optional[float]) -> bool:
+        """判断进程是否存活并校验启动时间（防止 PID 复用）"""
+        try:
+            process = psutil.Process(pid)
+            if expected_start_time is None:
+                return process.is_running()
+            actual_start_time = process.create_time()
+            return abs(actual_start_time - expected_start_time) < 1.0
+        except psutil.NoSuchProcess:
+            return False
+        except Exception as e:
+            logger.warning(f"Cache lock process check error: {e}")
+            return True
+
+    def _is_lock_stale(self, lock_path: str) -> bool:
+        """判断锁是否过期（进程已退出或超过最大存活时长）"""
+        if not os.path.exists(lock_path):
+            return False
+        now = time.time()
+        metadata = self._read_lock_metadata(lock_path)
+        if metadata:
+            created_at = metadata.get('created_at')
+            if isinstance(created_at, (int, float)) and now - created_at > self.LOCK_STALE_SECONDS:
+                return True
+            pid = metadata.get('pid')
+            pid_value = int(pid) if isinstance(pid, int) or (isinstance(pid, str) and pid.isdigit()) else None
+            expected_start_time = metadata.get('process_start_time')
+            expected_start_time_value = (
+                expected_start_time if isinstance(expected_start_time, (int, float)) else None
+            )
+            if pid_value is not None and not self._is_process_alive(pid_value, expected_start_time_value):
+                return True
+            return self._is_lock_stale_by_mtime(lock_path, now)
+        return self._is_lock_stale_by_mtime(lock_path, now)
+
+    def _is_lock_stale_by_mtime(self, lock_path: str, now: float) -> bool:
+        """基于文件时间判断锁是否过期"""
+        try:
+            mtime = os.path.getmtime(lock_path)
+            return now - mtime > self.LOCK_STALE_SECONDS
+        except Exception as e:
+            logger.warning(f"Cache lock stat error: {e}")
+            return False
+
+    def _remove_lock_file(self, lock_path: str, reason: str = "") -> bool:
+        """删除锁文件"""
+        try:
+            os.remove(lock_path)
+            if reason:
+                logger.info(f"Cache lock removed: {lock_path} ({reason})")
+            return True
+        except FileNotFoundError:
+            return True
+        except Exception as e:
+            logger.warning(f"Cache lock remove error: {e}")
+            return False
+
+    def _acquire_lock(self, cache_key: str) -> Optional[str]:
+        """获取缓存锁（跨进程安全）"""
+        if not self.enabled:
+            return None
+
+        lock_path = self._get_lock_path(cache_key)
+        deadline = time.monotonic() + self.LOCK_TIMEOUT_SEC
+
+        while True:
+            try:
+                fd = os.open(lock_path, os.O_CREAT | os.O_EXCL | os.O_WRONLY)
+                if not self._write_lock_metadata(fd, lock_path):
+                    return None
+                return lock_path
+            except FileExistsError:
+                if self._is_lock_stale(lock_path):
+                    removed = self._remove_lock_file(lock_path, "stale lock")
+                    if removed:
+                        continue
+                if time.monotonic() >= deadline:
+                    logger.warning(f"Cache lock timeout: {lock_path}")
+                    return None
+                time.sleep(self.LOCK_POLL_INTERVAL_SEC)
+            except Exception as e:
+                logger.warning(f"Cache lock error: {e}")
+                return None
+
+    def _release_lock(self, lock_path: Optional[str]) -> None:
+        """释放缓存锁"""
+        if not lock_path:
+            return
+        self._remove_lock_file(lock_path)
+
+    def is_cached(self, url: str) -> Tuple[bool, str]:
+        """
+        检查素材是否已缓存
+
+        Args:
+            url: 素材 URL
+
+        Returns:
+            (是否已缓存, 缓存文件路径)
+        """
+        if not self.enabled:
+            return False, ''
+
+        cache_path = self.get_cache_path(url)
+        exists = os.path.exists(cache_path) and os.path.getsize(cache_path) > 0
+        return exists, cache_path
+
+    def get_or_download(
+        self,
+        url: str,
+        dest: str,
+        timeout: int = 300,
+        max_retries: int = 5
+    ) -> bool:
+        """
+        从缓存获取素材，若未缓存则下载并缓存
+
+        Args:
+            url: 素材 URL
+            dest: 目标文件路径（任务工作目录中的路径）
+            timeout: 下载超时时间（秒）
+            max_retries: 最大重试次数
+
+        Returns:
+            是否成功
+        """
+        # 确保目标目录存在
+        dest_dir = os.path.dirname(dest)
+        if dest_dir:
+            os.makedirs(dest_dir, exist_ok=True)
+
+        # 缓存未启用时直接下载
+        if not self.enabled:
+            return storage.download_file(url, dest, max_retries=max_retries, timeout=timeout)
+
+        cache_key = _extract_cache_key(url)
+        lock_path = self._acquire_lock(cache_key)
+        if not lock_path:
+            logger.warning(f"Cache lock unavailable, downloading without cache: {url[:80]}...")
+            return storage.download_file(url, dest, max_retries=max_retries, timeout=timeout)
+
+        try:
+            cache_path = self.get_cache_path(url)
+            cached = os.path.exists(cache_path) and os.path.getsize(cache_path) > 0
+
+            if cached:
+                # 命中缓存，复制到目标路径
+                try:
+                    shutil.copy2(cache_path, dest)
+                    # 更新访问时间（用于 LRU 清理）
+                    os.utime(cache_path, None)
+                    file_size = os.path.getsize(dest)
+                    logger.info(f"Cache hit: {url[:80]}... -> {dest} ({file_size} bytes)")
+                    return True
+                except Exception as e:
+                    logger.warning(f"Failed to copy from cache: {e}, will re-download")
+                    # 缓存复制失败，删除可能损坏的缓存文件
+                    try:
+                        os.remove(cache_path)
+                    except Exception:
+                        pass
+
+            # 未命中缓存，下载到缓存目录
+            logger.debug(f"Cache miss: {url[:80]}...")
+
+            # 先下载到临时文件（唯一文件名，避免并发覆盖）
+            temp_cache_path = os.path.join(
+                self.cache_dir,
+                f"{cache_key}.{uuid.uuid4().hex}.downloading"
+            )
+            try:
+                if not storage.download_file(url, temp_cache_path, max_retries=max_retries, timeout=timeout):
+                    # 下载失败，清理临时文件
+                    if os.path.exists(temp_cache_path):
+                        os.remove(temp_cache_path)
+                    return False
+
+                if not os.path.exists(temp_cache_path) or os.path.getsize(temp_cache_path) <= 0:
+                    if os.path.exists(temp_cache_path):
+                        os.remove(temp_cache_path)
+                    return False
+
+                # 下载成功，原子替换缓存文件
+                os.replace(temp_cache_path, cache_path)
+
+                # 复制到目标路径
+                shutil.copy2(cache_path, dest)
+                file_size = os.path.getsize(dest)
+                logger.info(f"Downloaded and cached: {url[:80]}... ({file_size} bytes)")
+
+                # 检查是否需要清理缓存
+                if self.max_size_bytes > 0:
+                    self._cleanup_if_needed()
+
+                return True
+
+            except Exception as e:
+                logger.error(f"Cache download error: {e}")
+                # 清理临时文件
+                if os.path.exists(temp_cache_path):
+                    try:
+                        os.remove(temp_cache_path)
+                    except Exception:
+                        pass
+                return False
+        finally:
+            self._release_lock(lock_path)
+
+    def add_to_cache(self, url: str, source_path: str) -> bool:
+        """
+        将本地文件添加到缓存
+
+        Args:
+            url: 对应的 URL（用于生成缓存键）
+            source_path: 本地文件路径
+
+        Returns:
+            是否成功
+        """
+        if not self.enabled:
+            return False
+
+        if not os.path.exists(source_path):
+            logger.warning(f"Source file not found for cache: {source_path}")
+            return False
+
+        cache_key = _extract_cache_key(url)
+        lock_path = self._acquire_lock(cache_key)
+        if not lock_path:
+            logger.warning(f"Cache lock unavailable for adding: {url[:80]}...")
+            return False
+
+        try:
+            cache_path = self.get_cache_path(url)
+
+            # 先复制到临时文件
+            temp_cache_path = os.path.join(
+                self.cache_dir,
+                f"{cache_key}.{uuid.uuid4().hex}.adding"
+            )
+
+            shutil.copy2(source_path, temp_cache_path)
+
+            # 原子替换
+            os.replace(temp_cache_path, cache_path)
+
+            # 更新访问时间
+            os.utime(cache_path, None)
+
+            logger.info(f"Added to cache: {url[:80]}... <- {source_path}")
+
+            # 检查清理
+            if self.max_size_bytes > 0:
+                self._cleanup_if_needed()
+
+            return True
+
+        except Exception as e:
+            logger.error(f"Failed to add to cache: {e}")
+            if 'temp_cache_path' in locals() and os.path.exists(temp_cache_path):
+                try:
+                    os.remove(temp_cache_path)
+                except Exception:
+                    pass
+            return False
+        finally:
+            self._release_lock(lock_path)
+
+    def _cleanup_if_needed(self) -> None:
+        """
+        检查并清理缓存（LRU 策略）
+
+        当缓存大小超过限制时，删除最久未访问的文件。
+        """
+        if self.max_size_bytes <= 0:
+            return
+
+        try:
+            # 获取所有缓存文件及其信息
+            cache_files = []
+            total_size = 0
+
+            for filename in os.listdir(self.cache_dir):
+                if filename.endswith('.downloading') or filename.endswith('.lock'):
+                    continue
+                file_path = os.path.join(self.cache_dir, filename)
+                if os.path.isfile(file_path):
+                    stat = os.stat(file_path)
+                    cache_files.append({
+                        'path': file_path,
+                        'size': stat.st_size,
+                        'atime': stat.st_atime
+                    })
+                    total_size += stat.st_size
+
+            # 如果未超过限制，无需清理
+            if total_size <= self.max_size_bytes:
+                return
+
+            # 按访问时间排序（最久未访问的在前）
+            cache_files.sort(key=lambda x: x['atime'])
+
+            # 删除文件直到低于限制的 80%
+            target_size = int(self.max_size_bytes * 0.8)
+            deleted_count = 0
+
+            for file_info in cache_files:
+                if total_size <= target_size:
+                    break
+                # 从文件名提取 cache_key，检查是否有锁（说明正在被使用）
+                filename = os.path.basename(file_info['path'])
+                cache_key = os.path.splitext(filename)[0]
+                lock_path = self._get_lock_path(cache_key)
+                if os.path.exists(lock_path):
+                    if self._is_lock_stale(lock_path):
+                        self._remove_lock_file(lock_path, "cleanup stale lock")
+                    else:
+                        # 该文件正在被其他任务使用，跳过删除
+                        logger.debug(f"Cache cleanup: skipping locked file {filename}")
+                        continue
+                try:
+                    os.remove(file_info['path'])
+                    total_size -= file_info['size']
+                    deleted_count += 1
+                except Exception as e:
+                    logger.warning(f"Failed to delete cache file: {e}")
+
+            if deleted_count > 0:
+                logger.info(f"Cache cleanup: deleted {deleted_count} files, current size: {total_size / (1024*1024*1024):.2f} GB")
+
+        except Exception as e:
+            logger.warning(f"Cache cleanup error: {e}")
+
+    def clear(self) -> None:
+        """清空所有缓存"""
+        if not self.enabled:
+            return
+
+        try:
+            if os.path.exists(self.cache_dir):
+                shutil.rmtree(self.cache_dir)
+                os.makedirs(self.cache_dir, exist_ok=True)
+                logger.info("Cache cleared")
+        except Exception as e:
+            logger.error(f"Failed to clear cache: {e}")
+
+    def get_stats(self) -> dict:
+        """
+        获取缓存统计信息
+
+        Returns:
+            包含缓存统计的字典
+        """
+        if not self.enabled or not os.path.exists(self.cache_dir):
+            return {'enabled': False, 'file_count': 0, 'total_size_mb': 0}
+
+        file_count = 0
+        total_size = 0
+
+        for filename in os.listdir(self.cache_dir):
+            if filename.endswith('.downloading') or filename.endswith('.lock'):
+                continue
+            file_path = os.path.join(self.cache_dir, filename)
+            if os.path.isfile(file_path):
+                file_count += 1
+                total_size += os.path.getsize(file_path)
+
+        return {
+            'enabled': True,
+            'cache_dir': self.cache_dir,
+            'file_count': file_count,
+            'total_size_mb': round(total_size / (1024 * 1024), 2),
+            'max_size_gb': self.max_size_bytes / (1024 * 1024 * 1024) if self.max_size_bytes > 0 else 0
+        }

services/gpu_scheduler.py

@@ -0,0 +1,164 @@
+# -*- coding: utf-8 -*-
+"""
+GPU 调度器
+
+提供多 GPU 设备的轮询调度功能。
+"""
+
+import logging
+import threading
+from typing import List, Optional
+
+from domain.config import WorkerConfig
+from domain.gpu import GPUDevice
+from util.system import get_all_gpu_info, validate_gpu_device
+from constant import HW_ACCEL_CUDA, HW_ACCEL_QSV
+
+logger = logging.getLogger(__name__)
+
+
+class GPUScheduler:
+    """
+    GPU 调度器
+
+    实现多 GPU 设备的轮询（Round Robin）调度。
+    线程安全，支持并发任务执行。
+
+    使用方式：
+        scheduler = GPUScheduler(config)
+
+        # 在任务执行时
+        device_index = scheduler.acquire()
+        try:
+            # 执行任务
+            pass
+        finally:
+            scheduler.release(device_index)
+    """
+
+    def __init__(self, config: WorkerConfig):
+        """
+        初始化调度器
+
+        Args:
+            config: Worker 配置
+        """
+        self._config = config
+        self._devices: List[GPUDevice] = []
+        self._next_index: int = 0
+        self._lock = threading.Lock()
+        self._enabled = False
+
+        # 初始化设备列表
+        self._init_devices()
+
+    def _init_devices(self) -> None:
+        """初始化 GPU 设备列表"""
+        # 仅在启用硬件加速时才初始化
+        if self._config.hw_accel not in (HW_ACCEL_CUDA, HW_ACCEL_QSV):
+            logger.info("Hardware acceleration not enabled, GPU scheduler disabled")
+            return
+
+        configured_devices = self._config.gpu_devices
+
+        if configured_devices:
+            # 使用配置指定的设备
+            self._devices = self._validate_configured_devices(configured_devices)
+        else:
+            # 自动检测所有设备
+            self._devices = self._auto_detect_devices()
+
+        if self._devices:
+            self._enabled = True
+            device_info = ', '.join(str(d) for d in self._devices)
+            logger.info(f"GPU scheduler initialized with {len(self._devices)} device(s): {device_info}")
+        else:
+            logger.warning("No GPU devices available, scheduler disabled")
+
+    def _validate_configured_devices(self, indices: List[int]) -> List[GPUDevice]:
+        """
+        验证配置的设备列表
+
+        Args:
+            indices: 配置的设备索引列表
+
+        Returns:
+            验证通过的设备列表
+        """
+        devices = []
+        for index in indices:
+            if validate_gpu_device(index):
+                devices.append(GPUDevice(
+                    index=index,
+                    name=f"GPU-{index}",
+                    available=True
+                ))
+            else:
+                logger.warning(f"GPU device {index} is not available, skipping")
+        return devices
+
+    def _auto_detect_devices(self) -> List[GPUDevice]:
+        """
+        自动检测所有可用 GPU
+
+        Returns:
+            检测到的设备列表
+        """
+        all_devices = get_all_gpu_info()
+        # 过滤不可用设备
+        return [d for d in all_devices if d.available]
+
+    @property
+    def enabled(self) -> bool:
+        """调度器是否启用"""
+        return self._enabled
+
+    @property
+    def device_count(self) -> int:
+        """设备数量"""
+        return len(self._devices)
+
+    def acquire(self) -> Optional[int]:
+        """
+        获取下一个可用的 GPU 设备（轮询调度）
+
+        Returns:
+            GPU 设备索引，如果调度器未启用或无设备则返回 None
+        """
+        if not self._enabled or not self._devices:
+            return None
+
+        with self._lock:
+            device = self._devices[self._next_index]
+            self._next_index = (self._next_index + 1) % len(self._devices)
+            logger.debug(f"Acquired GPU device: {device.index}")
+            return device.index
+
+    def release(self, device_index: Optional[int]) -> None:
+        """
+        释放 GPU 设备
+
+        当前实现为无状态轮询，此方法仅用于日志记录。
+
+        Args:
+            device_index: 设备索引
+        """
+        if device_index is not None:
+            logger.debug(f"Released GPU device: {device_index}")
+
+    def get_status(self) -> dict:
+        """
+        获取调度器状态信息
+
+        Returns:
+            状态字典
+        """
+        return {
+            'enabled': self._enabled,
+            'device_count': len(self._devices),
+            'devices': [
+                {'index': d.index, 'name': d.name, 'available': d.available}
+                for d in self._devices
+            ],
+            'hw_accel': self._config.hw_accel,
+        }

services/storage.py

@@ -7,13 +7,38 @@
 
 import os
 import logging
+import subprocess
 from typing import Optional
+from urllib.parse import unquote
 
 import requests
 
 logger = logging.getLogger(__name__)
 
 
+# 文件扩展名到 Content-Type 的映射
+_CONTENT_TYPE_MAP = {
+    '.mp4': 'video/mp4',
+    '.aac': 'audio/aac',
+    '.ts': 'video/mp2t',
+    '.m4a': 'audio/mp4',
+}
+
+
+def _get_content_type(file_path: str) -> str:
+    """
+    根据文件扩展名获取 Content-Type
+
+    Args:
+        file_path: 文件路径
+
+    Returns:
+        Content-Type 字符串
+    """
+    ext = os.path.splitext(file_path)[1].lower()
+    return _CONTENT_TYPE_MAP.get(ext, 'application/octet-stream')
+
+
 def _apply_http_replace_map(url: str) -> str:
     """
     应用 HTTP_REPLACE_MAP 环境变量替换 URL
@@ -61,6 +86,7 @@ def upload_file(url: str, file_path: str, max_retries: int = 5, timeout: int = 6
 
     # 检查是否使用 rclone 上传
     if os.getenv("UPLOAD_METHOD") == "rclone":
+        logger.info(f"Uploading to: {url}")
         result = _upload_with_rclone(url, file_path)
         if result:
             return True
@@ -68,21 +94,23 @@ def upload_file(url: str, file_path: str, max_retries: int = 5, timeout: int = 6
 
     # 应用 HTTP_REPLACE_MAP 替换 URL
     http_url = _apply_http_replace_map(url)
+    content_type = _get_content_type(file_path)
+    logger.info(f"Uploading to: {http_url} (Content-Type: {content_type})")
 
     retries = 0
     while retries < max_retries:
         try:
             with open(file_path, 'rb') as f:
-                response = requests.put(
+                with requests.put(
                     http_url,
                     data=f,
                     stream=True,
                     timeout=timeout,
-                    headers={"Content-Type": "application/octet-stream"}
-                )
-                response.raise_for_status()
-                logger.info(f"Upload succeeded: {file_path}")
-                return True
+                    headers={"Content-Type": content_type}
+                ) as response:
+                    response.raise_for_status()
+                    logger.info(f"Upload succeeded: {file_path}")
+                    return True
 
         except requests.exceptions.Timeout:
             retries += 1
@@ -111,7 +139,6 @@ def _upload_with_rclone(url: str, file_path: str) -> bool:
         return False
 
     config_file = os.getenv("RCLONE_CONFIG_FILE", "")
-    rclone_config = f"--config {config_file}" if config_file else ""
 
     # 替换 URL
     new_url = url
@@ -119,23 +146,35 @@ def _upload_with_rclone(url: str, file_path: str) -> bool:
     for src, dst in replace_list:
         new_url = new_url.replace(src, dst)
     new_url = new_url.split("?", 1)[0]  # 移除查询参数
+    new_url = unquote(new_url)  # 解码 URL 编码的字符（如 %2F -> /）
 
     if new_url == url:
         return False
 
-    cmd = (
-        f"rclone copyto --no-check-dest --ignore-existing "
-        f"--multi-thread-chunk-size 8M --multi-thread-streams 8 "
-        f"{rclone_config} {file_path} {new_url}"
-    )
-    logger.debug(f"rclone command: {cmd}")
+    cmd = [
+        "rclone",
+        "copyto",
+        "--no-check-dest",
+        "--ignore-existing",
+        "--multi-thread-chunk-size",
+        "8M",
+        "--multi-thread-streams",
+        "8",
+    ]
+    if config_file:
+        cmd.extend(["--config", config_file])
+    cmd.extend([file_path, new_url])
 
-    result = os.system(cmd)
-    if result == 0:
+    logger.debug(f"rclone command: {' '.join(cmd)}")
+
+    result = subprocess.run(cmd, capture_output=True, text=True)
+    if result.returncode == 0:
         logger.info(f"rclone upload succeeded: {file_path}")
         return True
 
-    logger.warning(f"rclone upload failed (code={result}): {file_path}")
+    stderr = (result.stderr or '').strip()
+    stderr = stderr[:500] if stderr else ""
+    logger.warning(f"rclone upload failed (code={result.returncode}): {file_path} {stderr}")
     return False
 
 
@@ -177,13 +216,13 @@ def download_file(
     retries = 0
     while retries < max_retries:
         try:
-            response = requests.get(http_url, timeout=timeout, stream=True)
-            response.raise_for_status()
+            with requests.get(http_url, timeout=timeout, stream=True) as response:
+                response.raise_for_status()
 
-            with open(file_path, 'wb') as f:
-                for chunk in response.iter_content(chunk_size=8192):
-                    if chunk:
-                        f.write(chunk)
+                with open(file_path, 'wb') as f:
+                    for chunk in response.iter_content(chunk_size=8192):
+                        if chunk:
+                            f.write(chunk)
 
             file_size = os.path.getsize(file_path)
             logger.info(f"Download succeeded: {file_path} ({file_size} bytes)")

services/task_executor.py

@@ -12,9 +12,16 @@ from typing import Dict, Optional, TYPE_CHECKING
 
 from domain.task import Task, TaskType
 from domain.result import TaskResult, ErrorCode
+
+# 需要 GPU 加速的任务类型
+GPU_REQUIRED_TASK_TYPES = {
+    TaskType.RENDER_SEGMENT_VIDEO,
+    TaskType.COMPOSE_TRANSITION,
+}
 from domain.config import WorkerConfig
 from core.handler import TaskHandler
 from services.lease_service import LeaseService
+from services.gpu_scheduler import GPUScheduler
 
 if TYPE_CHECKING:
     from services.api_client import APIClientV2
@@ -60,6 +67,12 @@ class TaskExecutor:
         # 线程安全锁
         self.lock = threading.Lock()
 
+        # GPU 调度器（如果启用硬件加速）
+        self.gpu_scheduler = GPUScheduler(config)
+
+        if self.gpu_scheduler.enabled:
+            logger.info(f"GPU scheduler enabled with {self.gpu_scheduler.device_count} device(s)")
+
         # 注册处理器
         self._register_handlers()
 
@@ -130,6 +143,14 @@ class TaskExecutor:
                 logger.warning(f"[task:{task.task_id}] Task already running, skipping")
                 return False
 
+            # 检查并发上限
+            if len(self.current_tasks) >= self.config.max_concurrency:
+                logger.info(
+                    f"[task:{task.task_id}] Max concurrency reached "
+                    f"({self.config.max_concurrency}), rejecting task"
+                )
+                return False
+
             # 检查是否有对应的处理器
             if task.task_type not in self.handlers:
                 logger.error(f"[task:{task.task_id}] No handler for type: {task.task_type.value}")
@@ -164,15 +185,28 @@ class TaskExecutor:
         )
         lease_service.start()
 
+        # 获取 GPU 设备（仅对需要 GPU 的任务类型）
+        device_index = None
+        needs_gpu = task.task_type in GPU_REQUIRED_TASK_TYPES
+        if needs_gpu and self.gpu_scheduler.enabled:
+            device_index = self.gpu_scheduler.acquire()
+            if device_index is not None:
+                logger.info(f"[task:{task_id}] Assigned to GPU device {device_index}")
+
+        # 获取处理器（需要在设置 GPU 设备前获取）
+        handler = self.handlers.get(task.task_type)
+
         try:
             # 报告任务开始
             self.api_client.report_start(task_id)
 
-            # 获取处理器
-            handler = self.handlers.get(task.task_type)
             if not handler:
                 raise ValueError(f"No handler for task type: {task.task_type}")
 
+            # 设置 GPU 设备（线程本地存储）
+            if device_index is not None:
+                handler.set_gpu_device(device_index)
+
             # 执行前钩子
             handler.before_handle(task)
 
@@ -196,6 +230,14 @@ class TaskExecutor:
             self.api_client.report_fail(task_id, 'E_UNKNOWN', str(e))
 
         finally:
+            # 清除 GPU 设备设置
+            if handler:
+                handler.clear_gpu_device()
+
+            # 释放 GPU 设备（仅当实际分配了设备时）
+            if device_index is not None:
+                self.gpu_scheduler.release(device_index)
+
             # 停止租约续期
             lease_service.stop()
 

tests/unit/test_material_cache_lock.py

@@ -0,0 +1,15 @@
+# -*- coding: utf-8 -*-
+
+import os
+
+from services.cache import MaterialCache, _extract_cache_key
+
+
+def test_cache_lock_acquire_release(tmp_path):
+    cache = MaterialCache(cache_dir=str(tmp_path), enabled=True, max_size_gb=0)
+    cache_key = _extract_cache_key("https://example.com/path/file.mp4?token=abc")
+    lock_path = cache._acquire_lock(cache_key)
+    assert lock_path
+    assert os.path.exists(lock_path)
+    cache._release_lock(lock_path)
+    assert not os.path.exists(lock_path)

util/system.py

@@ -5,13 +5,17 @@
 提供系统信息采集功能。
 """
 
+import logging
 import os
 import platform
 import subprocess
-from typing import Optional
+from typing import Optional, Dict, Any, List
 
 import psutil
-from constant import SOFTWARE_VERSION, DEFAULT_CAPABILITIES
+from constant import SOFTWARE_VERSION, DEFAULT_CAPABILITIES, HW_ACCEL_NONE, HW_ACCEL_QSV, HW_ACCEL_CUDA
+from domain.gpu import GPUDevice
+
+logger = logging.getLogger(__name__)
 
 
 def get_sys_info():
@@ -101,3 +105,241 @@ def get_ffmpeg_version() -> str:
         pass
 
     return 'unknown'
+
+
+def check_ffmpeg_encoder(encoder: str) -> bool:
+    """
+    检查 FFmpeg 是否支持指定的编码器
+
+    Args:
+        encoder: 编码器名称，如 'h264_nvenc', 'h264_qsv'
+
+    Returns:
+        bool: 是否支持该编码器
+    """
+    try:
+        result = subprocess.run(
+            ['ffmpeg', '-hide_banner', '-encoders'],
+            capture_output=True,
+            text=True,
+            timeout=5
+        )
+        if result.returncode == 0:
+            return encoder in result.stdout
+    except Exception:
+        pass
+
+    return False
+
+
+def check_ffmpeg_decoder(decoder: str) -> bool:
+    """
+    检查 FFmpeg 是否支持指定的解码器
+
+    Args:
+        decoder: 解码器名称，如 'h264_cuvid', 'h264_qsv'
+
+    Returns:
+        bool: 是否支持该解码器
+    """
+    try:
+        result = subprocess.run(
+            ['ffmpeg', '-hide_banner', '-decoders'],
+            capture_output=True,
+            text=True,
+            timeout=5
+        )
+        if result.returncode == 0:
+            return decoder in result.stdout
+    except Exception:
+        pass
+
+    return False
+
+
+def check_ffmpeg_hwaccel(hwaccel: str) -> bool:
+    """
+    检查 FFmpeg 是否支持指定的硬件加速方法
+
+    Args:
+        hwaccel: 硬件加速方法，如 'cuda', 'qsv', 'dxva2', 'd3d11va'
+
+    Returns:
+        bool: 是否支持该硬件加速方法
+    """
+    try:
+        result = subprocess.run(
+            ['ffmpeg', '-hide_banner', '-hwaccels'],
+            capture_output=True,
+            text=True,
+            timeout=5
+        )
+        if result.returncode == 0:
+            return hwaccel in result.stdout
+    except Exception:
+        pass
+
+    return False
+
+
+def detect_hw_accel_support() -> Dict[str, Any]:
+    """
+    检测系统的硬件加速支持情况
+
+    Returns:
+        dict: 硬件加速支持信息
+        {
+            'cuda': {
+                'available': bool,
+                'gpu': str or None,
+                'encoder': bool,  # h264_nvenc
+                'decoder': bool,  # h264_cuvid
+            },
+            'qsv': {
+                'available': bool,
+                'encoder': bool,  # h264_qsv
+                'decoder': bool,  # h264_qsv
+            },
+            'recommended': str  # 推荐的加速方式: 'cuda', 'qsv', 'none'
+        }
+    """
+    result = {
+        'cuda': {
+            'available': False,
+            'gpu': None,
+            'encoder': False,
+            'decoder': False,
+        },
+        'qsv': {
+            'available': False,
+            'encoder': False,
+            'decoder': False,
+        },
+        'recommended': HW_ACCEL_NONE
+    }
+
+    # 检测 CUDA/NVENC 支持
+    gpu_info = get_gpu_info()
+    if gpu_info:
+        result['cuda']['gpu'] = gpu_info
+        result['cuda']['available'] = check_ffmpeg_hwaccel('cuda')
+        result['cuda']['encoder'] = check_ffmpeg_encoder('h264_nvenc')
+        result['cuda']['decoder'] = check_ffmpeg_decoder('h264_cuvid')
+
+    # 检测 QSV 支持
+    result['qsv']['available'] = check_ffmpeg_hwaccel('qsv')
+    result['qsv']['encoder'] = check_ffmpeg_encoder('h264_qsv')
+    result['qsv']['decoder'] = check_ffmpeg_decoder('h264_qsv')
+
+    # 推荐硬件加速方式（优先 CUDA，其次 QSV）
+    if result['cuda']['available'] and result['cuda']['encoder']:
+        result['recommended'] = HW_ACCEL_CUDA
+    elif result['qsv']['available'] and result['qsv']['encoder']:
+        result['recommended'] = HW_ACCEL_QSV
+
+    return result
+
+
+def get_hw_accel_info_str() -> str:
+    """
+    获取硬件加速支持信息的可读字符串
+
+    Returns:
+        str: 硬件加速支持信息描述
+    """
+    support = detect_hw_accel_support()
+
+    parts = []
+
+    if support['cuda']['available']:
+        gpu = support['cuda']['gpu'] or 'Unknown GPU'
+        status = 'encoder+decoder' if support['cuda']['encoder'] and support['cuda']['decoder'] else (
+            'encoder only' if support['cuda']['encoder'] else 'decoder only' if support['cuda']['decoder'] else 'hwaccel only'
+        )
+        parts.append(f"CUDA({gpu}, {status})")
+
+    if support['qsv']['available']:
+        status = 'encoder+decoder' if support['qsv']['encoder'] and support['qsv']['decoder'] else (
+            'encoder only' if support['qsv']['encoder'] else 'decoder only' if support['qsv']['decoder'] else 'hwaccel only'
+        )
+        parts.append(f"QSV({status})")
+
+    if not parts:
+        return "No hardware acceleration available"
+
+    return ', '.join(parts) + f" [recommended: {support['recommended']}]"
+
+
+def get_all_gpu_info() -> List[GPUDevice]:
+    """
+    获取所有 NVIDIA GPU 信息
+
+    使用 nvidia-smi 查询所有 GPU 设备。
+
+    Returns:
+        GPU 设备列表，失败返回空列表
+    """
+    try:
+        result = subprocess.run(
+            [
+                'nvidia-smi',
+                '--query-gpu=index,name,memory.total',
+                '--format=csv,noheader,nounits'
+            ],
+            capture_output=True,
+            text=True,
+            timeout=10
+        )
+
+        if result.returncode != 0:
+            return []
+
+        devices = []
+        for line in result.stdout.strip().split('\n'):
+            if not line.strip():
+                continue
+            parts = [p.strip() for p in line.split(',')]
+            if len(parts) >= 2:
+                index = int(parts[0])
+                name = parts[1]
+                memory = int(parts[2]) if len(parts) >= 3 else None
+                devices.append(GPUDevice(
+                    index=index,
+                    name=name,
+                    memory_total=memory,
+                    available=True
+                ))
+
+        return devices
+
+    except Exception as e:
+        logger.warning(f"Failed to detect GPUs: {e}")
+        return []
+
+
+def validate_gpu_device(index: int) -> bool:
+    """
+    验证指定索引的 GPU 设备是否可用
+
+    Args:
+        index: GPU 设备索引
+
+    Returns:
+        设备是否可用
+    """
+    try:
+        result = subprocess.run(
+            [
+                'nvidia-smi',
+                '-i', str(index),
+                '--query-gpu=name',
+                '--format=csv,noheader'
+            ],
+            capture_output=True,
+            text=True,
+            timeout=5
+        )
+        return result.returncode == 0 and bool(result.stdout.strip())
+    except Exception:
+        return False
+