fix(tts): AI Review 第三轮——约束类型一致 + OSS 孤儿文件清理 + import 规范

1. migration 058 改用 op.create_unique_constraint（与 model 的 UniqueConstraint
   定义一致，pg_constraint+pg_index 同时注册），避免 autogenerate schema drift。
2. save-to-library：asset_repository.create 失败时删除已上传的 OSS 文件，
   避免孤儿文件；补对应用例。
3. tempfile/Path import 移到模块顶部（PEP8）。

alembic/versions/058_uq_asset_lib_project_kind.py

@@ -36,13 +36,14 @@ def upgrade() -> None:
             WHERE t.rn > 1
         )
         """)
-    op.create_index(
+    # 与 model 的 UniqueConstraint 定义保持一致（pg_constraint + pg_index 同时注册），
+    # 避免 Alembic autogenerate 检测到 schema drift
+    op.create_unique_constraint(
         "uq_asset_libraries_project_kind",
         "asset_libraries",
         ["project_id", "kind"],
-        unique=True,
     )
 
 
 def downgrade() -> None:
-    op.drop_index("uq_asset_libraries_project_kind", table_name="asset_libraries")
+    op.drop_constraint("uq_asset_libraries_project_kind", "asset_libraries", type_="unique")

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

@@ -3,6 +3,8 @@
 from __future__ import annotations
 
 import logging
+import tempfile
+from pathlib import Path
 from typing import Any, Optional
 
 from app.auth import AuthenticatedUser, get_current_user
@@ -417,9 +419,6 @@ def save_tts_job_to_library(
 
     # 转存音频到素材 OSS 路径（tts-outputs/ 下的产物归 TTS 任务所有，
     # 素材独立持有副本，删除 TTS 任务不影响配音库素材）
-    import tempfile
-    from pathlib import Path
-
     audio_format = (job.format or "mp3").strip() or "mp3"
     content_type_map = {
         "mp3": "audio/mpeg",
@@ -488,7 +487,19 @@ def save_tts_job_to_library(
         classification_status=ClassificationStatus.PENDING,  # 音频不参与内容分类，保持 pending 与 ingest 链路一致
         uploaded_by_user_id=user_id,
     )
-    asset = asset_repository.create(asset)
+    try:
+        asset = asset_repository.create(asset)
+    except Exception as e:
+        # DB 写入失败：清理已上传到 OSS 的素材文件，避免产生无法索引的孤儿文件
+        logger.error("素材记录创建失败，清理 OSS 文件: %s, error=%s", storage_key, e, exc_info=True)
+        try:
+            storage_service.delete_file(storage_key)
+        except Exception:
+            logger.warning("清理孤儿 OSS 文件失败: %s", storage_key, exc_info=True)
+        raise HTTPException(
+            status_code=status.HTTP_502_BAD_GATEWAY,
+            detail="素材保存失败，请重试",
+        ) from e
 
     return SaveToLibraryResponse(
         id=asset.id,

tests/unit/test_tts_save_to_library_assets.py

@@ -129,6 +129,10 @@ class FakeStorage:
         self.uploads.append((storage_key, content_type))
         return storage_key
 
+    def delete_file(self, storage_key):
+        self.deleted = getattr(self, "deleted", [])
+        self.deleted.append(storage_key)
+
     def get_download_url(self, key, expires_seconds=3600):
         return f"https://oss.example.com/signed/{key}?sig=xxx"
 
@@ -267,6 +271,29 @@ class TestSaveToLibraryAssets:
         assert resp.status_code == 201, resp.text
         assert len(asset_repo.created) == 1
 
+    def test_save_db_failure_cleans_orphan_oss_file(self):
+        """asset_repository.create 抛异常 → 已上传的 OSS 文件被删除，返回 502。"""
+        from fastapi.testclient import TestClient
+
+        class FailAssetRepo(FakeAssetRepo):
+            def create(self, asset):
+                raise RuntimeError("DB connection lost")
+
+        voice_lib = AssetLibrary.create(project_id="proj-1", name="配音素材库", kind=AssetLibraryKind.VOICE)
+        storage = FakeStorage()
+        app, _, _ = _build_app(job=_completed_job(), libs=[voice_lib], storage=storage)
+        # 替换 asset repo 为会失败的实现
+        from app.dependencies import get_asset_repository
+
+        app.dependency_overrides[get_asset_repository] = lambda: FailAssetRepo()
+        client = TestClient(app)
+
+        resp = client.post("/tts/jobs/job-save-001/save-to-library", json={})
+        assert resp.status_code == 502, resp.text
+        # OSS 文件已上传后又被清理
+        assert storage.uploads, "音频应已上传"
+        assert hasattr(storage, "deleted") and storage.deleted, "失败后应删除孤儿 OSS 文件"
+
     def test_save_no_project_returns_400(self):
         """用户没有任何可访问项目 → 400。"""
         from fastapi.testclient import TestClient