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

"""
X (Twitter) 平台实现

后端：crawler.aiddit.com/crawler/x
"""

import json
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.registry import PlatformDef, register_platform

CRAWLER_URL = "http://crawler.aiddit.com/crawler/x/keyword"
DEFAULT_TIMEOUT = 60.0


async def search(
    platform_id: str,
    keyword: str,
    max_count: int = 20,
    cursor: str = "",
    extras: Optional[Dict[str, Any]] = None,
) -> ToolResult:
    try:
        async with httpx.AsyncClient(timeout=DEFAULT_TIMEOUT) as client:
            response = await client.post(CRAWLER_URL, json={"keyword": keyword})
            response.raise_for_status()
            data = response.json()

        if data.get("code") != 0:
            return ToolResult(title="X 搜索失败", output="", error=data.get("msg", "未知错误"))

        result_data = data.get("data", {})
        tweets = result_data.get("data", []) if isinstance(result_data, dict) else []

        summary_list = []
        for idx, tweet in enumerate(tweets[:max_count], 1):
            text = tweet.get("body_text", "")
            summary_list.append({
                "index": idx,
                "author": tweet.get("channel_account_name", ""),
                "body_text": text[:100] + ("..." if len(text) > 100 else ""),
                "like_count": tweet.get("like_count"),
                "comment_count": tweet.get("comment_count"),
            })

        # 拼图
        images = []
        collage_b64 = await _build_tweet_collage(tweets[:max_count])
        if collage_b64:
            images.append({"type": "base64", "media_type": "image/png", "data": collage_b64})

        return ToolResult(
            title=f"X: {keyword}",
            output=json.dumps({"data": summary_list}, ensure_ascii=False, indent=2),
            long_term_memory=f"Searched X for '{keyword}', {len(tweets)} results.",
            images=images,
            metadata={"posts": tweets[:max_count]},
        )

    except Exception as e:
        return ToolResult(title="X 搜索异常", output="", error=str(e))


async def detail(post: Dict[str, Any], extras: Optional[Dict[str, Any]] = None) -> ToolResult:
    """X 的详情直接从缓存的搜索结果取完整数据"""
    author = post.get("channel_account_name", "")
    text = post.get("body_text", "")[:30]

    all_images = []
    for img_item in post.get("image_url_list", []):
        url = img_item.get("image_url") if isinstance(img_item, dict) else img_item
        if url:
            all_images.append({"type": "url", "url": url})

    return ToolResult(
        title=f"X 详情: @{author}",
        output=json.dumps(post, ensure_ascii=False, indent=2),
        long_term_memory=f"Viewed X post by @{author}: {text}",
        images=all_images,
    )


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)
            titles.append(f"@{tweet.get('channel_account_name', '')}")

    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)
    b64, _ = encode_base64(grid, format="PNG")
    return b64


# ── 注册 ──

_X = PlatformDef(
    id="x",
    name="X (Twitter)",
    aliases=["twitter", "推特"],
)
_X.search_impl = search
_X.detail_impl = detail
register_platform(_X)