apps/api/app/api/routes/edit_plans_clips_batch.py

@@ -3,7 +3,7 @@
 from __future__ import annotations
 
 import logging
-from typing import Any, List, Optional
+from typing import Any, List
 
 from app.auth import AuthenticatedUser, get_current_user
 from app.dependencies import get_db_session, get_project_repository

apps/api/app/api/routes/edit_plans_cover.py

@@ -9,7 +9,7 @@
 from __future__ import annotations
 
 import logging
-from typing import Any, Dict, Optional
+from typing import Any, Optional
 
 from app.auth import AuthenticatedUser, get_current_user
 from app.core.storage import get_storage_service

apps/api/app/api/routes/edit_plans_filter.py

@@ -20,7 +20,6 @@ from sqlalchemy.orm import Session
 from packages.domain.config_schemas import normalize_plan_config
 from packages.domain.filter_presets import (
     FilterPreset,
-    build_ffmpeg_filter,
     get_filter_preset,
     list_filter_presets,
 )

apps/api/app/api/routes/edit_plans_timeline.py

@@ -20,7 +20,7 @@ from app.api.routes.edit_plans import (
 from app.auth import AuthenticatedUser, get_current_user
 from app.dependencies import get_db_session, get_project_repository
 from app.services import EditPlanService, PlanGeneratorService
-from fastapi import APIRouter, Depends, HTTPException, Query, status
+from fastapi import APIRouter, Depends, HTTPException, status
 from pydantic import BaseModel, Field
 from sqlalchemy.orm import Session
 

apps/api/app/api/routes/task_center.py

@@ -24,8 +24,6 @@ from app.schemas.task_center import (
 from fastapi import APIRouter, Depends, HTTPException, Query
 
 from packages.application import (
-    CreateGenerationTaskCommand,
-    CreateGenerationTaskUseCase,
     RetryGenerationTaskUseCase,
     SubmitIngestJobCommand,
     SubmitIngestJobUseCase,

apps/api/app/api/routes/videos.py

@@ -1,7 +1,5 @@
 import logging
-import uuid
 
-from app.api.routes._helpers import check_project_access
 from app.auth import AuthenticatedUser, get_current_user
 from app.core.celery_app import celery_app
 from app.core.storage import OSSStorageService, get_storage_service
@@ -110,7 +108,9 @@ def update_video_review_status(
     item = use_case.execute(video_id, request.review_status)
     if item is None:
         raise HTTPException(status_code=404, detail="Video not found")
-    logger.info("Video %s review status updated to %s by user %s", video_id, request.review_status, current_user.user_id)
+    logger.info(
+        "Video %s review status updated to %s by user %s", video_id, request.review_status, current_user.user_id
+    )
     return _to_video_response(item, storage)
 
 
@@ -121,7 +121,7 @@ def batch_download_videos(
     current_user: AuthenticatedUser = Depends(get_current_user),
 ):
     """批量下载成片，异步打包 zip。
-    
+
     传入 video_ids 列表，创建一个批量下载任务，任务完成后返回 zip 下载链接。
     """
     if not request.video_ids:
@@ -154,7 +154,7 @@ def get_batch_download_status(
     from celery.result import AsyncResult
 
     task = AsyncResult(job_id, app=celery_app)
-    
+
     status_map = {
         "PENDING": "pending",
         "STARTED": "running",

apps/api/app/services/cover_service.py

@@ -9,7 +9,7 @@ from __future__ import annotations
 import logging
 import tempfile
 from pathlib import Path
-from typing import Any, Dict, Optional
+from typing import Any, Dict
 
 logger = logging.getLogger(__name__)
 

apps/api/app/services/edit_plan_service.py

@@ -749,7 +749,7 @@ class EditPlanService:
 
         config["subtitles"] = subtitles
         clip.config = config
-        updated = self._clip_repo.update(clip)
+        self._clip_repo.update(clip)
 
         logger.info(
             "添加字幕: clip_id=%s subtitle_id=%s start=%.3fs end=%.3fs",

apps/api/app/services/edit_template_service.py

@@ -17,8 +17,6 @@ from packages.adapters.sqlalchemy_impl import (
     SQLAlchemyEditTemplateRepository,
     SQLAlchemyTemplateClipConfigRepository,
 )
-from packages.domain.edit_plan import EditPlan
-from packages.domain.edit_plan_clip import EditPlanClip
 from packages.domain.edit_template import EditTemplate, EditTemplateStatus
 from packages.domain.template_clip_config import (
     ClipType,

apps/worker/video_processing/bgm_mixer.py

@@ -88,14 +88,12 @@ def prepare_bgm_track(
 
     # 构建滤镜链
     filter_parts: list[str] = []
-    input_looped: bool = False
 
     if needs_loop:
         # 计算需要循环多少次才能铺满
         loop_count = max(1, int(target_duration / bgm_dur) + 2)
         # aloop 滤镜：循环指定次数
         filter_parts.append(f"aloop=loop={loop_count}:size=0")
-        input_looped = True
 
     # 音量调节
     volume = max(0.0, min(1.0, bgm.volume))

apps/worker/video_processing/color_grade_engine.py

@@ -10,7 +10,7 @@
 from __future__ import annotations
 
 import logging
-from dataclasses import dataclass, field
+from dataclasses import dataclass
 from typing import Any
 
 logger = logging.getLogger(__name__)

apps/worker/video_processing/concat_engine.py

@@ -18,10 +18,8 @@
 from __future__ import annotations
 
 import logging
-import tempfile
 from dataclasses import dataclass, field
 from pathlib import Path
-from typing import Any
 
 from video_processing.ffmpeg_utils import FFMPEG_BIN, probe_duration, probe_video_info, run_ffmpeg
 from video_processing.path_security import PathSecurityError, is_in_allowed_dirs, safe_resolve_path

apps/worker/video_processing/intro_outro_engine.py

@@ -160,7 +160,6 @@ class IntroOutroEngine:
             是否成功
         """
         duration = config.intro_duration
-        bg = config.intro_background.lstrip("#")
 
         # 转义文字
         title = config.intro_title.replace(":", "\\:").replace("'", "\\'")
@@ -178,9 +177,7 @@ class IntroOutroEngine:
         filter_parts = []
 
         # 背景
-        filter_parts.append(
-            f"color=c={config.intro_background}:s={output_width}x{output_height}:d={duration}[bg]"
-        )
+        filter_parts.append(f"color=c={config.intro_background}:s={output_width}x{output_height}:d={duration}[bg]")
 
         # 标题
         if title:
@@ -357,8 +354,6 @@ class IntroOutroEngine:
         只传了片头或片尾也可以，缺失的自动跳过。
         """
         # 收集所有片段
-        segments: list[tuple[Path, float]] = []  # (path, duration)
-
         # 简单探测时长（用 ffprobe，这里简化处理：直接用 xfade 的 offset）
         # 先添加到列表
         has_intro = intro_video is not None and intro_video.exists()

apps/worker/video_processing/render_audio.py

@@ -156,7 +156,6 @@ def mix_audio(
         from video_processing.bgm_mixer import BGMConfig, mix_bgm_with_main
 
         bgm_cfg = BGMConfig.from_config_dict(bgm_path, bgm_config)
-        bgm_output = ctx.work_dir / f"audio_with_bgm_{ctx.plan_id}.aac"
 
         try:
             # 这里 main_audio 就是 output_path，先有主音频再混 BGM

apps/worker/video_processing/subtitle_render_engine.py

@@ -24,13 +24,11 @@
 from __future__ import annotations
 
 import logging
-from dataclasses import dataclass, field
+from dataclasses import dataclass
 from pathlib import Path
 from typing import Any
 
 from video_processing.path_security import PathSecurityError, is_in_allowed_dirs, safe_resolve_path
-from video_processing.render_subtitles import generate_ass_subtitles
-from video_processing.subtitle_generator import generate_ass_from_timeline
 
 logger = logging.getLogger(__name__)
 

apps/worker/video_processing/unified_render_service.py

@@ -44,7 +44,7 @@ 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.speed_engine import SpeedEngine
 from video_processing.sticker_engine import StickerEngine
 from video_processing.subtitle_generator import generate_ass_from_timeline
 from video_processing.transition_engine import TransitionEngine

apps/worker/video_processing/watermark_engine.py

@@ -243,7 +243,6 @@ class WatermarkEngine:
 
         # 构建滤镜
         # 先缩放水印图
-        wm_input_idx = 1  # 假设水印图是第二个输入（索引1
         filter_parts = [
             f"[1:v]{wm_filter}{wm_pre_label}",
             f"{input_video_label}{wm_pre_label}overlay={overlay_expr}{output_label}",

apps/worker/worker_app/tasks/asset_analyzer.py

@@ -11,7 +11,6 @@ from __future__ import annotations
 import json
 import logging
 import os
-import subprocess
 import tempfile
 from dataclasses import dataclass, field
 

apps/worker/worker_app/tasks/batch_download.py

@@ -3,7 +3,6 @@
 from __future__ import annotations
 
 import logging
-import os
 import tempfile
 import uuid
 import zipfile
@@ -25,7 +24,7 @@ def batch_download_videos(self, video_ids: list[str], user_id: str = "") -> dict
     Returns:
         {"download_url": "...", "file_count": N, "total_size": total_bytes}
     """
-    from video_processing.oss_helpers import download_asset, upload_to_oss
+    from video_processing.oss_helpers import upload_to_oss
     from worker_app.db import SessionLocal
 
     from packages.adapters.sqlalchemy_impl.generated_video_repository import (

apps/worker/worker_app/tasks/edit_plan_generation.py

@@ -439,7 +439,6 @@ def render_edit_plan(self, plan_id: str) -> dict:
                     )
                     if task_status == "cancelled":
                         logger.info("任务已被取消，中止渲染: plan_id=%s task_id=%s", plan_id, generation_task_id)
-                        from packages.domain.edit_plan import EditPlanStatus
 
                         if plan.status.value == "rendering":
                             try:

apps/worker/worker_app/tasks/voice_extraction.py

@@ -2,7 +2,6 @@
 
 import logging
 import os
-import subprocess
 import tempfile
 
 from celery import Task

packages/adapters/sqlalchemy_impl/template_repository.py

@@ -5,7 +5,6 @@ from __future__ import annotations
 import uuid
 from typing import List, Optional
 
-from sqlalchemy import func
 from sqlalchemy.orm import Session
 
 from packages.adapters.sqlalchemy_impl.models import (

packages/adapters/tts/mock_tts_service.py

@@ -161,7 +161,6 @@ class MockTtsService(TtsService):
         tremolo_depth = 0.3  # 30% 深度
 
         # 构建滤镜链
-        filters: list[str] = []
 
         # 生成基频 + 泛音（让声音更丰富）
         # 用多个 sine 波叠加模拟更自然的音色

packages/application/tts_job/streaming_service.py

@@ -11,8 +11,6 @@ import asyncio
 import logging
 from typing import Any, Optional
 
-import httpx
-
 from packages.application.cosyvoice_service import CosyVoiceError, CosyVoiceService
 from packages.application.tts_job.text_splitter import split_text
 from packages.shared.url_security import ALLOWED_AUDIO_MIME_TYPES, safe_download_bytes

packages/application/tts_job/workflow.py

@@ -17,8 +17,6 @@ import tempfile
 from concurrent.futures import ThreadPoolExecutor, as_completed
 from typing import Optional
 
-import httpx
-
 from packages.application.cosyvoice_service import CosyVoiceAuthError, CosyVoiceError, CosyVoiceService
 from packages.application.tts_job.audio_merger import AudioMerger
 from packages.application.tts_job.text_splitter import split_text
@@ -27,10 +25,8 @@ from packages.ports.tts_job_repository import TTSJobRepository
 from packages.shared.storage import SharedStorageService, get_shared_storage_service
 from packages.shared.url_security import (
     ALLOWED_AUDIO_MIME_TYPES,
-    UrlSecurityError,
     safe_download_bytes,
     safe_download_file,
-    validate_url_safety,
 )
 
 logger = logging.getLogger(__name__)

packages/domain/filter_presets.py

@@ -7,7 +7,7 @@
 from __future__ import annotations
 
 from dataclasses import dataclass, field
-from typing import Dict, List, Optional
+from typing import List, Optional
 
 
 @dataclass(frozen=True)

packages/domain/tts_config.py

@@ -2,7 +2,7 @@
 
 from __future__ import annotations
 
-from dataclasses import dataclass, field
+from dataclasses import dataclass
 from typing import Any, Optional
 
 

packages/shared/ffmpeg_utils.py

@@ -11,8 +11,6 @@ from __future__ import annotations
 import logging
 import shutil
 import subprocess  # nosec B404
-from pathlib import Path
-from typing import Any
 
 logger = logging.getLogger(__name__)
 

pyproject.toml

@@ -89,10 +89,8 @@ ignore = [
   "W291",
   "W293",
   "B008",  # function-call-in-default-argument（FastAPI 依赖注入模式，大量使用）
-  "F401",  # unused-import
   "F403",
   "F405",
-  "F841",  # unused-variable
 ]
 
 [tool.ruff.lint.per-file-ignores]