Compare commits

..

1 Commits

Author SHA1 Message Date
CI Bot a248883c8e feat(ci): 集成gitleaks + pip-audit + vulture安全扫描到Validate阶段
CI/CD Pipeline / Unit Tests (pull_request) Successful in 1m13s
CI/CD Pipeline / Frontend Lint (pull_request) Successful in 1m13s
CI/CD Pipeline / Validate Code Quality And Tests (pull_request) Successful in 9m35s
CI/CD Pipeline / Build & Push Staging (Watchtower auto-deploy) (pull_request) Has been skipped
CI/CD Pipeline / Build Production Runtime Images (pull_request) Has been skipped
CI/CD Pipeline / Staging API Integration Tests (pull_request) Has been skipped
CI/CD Pipeline / Staging E2E Tests (pull_request) Has been skipped
CI/CD Pipeline / Deploy Production (pull_request) Has been skipped
CI/CD Pipeline / Production Browser E2E (pull_request) Has been skipped
CI/CD Pipeline / Integration Tests (pull_request) Successful in 2m26s
- gitleaks: 密钥检测(P0,阻断模式)
  - PR模式:增量扫描
  - Push模式:全量扫描
  - 双源下载(GitHub + Gitee镜像),失败优雅降级
- pip-audit: 依赖漏洞扫描(P1,告警模式,不阻断)
  - 扫描所有requirements文件
- vulture: 死代码检测(P2,告警模式,不阻断)
  - 置信度80%
  - 排除tests/migrations/scripts等
  - 框架代码白名单
2026-07-14 11:35:42 +08:00
18 changed files with 267 additions and 910 deletions
+92 -1
View File
@@ -101,6 +101,53 @@ jobs:
bandit --version
pytest --version
- name: Secret detection (gitleaks)
shell: sh
run: |
set -eu
echo "=== Installing gitleaks ==="
# 优先尝试 GitHub release,失败则用国内镜像
GITLEAKS_VERSION="8.18.4"
install_gitleaks() {
local url="$1"
curl -sSL -f -o /tmp/gitleaks.tar.gz "$url" || return 1
tar -xzf /tmp/gitleaks.tar.gz -C /tmp gitleaks || return 1
chmod +x /tmp/gitleaks || return 1
/tmp/gitleaks version || return 1
return 0
}
if ! install_gitleaks "https://github.com/gitleaks/gitleaks/releases/download/v${GITLEAKS_VERSION}/gitleaks_${GITLEAKS_VERSION}_linux_x64.tar.gz"; then
echo "GitHub release failed, trying mirror..."
if ! install_gitleaks "https://gitee.com/mirrors/gitleaks/releases/download/v${GITLEAKS_VERSION}/gitleaks_${GITLEAKS_VERSION}_linux_x64.tar.gz"; then
echo "WARN: Failed to install gitleaks from all sources, skipping secret scan"
exit 0
fi
fi
echo ""
echo "=== Running gitleaks scan ==="
if [ "${{ github.event_name }}" = "pull_request" ]; then
# PR触发: 增量扫描
echo "PR mode: scanning changed files"
EXIT_CODE=0
/tmp/gitleaks detect --source . --config .gitleaks.toml --verbose --exit-code 1 --log-opts="origin/${{ github.base_ref }}..HEAD" || EXIT_CODE=$?
if [ "$EXIT_CODE" = "1" ]; then
echo "ERROR: Secrets detected! Check the scan report above."
echo "If these are false positives, add them to .gitleaks.toml allowlist."
exit 1
fi
else
# Push触发: 全量扫描
echo "Push mode: full repository scan"
EXIT_CODE=0
/tmp/gitleaks detect --source . --config .gitleaks.toml --verbose --exit-code 1 || EXIT_CODE=$?
if [ "$EXIT_CODE" = "1" ]; then
echo "ERROR: Secrets detected! Check the scan report above."
echo "If these are false positives, add them to .gitleaks.toml allowlist."
exit 1
fi
fi
echo "gitleaks scan completed - no secrets detected"
- name: Run code quality checks
shell: sh
run: |
@@ -110,12 +157,56 @@ jobs:
python3 -m isort --check-only alembic apps packages tests scripts
python3 -m flake8 apps packages tests --count --statistics
- name: Run security scan
- name: Run security scan (bandit)
shell: sh
run: |
set -eu
bandit -r apps packages -q -ll
- name: Python dependency vulnerability scan (pip-audit)
shell: sh
run: |
set -eu
echo "=== Installing pip-audit ==="
python3 -m pip install -q pip-audit
pip-audit --version
echo ""
echo "=== Scanning Python dependencies ==="
EXIT_CODE=0
for req_file in requirements.txt requirements-base.txt requirements-dev.txt; do
if [ -f "$req_file" ]; then
echo "--- Scanning $req_file ---"
pip-audit -r "$req_file" --desc on --format text 2>&1 | head -30 || EXIT_CODE=$?
echo ""
fi
done
# 告警模式,不阻断CI(待稳定后再考虑改为阻断)
echo "pip-audit scan completed (advisory mode - warnings only, not blocking CI)"
if [ "$EXIT_CODE" != "0" ]; then
echo "WARNING: Potential vulnerabilities found in dependencies."
fi
exit 0
- name: Dead code detection (vulture)
shell: sh
run: |
set -eu
echo "=== Installing vulture ==="
python3 -m pip install -q vulture
vulture --version
echo ""
echo "=== Running vulture dead code scan ==="
# 告警模式,不阻断CI(P2级别,仅供参考)
EXIT_CODE=0
vulture --config vulture.conf vulture_whitelist.py || EXIT_CODE=$?
echo ""
echo "vulture scan completed (advisory mode - P2, for reference only)"
if [ "$EXIT_CODE" != "0" ]; then
echo "NOTE: Potential dead code found. Review results above."
echo "False positives can be added to vulture_whitelist.py"
fi
exit 0
- name: Validate release scripts syntax
shell: sh
run: |
+52
View File
@@ -0,0 +1,52 @@
# .gitleaks.toml - gitleaks 白名单配置
# 仓库: xiaoxia/xiaoxia-saas
# 用途: 排除已知的测试密钥、示例配置等误报
# 允许路径/文件排除
[allowlist]
description = "全局白名单 - 排除示例配置和测试文件"
paths = [
# 环境配置示例(无真实密钥)
'.env.example',
'.env.sample',
'*.env.example',
'*.env.sample',
# 测试文件
'tests/',
'test/',
'*/tests/',
'*/test/',
# 文档
'docs/',
'*.md',
'*.rst',
# 前端依赖
'node_modules/',
# Python包
'site-packages/',
# 锁定文件(自动生成)
'poetry.lock',
'Pipfile.lock',
'requirements*.txt.lock',
# CI配置本身
'.gitea/',
# Docker相关
'docker-compose*.yml',
# gitleaks配置自身
'.gitleaks.toml',
]
# 允许的密钥值/占位符正则
regexes = [
# 占位符模式
'''(?i)(your[_-]?password|your[_-]?secret|your[_-]?key|your[_-]?token|changeme|change[_-]?me|placeholder|example[_-]?key|test[_-]?key|dummy|fake|mock|xxx|none|not[_-]?set|TODO|FIXME)''',
# 数据库连接字符串中的通用密码(PostgreSQL示例配置)
'''postgresql://[^:]+:changeme@''',
'''postgresql://[^:]+:your-password@''',
'''postgresql://[^:]+:password@localhost''',
# Redis示例配置
'''redis://:changeme@''',
'''redis://:your-redis-password@''',
# JWT示例密钥
'''(?i)jwt[_-]?secret\s*[:=]\s*["']?(your[_-]?jwt|change|placeholder|secret|example)''',
]
@@ -1,29 +0,0 @@
"""add playback_speed to edit_plan_clips
Revision ID: 040_playback_speed
Revises: 039_transition_duration
Create Date: 2026-07-14 10:00:00.000000
"""
from __future__ import annotations
import sqlalchemy as sa
from alembic import op
# revision identifiers, used by Alembic.
revision = "040_playback_speed"
down_revision = "039_transition_duration"
branch_labels = None
depends_on = None
def upgrade() -> None:
op.add_column(
"edit_plan_clips",
sa.Column("playback_speed", sa.Float(), nullable=False, server_default="1.0"),
)
def downgrade() -> None:
op.drop_column("edit_plan_clips", "playback_speed")
-1
View File
@@ -211,7 +211,6 @@ class _PlanClipItem(BaseModel):
duration: float
transition_effect: str
transition_duration: float
playback_speed: float = 1.0
status: str
config: Optional[dict[str, Any]] = None
created_at: datetime
@@ -282,7 +282,6 @@ class EditPlanService:
duration: float = 0.0,
transition_effect: str = "cut",
transition_duration: float = 0.0,
playback_speed: float = 1.0,
config: Optional[dict[str, Any]] = None,
) -> EditPlanClip:
"""创建片段
@@ -304,7 +303,6 @@ class EditPlanService:
duration=duration,
transition_effect=transition_effect,
transition_duration=transition_duration,
playback_speed=playback_speed,
config=config,
)
created = self._clip_repo.create(clip)
@@ -329,7 +327,6 @@ class EditPlanService:
duration: Optional[float] = None,
transition_effect: Optional[str] = None,
transition_duration: Optional[float] = None,
playback_speed: Optional[float] = None,
config: Optional[dict[str, Any]] = None,
) -> EditPlanClip:
"""更新片段
@@ -339,15 +336,6 @@ class EditPlanService:
"""
existing = self.get_clip_or_raise(clip_id)
# 速度边界钳制
if playback_speed is not None:
if playback_speed <= 0:
playback_speed = 1.0
elif playback_speed < 0.25:
playback_speed = 0.25
elif playback_speed > 4.0:
playback_speed = 4.0
updated = EditPlanClip(
id=existing.id,
plan_id=existing.plan_id,
@@ -364,7 +352,6 @@ class EditPlanService:
transition_duration=(
transition_duration if transition_duration is not None else existing.transition_duration
),
playback_speed=playback_speed if playback_speed is not None else existing.playback_speed,
status=existing.status,
config=config if config is not None else existing.config,
created_at=existing.created_at,
+27 -93
View File
@@ -23,7 +23,6 @@ from typing import TYPE_CHECKING
from video_processing.ffmpeg_utils import FFMPEG_BIN, probe_has_audio, run_ffmpeg
from video_processing.reverse_engine import ReverseConfig, ReverseEngine
from video_processing.speed_engine import SpeedEngine
if TYPE_CHECKING:
from video_processing.unified_render_service import RenderLayer, ResolvedClip
@@ -226,118 +225,53 @@ def concat_main_audio(
clip = clips[0]
effective_duration = clip_effective_duration(clip)
trim_start = getattr(clip, "start_time", 0) or 0
speed = getattr(clip, "playback_speed", 1.0) or 1.0
if not isinstance(speed, (int, float)) or speed <= 0:
speed = 1.0
# 调速后时长
adjusted_duration = effective_duration / speed if abs(speed - 1.0) >= 1e-6 else effective_duration
# 最终时长:取调速后时长和视频总时长的较小值
final_duration = adjusted_duration
# 最终时长:取 clip 有效时长和视频总时长的较小值
# (视频总时长由主图层决定,但单 clip 场景下两者应该一致,仍做保护)
final_duration = effective_duration
if video_duration > 0 and (final_duration <= 0 or final_duration > video_duration):
final_duration = video_duration
# 音频倒放
reverse_config = ReverseConfig.from_dict(clip.config.get("reverse"))
has_reverse = reverse_config.enabled and reverse_config.reverse_audio
has_speed = abs(speed - 1.0) >= 1e-6
af_filters = []
if reverse_config.enabled and reverse_config.reverse_audio:
reverse_filter = ReverseEngine.build_audio_filter(reverse_config, duration=effective_duration)
if reverse_filter:
af_filters.append(reverse_filter)
if not has_speed and not has_reverse:
# 无调速无倒放:简单命令行,-ss 裁剪更高效
command = [
FFMPEG_BIN,
"-y",
"-i",
str(clip.local_path),
"-vn",
"-acodec",
"aac",
"-b:a",
"128k",
]
if trim_start > 0:
command.extend(["-ss", f"{trim_start:.3f}"])
if final_duration > 0:
command.extend(["-t", f"{final_duration:.3f}"])
command.append(str(output_path))
run_ffmpeg(command)
else:
# 有调速或倒放:用 filter_complex
speed_engine = SpeedEngine()
audio_filters = []
if effective_duration > 0:
audio_filters.append(f"atrim=start={trim_start:.3f}:duration={effective_duration:.3f}")
audio_filters.append("asetpts=PTS-STARTPTS")
# 音频调速
if has_speed:
from video_processing.speed_engine import SpeedConfig
config = SpeedConfig(speed=float(speed))
config.clamp()
atempo_filter = speed_engine.build_audio_filter(config)
if atempo_filter:
audio_filters.append(atempo_filter)
# 音频倒放
if has_reverse:
reverse_filter = ReverseEngine.build_audio_filter(reverse_config, duration=effective_duration)
if reverse_filter:
audio_filters.append(reverse_filter)
filter_parts: list[str] = [f"[0:a]{','.join(audio_filters)}[outa]"]
if video_duration > 0 and final_duration < adjusted_duration:
filter_parts.append(f"[outa]atrim=0:{final_duration:.3f}[final_audio]")
final_label = "final_audio"
else:
final_label = "outa"
filter_complex = ";".join(filter_parts)
command = [
FFMPEG_BIN,
"-y",
"-i",
str(clip.local_path),
"-filter_complex",
filter_complex,
"-map",
f"[{final_label}]",
"-acodec",
"aac",
"-b:a",
"128k",
str(output_path),
]
run_ffmpeg(command)
command = [
FFMPEG_BIN,
"-y",
"-i",
str(clip.local_path),
"-vn",
"-acodec",
"aac",
"-b:a",
"128k",
]
if trim_start > 0:
command.extend(["-ss", f"{trim_start:.3f}"])
if af_filters:
command.extend(["-af", ",".join(af_filters)])
if final_duration > 0:
command.extend(["-t", f"{final_duration:.3f}"])
command.append(str(output_path))
run_ffmpeg(command)
return
# 多 clip,用 filter_complex concat
input_args: list[str] = []
filter_parts: list[str] = []
speed_engine = SpeedEngine()
for i, clip in enumerate(clips):
input_args.extend(["-i", str(clip.local_path)])
effective_duration = clip_effective_duration(clip)
trim_start = getattr(clip, "start_time", 0) or 0
speed = getattr(clip, "playback_speed", 1.0) or 1.0
if not isinstance(speed, (int, float)) or speed <= 0:
speed = 1.0
audio_filters: list[str] = []
if effective_duration > 0:
audio_filters.append(f"atrim=start={trim_start:.3f}:duration={effective_duration:.3f}")
audio_filters.append("asetpts=PTS-STARTPTS")
# 音频调速 — atempo 多级串联
if abs(speed - 1.0) >= 1e-6:
from video_processing.speed_engine import SpeedConfig
config = SpeedConfig(speed=float(speed))
config.clamp()
atempo_filter = speed_engine.build_audio_filter(config)
if atempo_filter:
audio_filters.append(atempo_filter)
else:
audio_filters.append("asetpts=PTS-STARTPTS")
@@ -1,167 +0,0 @@
"""视频调速引擎 — 基于 FFmpeg setpts + atempo 的速度调整能力。
支持:
- 0.25x ~ 4x 变速范围
- 视频调速(setpts
- 音频调速(atempo,多级串联处理超范围值)
- 音调修正(pitch_correct,默认开启)
- 边界自动钳制,不阻断渲染
"""
from dataclasses import dataclass
from typing import Optional
# ─── 常量 ───────────────────────────────────────────────
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.01.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 调速滤镜链。
用法:
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:
"""生成视频调速滤镜字符串。
返回 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:
"""生成音频调速滤镜字符串。
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 串联,每级都在 [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:
"""计算调速后的时长。
加速 → 时长变短;减速 → 时长变长。
"""
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 值生成视频+音频滤镜。
返回 (video_filter, audio_filter, config)。
"""
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 或缺失则使用全局速度。"""
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)
@@ -45,7 +45,6 @@ from video_processing.pip_engine import PiPConfig, PiPEngine, PiPLayerConfig
from video_processing.render_audio import RenderContext, merge_audio_video, mix_audio
from video_processing.render_subtitles import generate_ass_subtitles
from video_processing.reverse_engine import ReverseConfig, ReverseEngine
from video_processing.speed_engine import SpeedConfig, SpeedEngine
from video_processing.sticker_engine import StickerEngine, parse_stickers_from_config
from video_processing.subtitle_generator import generate_ass_from_timeline
from video_processing.transition_engine import TransitionEngine
@@ -74,7 +73,6 @@ class ResolvedClip:
duration: float = 0.0 # 0 表示使用素材完整时长
transition_effect: str = "cut"
transition_duration: float = 0.0 # 0 表示使用全局默认值
playback_speed: float = 1.0 # 0 或 1.0 表示原速
config: dict[str, Any] = field(default_factory=dict)
# 运行时填充
@@ -188,7 +186,6 @@ class UnifiedRenderService:
self.asr_service = asr_service
self.bgm_path = bgm_path
self._transition_engine = TransitionEngine(default_duration=transition_duration)
self._speed_engine = SpeedEngine()
def render(self) -> RenderResult:
"""执行渲染,返回 RenderResult.
@@ -496,7 +493,7 @@ class UnifiedRenderService:
if not main_layer or not main_layer.clips:
return 0.0
total = sum(UnifiedRenderService._clip_adjusted_duration(c) for c in main_layer.clips)
total = sum(UnifiedRenderService._clip_effective_duration(c) for c in main_layer.clips)
# 减去转场重叠时间(粗略估算)
n_clips = len(main_layer.clips)
@@ -1200,7 +1197,6 @@ class UnifiedRenderService:
duration=final_duration,
transition_effect=clip.transition_effect or "cut",
transition_duration=getattr(clip, "transition_duration", 0.0) or 0.0,
playback_speed=getattr(clip, "playback_speed", 1.0) or 1.0,
config=clip_config,
actual_duration=actual_duration,
trim_config=effective_trim,
@@ -1298,11 +1294,6 @@ class UnifiedRenderService:
filters.append(f"trim=duration={effective_duration:.3f}")
filters.append("setpts=PTS-STARTPTS")
# 调速 — 基于 setpts 改变播放速度
speed = UnifiedRenderService._clip_speed(clip)
if abs(speed - 1.0) >= 1e-6:
filters.append(f"setpts=PTS/{speed:.4f}")
# 倒放滤镜(在 trim 之后、scale 之前应用)
reverse_config = ReverseConfig.from_dict(clip.config.get("reverse"))
if reverse_config.enabled and reverse_config.reverse_video:
@@ -1360,8 +1351,8 @@ class UnifiedRenderService:
for layer in layers:
layer_clip_indices = [all_clips.index(c) for c in layer.clips]
layer_labels = [preprocessed_labels[i] for i in layer_clip_indices]
# 使用调速后的实际时长,与 Step 1 的调速处理保持一致
layer_durations = [UnifiedRenderService._clip_adjusted_duration(all_clips[i]) for i in layer_clip_indices]
# 使用 trim 后的有效时长,与 Step 1 的 trim=duration 保持一致
layer_durations = [UnifiedRenderService._clip_effective_duration(all_clips[i]) for i in layer_clip_indices]
layer_transitions = [all_clips[i].transition_effect for i in layer_clip_indices]
layer_transition_durations = [all_clips[i].transition_duration for i in layer_clip_indices]
@@ -1606,7 +1597,7 @@ class UnifiedRenderService:
@staticmethod
def _clip_effective_duration(clip: ResolvedClip) -> float:
"""计算 clip 的有效时长(原速 trim 后时长)."""
"""计算 clip 的有效时长."""
if clip.duration > 0:
return min(clip.duration, clip.actual_duration) if clip.actual_duration > 0 else clip.duration
return clip.actual_duration if clip.actual_duration > 0 else 0.0
@@ -1703,20 +1694,3 @@ class UnifiedRenderService:
)
return new_filter, new_input_args
@staticmethod
def _clip_speed(clip: ResolvedClip) -> float:
"""获取 clip 的播放速度,无效值回退到 1.0."""
speed = getattr(clip, "playback_speed", 1.0)
if not isinstance(speed, (int, float)) or speed <= 0:
return 1.0
return float(speed)
@staticmethod
def _clip_adjusted_duration(clip: ResolvedClip) -> float:
"""计算调速后的 clip 实际时长(用于拼接计算)."""
base = UnifiedRenderService._clip_effective_duration(clip)
speed = UnifiedRenderService._clip_speed(clip)
if abs(speed - 1.0) < 1e-6:
return base
return base / speed
-8
View File
@@ -866,14 +866,6 @@
"type": "FLOAT",
"unique": false
},
{
"index": false,
"name": "playback_speed",
"nullable": false,
"primary_key": false,
"type": "FLOAT",
"unique": false
},
{
"index": true,
"name": "status",
@@ -55,7 +55,6 @@ class SQLAlchemyEditPlanClipRepository:
duration=clip.duration,
transition_effect=clip.transition_effect,
transition_duration=clip.transition_duration,
playback_speed=clip.playback_speed,
status=clip.status,
config=clip.config,
)
@@ -79,7 +78,6 @@ class SQLAlchemyEditPlanClipRepository:
model.duration = clip.duration
model.transition_effect = clip.transition_effect
model.transition_duration = clip.transition_duration
model.playback_speed = clip.playback_speed
model.status = clip.status
model.config = clip.config
model.updated_at = clip.updated_at
@@ -125,7 +123,6 @@ class SQLAlchemyEditPlanClipRepository:
duration=model.duration or 0.0,
transition_effect=model.transition_effect or "cut",
transition_duration=getattr(model, "transition_duration", 0.0) or 0.0,
playback_speed=model.playback_speed or 1.0,
status=EditPlanClipStatus(model.status) if model.status else EditPlanClipStatus.PENDING,
config=model.config or {},
created_at=model.created_at,
@@ -197,7 +197,6 @@ class EditPlanClipModel(Base):
duration = Column(Float, nullable=False, default=0.0)
transition_effect = Column(String(20), nullable=False, default="cut")
transition_duration = Column(Float, nullable=False, default=0.0)
playback_speed = Column(Float, nullable=False, default=1.0)
status = Column(String(20), nullable=False, default="pending", index=True)
config = Column(JSON, nullable=False, default=dict)
created_at = Column(DateTime, nullable=False, default=lambda: datetime.now(timezone.utc))
-10
View File
@@ -51,7 +51,6 @@ class EditPlanClip:
duration: float = 0.0
transition_effect: str = "cut"
transition_duration: float = 0.0 # 0 表示使用全局默认值
playback_speed: float = 1.0 # 0 或 1.0 表示原速,范围 0.25~4.0
status: EditPlanClipStatus = EditPlanClipStatus.PENDING
config: dict[str, Any] = field(default_factory=dict)
created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
@@ -71,7 +70,6 @@ class EditPlanClip:
duration: float = 0.0,
transition_effect: str = "cut",
transition_duration: float = 0.0,
playback_speed: float = 1.0,
config: dict[str, Any] | None = None,
) -> EditPlanClip:
"""创建剪辑计划片段"""
@@ -83,13 +81,6 @@ class EditPlanClip:
raise ValueError("start_time 不能为负数")
if duration < 0:
raise ValueError("duration 不能为负数")
# 速度边界钳制
if playback_speed <= 0:
playback_speed = 1.0
elif playback_speed < 0.25:
playback_speed = 0.25
elif playback_speed > 4.0:
playback_speed = 4.0
return cls(
id=uuid4().hex,
@@ -103,7 +94,6 @@ class EditPlanClip:
duration=duration,
transition_effect=transition_effect.strip() or "cut",
transition_duration=max(0.0, transition_duration),
playback_speed=playback_speed,
status=EditPlanClipStatus.PENDING,
config=config or {},
)
-117
View File
@@ -1,117 +0,0 @@
#!/bin/bash
# 灰度发布脚本:通过Nginx权重调整流量比例
# 用法: ./scripts/gray_deploy.sh <版本号> <灰度百分比>
#
# 需要在目标服务器上执行,或通过SSH执行
# 前提:服务器上运行两个版本的容器(stable + canary),Nginx做加权轮询
set -euo pipefail
VERSION="${1:-}"
GRAY_PCT="${2:-10}"
if [[ -z "$VERSION" ]]; then
echo "用法: $0 <版本号> [灰度百分比]"
echo "示例: $0 v0.1.129 5"
exit 1
fi
STABLE_VERSION="${STABLE_VERSION:-current}"
REGISTRY="${REGISTRY:-git.xiaoxiajianji.com/xiaoxia/xiaoxia-saas}"
echo "=========================================="
echo " 灰度发布"
echo " 新版本: $VERSION"
echo " 灰度比例: ${GRAY_PCT}%"
echo " 稳定版本: $STABLE_VERSION"
echo "=========================================="
# 1. 拉取新版本镜像
echo ""
echo ">>> 拉取新版本镜像..."
for component in api worker web; do
echo " 拉取 $component:$VERSION ..."
docker pull "${REGISTRY}-${component}:${VERSION}" 2>&1 | tail -1
done
# 2. 启动灰度版本容器(canary)
echo ""
echo ">>> 启动灰度版本容器..."
# API canary
CANARY_API_NAME="saas-api-canary"
if docker ps -a --format '{{.Names}}' | grep -q "^${CANARY_API_NAME}$"; then
echo " 停止旧 canary 容器..."
docker stop "$CANARY_API_NAME" 2>/dev/null || true
docker rm "$CANARY_API_NAME" 2>/dev/null || true
fi
echo " 启动 api canary..."
docker run -d \
--name "$CANARY_API_NAME" \
--network saas-network \
-e DATABASE_URL="${DATABASE_URL}" \
-e REDIS_URL="${REDIS_URL}" \
-e FEATURE_FLAG_PROVIDER=redis \
--restart unless-stopped \
"${REGISTRY}-api:${VERSION}"
# Worker canary
CANARY_WORKER_NAME="saas-worker-canary"
if docker ps -a --format '{{.Names}}' | grep -q "^${CANARY_WORKER_NAME}$"; then
echo " 停止旧 worker canary..."
docker stop "$CANARY_WORKER_NAME" 2>/dev/null || true
docker rm "$CANARY_WORKER_NAME" 2>/dev/null || true
fi
echo " 启动 worker canary..."
docker run -d \
--name "$CANARY_WORKER_NAME" \
--network saas-network \
-e DATABASE_URL="${DATABASE_URL}" \
-e REDIS_URL="${REDIS_URL}" \
--restart unless-stopped \
"${REGISTRY}-worker:${VERSION}"
# 3. 等待容器健康
echo ""
echo ">>> 等待容器健康..."
sleep 10
if ! docker ps --format '{{.Names}} {{.Status}}' | grep -q "$CANARY_API_NAME"; then
echo "错误: API canary 容器未运行"
docker logs "$CANARY_API_NAME" --tail 20
exit 1
fi
echo " ✅ API canary 运行中"
# 4. 更新Nginx权重
echo ""
echo ">>> 更新Nginx权重 (稳定: $((100-GRAY_PCT))% / 灰度: ${GRAY_PCT}%)..."
NGINX_CONF="${NGINX_CONF:-/etc/nginx/conf.d/saas-api.conf}"
if [[ -f "$NGINX_CONF" ]]; then
# 备份
cp "$NGINX_CONF" "${NGINX_CONF}.bak.$(date +%Y%m%d%H%M%S)"
# 更新 upstream 权重(需要根据实际配置调整)
echo " 请手动更新 Nginx upstream 配置中的权重"
echo " 示例配置:"
cat <<EOF
upstream saas_api_backend {
server saas-api:8000 weight=$((100-GRAY_PCT));
server saas-api-canary:8000 weight=${GRAY_PCT};
}
EOF
nginx -t && nginx -s reload
echo " ✅ Nginx 已reload"
else
echo " 警告: Nginx 配置文件不存在 ($NGINX_CONF)"
echo " 请手动配置灰度流量权重"
fi
echo ""
echo "=========================================="
echo " ✅ 灰度发布完成"
echo " 新版本: $VERSION (${GRAY_PCT}%流量)"
echo " 监控: Grafana / 日志"
echo "=========================================="
-126
View File
@@ -1,126 +0,0 @@
#!/bin/bash
# 一键发布脚本:打tag → 触发生产镜像构建 → 部署到灰度
# 用法: ./scripts/release.sh v0.1.129 [--gray 5]
set -euo pipefail
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
REPO_ROOT="$(cd "$SCRIPT_DIR/.." && pwd)"
usage() {
echo "用法: $0 <版本号> [--gray 百分比] [--no-deploy]"
echo ""
echo "示例:"
echo " $0 v0.1.129 # 打tag + 全量发布"
echo " $0 v0.1.129 --gray 5 # 打tag + 5%灰度发布"
echo " $0 v0.1.129 --no-deploy # 只打tag,不部署"
exit 1
}
# 参数解析
VERSION=""
GRAY_PCT=0
DEPLOY=true
while [[ $# -gt 0 ]]; do
case "$1" in
--gray)
GRAY_PCT="$2"
shift 2
;;
--no-deploy)
DEPLOY=false
shift
;;
-h|--help)
usage
;;
v*)
VERSION="$1"
shift
;;
*)
echo "未知参数: $1"
usage
;;
esac
done
if [[ -z "$VERSION" ]]; then
echo "错误: 请指定版本号(如 v0.1.129"
usage
fi
echo "=========================================="
echo " 发布版本: $VERSION"
echo " 灰度比例: ${GRAY_PCT}%"
echo " 自动部署: $DEPLOY"
echo "=========================================="
cd "$REPO_ROOT"
# 1. 确认在 develop 分支
CURRENT_BRANCH=$(git rev-parse --abbrev-ref HEAD)
if [[ "$CURRENT_BRANCH" != "develop" ]]; then
echo "错误: 请切换到 develop 分支后再发布"
exit 1
fi
# 2. 拉取最新代码
echo ""
echo ">>> 拉取最新代码..."
git pull origin develop
# 3. 生成 CHANGELOG
echo ""
echo ">>> 生成 CHANGELOG..."
if [[ -f "scripts/generate_changelog.py" ]]; then
PREV_TAG=$(git describe --tags --abbrev=0 HEAD^ 2>/dev/null || echo "")
if [[ -n "$PREV_TAG" ]]; then
python3 scripts/generate_changelog.py \
--from-tag "$PREV_TAG" \
--to-tag HEAD \
--gitea-token "${GITEA_TOKEN:-}" \
--output /tmp/changelog_$$.md
echo "CHANGELOG 已生成到 /tmp/changelog_$$.md"
fi
fi
# 4. 打tag
echo ""
echo ">>> 打 tag $VERSION ..."
if git rev-parse "$VERSION" >/dev/null 2>&1; then
echo "警告: tag $VERSION 已存在,跳过打tag"
else
git tag -a "$VERSION" -m "Release $VERSION"
git push origin "$VERSION"
echo "Tag $VERSION 已推送,触发生产镜像构建..."
fi
# 5. 等待镜像构建
if [[ "$DEPLOY" == "true" ]]; then
echo ""
echo ">>> 等待镜像构建完成(约10-15分钟)..."
echo " 镜像: git.xiaoxiajianji.com/xiaoxia/xiaoxia-saas-{api,worker,web}:$VERSION"
# 这里可以加镜像存在性检查
echo " (镜像构建由CI自动完成,请在Gitea Actions中确认)"
fi
# 6. 灰度部署
if [[ "$DEPLOY" == "true" && "$GRAY_PCT" -gt 0 ]]; then
echo ""
echo ">>> 灰度部署: ${GRAY_PCT}% 流量到 $VERSION"
if [[ -f "scripts/gray_deploy.sh" ]]; then
./scripts/gray_deploy.sh "$VERSION" "$GRAY_PCT"
else
echo "警告: gray_deploy.sh 不存在,跳过灰度部署"
fi
fi
echo ""
echo "=========================================="
echo " ✅ 发布流程完成"
echo " 版本: $VERSION"
echo " 灰度: ${GRAY_PCT}%"
echo "=========================================="
-42
View File
@@ -1,42 +0,0 @@
#!/bin/bash
# 灰度回滚脚本:切回稳定版本流量
# 用法: ./scripts/rollback.sh [稳定版本号]
set -euo pipefail
STABLE_VERSION="${1:-current}"
echo "=========================================="
echo " 灰度回滚"
echo " 切回稳定版本: $STABLE_VERSION"
echo "=========================================="
# 1. 恢复Nginx全量到稳定版本
echo ""
echo ">>> 恢复Nginx全量流量到稳定版本..."
NGINX_CONF="${NGINX_CONF:-/etc/nginx/conf.d/saas-api.conf}"
if [[ -f "$NGINX_CONF" ]]; then
# 找最近的备份
LATEST_BAK=$(ls -t "${NGINX_CONF}".bak.* 2>/dev/null | head -1)
if [[ -n "$LATEST_BAK" ]]; then
cp "$LATEST_BAK" "$NGINX_CONF"
echo " 从备份恢复: $LATEST_BAK"
else
echo " 未找到备份,请手动移除 canary upstream"
fi
nginx -t && nginx -s reload
echo " ✅ Nginx 已回滚"
fi
# 2. 停止灰度版本容器(保留30分钟以便排查)
echo ""
echo ">>> 灰度版本容器将在30分钟后停止(便于排查问题)"
echo " 立即停止请执行: docker stop saas-api-canary saas-worker-canary"
echo ""
echo "=========================================="
echo " ✅ 回滚完成"
echo " 流量已全部切回稳定版本"
echo "=========================================="
-269
View File
@@ -1,269 +0,0 @@
"""视频调速引擎单元测试."""
import pytest
from video_processing.speed_engine import (
MAX_SPEED,
MIN_SPEED,
SpeedConfig,
SpeedEngine,
)
# ─── SpeedConfig 解析与校验 ──────────────────────────────────
class TestSpeedConfig:
def test_default_values(self):
config = SpeedConfig()
assert config.speed == 1.0
assert config.pitch_correct is True
def test_parse_none(self):
config = SpeedConfig.parse(None)
assert config.speed == 1.0
assert config.pitch_correct is True
def test_parse_empty_dict(self):
config = SpeedConfig.parse({})
assert config.speed == 1.0
def test_parse_valid_speed(self):
config = SpeedConfig.parse({"speed": 2.0})
assert config.speed == 2.0
def test_parse_pitch_correct_false(self):
config = SpeedConfig.parse({"pitch_correct": False})
assert config.pitch_correct is False
def test_parse_invalid_speed_type(self):
config = SpeedConfig.parse({"speed": "fast"})
assert config.speed == 1.0
def test_parse_invalid_pitch_type(self):
config = SpeedConfig.parse({"pitch_correct": "yes"})
assert config.pitch_correct is True
def test_clamp_below_min(self):
config = SpeedConfig(speed=0.1)
config.clamp()
assert config.speed == MIN_SPEED
def test_clamp_zero(self):
config = SpeedConfig(speed=0)
config.clamp()
assert config.speed == 1.0
def test_clamp_negative(self):
config = SpeedConfig(speed=-1.0)
config.clamp()
assert config.speed == 1.0
def test_clamp_above_max(self):
config = SpeedConfig(speed=10.0)
config.clamp()
assert config.speed == MAX_SPEED
def test_clamp_within_range(self):
config = SpeedConfig(speed=1.5)
config.clamp()
assert config.speed == 1.5
def test_is_original_true(self):
config = SpeedConfig(speed=1.0)
assert config.is_original is True
def test_is_original_false(self):
config = SpeedConfig(speed=1.5)
assert config.is_original is False
def test_parse_clamps_automatically(self):
"""parse 方法应该自动调用 clamp."""
config = SpeedConfig.parse({"speed": 100.0})
assert config.speed == MAX_SPEED
# ─── SpeedEngine 视频滤镜 ────────────────────────────────────
class TestSpeedEngineVideoFilter:
def setup_method(self):
self.engine = SpeedEngine()
def test_original_speed_returns_empty(self):
config = SpeedConfig(speed=1.0)
assert self.engine.build_video_filter(config) == ""
def test_double_speed(self):
config = SpeedConfig(speed=2.0)
result = self.engine.build_video_filter(config)
assert "setpts=PTS/2.0" in result
def test_half_speed(self):
config = SpeedConfig(speed=0.5)
result = self.engine.build_video_filter(config)
assert "setpts=PTS/0.5" in result
def test_quarter_speed(self):
config = SpeedConfig(speed=0.25)
result = self.engine.build_video_filter(config)
assert "setpts=PTS/0.25" in result
def test_quad_speed(self):
config = SpeedConfig(speed=4.0)
result = self.engine.build_video_filter(config)
assert "setpts=PTS/4.0" in result
# ─── SpeedEngine 音频滤镜(atempo 多级串联) ─────────────────
class TestSpeedEngineAudioFilter:
def setup_method(self):
self.engine = SpeedEngine()
def test_original_speed_returns_empty(self):
config = SpeedConfig(speed=1.0)
assert self.engine.build_audio_filter(config) == ""
def test_double_speed_single_stage(self):
"""2x 在 atempo 单级范围内,只需一个 atempo."""
config = SpeedConfig(speed=2.0)
result = self.engine.build_audio_filter(config)
assert result == "atempo=2.0000"
def test_half_speed_single_stage(self):
config = SpeedConfig(speed=0.5)
result = self.engine.build_audio_filter(config)
assert result == "atempo=0.5000"
def test_quad_speed_two_stages(self):
"""4x 需要两级 atempo: 2.0 * 2.0."""
config = SpeedConfig(speed=4.0)
result = self.engine.build_audio_filter(config)
assert result == "atempo=2.0000,atempo=2.0000"
def test_quarter_speed_two_stages(self):
"""0.25x 需要两级 atempo: 0.5 * 0.5."""
config = SpeedConfig(speed=0.25)
result = self.engine.build_audio_filter(config)
assert result == "atempo=0.5000,atempo=0.5000"
def test_triple_speed_two_stages(self):
"""3x: 2.0 * 1.5."""
config = SpeedConfig(speed=3.0)
result = self.engine.build_audio_filter(config)
parts = result.split(",")
assert len(parts) == 2
assert "atempo=2.0000" in parts
assert "atempo=1.5000" in parts
def test_03_speed_two_stages(self):
"""0.3x: 0.5 * 0.6."""
config = SpeedConfig(speed=0.3)
result = self.engine.build_audio_filter(config)
parts = result.split(",")
assert len(parts) == 2
assert "atempo=0.5000" in parts
assert "atempo=0.6000" in parts
def test_split_atempo_inside_range(self):
"""0.5~2.0 范围内只返回一级."""
stages = SpeedEngine._split_atempo_stages(1.5)
assert len(stages) == 1
assert stages[0] == 1.5
def test_split_atempo_boundary_min(self):
stages = SpeedEngine._split_atempo_stages(0.5)
assert len(stages) == 1
assert stages[0] == 0.5
def test_split_atempo_boundary_max(self):
stages = SpeedEngine._split_atempo_stages(2.0)
assert len(stages) == 1
assert stages[0] == 2.0
def test_split_atempo_product_equals_speed(self):
"""所有级联的乘积应该等于原速度."""
test_cases = [0.25, 0.3, 0.5, 0.75, 1.0, 1.5, 2.0, 3.0, 4.0]
for speed in test_cases:
stages = SpeedEngine._split_atempo_stages(speed)
product = 1.0
for s in stages:
product *= s
assert abs(product - speed) < 1e-6, f"speed={speed}, stages={stages}, product={product}"
def test_split_atempo_all_in_range(self):
"""所有级都应该在 0.5~2.0 范围内."""
test_cases = [0.25, 0.3, 0.5, 0.75, 1.0, 1.5, 2.0, 3.0, 4.0]
for speed in test_cases:
stages = SpeedEngine._split_atempo_stages(speed)
for s in stages:
assert 0.5 <= s <= 2.0, f"speed={speed}, stage={s} out of range"
# ─── SpeedEngine 时长计算 ────────────────────────────────────
class TestSpeedEngineDuration:
def setup_method(self):
self.engine = SpeedEngine()
def test_original_speed_same_duration(self):
config = SpeedConfig(speed=1.0)
assert self.engine.adjust_duration(10.0, config) == 10.0
def test_double_speed_half_duration(self):
config = SpeedConfig(speed=2.0)
assert self.engine.adjust_duration(10.0, config) == 5.0
def test_half_speed_double_duration(self):
config = SpeedConfig(speed=0.5)
assert self.engine.adjust_duration(10.0, config) == 20.0
def test_quad_speed_quarter_duration(self):
config = SpeedConfig(speed=4.0)
assert self.engine.adjust_duration(10.0, config) == 2.5
def test_zero_duration(self):
config = SpeedConfig(speed=2.0)
assert self.engine.adjust_duration(0.0, config) == 0.0
def test_negative_duration(self):
config = SpeedConfig(speed=2.0)
assert self.engine.adjust_duration(-1.0, config) == -1.0
# ─── SpeedEngine 便捷方法 ────────────────────────────────────
class TestSpeedEngineHelper:
def setup_method(self):
self.engine = SpeedEngine()
def test_build_clip_speed_filter_original(self):
v_f, a_f, cfg = self.engine.build_clip_speed_filter(1.0)
assert v_f == ""
assert a_f == ""
assert cfg.speed == 1.0
def test_build_clip_speed_filter_2x(self):
v_f, a_f, cfg = self.engine.build_clip_speed_filter(2.0)
assert "setpts=PTS/2.0" in v_f
assert "atempo=2.0" in a_f
assert cfg.speed == 2.0
def test_build_clip_speed_clamped(self):
_, _, cfg = self.engine.build_clip_speed_filter(100.0)
assert cfg.speed == MAX_SPEED
def test_resolve_clip_speed_default(self):
assert SpeedEngine.resolve_clip_speed({}) == 1.0
assert SpeedEngine.resolve_clip_speed(None) == 1.0
def test_resolve_clip_speed_zero_uses_global(self):
assert SpeedEngine.resolve_clip_speed({"playback_speed": 0}, 1.5) == 1.5
def test_resolve_clip_speed_custom(self):
assert SpeedEngine.resolve_clip_speed({"playback_speed": 2.0}) == 2.0
def test_resolve_clip_speed_invalid_type(self):
assert SpeedEngine.resolve_clip_speed({"playback_speed": "fast"}) == 1.0
+35
View File
@@ -0,0 +1,35 @@
# vulture.conf - 死代码检测配置
# 仓库: xiaoxia/xiaoxia-saas
# 用途: 检测未使用的函数、变量、导入、类、方法、属性
# 扫描目录(空格分隔)
path = alembic apps packages scripts
# 排除路径(每个路径一行,相对于仓库根目录)
exclude =
tests
test
*/tests
*/test
site-packages
node_modules
migrations
.gitea
docs
scripts/check_*.py
scripts/init_*.py
# 最低置信度 (%)
# 0 = 报告所有可能的未使用代码
# 100 = 只报告确定未使用的代码
# 推荐从 80% 开始,逐步调高
min-confidence = 80
# 输出格式: string, json, yaml
format = text
# 按置信度排序
sort-by-size = False
# 显示置信度
show-uncertain = True
+57
View File
@@ -0,0 +1,57 @@
# vulture_whitelist.py - vulture 白名单文件
# 用途: 列出已知被框架/动态调用的代码,避免误报
# 参考: https://vulture.readthedocs.io/en/stable/whitelists.html
# FastAPI / Starlette 框架自动调用
# FastAPI route handlers (通过装饰器注册,vulture 可能无法识别)
apps.*.main.*
apps.*.api.*
apps.*.routes.*
apps.*.views.*
# SQLAlchemy ORM
# Model 类和字段通过 ORM 框架自动使用
apps.*.models.*
apps.*.schemas.*
packages.*.models.*
# Pydantic models
# Pydantic 字段通过序列化/反序列化使用
apps.*.schemas.*
packages.*.schemas.*
# Alembic migrations
# Migration 函数由 alembic 自动调用
alembic.versions.*.upgrade
alembic.versions.*.downgrade
# Celery tasks
# Task 函数通过 celery worker 调用
apps.*.tasks.*
packages.*.tasks.*
# CLI scripts / entry points
# 脚本通过命令行调用
scripts.*
# 中间件
apps.*.middleware.*
packages.*.middleware.*
# 异常类
apps.*.exceptions.*
packages.*.exceptions.*
# 配置类
apps.*.config.*
packages.*.config.*
# 工具函数(可能被多处间接调用,先白名单,后续清理)
apps.*.utils.*
packages.*.utils.*
apps.*.helpers.*
packages.*.helpers.*
# Dependencies (FastAPI Depends)
apps.*.dependencies.*
packages.*.dependencies.*