From 9e97473eecc8bfb3e905948facbeee6f9dc8f59c Mon Sep 17 00:00:00 2001
From: CI Bot <ci-bot@xiaoxiajianji.com>
Date: Mon, 17 Aug 2026 16:28:49 +0800
Subject: [PATCH] fix: batch query assets to resolve N+1 in
 _build_asset_url_map

- Add find_by_ids() to SQLAlchemyAssetRepository (single SQL IN query)
- Replace per-id find_by_id loop with single batch call
- Fixes AI Code Review blocking performance issue in PR #1404
---
 apps/api/app/api/routes/templates_editor/clips.py     | 10 +++++++++-
 packages/adapters/sqlalchemy_impl/asset_repository.py |  7 +++++++
 2 files changed, 16 insertions(+), 1 deletion(-)

diff --git a/apps/api/app/api/routes/templates_editor/clips.py b/apps/api/app/api/routes/templates_editor/clips.py
index 7c6b5e77a..5bbdc7c73 100755
--- a/apps/api/app/api/routes/templates_editor/clips.py
+++ b/apps/api/app/api/routes/templates_editor/clips.py
@@ -99,11 +99,19 @@ def _build_asset_url_map(
         logger.warning("获取存储服务失败，跳过asset_url生成")
         return {aid: None for aid in asset_ids}
 
+    # 批量查询所有 Asset（单次 SQL IN 查询，避免 N+1）
+    try:
+        assets = asset_repo.find_by_ids(asset_ids)
+        asset_map = {a.id: a for a in assets}
+    except Exception:
+        logger.warning("批量查询素材失败: asset_ids=%s", asset_ids, exc_info=True)
+        return {aid: None for aid in asset_ids if aid}
+
     for aid in asset_ids:
         if not aid:
             continue
         try:
-            asset = asset_repo.find_by_id(aid)
+            asset = asset_map.get(aid)
             if asset is None:
                 result[aid] = None
                 continue
diff --git a/packages/adapters/sqlalchemy_impl/asset_repository.py b/packages/adapters/sqlalchemy_impl/asset_repository.py
index 0c4077c14..b0c5d03b9 100755
--- a/packages/adapters/sqlalchemy_impl/asset_repository.py
+++ b/packages/adapters/sqlalchemy_impl/asset_repository.py
@@ -101,6 +101,13 @@ class SQLAlchemyAssetRepository:
             return None
         return self._to_domain(model)
 
+    def find_by_ids(self, asset_ids: list[str]) -> list[Asset]:
+        """批量查询素材（单次 SQL IN 查询，避免 N+1）。"""
+        if not asset_ids:
+            return []
+        models = self.session.query(AssetModel).filter(AssetModel.id.in_(asset_ids)).all()
+        return [self._to_domain(m) for m in models]
+
     def get(self, asset_id: str) -> Asset | None:
         return self.find_by_id(asset_id)