diff --git a/apps/api/app/api/routes/cover_templates.py b/apps/api/app/api/routes/cover_templates.py
index a8782cc8c..5c6d8402c 100644
--- a/apps/api/app/api/routes/cover_templates.py
+++ b/apps/api/app/api/routes/cover_templates.py
@@ -7,7 +7,6 @@ API:
   DELETE /api/v1/cover-templates/{id}  - 删除自定义模板（系统模板不可删）
 """
 
-
 import logging
 from typing import Any
 
diff --git a/apps/api/app/api/routes/generation_cover.py b/apps/api/app/api/routes/generation_cover.py
index ddc63df4e..c9bd2bf4d 100644
--- a/apps/api/app/api/routes/generation_cover.py
+++ b/apps/api/app/api/routes/generation_cover.py
@@ -86,7 +86,9 @@ def generate_cover(
     # 第二步：如果还没有，通过 generation_task_id 查找预览任务的产物
     if not rendered_storage_key:
         generation_task_id = (plan.config or {}).get("generation_task_id", "")
-        logger.info("[封面生成] 步骤2: 通过 generation_task_id 查找: plan_id=%s task_id=%s", plan_id, generation_task_id)
+        logger.info(
+            "[封面生成] 步骤2: 通过 generation_task_id 查找: plan_id=%s task_id=%s", plan_id, generation_task_id
+        )
         if generation_task_id:
             try:
                 gen_task_repo = SQLAlchemyGenerationTaskRepository(db)
@@ -170,7 +172,8 @@ def generate_cover(
         # MediaKit 的 HTTP 客户端会规范化 URL 导致 404
         if primary_video_url:
             import re as _re
-            primary_video_url = _re.sub(r'(?<!:)//', '/', primary_video_url)
+
+            primary_video_url = _re.sub(r"(?<!:)//", "/", primary_video_url)
         logger.info(
             "获取预览视频URL用于封面生成: plan_id=%s url=%s",
             plan_id,
@@ -187,7 +190,8 @@ def generate_cover(
     if cover_candidates and body.cover_type in ("ai_frame", "ai_regenerate"):
         logger.info(
             "[封面生成] 使用预存封面候选帧: plan_id=%s count=%d",
-            plan_id, len(cover_candidates),
+            plan_id,
+            len(cover_candidates),
         )
         first_frame = cover_candidates[0]
         cover_data = {
diff --git a/apps/worker/video_processing/render_adapter.py b/apps/worker/video_processing/render_adapter.py
index cb799bcca..8bf624ae8 100755
--- a/apps/worker/video_processing/render_adapter.py
+++ b/apps/worker/video_processing/render_adapter.py
@@ -74,7 +74,9 @@ class RenderAdapterResult:
     failed_clip_ids: list[str] = None  # 失败的 clip id 列表
     error_message: str = ""
     error_detail: str = ""  # 详细错误信息（如 ffmpeg stderr），用于排查
-    cover_candidates: list[dict] | None = None  # 封面候选帧 [{"image_url": "...", "frame_time": 5.0, "storage_key": "..."}]
+    cover_candidates: list[dict] | None = (
+        None  # 封面候选帧 [{"image_url": "...", "frame_time": 5.0, "storage_key": "..."}]
+    )
 
     def __post_init__(self):
         if self.rendered_clip_ids is None:
@@ -574,18 +576,18 @@ class RenderAdapter:
         try:
             from video_processing.thumbnail_generator import extract_and_upload_cover_frames
 
-            cover_candidates = extract_and_upload_cover_frames(
-                str(result.output_path), plan_id, num_frames=3
-            )
+            cover_candidates = extract_and_upload_cover_frames(str(result.output_path), plan_id, num_frames=3)
             if cover_candidates:
                 logger.info(
                     "[render-adapter] 封面候选帧生成成功: plan_id=%s count=%d",
-                    plan_id, len(cover_candidates),
+                    plan_id,
+                    len(cover_candidates),
                 )
         except Exception as cover_err:
             logger.warning(
                 "[render-adapter] 封面候选帧生成失败（不影响主流程）: plan_id=%s error=%s",
-                plan_id, cover_err,
+                plan_id,
+                cover_err,
             )
 
         self._report_progress(progress_cb, 100.0, "渲染完成")
diff --git a/apps/worker/video_processing/thumbnail_generator.py b/apps/worker/video_processing/thumbnail_generator.py
index 2730ab092..dec1481bd 100755
--- a/apps/worker/video_processing/thumbnail_generator.py
+++ b/apps/worker/video_processing/thumbnail_generator.py
@@ -191,7 +191,7 @@ def extract_cover_candidates(
         ratios.append(i / (num_frames + 1))
 
     results = []
-    for idx, ratio in enumerate(ratios):
+    for _idx, ratio in enumerate(ratios):
         frame_time = max(0.5, duration * ratio)
         tmp = tempfile.NamedTemporaryFile(suffix=".jpg", delete=False)
         tmp.close()
@@ -201,21 +201,29 @@ def extract_cover_candidates(
             seek_str = _format_seek_time(frame_time)
             scale_filter = f"scale={width}:-1:force_original_aspect_ratio=decrease,format=yuvj420p"
             cmd = [
-                FFMPEG_BIN, "-y",
-                "-ss", seek_str,
-                "-i", video_path,
-                "-vframes", "1",
-                "-vf", scale_filter,
-                "-q:v", "2",
+                FFMPEG_BIN,
+                "-y",
+                "-ss",
+                seek_str,
+                "-i",
+                video_path,
+                "-vframes",
+                "1",
+                "-vf",
+                scale_filter,
+                "-q:v",
+                "2",
                 output_path,
             ]
             run_ffmpeg(cmd, capture_output=True, timeout=timeout)
 
             if Path(output_path).exists() and Path(output_path).stat().st_size > 0:
-                results.append({
-                    "local_path": output_path,
-                    "frame_time": round(frame_time, 2),
-                })
+                results.append(
+                    {
+                        "local_path": output_path,
+                        "frame_time": round(frame_time, 2),
+                    }
+                )
             else:
                 Path(output_path).unlink(missing_ok=True)
         except Exception as e:
@@ -256,14 +264,18 @@ def extract_and_upload_cover_frames(
 
             url = upload_to_oss(local_path, storage_key)
             if url:
-                results.append({
-                    "image_url": url,
-                    "frame_time": frame_time,
-                    "storage_key": storage_key,
-                })
+                results.append(
+                    {
+                        "image_url": url,
+                        "frame_time": frame_time,
+                        "storage_key": storage_key,
+                    }
+                )
                 logger.info(
                     "封面候选帧上传成功: plan_id=%s idx=%d frame_time=%.2f",
-                    plan_id, idx, frame_time,
+                    plan_id,
+                    idx,
+                    frame_time,
                 )
         except Exception as e:
             logger.warning("封面候选帧上传失败: plan_id=%s idx=%d error=%s", plan_id, idx, e)
diff --git a/apps/worker/worker_app/tasks/edit_plan_generation.py b/apps/worker/worker_app/tasks/edit_plan_generation.py
index 0d740847c..58155e4ab 100644
--- a/apps/worker/worker_app/tasks/edit_plan_generation.py
+++ b/apps/worker/worker_app/tasks/edit_plan_generation.py
@@ -263,7 +263,8 @@ def _render_with_unified(
         plan.config = plan_config
         logger.info(
             "封面候选帧已写入 plan.config: plan_id=%s count=%d",
-            plan_id, len(result.cover_candidates),
+            plan_id,
+            len(result.cover_candidates),
         )
 
     return _finalize_render_success(
diff --git a/packages/shared/ai_service.py b/packages/shared/ai_service.py
index efe7f7c0a..2ae55042e 100755
--- a/packages/shared/ai_service.py
+++ b/packages/shared/ai_service.py
@@ -379,18 +379,25 @@ def _extract_frames_with_ffmpeg(
 
     # 先用 ffprobe 获取视频时长
     import subprocess as _subprocess
+
     from packages.shared.ffmpeg_utils import FFPROBE_BIN
 
     duration = 30.0  # 默认假设 30 秒
     try:
         probe_result = _subprocess.run(
             [
-                FFPROBE_BIN, "-v", "error",
-                "-show_entries", "format=duration",
-                "-of", "default=noprint_wrappers=1:nokey=1",
+                FFPROBE_BIN,
+                "-v",
+                "error",
+                "-show_entries",
+                "format=duration",
+                "-of",
+                "default=noprint_wrappers=1:nokey=1",
                 video_url,
             ],
-            capture_output=True, text=True, timeout=15,
+            capture_output=True,
+            text=True,
+            timeout=15,
         )
         if probe_result.returncode == 0 and probe_result.stdout.strip():
             duration = float(probe_result.stdout.strip())
@@ -400,7 +407,7 @@ def _extract_frames_with_ffmpeg(
     ratios = [i / (num_frames + 1) for i in range(1, num_frames + 1)]
     results = []
 
-    for idx, ratio in enumerate(ratios):
+    for _idx, ratio in enumerate(ratios):
         frame_time = max(0.5, duration * ratio)
         tmp = tempfile.NamedTemporaryFile(suffix=".jpg", delete=False)
         tmp.close()
@@ -409,11 +416,16 @@ def _extract_frames_with_ffmpeg(
         try:
             seek_str = f"{int(frame_time // 3600):02d}:{int((frame_time % 3600) // 60):02d}:{frame_time % 60:05.2f}"
             cmd = [
-                FFMPEG_BIN, "-y",
-                "-ss", seek_str,
-                "-i", video_url,
-                "-vframes", "1",
-                "-q:v", "2",
+                FFMPEG_BIN,
+                "-y",
+                "-ss",
+                seek_str,
+                "-i",
+                video_url,
+                "-vframes",
+                "1",
+                "-q:v",
+                "2",
                 output_path,
             ]
             run_ffmpeg(cmd, capture_output=True, timeout=timeout)
@@ -475,6 +487,7 @@ def _call_ai_cover_service(
     # ai_frame / ai_regenerate - 使用 FFmpeg 本地抽帧
     if primary_video_url:
         import re as _re
+
         primary_video_url = _re.sub(r"(?<!:)//", "/", primary_video_url)
 
         # 先检查视频 URL 是否可访问
@@ -483,11 +496,12 @@ def _call_ai_cover_service(
             if head_resp.status_code != 200:
                 logger.error(
                     "封面视频URL不可访问: plan_id=%s url=%s status=%d",
-                    plan_id, primary_video_url, head_resp.status_code,
+                    plan_id,
+                    primary_video_url,
+                    head_resp.status_code,
                 )
                 raise RuntimeError(
-                    f"封面生成失败: 预览视频URL不可访问 (HTTP {head_resp.status_code})。"
-                    f"请重新生成预览视频后再试。"
+                    f"封面生成失败: 预览视频URL不可访问 (HTTP {head_resp.status_code})。" f"请重新生成预览视频后再试。"
                 )
         except http_requests.RequestException as e:
             logger.error("封面视频URL连通性检查失败: plan_id=%s url=%s error=%s", plan_id, primary_video_url, e)
@@ -523,7 +537,9 @@ def _call_ai_cover_service(
 
                     logger.info(
                         "FFmpeg 抽帧成功: plan_id=%s frame_time=%.2f url=%s",
-                        plan_id, frame_time_val, public_url[:80],
+                        plan_id,
+                        frame_time_val,
+                        public_url[:80],
                     )
 
                     return {
@@ -546,9 +562,7 @@ def _call_ai_cover_service(
             logger.exception("FFmpeg 远程抽帧失败: %s", str(e))
 
     # 封面生成失败
-    raise RuntimeError(
-        f"封面生成失败: plan_id={plan_id}，无法从视频抽帧。请检查 primary_video_url 是否可访问。"
-    )
+    raise RuntimeError(f"封面生成失败: plan_id={plan_id}，无法从视频抽帧。请检查 primary_video_url 是否可访问。")
 
 
 # ── 公共入口 ────────────────────────────────────────────────────────────────
diff --git a/tests/unit/test_cover_frame_pre_extract.py b/tests/unit/test_cover_frame_pre_extract.py
index 0cd51b329..bef51877e 100644
--- a/tests/unit/test_cover_frame_pre_extract.py
+++ b/tests/unit/test_cover_frame_pre_extract.py
@@ -24,6 +24,7 @@ class TestExtractCoverCandidates:
     def test_extracts_3_frames_at_correct_positions(self, mock_probe, mock_run):
         """在 25%/50%/75% 处抽取 3 帧."""
         import tempfile
+
         from video_processing.thumbnail_generator import extract_cover_candidates
 
         # Create temp files that look like they were created
@@ -63,6 +64,7 @@ class TestExtractCoverCandidates:
     def test_handles_ffmpeg_failure_gracefully(self, mock_probe, mock_run):
         """FFmpeg 失败时跳过该帧，继续抽取其他帧."""
         import tempfile
+
         from video_processing.thumbnail_generator import extract_cover_candidates
 
         call_count = 0
@@ -96,6 +98,7 @@ class TestExtractCoverCandidates:
     def test_fallback_duration_when_probe_fails(self, mock_probe):
         """probe 失败时使用默认时长."""
         import tempfile
+
         from video_processing.thumbnail_generator import extract_cover_candidates
 
         # Mock run_ffmpeg to create output files
@@ -130,6 +133,7 @@ class TestExtractAndUploadCoverFrames:
     def test_uploads_and_returns_correct_format(self, mock_extract, mock_upload):
         """上传帧到 OSS 并返回正确格式."""
         import tempfile
+
         from video_processing.thumbnail_generator import extract_and_upload_cover_frames
 
         # Create actual temp files
@@ -172,6 +176,7 @@ class TestExtractAndUploadCoverFrames:
     def test_handles_upload_failure_gracefully(self, mock_extract, mock_upload):
         """上传失败时跳过该帧."""
         import tempfile
+
         from video_processing.thumbnail_generator import extract_and_upload_cover_frames
 
         tmp1 = tempfile.NamedTemporaryFile(suffix=".jpg", delete=False)