Compare commits

..

1 Commits

Author SHA1 Message Date
AI Bot fc11719e7d test(wave126): 小领域模块组合包单测 +53
CI/CD Pipeline / Check if frontend-only change (pull_request) Successful in 9s
CI/CD Pipeline / Validate - Code Quality (pull_request) Failing after 1m31s
CI/CD Pipeline / Validate - Type Check (mypy) (pull_request) Successful in 58s
CI/CD Pipeline / Frontend Lint (pull_request) Failing after 34s
CI/CD Pipeline / Validate - Migration (alembic) (pull_request) Successful in 1m7s
CI/CD Pipeline / PR Build Web Image (pull_request) Successful in 1m29s
CI/CD Pipeline / PR Build API Image (pull_request) Successful in 1m59s
AI Code Review / AI Code Review (pull_request) Successful in 2m2s
PR Automation / Auto Approve on CI Green (pull_request) Successful in 2m48s
CI/CD Pipeline / PR Build Worker Image (pull_request) Successful in 5m11s
Preview Deploy / Deploy Preview Environment (pull_request) Successful in 37s
CI/CD Pipeline / Unit Tests (pull_request) Failing after 0s
CI/CD Pipeline / Integration Tests (pull_request) Failing after 1s
CI/CD Pipeline / CI Gate (pull_request) Failing after 0s
PR Automation / Auto Merge on CI Green + Approved (pull_request) Successful in 27m48s
ACR Cleanup / ACR Image Cleanup (pull_request_target) Failing after 1s
Preview Cleanup / Cleanup Preview Environment (pull_request) Successful in 42s
CI/CD Pipeline / Canary Release to Production (pull_request) Has been cancelled
CI/CD Pipeline / Production Browser E2E (pull_request) Failing after 1218h23m58s
CI/CD Pipeline / ACR Image Cleanup (pull_request) Failing after 1218h24m45s
CI/CD Pipeline / Staging API Integration Tests (pull_request) Failing after 1218h24m46s
CI/CD Pipeline / Staging E2E Tests (pull_request) Failing after 1218h24m46s
CI/CD Pipeline / Deploy Production (pull_request) Failing after 1218h24m47s
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (pull_request) Failing after 1218h30m8s
CI/CD Pipeline / Build Production Worker Image (pull_request) Failing after 1218h30m10s
CI/CD Pipeline / Build Production Web Image (pull_request) Failing after 1218h30m12s
CI/CD Pipeline / Build Production API Image (pull_request) Failing after 1218h30m14s
CI/CD Pipeline / Frontend Unit Tests (pull_request) Failing after 1219h27m15s
CI/CD Pipeline / Build Staging Worker Image (pull_request) Failing after 1219h59m0s
CI/CD Pipeline / Build Staging API Image (pull_request) Failing after 1219h59m4s
CI/CD Pipeline / Build Staging Web Image (pull_request) Failing after 1220h31m49s
覆盖EditingMode/异常类/Tag/ClipType/TemplateClipConfig
2026-07-27 18:22:18 +08:00
8 changed files with 774 additions and 815 deletions
-16
View File
@@ -1,16 +0,0 @@
import React from "react"
import { Navigate } from "react-router-dom"
import { useAuthStore } from "@/store/authStore"
/** 受保护的路由组件 */
// eslint-disable-next-line react-refresh/only-export-components
export const ProtectedRoute = ({ children }: { children: React.ReactNode }) => {
const isAuthenticated = useAuthStore((state) => state.isAuthenticated)
const hasAccessToken = Boolean(localStorage.getItem("access_token"))
if (!isAuthenticated || !hasAccessToken) {
return <Navigate to="/login" replace />
}
return <>{children}</>
}
-225
View File
@@ -1,225 +0,0 @@
import { Navigate, type RouteObject } from "react-router-dom"
import MainLayout from "@/components/layout/MainLayout"
import { ProtectedRoute } from "./ProtectedRoute"
/**
* 受保护的 /app 子路由
* 所有页面使用 lazy 懒加载
*/
const appChildren: RouteObject[] = [
{
index: true,
element: <Navigate to="/app/dashboard" replace />,
},
{
path: "dashboard",
lazy: () =>
import("@/pages/dashboard/Dashboard").then((m) => ({
Component: m.default,
})),
},
{
path: "assets",
lazy: () =>
import("@/pages/assets/AssetLibrary").then((m) => ({
Component: m.default,
})),
},
{
path: "titles",
lazy: () =>
import("@/pages/titles/TitleLibrary").then((m) => ({
Component: m.default,
})),
},
{
path: "voices",
lazy: () =>
import("@/pages/voices/VoiceLibrary").then((m) => ({
Component: m.default,
})),
},
{
path: "templates",
lazy: () =>
import("@/pages/templates/TemplateLibrary").then((m) => ({
Component: m.default,
})),
},
{
path: "generate",
lazy: () =>
import("@/pages/generate/GeneratePage").then((m) => ({
Component: m.default,
})),
},
{
path: "history",
lazy: () =>
import("@/pages/history/TaskHistory").then((m) => ({
Component: m.default,
})),
},
{
path: "products",
lazy: () =>
import("@/pages/products/ProductLibrary").then((m) => ({
Component: m.default,
})),
},
{
path: "products/:id",
lazy: () =>
import("@/pages/products/ProductDetail").then((m) => ({
Component: m.default,
})),
},
{
path: "tasks",
lazy: () =>
import("@/pages/tasks/TaskCenter").then((m) => ({
Component: m.default,
})),
},
{
path: "editing-planner",
lazy: () =>
import("@/pages/editing-planner/EditingPlanner").then((m) => ({
Component: m.default,
})),
},
{
path: "my-templates",
lazy: () =>
import("@/pages/my-templates/MyTemplates").then((m) => ({
Component: m.default,
})),
},
{
path: "voice-clone",
lazy: () =>
import("@/pages/voice-clone/VoiceClone").then((m) => ({
Component: m.default,
})),
},
{
path: "voice-materials",
lazy: () =>
import("@/pages/voice-materials/VoiceMaterialLibrary").then((m) => ({
Component: m.default,
})),
},
{
path: "my-voices",
lazy: () =>
import("@/pages/my-voices/MyVoices").then((m) => ({
Component: m.default,
})),
},
{
path: "accounts",
lazy: () =>
import("@/pages/accounts/Accounts").then((m) => ({
Component: m.default,
})),
},
{
path: "duplication",
lazy: () =>
import("@/pages/duplication/DuplicationUpload").then((m) => ({
Component: m.default,
})),
},
{
path: "duplication/results",
lazy: () =>
import("@/pages/duplication/DuplicationResults").then((m) => ({
Component: m.default,
})),
},
{
path: "duplication/:id",
lazy: () =>
import("@/pages/duplication/DuplicationDetail").then((m) => ({
Component: m.default,
})),
},
{
path: "subscription",
lazy: () =>
import("@/pages/subscription/Plans").then((m) => ({
Component: m.default,
})),
},
{
path: "subscription/upgrade",
lazy: () =>
import("@/pages/subscription/UpgradeSubscription").then((m) => ({
Component: m.default,
})),
},
{
path: "subscription/billing",
lazy: () =>
import("@/pages/subscription/Billing").then((m) => ({
Component: m.default,
})),
},
{
path: "profile",
lazy: () =>
import("@/pages/profile/Settings").then((m) => ({
Component: m.default,
})),
},
{
path: "admin",
children: [
{
index: true,
lazy: () =>
import("@/pages/admin/AdminComingSoon").then((m) => ({
Component: m.default,
})),
},
{
path: "users",
lazy: () =>
import("@/pages/admin/AdminComingSoon").then((m) => ({
Component: m.default,
})),
},
{
path: "analytics",
lazy: () =>
import("@/pages/admin/AdminComingSoon").then((m) => ({
Component: m.default,
})),
},
{
path: "monitor",
lazy: () =>
import("@/pages/admin/AdminComingSoon").then((m) => ({
Component: m.default,
})),
},
{
path: "logs",
lazy: () =>
import("@/pages/admin/AdminComingSoon").then((m) => ({
Component: m.default,
})),
},
],
},
]
export const appRoutes: RouteObject = {
path: "/app",
element: (
<ProtectedRoute>
<MainLayout />
</ProtectedRoute>
),
children: appChildren,
}
+274 -4
View File
@@ -3,13 +3,283 @@
* 扁平化路由:去掉 Project 层级,所有资源直接归属用户
*/
import { createBrowserRouter, Navigate } from "react-router-dom"
import { publicRoutes } from "./publicRoutes"
import { appRoutes } from "./appRoutes"
import React from "react"
import MainLayout from "@/components/layout/MainLayout"
import Login from "@/pages/auth/Login"
import Register from "@/pages/auth/Register"
import ForgotPassword from "@/pages/auth/ForgotPassword"
import ResetPassword from "@/pages/auth/ResetPassword"
import WechatCallback from "@/pages/auth/WechatCallback"
import HomePage from "@/pages/home/HomePage"
import { useAuthStore } from "@/store/authStore"
/** 受保护的路由组件 */
// eslint-disable-next-line react-refresh/only-export-components
const ProtectedRoute = ({ children }: { children: React.ReactNode }) => {
const isAuthenticated = useAuthStore((state) => state.isAuthenticated)
const hasAccessToken = Boolean(localStorage.getItem("access_token"))
if (!isAuthenticated || !hasAccessToken) {
return <Navigate to="/login" replace />
}
return <>{children}</>
}
/** 首页路由组件:已登录跳 dashboard,未登录显示落地页 */
// eslint-disable-next-line react-refresh/only-export-components
const HomeRoute: React.FC = () => {
const isAuthenticated = useAuthStore((state) => state.isAuthenticated)
const hasAccessToken = Boolean(localStorage.getItem("access_token"))
if (isAuthenticated && hasAccessToken) {
return <Navigate to="/app/dashboard" replace />
}
return <HomePage />
}
/** 路由配置 */
export const router = createBrowserRouter([
...publicRoutes,
appRoutes,
{
path: "/",
element: <HomeRoute />,
},
{
path: "/login",
element: <Login />,
},
{
path: "/register",
element: <Register />,
},
{
path: "/forgot-password",
element: <ForgotPassword />,
},
{
path: "/reset-password",
element: <ResetPassword />,
},
{
path: "/auth/wechat/callback",
element: <WechatCallback />,
},
{
path: "/app",
element: (
<ProtectedRoute>
<MainLayout />
</ProtectedRoute>
),
children: [
{
index: true,
element: <Navigate to="/app/dashboard" replace />,
},
{
path: "dashboard",
lazy: () =>
import("@/pages/dashboard/Dashboard").then((m) => ({
Component: m.default,
})),
},
{
path: "assets",
lazy: () =>
import("@/pages/assets/AssetLibrary").then((m) => ({
Component: m.default,
})),
},
{
path: "titles",
lazy: () =>
import("@/pages/titles/TitleLibrary").then((m) => ({
Component: m.default,
})),
},
{
path: "voices",
lazy: () =>
import("@/pages/voices/VoiceLibrary").then((m) => ({
Component: m.default,
})),
},
{
path: "templates",
lazy: () =>
import("@/pages/templates/TemplateLibrary").then((m) => ({
Component: m.default,
})),
},
{
path: "generate",
lazy: () =>
import("@/pages/generate/GeneratePage").then((m) => ({
Component: m.default,
})),
},
{
path: "history",
lazy: () =>
import("@/pages/history/TaskHistory").then((m) => ({
Component: m.default,
})),
},
{
path: "products",
lazy: () =>
import("@/pages/products/ProductLibrary").then((m) => ({
Component: m.default,
})),
},
{
path: "products/:id",
lazy: () =>
import("@/pages/products/ProductDetail").then((m) => ({
Component: m.default,
})),
},
{
path: "tasks",
lazy: () =>
import("@/pages/tasks/TaskCenter").then((m) => ({
Component: m.default,
})),
},
{
path: "editing-planner",
lazy: () =>
import("@/pages/editing-planner/EditingPlanner").then((m) => ({
Component: m.default,
})),
},
{
path: "my-templates",
lazy: () =>
import("@/pages/my-templates/MyTemplates").then((m) => ({
Component: m.default,
})),
},
{
path: "voice-clone",
lazy: () =>
import("@/pages/voice-clone/VoiceClone").then((m) => ({
Component: m.default,
})),
},
{
path: "voice-materials",
lazy: () =>
import("@/pages/voice-materials/VoiceMaterialLibrary").then((m) => ({
Component: m.default,
})),
},
{
path: "my-voices",
lazy: () =>
import("@/pages/my-voices/MyVoices").then((m) => ({
Component: m.default,
})),
},
{
path: "accounts",
lazy: () =>
import("@/pages/accounts/Accounts").then((m) => ({
Component: m.default,
})),
},
{
path: "duplication",
lazy: () =>
import("@/pages/duplication/DuplicationUpload").then((m) => ({
Component: m.default,
})),
},
{
path: "duplication/results",
lazy: () =>
import("@/pages/duplication/DuplicationResults").then((m) => ({
Component: m.default,
})),
},
{
path: "duplication/:id",
lazy: () =>
import("@/pages/duplication/DuplicationDetail").then((m) => ({
Component: m.default,
})),
},
{
path: "subscription",
lazy: () =>
import("@/pages/subscription/Plans").then((m) => ({
Component: m.default,
})),
},
{
path: "subscription/upgrade",
lazy: () =>
import("@/pages/subscription/UpgradeSubscription").then((m) => ({
Component: m.default,
})),
},
{
path: "subscription/billing",
lazy: () =>
import("@/pages/subscription/Billing").then((m) => ({
Component: m.default,
})),
},
{
path: "profile",
lazy: () =>
import("@/pages/profile/Settings").then((m) => ({
Component: m.default,
})),
},
{
path: "admin",
children: [
{
index: true,
lazy: () =>
import("@/pages/admin/AdminComingSoon").then((m) => ({
Component: m.default,
})),
},
{
path: "users",
lazy: () =>
import("@/pages/admin/AdminComingSoon").then((m) => ({
Component: m.default,
})),
},
{
path: "analytics",
lazy: () =>
import("@/pages/admin/AdminComingSoon").then((m) => ({
Component: m.default,
})),
},
{
path: "monitor",
lazy: () =>
import("@/pages/admin/AdminComingSoon").then((m) => ({
Component: m.default,
})),
},
{
path: "logs",
lazy: () =>
import("@/pages/admin/AdminComingSoon").then((m) => ({
Component: m.default,
})),
},
],
},
],
},
{
path: "*",
element: <Navigate to="/" replace />,
-48
View File
@@ -1,48 +0,0 @@
import { Navigate, type RouteObject } from "react-router-dom"
import HomePage from "@/pages/home/HomePage"
import Login from "@/pages/auth/Login"
import Register from "@/pages/auth/Register"
import ForgotPassword from "@/pages/auth/ForgotPassword"
import ResetPassword from "@/pages/auth/ResetPassword"
import WechatCallback from "@/pages/auth/WechatCallback"
import { useAuthStore } from "@/store/authStore"
/** 首页路由组件:已登录跳 dashboard,未登录显示落地页 */
// eslint-disable-next-line react-refresh/only-export-components
const HomeRoute: React.FC = () => {
const isAuthenticated = useAuthStore((state) => state.isAuthenticated)
const hasAccessToken = Boolean(localStorage.getItem("access_token"))
if (isAuthenticated && hasAccessToken) {
return <Navigate to="/app/dashboard" replace />
}
return <HomePage />
}
export const publicRoutes: RouteObject[] = [
{
path: "/",
element: <HomeRoute />,
},
{
path: "/login",
element: <Login />,
},
{
path: "/register",
element: <Register />,
},
{
path: "/forgot-password",
element: <ForgotPassword />,
},
{
path: "/reset-password",
element: <ResetPassword />,
},
{
path: "/auth/wechat/callback",
element: <WechatCallback />,
},
]
+123 -30
View File
@@ -1,4 +1,4 @@
"""视频调速引擎 — 基于 FFmpeg setpts + atempo 的速度调整能力.
"""视频调速引擎 — 基于 FFmpeg setpts + atempo 的速度调整能力。
支持:
- 0.25x ~ 4x 变速范围
@@ -6,57 +6,147 @@
- 音频调速(atempo,多级串联处理超范围值)
- 音调修正(pitch_correct,默认开启)
- 边界自动钳制,不阻断渲染
注:核心领域模型已抽离到 packages/domain/speed_config.py,
本模块保留薄包装层,确保向后兼容。
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import Optional
from packages.domain.speed_config import ( # noqa: F401 — 向后兼容
DEFAULT_SPEED,
MAX_SPEED,
MIN_SPEED,
SpeedConfig,
_split_atempo_stages,
adjust_duration as _adjust_duration_base,
build_audio_filter as _build_audio_filter_base,
build_video_filter as _build_video_filter_base,
resolve_clip_speed as _resolve_clip_speed_base,
)
# ─── 常量 ───────────────────────────────────────────────
MIN_SPEED = 0.25
MAX_SPEED = 4.0
DEFAULT_SPEED = 1.0
# atempo 单级有效范围
_ATEMPO_MIN = 0.5
_ATEMPO_MAX = 2.0
@dataclass
class SpeedConfig:
"""调速配置。
Attributes:
speed: 播放速度,0.25~4.0,1.0 为原速
pitch_correct: 是否保持音调(默认 True,用 atempo 时间拉伸算法)
"""
speed: float = DEFAULT_SPEED
pitch_correct: bool = True
@classmethod
def parse(cls, data: Optional[dict]) -> "SpeedConfig":
"""从 dict 解析配置,无效值回退到默认。"""
if not data or not isinstance(data, dict):
return cls()
speed = data.get("speed", DEFAULT_SPEED)
if not isinstance(speed, (int, float)):
speed = DEFAULT_SPEED
pitch_correct = data.get("pitch_correct", True)
if not isinstance(pitch_correct, bool):
pitch_correct = True
config = cls(speed=float(speed), pitch_correct=pitch_correct)
config.clamp()
return config
def clamp(self) -> None:
"""将速度钳制到合法范围。"""
if self.speed <= 0:
self.speed = DEFAULT_SPEED
elif self.speed < MIN_SPEED:
self.speed = MIN_SPEED
elif self.speed > MAX_SPEED:
self.speed = MAX_SPEED
@property
def is_original(self) -> bool:
"""是否原速(无需调速)。"""
return abs(self.speed - 1.0) < 1e-6
class SpeedEngine:
"""调速引擎 — 生成 FFmpeg 调速滤镜链.
"""调速引擎 — 生成 FFmpeg 调速滤镜链。
薄包装层,实际逻辑委托给 packages.domain.speed_config。
用法:
engine = SpeedEngine()
video_filter = engine.build_video_filter(config)
audio_filter = engine.build_audio_filter(config)
new_duration = engine.adjust_duration(duration, config)
"""
def build_video_filter(self, config: SpeedConfig) -> str:
"""生成视频调速滤镜字符串."""
return _build_video_filter_base(config)
"""生成视频调速滤镜字符串。
返回 setpts 滤镜表达式,原速时返回空字符串。
"""
if config.is_original:
return ""
# setpts=PTS/speed — speed>1 加速,speed<1 减速
return f"setpts=PTS/{config.speed:.4f}"
def build_audio_filter(self, config: SpeedConfig) -> str:
"""生成音频调速滤镜字符串."""
return _build_audio_filter_base(config)
"""生成音频调速滤镜字符串。
atempo 单级范围 0.5~2.0,超出范围时自动多级串联:
- 0.25x → atempo=0.5,atempo=0.5
- 4x → atempo=2.0,atempo=2.0
- 0.3x → atempo=0.5,atempo=0.6
- 3x → atempo=2.0,atempo=1.5
原速时返回空字符串。
"""
if config.is_original:
return ""
speed = config.speed
stages: list[float] = self._split_atempo_stages(speed)
return ",".join(f"atempo={s:.4f}" for s in stages)
@staticmethod
def _split_atempo_stages(speed: float) -> list[float]:
"""将速度拆分为多级 atempo 串联(内部方法,向后兼容)."""
return _split_atempo_stages(speed)
"""将速度拆分为多级 atempo 串联,每级都在 [0.5, 2.0] 范围内。"""
if _ATEMPO_MIN <= speed <= _ATEMPO_MAX:
return [speed]
stages: list[float] = []
remaining = speed
# 加速场景(speed > 2.0)
if speed > _ATEMPO_MAX:
while remaining > _ATEMPO_MAX:
stages.append(_ATEMPO_MAX)
remaining /= _ATEMPO_MAX
stages.append(remaining)
# 减速场景(speed < 0.5)
else:
while remaining < _ATEMPO_MIN:
stages.append(_ATEMPO_MIN)
remaining /= _ATEMPO_MIN
stages.append(remaining)
return stages
def adjust_duration(self, original_duration: float, config: SpeedConfig) -> float:
"""计算调速后的时长."""
return _adjust_duration_base(original_duration, config)
"""计算调速后的时长。
加速 → 时长变短;减速 → 时长变长。
"""
if config.is_original or original_duration <= 0:
return original_duration
return original_duration / config.speed
def build_clip_speed_filter(
self,
speed: float,
pitch_correct: bool = True,
) -> tuple[str, str, SpeedConfig]:
"""便捷方法:从单一 speed 值生成视频+音频滤镜."""
"""便捷方法:从单一 speed 值生成视频+音频滤镜。
返回 (video_filter, audio_filter, config)。
"""
config = SpeedConfig(speed=speed, pitch_correct=pitch_correct)
config.clamp()
return (
@@ -70,5 +160,8 @@ class SpeedEngine:
clip_config: dict,
global_speed: float = DEFAULT_SPEED,
) -> float:
"""从 clip config 中解析 playback_speed,0 或缺失则使用全局速度."""
return _resolve_clip_speed_base(clip_config, global_speed)
"""从 clip config 中解析 playback_speed,0 或缺失则使用全局速度。"""
speed = clip_config.get("playback_speed", 0) if clip_config else 0
if not isinstance(speed, (int, float)) or speed <= 0:
return global_speed
return float(speed)
-179
View File
@@ -1,179 +0,0 @@
"""调速配置领域模型 — 纯逻辑,无FFmpeg依赖.
抽离自 speed_engine.py,包含:
- SpeedConfig 数据类(解析/钳制/原速判断)
- 视频/音频调速滤镜构建
- atempo 多级拆分算法
- 时长计算
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import Any
# ─── 常量 ───────────────────────────────────────────────
MIN_SPEED = 0.25
MAX_SPEED = 4.0
DEFAULT_SPEED = 1.0
# atempo 单级有效范围
_ATEMPO_MIN = 0.5
_ATEMPO_MAX = 2.0
@dataclass
class SpeedConfig:
"""调速配置.
Attributes:
speed: 播放速度,0.25~4.0,1.0 为原速
pitch_correct: 是否保持音调(默认 True,用 atempo 时间拉伸算法)
"""
speed: float = DEFAULT_SPEED
pitch_correct: bool = True
@classmethod
def parse(cls, data: dict[str, Any] | None) -> SpeedConfig:
"""从 dict 解析配置,无效值回退到默认."""
if not data or not isinstance(data, dict):
return cls()
speed = data.get("speed", DEFAULT_SPEED)
if not isinstance(speed, (int, float)):
speed = DEFAULT_SPEED
pitch_correct = data.get("pitch_correct", True)
if not isinstance(pitch_correct, bool):
pitch_correct = True
config = cls(speed=float(speed), pitch_correct=pitch_correct)
config.clamp()
return config
def clamp(self) -> None:
"""将速度钳制到合法范围."""
if self.speed <= 0:
self.speed = DEFAULT_SPEED
elif self.speed < MIN_SPEED:
self.speed = MIN_SPEED
elif self.speed > MAX_SPEED:
self.speed = MAX_SPEED
@property
def is_original(self) -> bool:
"""是否原速(无需调速)."""
return abs(self.speed - 1.0) < 1e-6
@property
def is_fast(self) -> bool:
"""是否加速播放."""
return self.speed > 1.0
@property
def is_slow(self) -> bool:
"""是否减速播放."""
return self.speed < 1.0
# ── 滤镜构建 ────────────────────────────────────────────────────────────────
def build_video_filter(config: SpeedConfig) -> str:
"""生成视频调速滤镜字符串.
返回 setpts 滤镜表达式,原速时返回空字符串。
"""
if config.is_original:
return ""
# setpts=PTS/speed — speed>1 加速,speed<1 减速
return f"setpts=PTS/{config.speed:.4f}"
def build_audio_filter(config: SpeedConfig) -> str:
"""生成音频调速滤镜字符串.
atempo 单级范围 0.5~2.0,超出范围时自动多级串联:
- 0.25x → atempo=0.5,atempo=0.5
- 4x → atempo=2.0,atempo=2.0
- 0.3x → atempo=0.5,atempo=0.6
- 3x → atempo=2.0,atempo=1.5
原速时返回空字符串。
"""
if config.is_original:
return ""
speed = config.speed
stages: list[float] = _split_atempo_stages(speed)
return ",".join(f"atempo={s:.4f}" for s in stages)
def _split_atempo_stages(speed: float) -> list[float]:
"""将速度拆分为多级 atempo 串联,每级都在 [0.5, 2.0] 范围内."""
if _ATEMPO_MIN <= speed <= _ATEMPO_MAX:
return [speed]
stages: list[float] = []
remaining = speed
# 加速场景(speed > 2.0)
if speed > _ATEMPO_MAX:
while remaining > _ATEMPO_MAX:
stages.append(_ATEMPO_MAX)
remaining /= _ATEMPO_MAX
stages.append(remaining)
# 减速场景(speed < 0.5)
else:
while remaining < _ATEMPO_MIN:
stages.append(_ATEMPO_MIN)
remaining /= _ATEMPO_MIN
stages.append(remaining)
return stages
# ── 时长计算 ────────────────────────────────────────────────────────────────
def adjust_duration(original_duration: float, config: SpeedConfig) -> float:
"""计算调速后的时长.
加速 → 时长变短;减速 → 时长变长。
"""
if config.is_original or original_duration <= 0:
return original_duration
return original_duration / config.speed
# ── 便捷方法 ────────────────────────────────────────────────────────────────
def build_clip_speed_filter(
speed: float,
pitch_correct: bool = True,
) -> tuple[str, str, SpeedConfig]:
"""便捷方法:从单一 speed 值生成视频+音频滤镜.
返回 (video_filter, audio_filter, config)。
"""
config = SpeedConfig(speed=speed, pitch_correct=pitch_correct)
config.clamp()
return (
build_video_filter(config),
build_audio_filter(config),
config,
)
def resolve_clip_speed(
clip_config: dict[str, Any] | None,
global_speed: float = DEFAULT_SPEED,
) -> float:
"""从 clip config 中解析 playback_speed,0 或缺失则使用全局速度."""
speed = clip_config.get("playback_speed", 0) if clip_config else 0
if not isinstance(speed, (int, float)) or speed <= 0:
return global_speed
return float(speed)
+377
View File
@@ -0,0 +1,377 @@
"""小领域模块组合单测 — template_clip_config / tag / exceptions / editing_mode."""
from __future__ import annotations
import pytest
from packages.domain.editing_mode import EditingMode
from packages.domain.exceptions import (
DomainError,
NotFoundError,
QuotaExceededError,
ValidationError,
)
from packages.domain.tag import Tag
from packages.domain.template_clip_config import (
ClipType,
TemplateClipConfig,
TransitionEffect,
)
# ── EditingMode ────────────────────────────────────────────────────────────
class TestEditingMode:
def test_one_take(self):
assert EditingMode.ONE_TAKE == "one_take"
def test_pip(self):
assert EditingMode.PIP == "pip"
def test_voice_over(self):
assert EditingMode.VOICE_OVER == "voice_over"
def test_voice_pip(self):
assert EditingMode.VOICE_PIP == "voice_pip"
def test_from_string(self):
assert EditingMode("one_take") == EditingMode.ONE_TAKE
assert EditingMode("pip") == EditingMode.PIP
def test_invalid_raises(self):
with pytest.raises(ValueError):
EditingMode("invalid_mode")
def test_is_str(self):
# StrEnum 实例本身就是 str
assert isinstance(EditingMode.ONE_TAKE, str)
assert EditingMode.ONE_TAKE + "_suffix" == "one_take_suffix"
# ── 异常类 ─────────────────────────────────────────────────────────────────
class TestExceptions:
def test_domain_error_is_exception(self):
err = DomainError("test")
assert isinstance(err, Exception)
assert str(err) == "test"
def test_not_found_inherits_domain(self):
err = NotFoundError("not found")
assert isinstance(err, DomainError)
assert str(err) == "not found"
def test_validation_error_inherits_domain(self):
err = ValidationError("bad input")
assert isinstance(err, DomainError)
assert str(err) == "bad input"
def test_quota_exceeded_error_attributes(self):
err = QuotaExceededError(dimension="storage_gb", limit=100, used=150)
assert err.dimension == "storage_gb"
assert err.limit == 100
assert err.used == 150
assert "storage_gb" in str(err)
assert "150" in str(err)
assert "100" in str(err)
def test_quota_exceeded_is_domain_error(self):
err = QuotaExceededError("x", 10, 20)
assert isinstance(err, DomainError)
def test_catch_domain_error_catches_all(self):
"""所有领域异常都能被 DomainError catch."""
for cls in [NotFoundError, ValidationError, QuotaExceededError]:
try:
if cls == QuotaExceededError:
raise cls("dim", 10, 20)
raise cls("msg")
except DomainError:
pass
else:
pytest.fail(f"{cls.__name__} 未被 DomainError 捕获")
# ── Tag ────────────────────────────────────────────────────────────────────
class TestTag:
def test_create_basic(self):
tag = Tag.create(user_id="user1", name="旅行")
assert tag.id
assert tag.user_id == "user1"
assert tag.name == "旅行"
def test_create_name_stripped(self):
tag = Tag.create(user_id="u1", name=" 美食 ")
assert tag.name == "美食"
def test_create_empty_name_raises(self):
with pytest.raises(ValueError, match="标签名称不能为空"):
Tag.create(user_id="u1", name="")
def test_create_whitespace_name_raises(self):
with pytest.raises(ValueError, match="标签名称不能为空"):
Tag.create(user_id="u1", name=" ")
def test_create_has_created_at(self):
tag = Tag.create(user_id="u1", name="t1")
assert tag.created_at is not None
def test_unique_ids(self):
t1 = Tag.create(user_id="u1", name="t1")
t2 = Tag.create(user_id="u1", name="t2")
assert t1.id != t2.id
# ── ClipType 枚举 ─────────────────────────────────────────────────────────
class TestClipType:
def test_all_types(self):
assert ClipType.INTRO == "intro"
assert ClipType.MAIN == "main"
assert ClipType.TRANSITION == "transition"
assert ClipType.OUTRO == "outro"
assert ClipType.TITLE == "title"
assert ClipType.SUBTITLE == "subtitle"
def test_from_string(self):
assert ClipType("main") == ClipType.MAIN
assert ClipType("intro") == ClipType.INTRO
def test_invalid_raises(self):
with pytest.raises(ValueError):
ClipType("invalid")
# ── TransitionEffect 枚举 ─────────────────────────────────────────────────
class TestTransitionEffect:
def test_all_effects(self):
assert TransitionEffect.CUT == "cut"
assert TransitionEffect.FADE == "fade"
assert TransitionEffect.SLIDE_LEFT == "slide_left"
assert TransitionEffect.SLIDE_RIGHT == "slide_right"
assert TransitionEffect.DISSOLVE == "dissolve"
assert TransitionEffect.WIPE == "wipe"
def test_from_string(self):
assert TransitionEffect("fade") == TransitionEffect.FADE
assert TransitionEffect("cut") == TransitionEffect.CUT
def test_invalid_raises(self):
with pytest.raises(ValueError):
TransitionEffect("invalid_effect")
# ── TemplateClipConfig.create ──────────────────────────────────────────────
class TestTemplateClipConfigCreate:
def test_basic_creation(self):
clip = TemplateClipConfig.create(
template_id="tmpl_1",
clip_type=ClipType.MAIN,
order=1,
)
assert clip.id
assert clip.template_id == "tmpl_1"
assert clip.clip_type == ClipType.MAIN
assert clip.order == 1
assert clip.min_duration == 0.0
assert clip.max_duration == 0.0
assert clip.text_template == ""
assert clip.material_requirements == {}
assert clip.transition_effect == TransitionEffect.CUT
assert clip.config == {}
def test_clip_type_string(self):
clip = TemplateClipConfig.create(
template_id="t1", clip_type="intro", order=0,
)
assert clip.clip_type == ClipType.INTRO
def test_template_id_stripped(self):
clip = TemplateClipConfig.create(
template_id=" tmpl_1 ", clip_type=ClipType.MAIN, order=1,
)
assert clip.template_id == "tmpl_1"
def test_empty_template_id_raises(self):
with pytest.raises(ValueError, match="template_id 不能为空"):
TemplateClipConfig.create(template_id="", clip_type=ClipType.MAIN, order=1)
def test_whitespace_template_id_raises(self):
with pytest.raises(ValueError, match="template_id 不能为空"):
TemplateClipConfig.create(template_id=" ", clip_type=ClipType.MAIN, order=1)
def test_negative_min_duration_raises(self):
with pytest.raises(ValueError, match="min_duration 不能为负数"):
TemplateClipConfig.create(
template_id="t1", clip_type=ClipType.MAIN, order=1,
min_duration=-1.0,
)
def test_negative_max_duration_raises(self):
with pytest.raises(ValueError, match="max_duration 不能为负数"):
TemplateClipConfig.create(
template_id="t1", clip_type=ClipType.MAIN, order=1,
max_duration=-1.0,
)
def test_min_greater_than_max_raises(self):
with pytest.raises(ValueError, match="min_duration 不能大于 max_duration"):
TemplateClipConfig.create(
template_id="t1", clip_type=ClipType.MAIN, order=1,
min_duration=10.0, max_duration=5.0,
)
def test_zero_min_and_max_ok(self):
clip = TemplateClipConfig.create(
template_id="t1", clip_type=ClipType.MAIN, order=1,
min_duration=0, max_duration=0,
)
assert clip.min_duration == 0
assert clip.max_duration == 0
def test_min_zero_max_positive_ok(self):
clip = TemplateClipConfig.create(
template_id="t1", clip_type=ClipType.MAIN, order=1,
min_duration=0, max_duration=10.0,
)
assert clip.max_duration == 10.0
def test_min_equals_max_ok(self):
clip = TemplateClipConfig.create(
template_id="t1", clip_type=ClipType.MAIN, order=1,
min_duration=5.0, max_duration=5.0,
)
assert clip.min_duration == 5.0
assert clip.max_duration == 5.0
def test_with_text_template(self):
clip = TemplateClipConfig.create(
template_id="t1", clip_type=ClipType.MAIN, order=1,
text_template=" 欢迎收看 {channel} ",
)
# text_template 会 strip
assert clip.text_template == "欢迎收看 {channel}"
def test_with_material_requirements(self):
reqs = {"material_type": "video", "min_duration": 3}
clip = TemplateClipConfig.create(
template_id="t1", clip_type=ClipType.MAIN, order=1,
material_requirements=reqs,
)
assert clip.material_requirements == reqs
def test_material_requirements_none_defaults_empty(self):
clip = TemplateClipConfig.create(
template_id="t1", clip_type=ClipType.MAIN, order=1,
material_requirements=None,
)
assert clip.material_requirements == {}
def test_transition_effect_string(self):
clip = TemplateClipConfig.create(
template_id="t1", clip_type=ClipType.MAIN, order=1,
transition_effect="fade",
)
assert clip.transition_effect == TransitionEffect.FADE
def test_with_config(self):
config = {"speed": 1.5, "filter": "vibrance"}
clip = TemplateClipConfig.create(
template_id="t1", clip_type=ClipType.MAIN, order=1,
config=config,
)
assert clip.config == config
def test_config_none_defaults_empty(self):
clip = TemplateClipConfig.create(
template_id="t1", clip_type=ClipType.MAIN, order=1,
config=None,
)
assert clip.config == {}
def test_created_at_and_updated_at(self):
clip = TemplateClipConfig.create(
template_id="t1", clip_type=ClipType.MAIN, order=1,
)
assert clip.created_at is not None
assert clip.updated_at is not None
def test_unique_ids(self):
c1 = TemplateClipConfig.create(template_id="t1", clip_type=ClipType.MAIN, order=1)
c2 = TemplateClipConfig.create(template_id="t1", clip_type=ClipType.MAIN, order=2)
assert c1.id != c2.id
# ── TemplateClipConfig 属性 ────────────────────────────────────────────────
class TestTemplateClipConfigProperties:
def test_has_duration_range_false_both_zero(self):
clip = TemplateClipConfig.create(
template_id="t1", clip_type=ClipType.MAIN, order=1,
)
assert clip.has_duration_range is False
def test_has_duration_range_true_min_only(self):
clip = TemplateClipConfig.create(
template_id="t1", clip_type=ClipType.MAIN, order=1,
min_duration=2.0,
)
assert clip.has_duration_range is True
def test_has_duration_range_true_max_only(self):
clip = TemplateClipConfig.create(
template_id="t1", clip_type=ClipType.MAIN, order=1,
max_duration=10.0,
)
assert clip.has_duration_range is True
def test_has_duration_range_true_both_set(self):
clip = TemplateClipConfig.create(
template_id="t1", clip_type=ClipType.MAIN, order=1,
min_duration=2.0, max_duration=10.0,
)
assert clip.has_duration_range is True
def test_default_duration_zero(self):
clip = TemplateClipConfig.create(
template_id="t1", clip_type=ClipType.MAIN, order=1,
)
assert clip.default_duration == 0.0
def test_default_duration_min_only(self):
clip = TemplateClipConfig.create(
template_id="t1", clip_type=ClipType.MAIN, order=1,
min_duration=3.0,
)
assert clip.default_duration == 3.0
def test_default_duration_max_only(self):
clip = TemplateClipConfig.create(
template_id="t1", clip_type=ClipType.MAIN, order=1,
max_duration=10.0,
)
assert clip.default_duration == 10.0
def test_default_duration_both_midpoint(self):
clip = TemplateClipConfig.create(
template_id="t1", clip_type=ClipType.MAIN, order=1,
min_duration=2.0, max_duration=8.0,
)
assert clip.default_duration == 5.0
def test_default_duration_min_equals_max(self):
clip = TemplateClipConfig.create(
template_id="t1", clip_type=ClipType.MAIN, order=1,
min_duration=5.0, max_duration=5.0,
)
assert clip.default_duration == 5.0
-313
View File
@@ -1,313 +0,0 @@
"""speed_config 领域模型单测."""
from __future__ import annotations
import pytest
from packages.domain.speed_config import (
DEFAULT_SPEED,
MAX_SPEED,
MIN_SPEED,
SpeedConfig,
adjust_duration,
build_audio_filter,
build_video_filter,
build_clip_speed_filter,
resolve_clip_speed,
)
# ── 常量测试 ────────────────────────────────────────────────────────────────
class TestConstants:
def test_min_speed(self):
assert MIN_SPEED == 0.25
def test_max_speed(self):
assert MAX_SPEED == 4.0
def test_default_speed(self):
assert DEFAULT_SPEED == 1.0
# ── SpeedConfig.parse 测试 ────────────────────────────────────────────────
class TestSpeedConfigParse:
def test_none_returns_default(self):
cfg = SpeedConfig.parse(None)
assert cfg.speed == DEFAULT_SPEED
assert cfg.pitch_correct is True
def test_empty_dict_returns_default(self):
cfg = SpeedConfig.parse({})
assert cfg.speed == DEFAULT_SPEED
def test_invalid_type_returns_default(self):
cfg = SpeedConfig.parse("not_a_dict")
assert cfg.speed == DEFAULT_SPEED
def test_valid_speed(self):
cfg = SpeedConfig.parse({"speed": 2.0})
assert cfg.speed == 2.0
def test_speed_clamped_low(self):
cfg = SpeedConfig.parse({"speed": 0.1})
assert cfg.speed == MIN_SPEED
def test_speed_clamped_high(self):
cfg = SpeedConfig.parse({"speed": 5.0})
assert cfg.speed == MAX_SPEED
def test_zero_speed_returns_default(self):
cfg = SpeedConfig.parse({"speed": 0})
assert cfg.speed == DEFAULT_SPEED
def test_negative_speed_returns_default(self):
cfg = SpeedConfig.parse({"speed": -1.0})
assert cfg.speed == DEFAULT_SPEED
def test_pitch_correct_false(self):
cfg = SpeedConfig.parse({"pitch_correct": False})
assert cfg.pitch_correct is False
def test_pitch_correct_invalid_type_defaults_true(self):
cfg = SpeedConfig.parse({"pitch_correct": "yes"})
assert cfg.pitch_correct is True
def test_string_speed_invalid_uses_default(self):
cfg = SpeedConfig.parse({"speed": "fast"})
assert cfg.speed == DEFAULT_SPEED
# ── SpeedConfig.clamp 测试 ────────────────────────────────────────────────
class TestClamp:
def test_already_valid_unchanged(self):
cfg = SpeedConfig(speed=1.5)
cfg.clamp()
assert cfg.speed == 1.5
def test_below_min_clamped(self):
cfg = SpeedConfig(speed=0.1)
cfg.clamp()
assert cfg.speed == MIN_SPEED
def test_above_max_clamped(self):
cfg = SpeedConfig(speed=10.0)
cfg.clamp()
assert cfg.speed == MAX_SPEED
def test_zero_defaults(self):
cfg = SpeedConfig(speed=0.0)
cfg.clamp()
assert cfg.speed == DEFAULT_SPEED
def test_negative_defaults(self):
cfg = SpeedConfig(speed=-2.0)
cfg.clamp()
assert cfg.speed == DEFAULT_SPEED
def test_exact_min_stays(self):
cfg = SpeedConfig(speed=MIN_SPEED)
cfg.clamp()
assert cfg.speed == MIN_SPEED
def test_exact_max_stays(self):
cfg = SpeedConfig(speed=MAX_SPEED)
cfg.clamp()
assert cfg.speed == MAX_SPEED
# ── is_original / is_fast / is_slow 测试 ─────────────────────────────────
class TestSpeedProperties:
def test_is_original_true(self):
cfg = SpeedConfig(speed=1.0)
assert cfg.is_original is True
def test_is_original_false_fast(self):
cfg = SpeedConfig(speed=2.0)
assert cfg.is_original is False
def test_is_original_false_slow(self):
cfg = SpeedConfig(speed=0.5)
assert cfg.is_original is False
def test_is_original_near_one(self):
cfg = SpeedConfig(speed=1.0000001)
assert cfg.is_original is True
def test_is_fast_true(self):
cfg = SpeedConfig(speed=2.0)
assert cfg.is_fast is True
def test_is_fast_false(self):
cfg = SpeedConfig(speed=0.5)
assert cfg.is_fast is False
def test_is_false_for_original(self):
cfg = SpeedConfig(speed=1.0)
assert cfg.is_fast is False
assert cfg.is_slow is False
def test_is_slow_true(self):
cfg = SpeedConfig(speed=0.5)
assert cfg.is_slow is True
def test_is_slow_false(self):
cfg = SpeedConfig(speed=2.0)
assert cfg.is_slow is False
# ── build_video_filter 测试 ───────────────────────────────────────────────
class TestBuildVideoFilter:
def test_original_speed_empty(self):
cfg = SpeedConfig(speed=1.0)
assert build_video_filter(cfg) == ""
def test_fast_speed_setpts(self):
cfg = SpeedConfig(speed=2.0)
result = build_video_filter(cfg)
assert "setpts=PTS/2.0000" in result
def test_slow_speed_setpts(self):
cfg = SpeedConfig(speed=0.5)
result = build_video_filter(cfg)
assert "setpts=PTS/0.5000" in result
def test_format_four_decimals(self):
cfg = SpeedConfig(speed=1.5)
result = build_video_filter(cfg)
assert "1.5000" in result
# ── build_audio_filter 测试 ───────────────────────────────────────────────
class TestBuildAudioFilter:
def test_original_speed_empty(self):
cfg = SpeedConfig(speed=1.0)
assert build_audio_filter(cfg) == ""
def test_single_stage_within_range(self):
cfg = SpeedConfig(speed=1.5)
result = build_audio_filter(cfg)
assert result == "atempo=1.5000"
assert result.count("atempo") == 1
def test_fast_two_stages(self):
cfg = SpeedConfig(speed=3.0)
result = build_audio_filter(cfg)
assert result.count("atempo") == 2
# 2.0 * 1.5 = 3.0
assert "atempo=2.0000" in result
assert "atempo=1.5000" in result
def test_max_speed_two_stages(self):
cfg = SpeedConfig(speed=4.0)
result = build_audio_filter(cfg)
assert result.count("atempo") == 2
# 2.0 * 2.0 = 4.0
assert result == "atempo=2.0000,atempo=2.0000"
def test_slow_two_stages(self):
cfg = SpeedConfig(speed=0.25)
result = build_audio_filter(cfg)
assert result.count("atempo") == 2
# 0.5 * 0.5 = 0.25
assert result == "atempo=0.5000,atempo=0.5000"
def test_slow_single_stage(self):
cfg = SpeedConfig(speed=0.8)
result = build_audio_filter(cfg)
assert result == "atempo=0.8000"
assert result.count("atempo") == 1
def test_exactly_two_point_zero_single(self):
cfg = SpeedConfig(speed=2.0)
result = build_audio_filter(cfg)
assert result.count("atempo") == 1
assert "atempo=2.0000" in result
def test_exactly_half_single(self):
cfg = SpeedConfig(speed=0.5)
result = build_audio_filter(cfg)
assert result.count("atempo") == 1
assert "atempo=0.5000" in result
# ── adjust_duration 测试 ──────────────────────────────────────────────────
class TestAdjustDuration:
def test_original_speed_unchanged(self):
cfg = SpeedConfig(speed=1.0)
assert adjust_duration(10.0, cfg) == 10.0
def test_double_speed_halved(self):
cfg = SpeedConfig(speed=2.0)
assert adjust_duration(10.0, cfg) == 5.0
def test_half_speed_doubled(self):
cfg = SpeedConfig(speed=0.5)
assert adjust_duration(10.0, cfg) == 20.0
def test_zero_duration_unchanged(self):
cfg = SpeedConfig(speed=2.0)
assert adjust_duration(0.0, cfg) == 0.0
def test_negative_duration_unchanged(self):
cfg = SpeedConfig(speed=2.0)
assert adjust_duration(-5.0, cfg) == -5.0
# ── build_clip_speed_filter 测试 ─────────────────────────────────────────
class TestBuildClipSpeedFilter:
def test_normal_speed(self):
vf, af, cfg = build_clip_speed_filter(2.0)
assert vf == "setpts=PTS/2.0000"
assert "atempo=2.0000" in af
assert cfg.speed == 2.0
def test_clamped_speed(self):
vf, af, cfg = build_clip_speed_filter(10.0)
assert cfg.speed == MAX_SPEED
def test_pitch_correct_param(self):
vf, af, cfg = build_clip_speed_filter(1.5, pitch_correct=False)
assert cfg.pitch_correct is False
def test_original_speed_empty_filters(self):
vf, af, cfg = build_clip_speed_filter(1.0)
assert vf == ""
assert af == ""
# ── resolve_clip_speed 测试 ──────────────────────────────────────────────
class TestResolveClipSpeed:
def test_none_config_uses_global(self):
assert resolve_clip_speed(None, 1.5) == 1.5
def test_no_playback_speed_uses_global(self):
assert resolve_clip_speed({}, 1.5) == 1.5
def test_zero_speed_uses_global(self):
assert resolve_clip_speed({"playback_speed": 0}, 1.5) == 1.5
def test_valid_speed_returns_speed(self):
assert resolve_clip_speed({"playback_speed": 2.0}, 1.0) == 2.0
def test_invalid_type_uses_global(self):
assert resolve_clip_speed({"playback_speed": "fast"}, 1.0) == 1.0
def test_default_global_speed(self):
assert resolve_clip_speed({}) == DEFAULT_SPEED