style: auto-format with black + isort + ruff + prettier [skip ci-format-check]

apps/api/app/api/routes/tts.py

@@ -41,8 +41,6 @@ from sqlalchemy.orm import Session
 from packages.adapters.sqlalchemy_impl.tts_job_repository import (
     SQLAlchemyTTSJobRepository,
 )
-from packages.domain.points_rules import calculate_points_cost
-from packages.domain.points_service import PointsService
 from packages.application.cosyvoice_service import CosyVoiceError, CosyVoiceService
 from packages.application.tts_job.streaming_service import TTSStreamingService
 from packages.application.tts_job.use_cases import (
@@ -55,6 +53,8 @@ from packages.application.tts_job.use_cases import (
 )
 from packages.application.tts_job.workflow import TTSWorkflowService
 from packages.domain import Asset, AssetLibrary, AssetLibraryKind, AssetStatus, ClassificationStatus
+from packages.domain.points_rules import calculate_points_cost
+from packages.domain.points_service import PointsService
 from packages.domain.voice_presets import list_voices
 from packages.ports.asset_library_repository import AssetLibraryRepository
 from packages.ports.asset_repository import AssetRepository

apps/api/app/api/routes/voice_clones.py

@@ -25,6 +25,7 @@ from app.schemas.voice_clone import (
     VoiceCloneStatusResponse,
 )
 from fastapi import APIRouter, Depends, HTTPException, Query, Response, status
+from sqlalchemy.orm import Session
 
 from packages.adapters.sqlalchemy_impl.voice_clone_profile_repository import (
     SQLAlchemyVoiceCloneProfileRepository,
@@ -43,7 +44,6 @@ from packages.application.voice_clone.workflow import (
 )
 from packages.domain.points_rules import calculate_points_cost
 from packages.domain.points_service import PointsService
-from sqlalchemy.orm import Session
 
 # remove duplicate
 _DUMMY_DELETED = ()

packages/middleware/points_gate.py

@@ -91,13 +91,16 @@ def points_gate(
         is_async = asyncio.iscoroutinefunction(func)
 
         if is_async:
+
             async def wrapper(*args: Any, **kwargs: Any) -> Any:
                 if not _pg_enabled():  # noqa: F821
                     return await func(*args, **kwargs)
                 return await _pg_execute(  # noqa: F821
                     func, args, kwargs, scene_key, per_unit, unit_field, quantity_field, True
                 )
+
         else:
+
             def wrapper(*args: Any, **kwargs: Any) -> Any:
                 if not _pg_enabled():  # noqa: F821
                     return func(*args, **_pg_filter(func, kwargs))  # noqa: F821
@@ -126,8 +129,7 @@ def points_gate(
                 # 为稳妥起见，直接把 wrapper code 的 co_names 映射到新名——复杂度过高，
                 # 这里采用「确保短名没冲突」策略：如果冲突就抛异常让开发者改名。
                 raise RuntimeError(
-                    f"points_gate: name collision in {func.__module__}.{func.__name__}: "
-                    f"'{k}' already defined"
+                    f"points_gate: name collision in {func.__module__}.{func.__name__}: " f"'{k}' already defined"
                 )
             merged_globals[k] = v
 

tests/unit/test_tts_voice_clone_points.py

@@ -58,14 +58,15 @@ class TestEstimateMinutes:
 
 
 class TestTtsSynthesizePointsDeduction:
-    def _setup(self, text="你好", deduct_success=True, balance=0,
-               start_synth_raises=None, send_task_raises=None):
+    def _setup(self, text="你好", deduct_success=True, balance=0, start_synth_raises=None, send_task_raises=None):
         db = MagicMock()
         cu = _make_cu()
         repo = MagicMock()
         import enum
+
         class _S(enum.Enum):
             processing = "processing"
+
         job = SimpleNamespace(id="job-1", status=_S.processing, metadata={})
         uc = MagicMock()
         uc.execute.return_value = job
@@ -82,18 +83,23 @@ class TestTtsSynthesizePointsDeduction:
         return db, cu, repo, uc, wf, vc_repo, svc, fake_settings, job
 
     def test_insufficient_raises_402(self):
-        db, cu, repo, uc, wf, vc_repo, svc, fs, _ = self._setup(
-            text="你好" * 200, deduct_success=False, balance=0)
+        db, cu, repo, uc, wf, vc_repo, svc, fs, _ = self._setup(text="你好" * 200, deduct_success=False, balance=0)
         from app.api.routes.tts import synthesize
-        with patch("app.api.routes.tts.CreateTTSJobUseCase", return_value=uc), \
-             patch("app.api.routes.tts.TTSWorkflowService", return_value=wf), \
-             patch("app.api.routes.tts.PointsService", return_value=svc), \
-             patch("app.api.routes.tts.settings", fs):
+
+        with (
+            patch("app.api.routes.tts.CreateTTSJobUseCase", return_value=uc),
+            patch("app.api.routes.tts.TTSWorkflowService", return_value=wf),
+            patch("app.api.routes.tts.PointsService", return_value=svc),
+            patch("app.api.routes.tts.settings", fs),
+        ):
             with pytest.raises(HTTPException) as ei:
                 synthesize(
                     request=_make_request(text="你好" * 200),
-                    authenticated_user=cu, db=db, repository=repo,
-                    cosyvoice_service=MagicMock(), voice_clone_repo=vc_repo,
+                    authenticated_user=cu,
+                    db=db,
+                    repository=repo,
+                    cosyvoice_service=MagicMock(),
+                    voice_clone_repo=vc_repo,
                 )
             assert ei.value.status_code == 402
             assert ei.value.detail["code"] == "INSUFFICIENT_POINTS"
@@ -101,15 +107,21 @@ class TestTtsSynthesizePointsDeduction:
     def test_success_deducts_points(self):
         db, cu, repo, uc, wf, vc_repo, svc, fs, job = self._setup()
         from app.api.routes.tts import synthesize
-        with patch("app.api.routes.tts.CreateTTSJobUseCase", return_value=uc), \
-             patch("app.api.routes.tts.TTSWorkflowService", return_value=wf), \
-             patch("app.api.routes.tts.PointsService", return_value=svc), \
-             patch("app.api.routes.tts.celery_app.send_task") as _st, \
-             patch("app.api.routes.tts.settings", fs):
+
+        with (
+            patch("app.api.routes.tts.CreateTTSJobUseCase", return_value=uc),
+            patch("app.api.routes.tts.TTSWorkflowService", return_value=wf),
+            patch("app.api.routes.tts.PointsService", return_value=svc),
+            patch("app.api.routes.tts.celery_app.send_task") as _st,
+            patch("app.api.routes.tts.settings", fs),
+        ):
             resp = synthesize(
                 request=_make_request(text="测试"),
-                authenticated_user=cu, db=db, repository=repo,
-                cosyvoice_service=MagicMock(), voice_clone_repo=vc_repo,
+                authenticated_user=cu,
+                db=db,
+                repository=repo,
+                cosyvoice_service=MagicMock(),
+                voice_clone_repo=vc_repo,
             )
             svc.deduct_points.assert_called_once()
             assert resp.job_id == job.id
@@ -117,35 +129,48 @@ class TestTtsSynthesizePointsDeduction:
     def test_synthesis_failure_refunds(self):
         db, cu, repo, uc, wf, vc_repo, svc, fs, _ = self._setup(start_synth_raises=RuntimeError("boom"))
         from app.api.routes.tts import synthesize
-        with patch("app.api.routes.tts.CreateTTSJobUseCase", return_value=uc), \
-             patch("app.api.routes.tts.TTSWorkflowService", return_value=wf), \
-             patch("app.api.routes.tts.PointsService", return_value=svc), \
-             patch("app.api.routes.tts.celery_app.send_task"), \
-             patch("app.api.routes.tts.settings", fs):
+
+        with (
+            patch("app.api.routes.tts.CreateTTSJobUseCase", return_value=uc),
+            patch("app.api.routes.tts.TTSWorkflowService", return_value=wf),
+            patch("app.api.routes.tts.PointsService", return_value=svc),
+            patch("app.api.routes.tts.celery_app.send_task"),
+            patch("app.api.routes.tts.settings", fs),
+        ):
             synthesize(
                 request=_make_request(text="测试"),
-                authenticated_user=cu, db=db, repository=repo,
-                cosyvoice_service=MagicMock(), voice_clone_repo=vc_repo,
+                authenticated_user=cu,
+                db=db,
+                repository=repo,
+                cosyvoice_service=MagicMock(),
+                voice_clone_repo=vc_repo,
             )
             assert svc.refund_points.called
 
     def test_celery_send_failure_refunds(self):
         db, cu, repo, uc, wf, vc_repo, svc, fs, _ = self._setup(send_task_raises=RuntimeError("celery down"))
         from app.api.routes.tts import synthesize
-        with patch("app.api.routes.tts.CreateTTSJobUseCase", return_value=uc), \
-             patch("app.api.routes.tts.TTSWorkflowService", return_value=wf), \
-             patch("app.api.routes.tts.PointsService", return_value=svc), \
-             patch("app.api.routes.tts.celery_app.send_task", side_effect=RuntimeError("celery down")), \
-             patch("app.api.routes.tts.settings", fs):
+
+        with (
+            patch("app.api.routes.tts.CreateTTSJobUseCase", return_value=uc),
+            patch("app.api.routes.tts.TTSWorkflowService", return_value=wf),
+            patch("app.api.routes.tts.PointsService", return_value=svc),
+            patch("app.api.routes.tts.celery_app.send_task", side_effect=RuntimeError("celery down")),
+            patch("app.api.routes.tts.settings", fs),
+        ):
             synthesize(
                 request=_make_request(text="测试"),
-                authenticated_user=cu, db=db, repository=repo,
-                cosyvoice_service=MagicMock(), voice_clone_repo=vc_repo,
+                authenticated_user=cu,
+                db=db,
+                repository=repo,
+                cosyvoice_service=MagicMock(),
+                voice_clone_repo=vc_repo,
             )
             assert svc.refund_points.called
 
     def test_member_cheaper(self):
         from packages.domain.points_rules import calculate_points_cost
+
         cf = calculate_points_cost("ai_voice", is_member=False, duration_minutes=2)
         cm = calculate_points_cost("ai_voice", is_member=True, member_type="monthly", duration_minutes=2)
         assert cm < cf
@@ -172,29 +197,48 @@ class TestVoiceClonePreviewPoints:
     def test_insufficient_raises_402(self):
         db, cu, repo, uc, cosy, svc, fs = self._setup(deduct_success=False, balance=0)
         from app.api.routes.voice_clones import get_voice_clone_preview
-        with patch("app.api.routes.voice_clones.GetVoiceCloneUseCase", return_value=uc), \
-             patch("app.api.routes.voice_clones.PointsService", return_value=svc), \
-             patch("app.api.routes.voice_clones.settings", fs), \
-             patch("app.api.routes.voice_clones._clone_preview_cache", {}):
+
+        with (
+            patch("app.api.routes.voice_clones.GetVoiceCloneUseCase", return_value=uc),
+            patch("app.api.routes.voice_clones.PointsService", return_value=svc),
+            patch("app.api.routes.voice_clones.settings", fs),
+            patch("app.api.routes.voice_clones._clone_preview_cache", {}),
+        ):
             with pytest.raises(HTTPException) as ei:
                 get_voice_clone_preview(
-                    clone_id="c1", text="你好", speed=1.0, emotion="",
-                    authenticated_user=cu, db=db, repository=repo, cosyvoice=cosy,
+                    clone_id="c1",
+                    text="你好",
+                    speed=1.0,
+                    emotion="",
+                    authenticated_user=cu,
+                    db=db,
+                    repository=repo,
+                    cosyvoice=cosy,
                 )
             assert ei.value.status_code == 402
 
     def test_synth_cosyvoice_error_refunds_and_raises_502(self):
         from packages.application.cosyvoice_service import CosyVoiceError
+
         db, cu, repo, uc, cosy, svc, fs = self._setup(synth_raises=CosyVoiceError("fail"))
         from app.api.routes.voice_clones import get_voice_clone_preview
-        with patch("app.api.routes.voice_clones.GetVoiceCloneUseCase", return_value=uc), \
-             patch("app.api.routes.voice_clones.PointsService", return_value=svc), \
-             patch("app.api.routes.voice_clones.settings", fs), \
-             patch("app.api.routes.voice_clones._clone_preview_cache", {}):
+
+        with (
+            patch("app.api.routes.voice_clones.GetVoiceCloneUseCase", return_value=uc),
+            patch("app.api.routes.voice_clones.PointsService", return_value=svc),
+            patch("app.api.routes.voice_clones.settings", fs),
+            patch("app.api.routes.voice_clones._clone_preview_cache", {}),
+        ):
             with pytest.raises(HTTPException) as ei:
                 get_voice_clone_preview(
-                    clone_id="c1", text="你好", speed=1.0, emotion="",
-                    authenticated_user=cu, db=db, repository=repo, cosyvoice=cosy,
+                    clone_id="c1",
+                    text="你好",
+                    speed=1.0,
+                    emotion="",
+                    authenticated_user=cu,
+                    db=db,
+                    repository=repo,
+                    cosyvoice=cosy,
                 )
             assert ei.value.status_code == 502
             assert svc.refund_points.called
@@ -202,13 +246,22 @@ class TestVoiceClonePreviewPoints:
     def test_success_returns_audio(self):
         db, cu, repo, uc, cosy, svc, fs = self._setup()
         from app.api.routes.voice_clones import get_voice_clone_preview
-        with patch("app.api.routes.voice_clones.GetVoiceCloneUseCase", return_value=uc), \
-             patch("app.api.routes.voice_clones.PointsService", return_value=svc), \
-             patch("app.api.routes.voice_clones.settings", fs), \
-             patch("app.api.routes.voice_clones._clone_preview_cache", {}):
+
+        with (
+            patch("app.api.routes.voice_clones.GetVoiceCloneUseCase", return_value=uc),
+            patch("app.api.routes.voice_clones.PointsService", return_value=svc),
+            patch("app.api.routes.voice_clones.settings", fs),
+            patch("app.api.routes.voice_clones._clone_preview_cache", {}),
+        ):
             resp = get_voice_clone_preview(
-                clone_id="c1", text="你好", speed=1.0, emotion="",
-                authenticated_user=cu, db=db, repository=repo, cosyvoice=cosy,
+                clone_id="c1",
+                text="你好",
+                speed=1.0,
+                emotion="",
+                authenticated_user=cu,
+                db=db,
+                repository=repo,
+                cosyvoice=cosy,
             )
             svc.deduct_points.assert_called_once()
             assert resp.audio_url.startswith("http")