Files
xiaoxia-saas/apps/worker/video_processing/speed_engine.py
T
xiaoxia 39187a0660
CI/CD Pipeline / Check if frontend-only change (push) Has been cancelled
CI/CD Pipeline / Validate - Code Quality (push) Has been cancelled
CI/CD Pipeline / Validate - Type Check (mypy) (push) Has been cancelled
CI/CD Pipeline / Validate - Migration (alembic) (push) Has been cancelled
CI/CD Pipeline / Unit Tests (push) Has been cancelled
CI/CD Pipeline / Integration Tests (push) Has been cancelled
CI/CD Pipeline / Frontend Lint (push) Has been cancelled
CI/CD Pipeline / Frontend Unit Tests (push) Has been cancelled
CI/CD Pipeline / PR Build API Image (push) Has been cancelled
CI/CD Pipeline / PR Build Web Image (push) Has been cancelled
CI/CD Pipeline / PR Build Worker Image (push) Has been cancelled
CI/CD Pipeline / Build Staging API Image (push) Has been cancelled
CI/CD Pipeline / Build Staging Web Image (push) Has been cancelled
CI/CD Pipeline / Build Staging Worker Image (push) Has been cancelled
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (push) Has been cancelled
CI/CD Pipeline / Staging E2E Tests (push) Has been cancelled
CI/CD Pipeline / Staging API Integration Tests (push) Has been cancelled
CI/CD Pipeline / Build Production API Image (push) Has been cancelled
CI/CD Pipeline / Build Production Web Image (push) Has been cancelled
CI/CD Pipeline / Build Production Worker Image (push) Has been cancelled
CI/CD Pipeline / Deploy Production (push) Has been cancelled
CI/CD Pipeline / Production Browser E2E (push) Has been cancelled
CI/CD Pipeline / ACR Image Cleanup (push) Has been cancelled
CI/CD Pipeline / Canary Release to Production (push) Has been cancelled
CI/CD Pipeline / CI Gate (push) Has been cancelled
fix: 修复 20 个 ruff lint 错误,解除所有 PR CI 阻塞
F401 (14): 移除未使用的 import
  - multi_track_mixer_pure.py: Any, Optional, math
  - pip_engine_pure.py: Any
  - speed_engine.py: Optional, MAX_SPEED, MIN_SPEED
  - url_security.py: ALLOWED_AUDIO_MIME_TYPES, ALLOWED_IMAGE_MIME_TYPES,
    ALLOWED_VIDEO_MIME_TYPES, MAGIC_NUMBERS, MAX_URL_LENGTH,
    check_internal_hostname, is_trusted_domain

E741 (3): 重命名模糊变量 l → layer
  - pip_engine_pure.py: lambda 参数
  - test_pip_engine_pure.py: 两处列表推导

B017 (2): pytest.raises(Exception) → FrozenInstanceError
  - test_video_filter_builder.py
  - test_transition_presets.py

B905 (1): zip() 补充 strict=True
  - pip_engine_pure.py
2026-07-28 16:03:01 +08:00

72 lines
2.5 KiB
Python
Executable File
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""视频调速引擎 — 基于 FFmpeg setpts + atempo 的速度调整能力.
支持:
- 0.25x ~ 4x 变速范围
- 视频调速(setpts
- 音频调速(atempo,多级串联处理超范围值)
- 音调修正(pitch_correct,默认开启)
- 边界自动钳制,不阻断渲染
注:核心领域模型已抽离到 packages/domain/speed_config.py
本模块保留薄包装层,确保向后兼容。
"""
from __future__ import annotations
from packages.domain.speed_config import (
DEFAULT_SPEED,
SpeedConfig,
_split_atempo_stages,
)
from packages.domain.speed_config import adjust_duration as _adjust_duration_base # noqa: F401 — 向后兼容
from packages.domain.speed_config import build_audio_filter as _build_audio_filter_base
from packages.domain.speed_config import build_video_filter as _build_video_filter_base
from packages.domain.speed_config import resolve_clip_speed as _resolve_clip_speed_base
class SpeedEngine:
"""调速引擎 — 生成 FFmpeg 调速滤镜链.
薄包装层,实际逻辑委托给 packages.domain.speed_config。
"""
def build_video_filter(self, config: SpeedConfig) -> str:
"""生成视频调速滤镜字符串."""
return _build_video_filter_base(config)
def build_audio_filter(self, config: SpeedConfig) -> str:
"""生成音频调速滤镜字符串."""
return _build_audio_filter_base(config)
@staticmethod
def _split_atempo_stages(speed: float) -> list[float]:
"""将速度拆分为多级 atempo 串联(内部方法,向后兼容)."""
return _split_atempo_stages(speed)
def adjust_duration(self, original_duration: float, config: SpeedConfig) -> float:
"""计算调速后的时长."""
return _adjust_duration_base(original_duration, config)
def build_clip_speed_filter(
self,
speed: float,
pitch_correct: bool = True,
) -> tuple[str, str, SpeedConfig]:
"""便捷方法:从单一 speed 值生成视频+音频滤镜."""
config = SpeedConfig(speed=speed, pitch_correct=pitch_correct)
config.clamp()
return (
self.build_video_filter(config),
self.build_audio_filter(config),
config,
)
@staticmethod
def resolve_clip_speed(
clip_config: dict,
global_speed: float = DEFAULT_SPEED,
) -> float:
"""从 clip config 中解析 playback_speed0 或缺失则使用全局速度."""
return _resolve_clip_speed_base(clip_config, global_speed)