重构(内容工具): 统一平台图片拼图管线

抽出图片加载、失败过滤、网格渲染、CDN 上传和 base64 降级组件；AIGC Channel、X、YouTube 继续保留各自选图与标题规则，并整理内容缓存和媒体处理异常边界。

agent/agent/tools/builtin/content/cache.py

@@ -7,11 +7,14 @@
 """
 
 import json
+import logging
 import os
 import time
 from pathlib import Path
 from typing import Any, Dict, List, Optional
 
+logger = logging.getLogger(__name__)
+
 # CWD 锚定 —— 每个调用项目有独立缓存目录，避免 /tmp 跨会话污染
 _CACHE_DIR = Path(os.getcwd()) / ".cache" / "content_search"
 _CACHE_DIR.mkdir(parents=True, exist_ok=True)
@@ -34,7 +37,8 @@ def _load_raw(trace_id: str) -> dict:
             p.unlink(missing_ok=True)
             return {}
         return data
-    except Exception:
+    except (OSError, json.JSONDecodeError, TypeError) as exc:
+        logger.debug("Failed to read content cache %s: %s", p, exc)
         return {}
 
 
@@ -44,8 +48,8 @@ def _save_raw(trace_id: str, data: dict) -> None:
         _cache_path(trace_id).write_text(
             json.dumps(data, ensure_ascii=False), encoding="utf-8"
         )
-    except Exception:
-        pass
+    except OSError as exc:
+        logger.warning("Failed to write content cache for trace %s: %s", trace_id, exc)
 
 
 def save_search_results(

agent/agent/tools/builtin/content/collage.py

@@ -0,0 +1,69 @@
+"""Shared image-collage rendering for content platform adapters."""
+
+from __future__ import annotations
+
+import hashlib
+import io
+import logging
+from typing import Any, Dict, List, Optional, Sequence
+
+from agent.tools.utils.image import build_image_grid, encode_base64, load_images
+
+
+async def _upload_bytes(image_bytes: bytes, filename: str) -> str:
+    from agent.tools.builtin.file.image_cdn import _upload_bytes_to_oss
+
+    return await _upload_bytes_to_oss(image_bytes, filename)
+
+
+async def render_image_collage(
+    urls: Sequence[str],
+    *,
+    labels: Optional[Sequence[str]] = None,
+    filename_prefix: str = "collage",
+    logger: Optional[logging.Logger] = None,
+) -> Optional[Dict[str, Any]]:
+    """Load, filter, render and publish a platform-selected image set."""
+
+    if not urls:
+        return None
+    if labels is not None and len(labels) != len(urls):
+        raise ValueError("labels must match urls length")
+
+    loaded = await load_images(list(urls))
+    valid_images: List[Any] = []
+    valid_labels: List[str] = []
+    for index, (_, image) in enumerate(loaded):
+        if image is None:
+            continue
+        valid_images.append(image)
+        if labels is not None:
+            valid_labels.append(labels[index])
+
+    if not valid_images:
+        return None
+
+    grid = build_image_grid(
+        images=valid_images,
+        labels=valid_labels if labels is not None else None,
+    )
+    buffer = io.BytesIO()
+    grid.save(buffer, format="PNG")
+    image_bytes = buffer.getvalue()
+
+    try:
+        digest = hashlib.md5(image_bytes).hexdigest()[:12]
+        url = await _upload_bytes(
+            image_bytes,
+            f"{filename_prefix}_{digest}.png",
+        )
+        return {"type": "url", "url": url}
+    except Exception as exc:
+        (logger or logging.getLogger(__name__)).warning(
+            "Failed to upload collage to CDN: %s", exc
+        )
+        encoded, _ = encode_base64(grid, format="PNG")
+        return {"type": "base64", "media_type": "image/png", "data": encoded}
+
+
+__all__ = ["render_image_collage"]

agent/agent/tools/builtin/content/media.py

@@ -7,6 +7,7 @@
 
 import asyncio
 import json
+import logging
 import subprocess
 import tempfile
 from pathlib import Path
@@ -14,6 +15,8 @@ from typing import Dict, List, Optional
 
 from agent.tools import tool, ToolResult
 
+logger = logging.getLogger(__name__)
+
 VIDEO_DOWNLOAD_DIR = Path(tempfile.gettempdir()) / "youtube_videos"
 VIDEO_DOWNLOAD_DIR.mkdir(exist_ok=True)
 
@@ -37,7 +40,8 @@ def download_youtube_video(video_id: str) -> Optional[str]:
         if result.returncode == 0 and output_path.exists():
             return str(output_path)
         return None
-    except Exception:
+    except (OSError, subprocess.SubprocessError) as exc:
+        logger.warning("YouTube video download failed for %s: %s", video_id, exc)
         return None
 
 

agent/agent/tools/builtin/content/platforms/aigc_channel.py

@@ -6,6 +6,7 @@ AIGC-Channel 平台实现（9 个中文平台）
 """
 
 import json
+import logging
 import re
 from typing import Any, Dict, List, Optional
 
@@ -13,11 +14,13 @@ import httpx
 
 from agent.tools.models import ToolResult
 from agent.tools.builtin.content.quality import get_post_evaluator
-from agent.tools.utils.image import build_image_grid, encode_base64, load_images
+from agent.tools.builtin.content.collage import render_image_collage
 from agent.tools.builtin.content.registry import (
     PlatformDef, ParamSpec, register_platform,
 )
 
+logger = logging.getLogger(__name__)
+
 BASE_URL = "http://aigc-channel.aiddit.com/aigc/channel"
 DEFAULT_TIMEOUT = 60.0
 
@@ -188,8 +191,8 @@ async def search(
                 }
                 post["_quality_score"] = eval_res["total_score"]
                 post["_quality_grade"] = eval_res["grade"]
-            except Exception:
-                pass
+            except Exception as exc:
+                logger.debug("AIGC post quality evaluation failed: %s", exc)
                 
         summary_item = {
             "index": idx,
@@ -230,34 +233,13 @@ KEEP_INDIVIDUAL = 8     # 单张图片保留数量；剩余图片合并为 1 张
 
 
 async def _build_images_collage(urls: List[str]) -> Optional[Dict[str, Any]]:
-    """将一组图片 URL 拼成单张网格图"""
-    if not urls:
-        return None
-
-    loaded = await load_images(urls)
-    valid_images = [img for (_, img) in loaded if img is not None]
-    if not valid_images:
-        return None
-
-    grid = build_image_grid(images=valid_images, labels=None)
-    import io
-    buf = io.BytesIO()
-    grid.save(buf, format="PNG")
-    img_bytes = buf.getvalue()
-
-    try:
-        from agent.tools.builtin.file.image_cdn import _upload_bytes_to_oss
-        import hashlib
+    """Render detail overflow images as one collage."""
 
-        md5_hash = hashlib.md5(img_bytes).hexdigest()[:12]
-        filename = f"collage_detail_{md5_hash}.png"
-        cdn_url = await _upload_bytes_to_oss(img_bytes, filename)
-        return {"type": "url", "url": cdn_url}
-    except Exception as e:
-        import logging
-        logging.getLogger(__name__).warning("Failed to upload detail collage to CDN: %s", e)
-        b64, _ = encode_base64(grid, format="PNG")
-        return {"type": "base64", "media_type": "image/png", "data": b64}
+    return await render_image_collage(
+        urls,
+        filename_prefix="collage_detail",
+        logger=logger,
+    )
 
 
 async def detail(
@@ -375,54 +357,23 @@ async def suggest(channel: str, keyword: str) -> ToolResult:
 # ── 拼图辅助 ──
 
 async def _build_collage(posts: List[Dict[str, Any]]) -> Optional[str]:
-    """封面图网格拼图"""
+    """Select post covers and render the search-result collage."""
+
     urls, titles = [], []
     for post in posts:
-        imgs = post.get("images", [])
-        if imgs and imgs[0]:
-            urls.append(imgs[0])
-            base_title = _strip_html(post.get("title", ""))
+        images = post.get("images", [])
+        if images and images[0]:
+            urls.append(images[0])
+            title = _strip_html(post.get("title", ""))
             score = post.get("_quality_score")
-            if score is not None:
-                title_with_score = f"[{score}分] {base_title}"
-            else:
-                title_with_score = base_title
-            titles.append(title_with_score)
-
-    if not urls:
-        return None
-
-    loaded = await load_images(urls)
-    valid_images, valid_labels = [], []
-    for (_, img), title in zip(loaded, titles):
-        if img is not None:
-            valid_images.append(img)
-            valid_labels.append(title)
-
-    if not valid_images:
-        return None
-
-    grid = build_image_grid(images=valid_images, labels=valid_labels)
-    import io
-    buf = io.BytesIO()
-    grid.save(buf, format="PNG")
-    img_bytes = buf.getvalue()
-    
-    # 尝试上传到 CDN，替换冗长的 base64
-    try:
-        from agent.tools.builtin.file.image_cdn import _upload_bytes_to_oss
-        import hashlib
-        
-        md5_hash = hashlib.md5(img_bytes).hexdigest()[:12]
-        filename = f"collage_search_{md5_hash}.png"
-        cdn_url = await _upload_bytes_to_oss(img_bytes, filename)
-        return {"type": "url", "url": cdn_url}
-    except Exception as e:
-        import logging
-        logging.getLogger(__name__).warning("Failed to upload collage to CDN: %s", e)
-        # 降级：还是用 base64 但可能会超长
-        b64, _ = encode_base64(grid, format="PNG")
-        return {"type": "base64", "media_type": "image/png", "data": b64}
+            titles.append(f"[{score}分] {title}" if score is not None else title)
+
+    return await render_image_collage(
+        urls,
+        labels=titles,
+        filename_prefix="collage_search",
+        logger=logger,
+    )
 
 
 # ── 注册所有 AIGC 平台 ──

agent/agent/tools/builtin/content/platforms/x.py

@@ -5,15 +5,18 @@ X (Twitter) 平台实现
 """
 
 import json
+import logging
 from typing import Any, Dict, List, Optional
 
 import httpx
 
 from agent.tools.models import ToolResult
-from agent.tools.utils.image import build_image_grid, encode_base64, load_images
+from agent.tools.builtin.content.collage import render_image_collage
 from agent.tools.builtin.content.registry import PlatformDef, register_platform
 from agent.tools.builtin.content.quality import get_post_evaluator
 
+logger = logging.getLogger(__name__)
+
 CRAWLER_URL = "http://crawler.aiddit.com/crawler/x/keyword"
 COMMENT_URL = "http://crawler.aiddit.com/crawler/x/comment"
 DEFAULT_TIMEOUT = 60.0
@@ -64,8 +67,8 @@ async def search(
                         "quality_grade": eval_res["grade"]
                     }
                     tweet["_quality_score"] = eval_res["total_score"]
-                except Exception:
-                    pass
+                except Exception as exc:
+                    logger.debug("X post quality evaluation failed: %s", exc)
 
             summary_item = {
                 "index": idx,
@@ -101,34 +104,13 @@ KEEP_INDIVIDUAL = 8
 
 
 async def _build_images_collage(urls: List[str]) -> Optional[Dict[str, Any]]:
-    """将一组图片 URL 拼成单张网格图"""
-    if not urls:
-        return None
-
-    loaded = await load_images(urls)
-    valid_images = [img for (_, img) in loaded if img is not None]
-    if not valid_images:
-        return None
+    """Render detail overflow images as one collage."""
 
-    grid = build_image_grid(images=valid_images, labels=None)
-    import io
-    buf = io.BytesIO()
-    grid.save(buf, format="PNG")
-    img_bytes = buf.getvalue()
-
-    try:
-        from agent.tools.builtin.file.image_cdn import _upload_bytes_to_oss
-        import hashlib
-
-        md5_hash = hashlib.md5(img_bytes).hexdigest()[:12]
-        filename = f"x_detail_collage_{md5_hash}.png"
-        cdn_url = await _upload_bytes_to_oss(img_bytes, filename)
-        return {"type": "url", "url": cdn_url}
-    except Exception as e:
-        import logging
-        logging.getLogger(__name__).warning("Failed to upload x detail collage to CDN: %s", e)
-        b64, _ = encode_base64(grid, format="PNG")
-        return {"type": "base64", "media_type": "image/png", "data": b64}
+    return await render_image_collage(
+        urls,
+        filename_prefix="x_detail_collage",
+        logger=logger,
+    )
 
 
 async def _fetch_author_comments(content_id: str, author_id: str) -> List[Dict[str, Any]]:
@@ -247,56 +229,26 @@ async def detail(post: Dict[str, Any], extras: Optional[Dict[str, Any]] = None)
 async def _build_tweet_collage(tweets: List[Dict[str, Any]]) -> Optional[str]:
     urls, titles = [], []
     for tweet in tweets:
-        thumb = None
-        for img_item in tweet.get("image_url_list", []):
-            url = img_item.get("image_url") if isinstance(img_item, dict) else img_item
-            if url:
-                thumb = url
-                break
-        if not thumb:
-            thumb = tweet.get("cover_url")
-        if thumb:
-            urls.append(thumb)
-            base_title = f"@{tweet.get('channel_account_name', '')}"
+        thumbnail = next(
+            (
+                item.get("image_url") if isinstance(item, dict) else item
+                for item in tweet.get("image_url_list", [])
+                if (item.get("image_url") if isinstance(item, dict) else item)
+            ),
+            None,
+        ) or tweet.get("cover_url")
+        if thumbnail:
+            urls.append(thumbnail)
+            title = f"@{tweet.get('channel_account_name', '')}"
             score = tweet.get("_quality_score")
-            if score is not None:
-                title_with_score = f"[{score}分] {base_title}"
-            else:
-                title_with_score = base_title
-            titles.append(title_with_score)
-
-    if not urls:
-        return None
-
-    loaded = await load_images(urls)
-    valid_images, valid_labels = [], []
-    for (_, img), title in zip(loaded, titles):
-        if img is not None:
-            valid_images.append(img)
-            valid_labels.append(title)
-
-    if not valid_images:
-        return None
-
-    grid = build_image_grid(images=valid_images, labels=valid_labels)
-    import io
-    buf = io.BytesIO()
-    grid.save(buf, format="PNG")
-    img_bytes = buf.getvalue()
-    
-    try:
-        from agent.tools.builtin.file.image_cdn import _upload_bytes_to_oss
-        import hashlib
-        
-        md5_hash = hashlib.md5(img_bytes).hexdigest()[:12]
-        filename = f"x_collage_{md5_hash}.png"
-        cdn_url = await _upload_bytes_to_oss(img_bytes, filename)
-        return {"type": "url", "url": cdn_url}
-    except Exception as e:
-        import logging
-        logging.getLogger(__name__).warning("Failed to upload x collage to CDN: %s", e)
-        b64, _ = encode_base64(grid, format="PNG")
-        return {"type": "base64", "media_type": "image/png", "data": b64}
+            titles.append(f"[{score}分] {title}" if score is not None else title)
+
+    return await render_image_collage(
+        urls,
+        labels=titles,
+        filename_prefix="x_collage",
+        logger=logger,
+    )
 
 
 # ── 注册 ──

agent/agent/tools/builtin/content/platforms/youtube.py

@@ -5,6 +5,7 @@ YouTube 平台实现
 """
 
 import json
+import logging
 import re
 import time
 from typing import Any, Dict, List, Optional
@@ -13,11 +14,13 @@ import httpx
 
 from agent.tools.models import ToolResult
 from agent.tools.builtin.content.quality import get_post_evaluator
-from agent.tools.utils.image import build_image_grid, encode_base64, load_images
+from agent.tools.builtin.content.collage import render_image_collage
 from agent.tools.builtin.content.registry import (
     PlatformDef, ParamSpec, register_platform,
 )
 
+logger = logging.getLogger(__name__)
+
 CRAWLER_BASE_URL = "http://crawler.aiddit.com/crawler"
 DEFAULT_TIMEOUT = 60.0
 
@@ -198,8 +201,8 @@ async def search(
                     }
                     video["_quality_score"] = eval_res["total_score"]
                     video["_quality_grade"] = eval_res["grade"]
-                except Exception:
-                    pass
+                except Exception as exc:
+                    logger.debug("YouTube quality evaluation failed: %s", exc)
             
             summary_item = {
                 "index": idx,
@@ -282,8 +285,8 @@ async def detail(post: Dict[str, Any], extras: Optional[Dict[str, Any]] = None)
                         inner2 = inner.get("data", {})
                         if isinstance(inner2, dict):
                             captions_text = inner2.get("content")
-        except Exception:
-            pass
+        except Exception as exc:
+            logger.debug("YouTube caption retrieval failed: %s", exc)
 
     # ── 3) 视频文件下载（用户显式 extras.download_video=True 时才跑） ──
     video_path = None
@@ -375,7 +378,7 @@ async def detail(post: Dict[str, Any], extras: Optional[Dict[str, Any]] = None)
     if transcript_text and transcript_text != captions_text:
         memory_parts.append("transcript")
     if detail_error:
-        memory_parts.append(f"degraded(detail backend down)")
+        memory_parts.append("degraded(detail backend down)")
     memory_extra = f" with {'+'.join(memory_parts)}" if memory_parts else ""
 
     title = video_info.get("title") or post.get("title") or content_id
@@ -391,56 +394,22 @@ async def detail(post: Dict[str, Any], extras: Optional[Dict[str, Any]] = None)
 async def _build_video_collage(videos: List[Dict[str, Any]]) -> Optional[str]:
     urls, titles = [], []
     for video in videos:
-        thumb = None
-        if "thumbnails" in video and isinstance(video["thumbnails"], list) and video["thumbnails"]:
-            thumb = video["thumbnails"][0].get("url")
-        elif "thumbnail" in video:
-            thumb = video.get("thumbnail")
-        elif "cover_url" in video:
-            thumb = video.get("cover_url")
-
-        if thumb:
-            urls.append(thumb)
-            base_title = video.get("title", "")
+        thumbnail = None
+        if isinstance(video.get("thumbnails"), list) and video["thumbnails"]:
+            thumbnail = video["thumbnails"][0].get("url")
+        thumbnail = thumbnail or video.get("thumbnail") or video.get("cover_url")
+        if thumbnail:
+            urls.append(thumbnail)
+            title = video.get("title", "")
             score = video.get("_quality_score")
-            if score is not None:
-                title_with_score = f"[{score}分] {base_title}"
-            else:
-                title_with_score = base_title
-            titles.append(title_with_score)
-
-    if not urls:
-        return None
+            titles.append(f"[{score}分] {title}" if score is not None else title)
 
-    loaded = await load_images(urls)
-    valid_images, valid_labels = [], []
-    for (_, img), title in zip(loaded, titles):
-        if img is not None:
-            valid_images.append(img)
-            valid_labels.append(title)
-
-    if not valid_images:
-        return None
-
-    grid = build_image_grid(images=valid_images, labels=valid_labels)
-    import io
-    buf = io.BytesIO()
-    grid.save(buf, format="PNG")
-    img_bytes = buf.getvalue()
-    
-    try:
-        from agent.tools.builtin.file.image_cdn import _upload_bytes_to_oss
-        import hashlib
-        
-        md5_hash = hashlib.md5(img_bytes).hexdigest()[:12]
-        filename = f"youtube_collage_{md5_hash}.png"
-        cdn_url = await _upload_bytes_to_oss(img_bytes, filename)
-        return {"type": "url", "url": cdn_url}
-    except Exception as e:
-        import logging
-        logging.getLogger(__name__).warning("Failed to upload youtube collage to CDN: %s", e)
-        b64, _ = encode_base64(grid, format="PNG")
-        return {"type": "base64", "media_type": "image/png", "data": b64}
+    return await render_image_collage(
+        urls,
+        labels=titles,
+        filename_prefix="youtube_collage",
+        logger=logger,
+    )
 
 
 # ── 注册 ──

agent/agent/tools/builtin/content/tools.py

@@ -11,6 +11,7 @@
 """
 
 import json
+import logging
 import os
 import uuid
 from typing import Any, Dict, Optional
@@ -26,6 +27,8 @@ import agent.tools.builtin.content.platforms.aigc_channel  # noqa: F401
 import agent.tools.builtin.content.platforms.youtube       # noqa: F401
 import agent.tools.builtin.content.platforms.x             # noqa: F401
 
+logger = logging.getLogger(__name__)
+
 
 def _get_trace_id(context: Optional[Dict[str, Any]]) -> str:
     """从 context 取 trace_id，回退到环境变量或自动生成"""
@@ -50,8 +53,8 @@ def _convert_post_timestamps(post: dict) -> None:
             if ts > 1000000000000:
                 ts = ts / 1000
             post[field] = datetime.fromtimestamp(ts).strftime("%Y-%m-%d %H:%M:%S")
-        except Exception:
-            pass
+        except (TypeError, ValueError, OSError, OverflowError) as exc:
+            logger.debug("Ignored invalid post timestamp %r: %s", ts, exc)
 
 
 # ── content_platforms ──
@@ -91,7 +94,7 @@ async def content_platforms(
         result = [p.summary() for p in hits]
 
     return ToolResult(
-        title=f"内容平台" + (f" ({platform})" if platform else ""),
+        title="内容平台" + (f" ({platform})" if platform else ""),
         output=json.dumps(result, ensure_ascii=False, indent=2),
     )
 

agent/agent/tools/utils/image.py

@@ -14,13 +14,15 @@
 import asyncio
 import base64
 import io
+import logging
 import math
-from pathlib import Path
 from typing import List, Optional, Sequence, Tuple
 
 import httpx
 from PIL import Image, ImageDraw, ImageFont
 
+logger = logging.getLogger(__name__)
+
 
 # ── 网格拼图默认参数 ──
 DEFAULT_THUMB_SIZE = 250         # 每格缩略图边长
@@ -58,7 +60,7 @@ def _load_fonts(title_size: int = 16, index_size: int = 32):
                 ImageFont.truetype(path, title_size),
                 ImageFont.truetype(path, index_size),
             )
-        except Exception:
+        except OSError:
             continue
     default = ImageFont.load_default()
     return default, default
@@ -72,7 +74,8 @@ async def _load_image_from_url(client: httpx.AsyncClient, url: str) -> Optional[
         resp = await client.get(url, timeout=15.0)
         resp.raise_for_status()
         return Image.open(io.BytesIO(resp.content)).convert("RGB")
-    except Exception:
+    except (httpx.HTTPError, OSError, ValueError) as exc:
+        logger.debug("Failed to load image URL %s: %s", url, exc)
         return None
 
 
@@ -80,7 +83,8 @@ def _load_image_from_path(path: str) -> Optional[Image.Image]:
     """从本地路径加载图片，失败返回 None"""
     try:
         return Image.open(path).convert("RGB")
-    except Exception:
+    except (OSError, ValueError) as exc:
+        logger.debug("Failed to load image path %s: %s", path, exc)
         return None
 
 

agent/tests/test_content_collage.py

@@ -0,0 +1,107 @@
+from PIL import Image
+import pytest
+
+from agent.tools.builtin.content import collage
+from agent.tools.builtin.content.platforms import aigc_channel, x, youtube
+
+
+@pytest.mark.asyncio
+async def test_collage_filters_failed_images_and_keeps_matching_labels(monkeypatch):
+    image = Image.new("RGB", (4, 4), "white")
+    captured = {}
+
+    async def fake_load(urls):
+        return [(urls[0], image), (urls[1], None)]
+
+    def fake_grid(*, images, labels):
+        captured["images"] = images
+        captured["labels"] = labels
+        return image
+
+    async def fake_upload(image_bytes, filename):
+        captured["filename"] = filename
+        return "https://cdn.example/collage.png"
+
+    monkeypatch.setattr(collage, "load_images", fake_load)
+    monkeypatch.setattr(collage, "build_image_grid", fake_grid)
+    monkeypatch.setattr(collage, "_upload_bytes", fake_upload)
+
+    result = await collage.render_image_collage(
+        ["one", "two"], labels=["first", "second"], filename_prefix="test"
+    )
+
+    assert result == {"type": "url", "url": "https://cdn.example/collage.png"}
+    assert captured["images"] == [image]
+    assert captured["labels"] == ["first"]
+    assert captured["filename"].startswith("test_")
+
+
+@pytest.mark.asyncio
+async def test_collage_falls_back_to_base64_when_upload_fails(monkeypatch):
+    image = Image.new("RGB", (2, 2), "white")
+
+    async def fake_load(urls):
+        return [(urls[0], image)]
+
+    async def failed_upload(image_bytes, filename):
+        raise RuntimeError("offline")
+
+    monkeypatch.setattr(collage, "load_images", fake_load)
+    monkeypatch.setattr(collage, "build_image_grid", lambda **kwargs: image)
+    monkeypatch.setattr(collage, "_upload_bytes", failed_upload)
+
+    result = await collage.render_image_collage(["one"])
+
+    assert result["type"] == "base64"
+    assert result["media_type"] == "image/png"
+    assert result["data"]
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ("adapter", "args", "expected_prefix"),
+    [
+        (aigc_channel._build_images_collage, (["u"],), "collage_detail"),
+        (x._build_images_collage, (["u"],), "x_detail_collage"),
+    ],
+)
+async def test_detail_adapters_delegate_to_shared_collage(
+    monkeypatch, adapter, args, expected_prefix
+):
+    module = __import__(adapter.__module__, fromlist=["render_image_collage"])
+    captured = {}
+
+    async def fake_render(urls, **kwargs):
+        captured.update(kwargs)
+        return {"type": "url", "url": "ok"}
+
+    monkeypatch.setattr(module, "render_image_collage", fake_render)
+
+    assert await adapter(*args) == {"type": "url", "url": "ok"}
+    assert captured["filename_prefix"] == expected_prefix
+
+
+@pytest.mark.asyncio
+async def test_search_adapters_keep_platform_selection_rules(monkeypatch):
+    calls = []
+
+    async def fake_render(urls, **kwargs):
+        calls.append((urls, kwargs))
+        return {"type": "url", "url": "ok"}
+
+    monkeypatch.setattr(x, "render_image_collage", fake_render)
+    monkeypatch.setattr(youtube, "render_image_collage", fake_render)
+
+    await x._build_tweet_collage(
+        [{"image_url_list": [{"image_url": "x.jpg"}], "channel_account_name": "a"}]
+    )
+    await youtube._build_video_collage(
+        [{"thumbnails": [{"url": "yt.jpg"}], "title": "video"}]
+    )
+
+    assert calls[0][0] == ["x.jpg"]
+    assert calls[0][1]["labels"] == ["@a"]
+    assert calls[0][1]["filename_prefix"] == "x_collage"
+    assert calls[1][0] == ["yt.jpg"]
+    assert calls[1][1]["labels"] == ["video"]
+    assert calls[1][1]["filename_prefix"] == "youtube_collage"