chore(ci): restore -x flag after diagnosis

.flake8

@@ -2,6 +2,7 @@
 max-line-length = 120
 exclude =
     .git,
+    .cache,
     __pycache__,
     .venv,
     venv,

.gitea/workflows/ci-cd.yml

@@ -165,7 +165,7 @@ jobs:
         run: |
           set -eu
           PYTHONPATH="$PWD/apps/api:$PWD" python3 -m pytest tests/unit -q \
-            --cov=apps --cov-report=term --cov-report=xml
+            --cov=apps --cov-report=term --cov-report=xml --cov-fail-under=50
 
       - name: Start PostgreSQL for integration tests
         shell: sh

.gitignore

@@ -6,6 +6,7 @@ dist/
 coverage/
 
 # Python / backend
+.cache/
 .venv/
 venv/
 .venv-ci-root/

apps/worker/video_processing/unified_render_service.py

@@ -314,7 +314,9 @@ class UnifiedRenderService:
             # 有效时长 = min(指定时长, 实际时长)；若均未设置则跳过
             effective_duration = 0.0
             if clip.duration > 0:
-                effective_duration = min(clip.duration, clip.actual_duration) if clip.actual_duration > 0 else clip.duration
+                effective_duration = (
+                    min(clip.duration, clip.actual_duration) if clip.actual_duration > 0 else clip.duration
+                )
             elif clip.actual_duration > 0:
                 effective_duration = clip.actual_duration
 

apps/worker/worker_app/tasks/generation.py

@@ -778,14 +778,12 @@ def generate_video(self, task_id: str) -> dict:
             verify_url = get_signed_download_url(file_url, expires_seconds=300) or file_url
             if not _verify_url_accessible(verify_url):
                 # 预签名 URL 也访问失败时，退一步用 object_exists 确认上传成功
-                from video_processing.oss_helpers import oss_bucket, normalize_storage_key
+                from video_processing.oss_helpers import normalize_storage_key, oss_bucket
 
                 bucket = oss_bucket()
                 key = normalize_storage_key(file_url)
                 if bucket and bucket.object_exists(key):
-                    logger.info(
-                        "URL 校验失败但 object_exists 确认文件存在，视为上传成功: storage_key=%s", key
-                    )
+                    logger.info("URL 校验失败但 object_exists 确认文件存在，视为上传成功: storage_key=%s", key)
                     if gen_task:
                         gen_task.append_log("OSS上传", "URL校验降级: object_exists确认存在", level="WARN")
                 else:

apps/worker/worker_app/tasks/voice_clone.py

@@ -4,6 +4,7 @@ import logging
 
 from celery import Task
 from celery.exceptions import Retry
+from video_processing.oss_helpers import get_signed_download_url
 from worker_app.celery_app import celery_app
 from worker_app.db import SessionLocal
 
@@ -16,7 +17,6 @@ from packages.application.cosyvoice_service import (
     CosyVoiceTimeoutError,
 )
 from packages.application.voice_clone.workflow import VoiceCloneWorkflowService
-from video_processing.oss_helpers import get_signed_download_url
 
 logger = logging.getLogger(__name__)
 

packages/application/cosyvoice_service.py

@@ -121,9 +121,7 @@ class CosyVoiceService:
         self._api_key = api_key or settings.cosyvoice_api_key
         self._base_url = base_url or settings.cosyvoice_base_url
         self._model = model or settings.cosyvoice_model
-        self._clone_model = clone_model or getattr(
-            settings, "cosyvoice_clone_model", "voice-enrollment"
-        )
+        self._clone_model = clone_model or getattr(settings, "cosyvoice_clone_model", "voice-enrollment")
         self._audio_url_signer = audio_url_signer
 
         # base_url 规范化：去掉末尾的路径残留（兼容旧版配置）
@@ -134,11 +132,11 @@ class CosyVoiceService:
             # 截取到 /api/v1 为止
             idx = self._base_url.find("/api/v1")
             if idx >= 0:
-                self._base_url = self._base_url[:idx + len("/api/v1")]
+                self._base_url = self._base_url[: idx + len("/api/v1")]
             logger.warning(
-                "[CosyVoice Config] base_url包含旧版text2audio路径，已自动修正: "
-                "%s -> %s",
-                old_url, self._base_url,
+                "[CosyVoice Config] base_url包含旧版text2audio路径，已自动修正: " "%s -> %s",
+                old_url,
+                self._base_url,
             )
 
         self._client = http_client or httpx.Client(
@@ -236,8 +234,7 @@ class CosyVoiceService:
         if self._audio_url_signer:
             try:
                 signed_audio_url = self._audio_url_signer(audio_url)
-                logger.info("音频URL已预签名: original=%s signed_prefix=%s",
-                            audio_url[:80], signed_audio_url[:80])
+                logger.info("音频URL已预签名: original=%s signed_prefix=%s", audio_url[:80], signed_audio_url[:80])
             except Exception as e:
                 logger.warning("音频URL预签名失败，使用原始URL: %s", e)
 
@@ -358,9 +355,7 @@ class CosyVoiceService:
         while attempts < self.CLONE_MAX_POLL_ATTEMPTS:
             elapsed = time.time() - start_time
             if elapsed > timeout:
-                raise CosyVoiceTimeoutError(
-                    f"音色克隆任务超时（{timeout}秒）: voice_id={voice_id}"
-                )
+                raise CosyVoiceTimeoutError(f"音色克隆任务超时（{timeout}秒）: voice_id={voice_id}")
 
             result = self.query_voice_status(voice_id)
             status = result.get("status", "").upper()
@@ -368,9 +363,7 @@ class CosyVoiceService:
             if status == "OK":
                 return {"voice_id": voice_id}
             elif status == "UNDEPLOYED":
-                raise CosyVoiceError(
-                    f"音色克隆任务失败（审核未通过）: voice_id={voice_id}"
-                )
+                raise CosyVoiceError(f"音色克隆任务失败（审核未通过）: voice_id={voice_id}")
             elif status in ("DEPLOYING", "PENDING", "PROCESSING", ""):
                 # 继续轮询
                 time.sleep(self.CLONE_POLL_INTERVAL)
@@ -380,9 +373,7 @@ class CosyVoiceService:
                 time.sleep(self.CLONE_POLL_INTERVAL)
                 attempts += 1
 
-        raise CosyVoiceTimeoutError(
-            f"音色克隆任务轮询次数超限: voice_id={voice_id}"
-        )
+        raise CosyVoiceTimeoutError(f"音色克隆任务轮询次数超限: voice_id={voice_id}")
 
     def clone_voice(
         self,
@@ -497,9 +488,7 @@ class CosyVoiceService:
         request_id = response.get("request_id", "")
 
         if not audio_url:
-            raise CosyVoiceError(
-                f"CosyVoice API 未返回 audio_url: {response}"
-            )
+            raise CosyVoiceError(f"CosyVoice API 未返回 audio_url: {response}")
 
         return {
             "task_id": "",  # 同步接口无 task_id，兼容旧接口
@@ -509,9 +498,7 @@ class CosyVoiceService:
             "request_id": request_id,
         }
 
-    def poll_synthesize_task(
-        self, task_id: str, timeout: float = 120.0
-    ) -> dict:
+    def poll_synthesize_task(self, task_id: str, timeout: float = 120.0) -> dict:
         """轮询合成任务（同步接口无需轮询，保留兼容）.
 
         CosyVoice SpeechSynthesizer 非流式接口是同步的，
@@ -520,10 +507,7 @@ class CosyVoiceService:
         Raises:
             CosyVoiceError: 同步接口无需轮询
         """
-        raise CosyVoiceError(
-            "CosyVoice 非流式合成接口是同步的，无需轮询. "
-            "请直接使用 submit_synthesize_task()."
-        )
+        raise CosyVoiceError("CosyVoice 非流式合成接口是同步的，无需轮询. " "请直接使用 submit_synthesize_task().")
 
     def synthesize_speech(
         self,
@@ -626,15 +610,17 @@ class CosyVoiceService:
 
         # DEBUG: 打印完整请求信息，用于排查418错误
         import json as json_lib
+
         safe_headers = {k: v for k, v in headers.items()}
         if "Authorization" in safe_headers:
             token = safe_headers["Authorization"]
             if len(token) > 20:
                 safe_headers["Authorization"] = token[:13] + "..." + token[-4:]
         logger.info(
-            "[CosyVoice Debug] 请求详情: "
-            "method=%s, url=%s, headers=%s, body=%s",
-            method, url, safe_headers,
+            "[CosyVoice Debug] 请求详情: " "method=%s, url=%s, headers=%s, body=%s",
+            method,
+            url,
+            safe_headers,
             json_lib.dumps(json, ensure_ascii=False) if json else "None",
         )
 
@@ -652,8 +638,7 @@ class CosyVoiceService:
 
                 # DEBUG: 打印响应状态和完整响应体
                 logger.info(
-                    "[CosyVoice Debug] 响应详情: "
-                    "status=%d, body=%s",
+                    "[CosyVoice Debug] 响应详情: " "status=%d, body=%s",
                     response.status_code,
                     response.text[:2000],  # 最多2000字符，避免日志过大
                 )
@@ -662,9 +647,7 @@ class CosyVoiceService:
                 if response.status_code == 200:
                     return response.json()
                 elif response.status_code in (401, 403):
-                    raise CosyVoiceAuthError(
-                        f"CosyVoice API 认证失败: HTTP {response.status_code}"
-                    )
+                    raise CosyVoiceAuthError(f"CosyVoice API 认证失败: HTTP {response.status_code}")
                 elif response.status_code == 400:
                     # 客户端错误，不重试
                     body_text = response.text
@@ -672,19 +655,12 @@ class CosyVoiceService:
                         body = response.json()
                         code = body.get("code", "")
                         message = body.get("message", "")
-                        raise CosyVoiceError(
-                            f"CosyVoice API 参数错误: HTTP 400, "
-                            f"code={code}, message={message}"
-                        )
+                        raise CosyVoiceError(f"CosyVoice API 参数错误: HTTP 400, " f"code={code}, message={message}")
                     except ValueError:
-                        raise CosyVoiceError(
-                            f"CosyVoice API 调用失败: HTTP 400, body={body_text}"
-                        )
+                        raise CosyVoiceError(f"CosyVoice API 调用失败: HTTP 400, body={body_text}")
                 elif response.status_code >= 500:
                     # 服务端错误，可重试
-                    last_error = CosyVoiceError(
-                        f"CosyVoice API 服务端错误: HTTP {response.status_code}"
-                    )
+                    last_error = CosyVoiceError(f"CosyVoice API 服务端错误: HTTP {response.status_code}")
                     logger.warning(
                         "CosyVoice API 失败 (尝试 %d/%d): HTTP %d",
                         attempt + 1,
@@ -694,8 +670,7 @@ class CosyVoiceService:
                 else:
                     # 其他客户端错误，不重试
                     raise CosyVoiceError(
-                        f"CosyVoice API 调用失败: HTTP {response.status_code}, "
-                        f"body={response.text}"
+                        f"CosyVoice API 调用失败: HTTP {response.status_code}, " f"body={response.text}"
                     )
 
             except httpx.TimeoutException as e:

packages/application/tts_job/workflow.py

@@ -219,9 +219,7 @@ class TTSWorkflowService:
 
         # 新接口（同步）：没有 task_id，重新合成
         if not task_id:
-            logger.info(
-                f"TTS 任务无 task_id，重新同步合成: job_id={job_id}"
-            )
+            logger.info(f"TTS 任务无 task_id，重新同步合成: job_id={job_id}")
             return self._resynthesize_and_complete(job)
 
         # 旧接口遗留的 task_id，尝试轮询（兼容过渡）
@@ -235,9 +233,7 @@ class TTSWorkflowService:
             )
         except CosyVoiceError:
             # 旧接口轮询失败，重新同步合成
-            logger.warning(
-                f"旧 task_id 轮询失败，重新同步合成: job_id={job_id}, task_id={task_id}"
-            )
+            logger.warning(f"旧 task_id 轮询失败，重新同步合成: job_id={job_id}, task_id={task_id}")
             return self._resynthesize_and_complete(job)
 
     def process_synthesis_result(
@@ -524,10 +520,7 @@ class TTSWorkflowService:
         missing_indices = [i for i in range(segment_count) if results[i] is None]
 
         if missing_indices:
-            logger.info(
-                f"分段任务重新合成缺失段: job_id={job.id}, "
-                f"缺失={len(missing_indices)}/{segment_count}"
-            )
+            logger.info(f"分段任务重新合成缺失段: job_id={job.id}, " f"缺失={len(missing_indices)}/{segment_count}")
             # 并发重新合成缺失分段
             max_workers = min(len(missing_indices), _MAX_SEGMENT_WORKERS)
             with ThreadPoolExecutor(max_workers=max_workers) as executor:
@@ -550,13 +543,8 @@ class TTSWorkflowService:
                     try:
                         results[idx] = future.result()
                     except Exception as e:
-                        logger.error(
-                            f"分段重新合成失败: job_id={job.id}, "
-                            f"segment={idx}, error={e}"
-                        )
-                        self._handle_segment_failure(
-                            job, f"分段 {idx + 1} 重新合成失败: {e}"
-                        )
+                        logger.error(f"分段重新合成失败: job_id={job.id}, " f"segment={idx}, error={e}")
+                        self._handle_segment_failure(job, f"分段 {idx + 1} 重新合成失败: {e}")
                         return self.repository.get(job.id)
 
         # 所有分段完成，下载合并
@@ -564,9 +552,7 @@ class TTSWorkflowService:
             try:
                 merged_data, total_duration = self._download_and_merge_segments(results, job)
 
-                permanent_url, storage_key = self._upload_merged_to_oss(
-                    merged_data, job.user_id, job.id, job.format
-                )
+                permanent_url, storage_key = self._upload_merged_to_oss(merged_data, job.user_id, job.id, job.format)
 
                 job.mark_completed(
                     output_audio_url=permanent_url,
@@ -575,10 +561,7 @@ class TTSWorkflowService:
                     file_size=len(merged_data),
                 )
                 job = self.repository.update(job)
-                logger.info(
-                    f"分段合成完成（重新合成路径）: job_id={job.id}, "
-                    f"merged_size={len(merged_data)}"
-                )
+                logger.info(f"分段合成完成（重新合成路径）: job_id={job.id}, " f"merged_size={len(merged_data)}")
                 return job
 
             except Exception as e:

packages/application/voice_clone/workflow.py

@@ -133,7 +133,9 @@ class VoiceCloneWorkflowService:
 
                 profile.metadata = task_metadata
                 profile = self.repository.update(profile)
-                logger.info(f"音色克隆任务已提交: profile_id={profile.id}, " f"voice_id={submit_result.get('voice_id')}")
+                logger.info(
+                    f"音色克隆任务已提交: profile_id={profile.id}, " f"voice_id={submit_result.get('voice_id')}"
+                )
 
             except (CosyVoiceError, CosyVoiceAuthError) as e:
                 # CosyVoice 提交失败，标记为 failed

pyproject.toml

@@ -1,7 +1,39 @@
 [tool.black]
 line-length = 120
 target-version = ["py312"]
+extend-exclude = '''
+(
+  \.git
+  | \.cache
+  | \.pytest_cache
+  | \.mypy_cache
+  | __pycache__
+  | node_modules
+  | \.venv
+  | venv
+  | build
+  | dist
+  | \.next
+  | out
+  | coverage
+)
+'''
 
 [tool.isort]
 profile = "black"
 line_length = 120
+extend_skip_glob = [
+    ".git/**",
+    ".cache/**",
+    ".pytest_cache/**",
+    ".mypy_cache/**",
+    "__pycache__/**",
+    "node_modules/**",
+    ".venv/**",
+    "venv/**",
+    "build/**",
+    "dist/**",
+    ".next/**",
+    "out/**",
+    "coverage/**",
+]

setup.cfg

@@ -3,6 +3,7 @@ max-line-length = 120
 extend-ignore = E203,W503,E501,E302,E402,E722,W291,W293,F401,F403,F405,F841
 exclude =
     .git,
+    .cache,
     __pycache__,
     .venv,
     .venv-ci-root,

tests/integration/test_generation_api.py

@@ -241,7 +241,7 @@ def client():
 class TestCreateGenerationTask:
     """创建生成任务端点测试。"""
 
-    @patch("app.api.routes.generation_tasks.celery_app")
+    @patch("app.core.task_enqueue.celery_app")
     def test_create_task_success(self, mock_celery, client):
         """正常创建生成任务成功。"""
         mock_celery.send_task = MagicMock()
@@ -270,7 +270,7 @@ class TestCreateGenerationTask:
         assert mock_celery.send_task.called
         assert mock_celery.send_task.call_args[0][0] == "worker.generate_video"
 
-    @patch("app.api.routes.generation_tasks.celery_app")
+    @patch("app.core.task_enqueue.celery_app")
     def test_create_batch_tasks(self, mock_celery, client):
         """批量创建多个生成任务。"""
         mock_celery.send_task = MagicMock()
@@ -347,7 +347,7 @@ class TestListGenerationTasks:
 
     def _create_task(self, client, task_suffix: str = "1"):
         """辅助方法：创建一个生成任务。"""
-        with patch("app.api.routes.generation_tasks.celery_app") as mock_celery:
+        with patch("app.core.task_enqueue.celery_app") as mock_celery:
             mock_celery.send_task = MagicMock()
             resp = client.post(
                 "/api/v1/generation/tasks",
@@ -368,7 +368,7 @@ class TestListGenerationTasks:
         assert "items" in data
         assert data["items"] == []
 
-    @patch("app.api.routes.generation_tasks.celery_app")
+    @patch("app.core.task_enqueue.celery_app")
     def test_list_returns_user_tasks(self, mock_celery, client):
         """返回当前用户的生成任务列表。"""
         mock_celery.send_task = MagicMock()
@@ -406,7 +406,7 @@ class TestGetGenerationTask:
     """获取生成任务详情端点测试。"""
 
     def _create_task(self, client) -> str:
-        with patch("app.api.routes.generation_tasks.celery_app") as mock_celery:
+        with patch("app.core.task_enqueue.celery_app") as mock_celery:
             mock_celery.send_task = MagicMock()
             resp = client.post(
                 "/api/v1/generation/tasks",
@@ -449,7 +449,7 @@ class TestListGenerationResults:
     """列出生成结果端点测试。"""
 
     def _create_task(self, client) -> str:
-        with patch("app.api.routes.generation_tasks.celery_app") as mock_celery:
+        with patch("app.core.task_enqueue.celery_app") as mock_celery:
             mock_celery.send_task = MagicMock()
             resp = client.post(
                 "/api/v1/generation/tasks",
@@ -489,7 +489,7 @@ class TestRetryGenerationTask:
 
     def _create_failed_task(self, client) -> str:
         """创建一个失败状态的任务。"""
-        with patch("app.api.routes.generation_tasks.celery_app") as mock_celery:
+        with patch("app.core.task_enqueue.celery_app") as mock_celery:
             mock_celery.send_task = MagicMock()
             resp = client.post(
                 "/api/v1/generation/tasks",
@@ -509,7 +509,7 @@ class TestRetryGenerationTask:
         # 让我们直接通过 retry 测试来验证
         return task_id
 
-    @patch("app.api.routes.generation_tasks.celery_app")
+    @patch("app.core.task_enqueue.celery_app")
     def test_retry_failed_task(self, mock_celery, client):
         """重试失败的任务成功。"""
         mock_celery.send_task = MagicMock()
@@ -539,7 +539,7 @@ class TestRetryGenerationTask:
         assert resp.status_code == 404
         assert "not found" in resp.json()["detail"].lower()
 
-    @patch("app.api.routes.generation_tasks.celery_app")
+    @patch("app.core.task_enqueue.celery_app")
     def test_retry_completed_task_returns_409(self, mock_celery, client):
         """重试已完成的任务返回 409。"""
         mock_celery.send_task = MagicMock()
@@ -568,7 +568,7 @@ class TestRetryGenerationTask:
 class TestGenerationTaskFlow:
     """生成任务完整流程集成测试。"""
 
-    @patch("app.api.routes.generation_tasks.celery_app")
+    @patch("app.core.task_enqueue.celery_app")
     def test_create_list_detail_results_flow(self, mock_celery, client):
         """测试创建 → 列表 → 详情 → 结果 完整流程。"""
         mock_celery.send_task = MagicMock()

tests/integration/test_task_center_api.py

@@ -428,7 +428,7 @@ class TestRetryProjectTask:
         assert resp.status_code == 400
         assert "Unsupported" in resp.json()["detail"]
 
-    @patch("app.api.routes.task_center.celery_app")
+    @patch("app.core.task_enqueue.celery_app")
     def test_retry_failed_generation_task(self, mock_celery, client):
         """重试失败的 generation 任务成功。"""
         mock_celery.send_task = MagicMock()
@@ -581,7 +581,7 @@ class TestRetryProjectTask:
 class TestTaskCenterCrossEndpoint:
     """任务中心跨端点集成测试。"""
 
-    @patch("app.api.routes.task_center.celery_app")
+    @patch("app.core.task_enqueue.celery_app")
     def test_list_then_retry_then_list(self, mock_celery, client):
         """列出任务 → 重试失败任务 → 再列出验证新任务。"""
         mock_celery.send_task = MagicMock()

tests/unit/test_cosyvoice_service.py

@@ -197,7 +197,11 @@ class TestSubmitCloneTask:
 
         payload = mock_client.request.call_args.kwargs["json"]
         # 中文和特殊字符被过滤，剩下字母数字
-        assert payload["input"]["prefix"] == "2024" or payload["input"]["prefix"] == "clone" or len(payload["input"]["prefix"]) <= 10
+        assert (
+            payload["input"]["prefix"] == "2024"
+            or payload["input"]["prefix"] == "clone"
+            or len(payload["input"]["prefix"]) <= 10
+        )
 
     def test_submit_auth_401_raises(self) -> None:
         mock_client = MagicMock()
@@ -305,9 +309,7 @@ class TestPollCloneTask:
 
     def test_poll_undeployed_raises_error(self) -> None:
         mock_client = MagicMock()
-        mock_client.request.return_value = _mock_response(
-            200, {"output": {"status": "UNDEPLOYED"}}
-        )
+        mock_client.request.return_value = _mock_response(200, {"output": {"status": "UNDEPLOYED"}})
 
         service = _make_service(http_client=mock_client)
         service.CLONE_POLL_INTERVAL = 0.01
@@ -317,9 +319,7 @@ class TestPollCloneTask:
 
     def test_poll_timeout_raises(self) -> None:
         mock_client = MagicMock()
-        mock_client.request.return_value = _mock_response(
-            200, {"output": {"status": "DEPLOYING"}}
-        )
+        mock_client.request.return_value = _mock_response(200, {"output": {"status": "DEPLOYING"}})
 
         service = _make_service(http_client=mock_client)
         service.CLONE_POLL_INTERVAL = 0.01
@@ -396,9 +396,7 @@ class TestSynthesizeSpeech:
         )
 
         service = _make_service(http_client=mock_client)
-        result = service.synthesize_speech(
-            text="你好世界", voice_id="longxiaochun_v3"
-        )
+        result = service.synthesize_speech(text="你好世界", voice_id="longxiaochun_v3")
 
         assert isinstance(result, SynthesizeResult)
         assert result.audio_url == "https://dashscope-result.oss.com/output.mp3"
@@ -426,8 +424,12 @@ class TestSynthesizeSpeech:
 
         service = _make_service(http_client=mock_client)
         service.synthesize_speech(
-            text="test", voice_id="v1", sample_rate=44100,
-            format="wav", speed=1.5, volume=80,
+            text="test",
+            voice_id="v1",
+            sample_rate=44100,
+            format="wav",
+            speed=1.5,
+            volume=80,
         )
 
         payload = mock_client.request.call_args.kwargs["json"]
@@ -463,7 +465,8 @@ class TestSynthesizeSpeech:
         """同步接口的 submit_synthesize_task 返回空 task_id 字段（兼容旧接口）."""
         mock_client = MagicMock()
         mock_client.request.return_value = _mock_response(
-            200, {"output": {"audio": {"url": "https://e.com/a.mp3"}}},
+            200,
+            {"output": {"audio": {"url": "https://e.com/a.mp3"}}},
         )
 
         service = _make_service(http_client=mock_client)
@@ -490,9 +493,7 @@ class TestRetryLogic:
         mock_client.request.side_effect = [
             _mock_response(500, text="Server Error"),
             _mock_response(502, text="Bad Gateway"),
-            _mock_response(
-                200, {"output": {"audio": {"url": "https://e.com/a.mp3"}}}
-            ),
+            _mock_response(200, {"output": {"audio": {"url": "https://e.com/a.mp3"}}}),
         ]
 
         service = _make_service(http_client=mock_client)
@@ -547,9 +548,7 @@ class TestSanitizePrefix:
 class TestCheckTaskStatus:
     def test_check_task_status_uses_query_voice(self) -> None:
         mock_client = MagicMock()
-        mock_client.request.return_value = _mock_response(
-            200, {"output": {"status": "OK"}}
-        )
+        mock_client.request.return_value = _mock_response(200, {"output": {"status": "OK"}})
 
         service = _make_service(http_client=mock_client)
         result = service.check_task_status("voice-123")

tests/unit/test_p02_p03_fixes.py

@@ -15,7 +15,6 @@ from video_processing.ffmpeg_utils import build_xfade_filter_chain
 # ── P0-3: build_xfade_filter_chain 安全钳制 ──────────────────────────────────
 
 
-
 class TestBuildXfadeFilterChainSafetyClamp:
     """验证 xfade 滤镜链的安全钳制逻辑，防止 exit 234。"""
 
@@ -124,9 +123,7 @@ class TestBuildXfadeFilterChainSafetyClamp:
             durations_found.append(float(m.group(1)))
 
         # 第一个 xfade: td 必须 ≤ 0.3 (第二个输入 clip_durations[1]=0.3)
-        assert durations_found[0] <= 0.3 + 0.001, (
-            f"第一个 xfade td={durations_found[0]} 超过 clip_durations[1]=0.3"
-        )
+        assert durations_found[0] <= 0.3 + 0.001, f"第一个 xfade td={durations_found[0]} 超过 clip_durations[1]=0.3"
         # 第二个 xfade: td 可以 = 0.5 (clip_durations[2]=5.0)
         assert durations_found[1] <= 0.5 + 0.001
         assert dur > 0
@@ -173,9 +170,9 @@ class TestBuildXfadeFilterChainSafetyClamp:
                 assert dur_val >= 0.001  # 至少 1ms
                 # P1 修复验证: td 不能超过第二个输入片段时长
                 second_input_idx = xfade_idx + 1
-                assert dur_val <= durations[second_input_idx] + 0.001, (
-                    f"td={dur_val} > clip_durations[{second_input_idx}]={durations[second_input_idx]}"
-                )
+                assert (
+                    dur_val <= durations[second_input_idx] + 0.001
+                ), f"td={dur_val} > clip_durations[{second_input_idx}]={durations[second_input_idx]}"
                 xfade_idx += 1
 
 

tests/unit/test_p02_worker_oss_fix.py

@@ -13,7 +13,6 @@ from unittest.mock import MagicMock, patch
 
 import pytest
 
-
 # ── oss_bucket endpoint scheme 修复 ──────────────────────────────────────────
 
 
@@ -25,17 +24,19 @@ class TestOSSBucketEndpointScheme:
         from video_processing.oss_helpers import oss_bucket
 
         mock_bucket_instance = MagicMock()
-        with patch.dict(
-            os.environ,
-            {
-                "OSS_ACCESS_KEY_ID": "test-key",
-                "OSS_ACCESS_KEY_SECRET": "test-secret",
-                "OSS_ENDPOINT": "oss-cn-hangzhou.aliyuncs.com",
-                "OSS_BUCKET_NAME": "test-bucket",
-            },
-        ), patch("video_processing.oss_helpers.oss2.Auth") as mock_auth, patch(
-            "video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket_instance
-        ) as mock_bucket_cls:
+        with (
+            patch.dict(
+                os.environ,
+                {
+                    "OSS_ACCESS_KEY_ID": "test-key",
+                    "OSS_ACCESS_KEY_SECRET": "test-secret",
+                    "OSS_ENDPOINT": "oss-cn-hangzhou.aliyuncs.com",
+                    "OSS_BUCKET_NAME": "test-bucket",
+                },
+            ),
+            patch("video_processing.oss_helpers.oss2.Auth") as mock_auth,
+            patch("video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket_instance) as mock_bucket_cls,
+        ):
             # 清除缓存，确保重新创建
             import video_processing.oss_helpers as oss_mod
 
@@ -45,9 +46,7 @@ class TestOSSBucketEndpointScheme:
             # 验证 endpoint 传的是带 https:// 的
             call_args = mock_bucket_cls.call_args
             endpoint_arg = call_args[0][1]  # 第 2 个位置参数是 endpoint
-            assert endpoint_arg.startswith("https://"), (
-                f"endpoint 应该带 https:// 前缀，实际为: {endpoint_arg}"
-            )
+            assert endpoint_arg.startswith("https://"), f"endpoint 应该带 https:// 前缀，实际为: {endpoint_arg}"
             assert "oss-cn-hangzhou.aliyuncs.com" in endpoint_arg
 
     def test_endpoint_with_https_keeps_as_is(self):
@@ -55,17 +54,19 @@ class TestOSSBucketEndpointScheme:
         from video_processing.oss_helpers import oss_bucket
 
         mock_bucket_instance = MagicMock()
-        with patch.dict(
-            os.environ,
-            {
-                "OSS_ACCESS_KEY_ID": "test-key",
-                "OSS_ACCESS_KEY_SECRET": "test-secret",
-                "OSS_ENDPOINT": "https://oss-cn-hangzhou.aliyuncs.com",
-                "OSS_BUCKET_NAME": "test-bucket",
-            },
-        ), patch("video_processing.oss_helpers.oss2.Auth"), patch(
-            "video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket_instance
-        ) as mock_bucket_cls:
+        with (
+            patch.dict(
+                os.environ,
+                {
+                    "OSS_ACCESS_KEY_ID": "test-key",
+                    "OSS_ACCESS_KEY_SECRET": "test-secret",
+                    "OSS_ENDPOINT": "https://oss-cn-hangzhou.aliyuncs.com",
+                    "OSS_BUCKET_NAME": "test-bucket",
+                },
+            ),
+            patch("video_processing.oss_helpers.oss2.Auth"),
+            patch("video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket_instance) as mock_bucket_cls,
+        ):
             import video_processing.oss_helpers as oss_mod
 
             bucket = oss_bucket()
@@ -81,17 +82,19 @@ class TestOSSBucketEndpointScheme:
         from video_processing.oss_helpers import oss_bucket
 
         mock_bucket_instance = MagicMock()
-        with patch.dict(
-            os.environ,
-            {
-                "OSS_ACCESS_KEY_ID": "test-key",
-                "OSS_ACCESS_KEY_SECRET": "test-secret",
-                "OSS_ENDPOINT": "http://oss-cn-hangzhou.aliyuncs.com",
-                "OSS_BUCKET_NAME": "test-bucket",
-            },
-        ), patch("video_processing.oss_helpers.oss2.Auth"), patch(
-            "video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket_instance
-        ) as mock_bucket_cls:
+        with (
+            patch.dict(
+                os.environ,
+                {
+                    "OSS_ACCESS_KEY_ID": "test-key",
+                    "OSS_ACCESS_KEY_SECRET": "test-secret",
+                    "OSS_ENDPOINT": "http://oss-cn-hangzhou.aliyuncs.com",
+                    "OSS_BUCKET_NAME": "test-bucket",
+                },
+            ),
+            patch("video_processing.oss_helpers.oss2.Auth"),
+            patch("video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket_instance) as mock_bucket_cls,
+        ):
             import video_processing.oss_helpers as oss_mod
 
             bucket = oss_bucket()
@@ -133,16 +136,18 @@ class TestGetSignedDownloadUrl:
         mock_bucket = MagicMock()
         mock_bucket.sign_url.return_value = "https://test-bucket.oss-cn-hangzhou.aliyuncs.com/generated/test.mp4?OSSAccessKeyId=xxx&Expires=xxx&Signature=xxx"
 
-        with patch.dict(
-            os.environ,
-            {
-                "OSS_ACCESS_KEY_ID": "test-key",
-                "OSS_ACCESS_KEY_SECRET": "test-secret",
-                "OSS_ENDPOINT": "oss-cn-hangzhou.aliyuncs.com",
-                "OSS_BUCKET_NAME": "test-bucket",
-            },
-        ), patch("video_processing.oss_helpers.oss2.Auth"), patch(
-            "video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket
+        with (
+            patch.dict(
+                os.environ,
+                {
+                    "OSS_ACCESS_KEY_ID": "test-key",
+                    "OSS_ACCESS_KEY_SECRET": "test-secret",
+                    "OSS_ENDPOINT": "oss-cn-hangzhou.aliyuncs.com",
+                    "OSS_BUCKET_NAME": "test-bucket",
+                },
+            ),
+            patch("video_processing.oss_helpers.oss2.Auth"),
+            patch("video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket),
         ):
             result = get_signed_download_url("generated/test.mp4", expires_seconds=3600)
 
@@ -155,22 +160,24 @@ class TestGetSignedDownloadUrl:
         from video_processing.oss_helpers import get_signed_download_url
 
         mock_bucket = MagicMock()
-        mock_bucket.sign_url.return_value = "https://test-bucket.oss-cn-hangzhou.aliyuncs.com/generated/test.mp4?sign=xxx"
+        mock_bucket.sign_url.return_value = (
+            "https://test-bucket.oss-cn-hangzhou.aliyuncs.com/generated/test.mp4?sign=xxx"
+        )
 
-        with patch.dict(
-            os.environ,
-            {
-                "OSS_ACCESS_KEY_ID": "test-key",
-                "OSS_ACCESS_KEY_SECRET": "test-secret",
-                "OSS_ENDPOINT": "oss-cn-hangzhou.aliyuncs.com",
-                "OSS_BUCKET_NAME": "test-bucket",
-            },
-        ), patch("video_processing.oss_helpers.oss2.Auth"), patch(
-            "video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket
+        with (
+            patch.dict(
+                os.environ,
+                {
+                    "OSS_ACCESS_KEY_ID": "test-key",
+                    "OSS_ACCESS_KEY_SECRET": "test-secret",
+                    "OSS_ENDPOINT": "oss-cn-hangzhou.aliyuncs.com",
+                    "OSS_BUCKET_NAME": "test-bucket",
+                },
+            ),
+            patch("video_processing.oss_helpers.oss2.Auth"),
+            patch("video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket),
         ):
-            result = get_signed_download_url(
-                "https://test-bucket.oss-cn-hangzhou.aliyuncs.com/generated/test.mp4"
-            )
+            result = get_signed_download_url("https://test-bucket.oss-cn-hangzhou.aliyuncs.com/generated/test.mp4")
 
             mock_bucket.sign_url.assert_called_once()
             # 验证传给 sign_url 的是纯 storage key，不是完整 URL
@@ -193,16 +200,18 @@ class TestGetSignedDownloadUrl:
         mock_bucket = MagicMock()
         mock_bucket.sign_url.side_effect = Exception("sign failed")
 
-        with patch.dict(
-            os.environ,
-            {
-                "OSS_ACCESS_KEY_ID": "test-key",
-                "OSS_ACCESS_KEY_SECRET": "test-secret",
-                "OSS_ENDPOINT": "oss-cn-hangzhou.aliyuncs.com",
-                "OSS_BUCKET_NAME": "test-bucket",
-            },
-        ), patch("video_processing.oss_helpers.oss2.Auth"), patch(
-            "video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket
+        with (
+            patch.dict(
+                os.environ,
+                {
+                    "OSS_ACCESS_KEY_ID": "test-key",
+                    "OSS_ACCESS_KEY_SECRET": "test-secret",
+                    "OSS_ENDPOINT": "oss-cn-hangzhou.aliyuncs.com",
+                    "OSS_BUCKET_NAME": "test-bucket",
+                },
+            ),
+            patch("video_processing.oss_helpers.oss2.Auth"),
+            patch("video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket),
         ):
             result = get_signed_download_url("generated/test.mp4")
             assert result is None
@@ -223,16 +232,18 @@ class TestUploadToOSSReturnsHTTPS:
 
         from pathlib import Path
 
-        with patch.dict(
-            os.environ,
-            {
-                "OSS_ACCESS_KEY_ID": "test-key",
-                "OSS_ACCESS_KEY_SECRET": "test-secret",
-                "OSS_ENDPOINT": "oss-cn-hangzhou.aliyuncs.com",
-                "OSS_BUCKET_NAME": "test-bucket",
-            },
-        ), patch("video_processing.oss_helpers.oss2.Auth"), patch(
-            "video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket
+        with (
+            patch.dict(
+                os.environ,
+                {
+                    "OSS_ACCESS_KEY_ID": "test-key",
+                    "OSS_ACCESS_KEY_SECRET": "test-secret",
+                    "OSS_ENDPOINT": "oss-cn-hangzhou.aliyuncs.com",
+                    "OSS_BUCKET_NAME": "test-bucket",
+                },
+            ),
+            patch("video_processing.oss_helpers.oss2.Auth"),
+            patch("video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket),
         ):
             result = upload_to_oss(Path("/tmp/test.mp4"), "generated/test.mp4")
 
@@ -249,16 +260,18 @@ class TestUploadToOSSReturnsHTTPS:
 
         from pathlib import Path
 
-        with patch.dict(
-            os.environ,
-            {
-                "OSS_ACCESS_KEY_ID": "test-key",
-                "OSS_ACCESS_KEY_SECRET": "test-secret",
-                "OSS_ENDPOINT": "https://oss-cn-hangzhou.aliyuncs.com",
-                "OSS_BUCKET_NAME": "test-bucket",
-            },
-        ), patch("video_processing.oss_helpers.oss2.Auth"), patch(
-            "video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket
+        with (
+            patch.dict(
+                os.environ,
+                {
+                    "OSS_ACCESS_KEY_ID": "test-key",
+                    "OSS_ACCESS_KEY_SECRET": "test-secret",
+                    "OSS_ENDPOINT": "https://oss-cn-hangzhou.aliyuncs.com",
+                    "OSS_BUCKET_NAME": "test-bucket",
+                },
+            ),
+            patch("video_processing.oss_helpers.oss2.Auth"),
+            patch("video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket),
         ):
             result = upload_to_oss(Path("/tmp/test.mp4"), "generated/test.mp4")
 

tests/unit/test_unified_render_service.py

@@ -435,8 +435,7 @@ class TestBuildFilterComplex:
         last_setpts = max(setpts_positions)
         first_fps = min(fps_positions)
         assert last_setpts < first_fps, (
-            f"单视频: setpts(position={last_setpts}) 应该在 fps(position={first_fps}) 之前。"
-            f"滤镜链: {chain_str}"
+            f"单视频: setpts(position={last_setpts}) 应该在 fps(position={first_fps}) 之前。" f"滤镜链: {chain_str}"
         )
 
     def test_empty_layers_raises(self):