""" AIGC-Channel 平台实现（9 个中文平台） 后端：aigc-channel.aiddit.com 平台：xhs / gzh / sph / github / toutiao / douyin / bili / zhihu / weibo """ 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, ParamSpec, register_platform, ) BASE_URL = "http://aigc-channel.aiddit.com/aigc/channel" DEFAULT_TIMEOUT = 60.0 # ── 平台注册 ── _XHS_SEARCH_PARAMS = { "sort_type": ParamSpec( values=["综合排序", "最新发布", "最多点赞"], default="综合排序", ), "publish_time": ParamSpec( values=["不限", "近1天", "近7天", "近30天"], default="不限", ), "content_type": ParamSpec( values=["不限", "图文", "视频", "文章"], default="不限", ), "filter_note_range": ParamSpec( values=["不限", "1分钟以内", "1-5分钟", "5分钟以上"], default="不限", note="仅视频内容生效", ), } _COMMON_CONTENT_TYPE = { "content_type": ParamSpec( values=["视频", "图文"], default="", note="留空不限", ), } # 9 个中文平台定义 _AIGC_PLATFORMS = [ PlatformDef(id="xhs", name="小红书", aliases=["RED", "xiaohongshu"], search_params=_XHS_SEARCH_PARAMS, supports_suggest=True), PlatformDef(id="gzh", name="公众号", aliases=["微信公众号", "wechat"], search_params=_COMMON_CONTENT_TYPE), PlatformDef(id="sph", name="视频号", aliases=["微信视频号"], search_params=_COMMON_CONTENT_TYPE), PlatformDef(id="github", name="GitHub", aliases=["gh"], search_params=_COMMON_CONTENT_TYPE), PlatformDef(id="toutiao", name="头条", aliases=["今日头条", "toutiao"], search_params=_COMMON_CONTENT_TYPE, supports_suggest=True), PlatformDef(id="douyin", name="抖音", aliases=["TikTok"], search_params=_COMMON_CONTENT_TYPE, supports_suggest=True), PlatformDef(id="bili", name="B站", aliases=["哔哩哔哩", "bilibili"], search_params=_COMMON_CONTENT_TYPE, supports_suggest=True), PlatformDef(id="zhihu", name="知乎", aliases=[], search_params=_COMMON_CONTENT_TYPE, supports_suggest=True), PlatformDef(id="weibo", name="微博", aliases=["sina"], search_params=_COMMON_CONTENT_TYPE), ] # suggest API 额外支持 wx（微信搜一搜），但它不是搜索平台 _SUGGEST_ONLY_CHANNELS = {"wx": "微信"} # ── 搜索实现 ── async def search( platform_id: str, keyword: str, max_count: int = 20, cursor: str = "", extras: Optional[Dict[str, Any]] = None, ) -> ToolResult: """AIGC-Channel 统一搜索""" extras = extras or {} if platform_id == "xhs": payload = { "type": platform_id, "keyword": keyword, "cursor": cursor, "content_type": extras.get("content_type", "不限"), "sort_type": extras.get("sort_type", "综合排序"), "publish_time": extras.get("publish_time", "不限"), "filter_note_range": extras.get("filter_note_range", "不限"), } else: payload = { "type": platform_id, "keyword": keyword, "cursor": cursor or "0", "max_count": max_count, "content_type": extras.get("content_type", ""), } try: async with httpx.AsyncClient(timeout=DEFAULT_TIMEOUT) as client: response = await client.post( f"{BASE_URL}/data", json=payload, headers={"Content-Type": "application/json"}, ) response.raise_for_status() data = response.json() except httpx.HTTPStatusError as e: return ToolResult(title="搜索失败", output="", error=f"HTTP {e.response.status_code}: {e.response.text}") except Exception as e: return ToolResult(title="搜索失败", output="", error=str(e)) posts = data.get("data", []) # 构建概览摘要 summary_list = [] for idx, post in enumerate(posts, 1): body = post.get("body_text", "") or "" title = post.get("title") or body[:20] or "" summary_list.append({ "index": idx, "title": title, "body_text": body[:100] + ("..." if len(body) > 100 else ""), "like_count": post.get("like_count"), "comment_count": post.get("comment_count"), "channel": post.get("channel"), "link": post.get("link"), "content_type": post.get("content_type"), }) # 封面拼图 images = [] try: collage_b64 = await _build_collage(posts) if collage_b64: images.append({"type": "base64", "media_type": "image/png", "data": collage_b64}) except Exception: pass return ToolResult( title=f"搜索: {keyword} ({platform_id})", output=json.dumps({"data": summary_list}, ensure_ascii=False, indent=2), long_term_memory=f"Searched '{keyword}' on {platform_id}, {len(posts)} results. Use content_detail to view full details.", images=images, metadata={"posts": posts}, # 完整数据传给上层缓存 ) # ── 详情实现（从缓存获取，不需要额外 HTTP） ── async def detail(post: Dict[str, Any], extras: Optional[Dict[str, Any]] = None) -> ToolResult: """返回单条帖子的完整内容""" title = post.get("title") or post.get("body_text", "")[:30] or "无标题" images = [] for img_url in post.get("images", []): if img_url: images.append({"type": "url", "url": img_url}) return ToolResult( title=f"详情: {title}", output=json.dumps(post, ensure_ascii=False, indent=2), long_term_memory=f"Viewed detail: {title}", images=images, ) # ── 建议词实现 ── async def suggest(channel: str, keyword: str) -> ToolResult: """获取搜索建议词""" try: async with httpx.AsyncClient(timeout=DEFAULT_TIMEOUT) as client: response = await client.post( f"{BASE_URL}/suggest", json={"type": channel, "keyword": keyword}, headers={"Content-Type": "application/json"}, ) response.raise_for_status() data = response.json() except Exception as e: return ToolResult(title="建议词获取失败", output="", error=str(e)) suggestion_count = sum(len(item.get("list", [])) for item in data.get("data", [])) return ToolResult( title=f"建议词: {keyword} ({channel})", output=json.dumps(data, ensure_ascii=False, indent=2), long_term_memory=f"Got {suggestion_count} suggestions for '{keyword}' on {channel}", ) # ── 拼图辅助 ── async def _build_collage(posts: List[Dict[str, Any]]) -> Optional[str]: """封面图网格拼图""" urls, titles = [], [] for post in posts: imgs = post.get("images", []) if imgs and imgs[0]: urls.append(imgs[0]) titles.append(post.get("title", "") or "") 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 # ── 注册所有 AIGC 平台 ── def _register_all(): for p in _AIGC_PLATFORMS: p.search_impl = search p.detail_impl = detail if p.supports_suggest: p.suggest_impl = suggest p.suggest_channels = [p.id] register_platform(p) # wx 只有 suggest，没有搜索 # suggest 调用时 channel 传 "wx"，但不注册为独立平台 _register_all()