From 7adcf28b76c679e6145fc23c620daa92f3658073 Mon Sep 17 00:00:00 2001 From: CI Bot Date: Wed, 9 Sep 2026 12:11:22 +0000 Subject: [PATCH] style: auto-format with black + isort + prettier [skip ci-format-check] --- .../app/services/ai_avatar_render_service.py | 4 +-- apps/api/app/services/lipsync_service.py | 11 +++----- .../test_ai_avatar_emotion_tts_lipsync.py | 27 ++++++++++++++----- 3 files changed, 24 insertions(+), 18 deletions(-) diff --git a/apps/api/app/services/ai_avatar_render_service.py b/apps/api/app/services/ai_avatar_render_service.py index 2baa5cac2..21266333b 100644 --- a/apps/api/app/services/ai_avatar_render_service.py +++ b/apps/api/app/services/ai_avatar_render_service.py @@ -299,9 +299,7 @@ class AiAvatarRenderService: generate_smart_cover, ) - smart_cover_url = generate_smart_cover( - output_video_url, job_id=job_id, max_frames=5 - ) + smart_cover_url = generate_smart_cover(output_video_url, job_id=job_id, max_frames=5) except Exception: logger.warning("智能封面(MediaKit)失败，回退 FFmpeg 封面 job_id=%s", job_id, exc_info=True) diff --git a/apps/api/app/services/lipsync_service.py b/apps/api/app/services/lipsync_service.py index 3b8a02b5d..f8db385aa 100644 --- a/apps/api/app/services/lipsync_service.py +++ b/apps/api/app/services/lipsync_service.py @@ -9,13 +9,12 @@ from __future__ import annotations +import io import logging import uuid from datetime import datetime, timezone from typing import Optional -import io - from app.services.mediakit_client import ( STATUS_COMPLETED, STATUS_FAILED, @@ -127,9 +126,7 @@ class LipsyncService: ) storage = get_shared_storage_service() storage_key = f"lipsync-tts/{user_id}/{job_id}.mp3" - permanent_url = storage.upload_file( - io.BytesIO(audio_data), storage_key, content_type="audio/mpeg" - ) + permanent_url = storage.upload_file(io.BytesIO(audio_data), storage_key, content_type="audio/mpeg") logger.info("对口型 TTS 音频已转存 OSS: job_id=%s key=%s", job_id, storage_key) return permanent_url except Exception as exc: @@ -316,9 +313,7 @@ class LipsyncService: data = resp.content storage = get_shared_storage_service() storage_key = f"lipsync-outputs/{user_id}/{job_id}.mp4" - permanent_url = storage.upload_file( - io.BytesIO(data), storage_key, content_type="video/mp4" - ) + permanent_url = storage.upload_file(io.BytesIO(data), storage_key, content_type="video/mp4") logger.info("对口型输出视频已转存 OSS: job_id=%s key=%s", job_id, storage_key) return permanent_url or temp_url except Exception as exc: diff --git a/tests/unit/test_ai_avatar_emotion_tts_lipsync.py b/tests/unit/test_ai_avatar_emotion_tts_lipsync.py index f81041d5c..d7f426241 100644 --- a/tests/unit/test_ai_avatar_emotion_tts_lipsync.py +++ b/tests/unit/test_ai_avatar_emotion_tts_lipsync.py @@ -16,6 +16,7 @@ os.environ.setdefault("JWT_SECRET_KEY", "dev-secret-key-for-testing") # ── 情绪归一化 ────────────────────────────────────────────────────────── + def test_normalize_emotion_english_values(): from packages.application.cosyvoice_service import normalize_emotion @@ -46,6 +47,7 @@ def test_normalize_emotion_invalid_returns_empty(): # ── CosyVoice payload 携带 emotion + rate ────────────────────────────── + def _make_service_with_captured_client(captured: dict): """构造 CosyVoiceService，拦截 post 请求体到 captured['json'].""" import httpx as _httpx @@ -103,6 +105,7 @@ def test_submit_synthesize_payload_omits_emotion_when_empty(): # ── 对口型 TTS 直生分支 ───────────────────────────────────────────────── + def _lipsync_service_with_mocks(): from app.services.lipsync_service import LipsyncService @@ -129,8 +132,10 @@ def _lipsync_service_with_mocks(): def test_create_job_tts_direct_mode_synthesizes_audio(): svc, client, cosy = _lipsync_service_with_mocks() - with patch("app.services.lipsync_service.get_shared_storage_service") as storage_patch, \ - patch("app.services.lipsync_service.safe_download_bytes") as dl_patch: + with ( + patch("app.services.lipsync_service.get_shared_storage_service") as storage_patch, + patch("app.services.lipsync_service.safe_download_bytes") as dl_patch, + ): storage = MagicMock() storage.upload_file.return_value = "https://oss/tts.mp3" storage_patch.return_value = storage @@ -176,6 +181,7 @@ def test_create_job_direct_audio_mode_skips_tts(): def test_create_job_tts_failure_raises(): from app.services.mediakit_client import MediaKitError + from packages.application.cosyvoice_service import CosyVoiceError svc, client, cosy = _lipsync_service_with_mocks() @@ -195,6 +201,7 @@ def test_create_job_tts_failure_raises(): # ── refresh 同步中间状态 ──────────────────────────────────────────────── + def test_refresh_syncs_running_status(): from app.services.lipsync_service import LipsyncService @@ -215,6 +222,7 @@ def test_refresh_syncs_running_status(): # ── 智能封面 ──────────────────────────────────────────────────────────── + def test_smart_cover_selects_best_frame_and_persists(): from app.services import ai_avatar_cover_service as cov @@ -222,10 +230,12 @@ def test_smart_cover_selects_best_frame_and_persists(): {"image_url": "https://mk/f0.jpg"}, {"image_url": "https://mk/f1.jpg"}, ] - with patch("packages.shared.mediakit_client.get_mediakit_client") as mk_patch, \ - patch("packages.shared.cover_frame_scorer.score_frames") as score_patch, \ - patch("httpx.get") as http_get, \ - patch("packages.shared.storage.get_shared_storage_service") as storage_patch: + with ( + patch("packages.shared.mediakit_client.get_mediakit_client") as mk_patch, + patch("packages.shared.cover_frame_scorer.score_frames") as score_patch, + patch("httpx.get") as http_get, + patch("packages.shared.storage.get_shared_storage_service") as storage_patch, + ): mk = MagicMock() mk.is_available = True mk.extract_frames.return_value = snapshots @@ -263,8 +273,11 @@ def test_smart_cover_returns_empty_when_mediakit_unavailable(): # ── 渲染 script_id 可选（手动文案直生场景）────────────────────────────── + def test_render_request_script_id_optional(): - import sys, os + import os + import sys + sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "apps", "api")) from app.schemas.ai_avatar_render import CreateAiAvatarRenderRequest