Viral Shorts

Instructions

Implementation Reference

• src/server.py:375-455 (handler)

  The main handler function that implements get_youtube_shorts_trends. It orchestrates fetching trending videos from YouTube, analyzing them with ViralAnalyzer, filtering by minimum views, ranking, and generating a Markdown report with statistics.
  async def get_youtube_shorts_trends( keyword: str = "", hours_ago: int = 24, max_results: int = 10, min_views: int = 100000 ) -> list[TextContent]: """ 工具 1 的实现:获取 YouTube Shorts 热门趋势 这里是真正的业务逻辑: 1. 调用 YouTube API 搜索视频 2. 使用爆款分析算法计算指标 3. 筛选和排序 4. 生成 Markdown 格式的报告 """ try: # 初始化客户端和分析器 client = YouTubeClient() analyzer = ViralAnalyzer() # 获取视频数据 videos = client.get_trending_shorts( keyword=keyword, hours_ago=hours_ago, max_results=max_results ) if not videos: return [TextContent( type="text", text=f"未找到符合条件的视频。\n\n搜索参数:\n- 关键词: {keyword or '全局'}\n- 时间范围: 最近 {hours_ago} 小时" )] # 分析视频(计算互动率) videos = analyzer.analyze_videos(videos) # 筛选和排序 videos = analyzer.filter_by_views(videos, min_views=min_views) videos = analyzer.rank_videos(videos) if not videos: return [TextContent( type="text", text=f"未找到满足条件的视频(播放量 >= {min_views:,})。\n\n建议:\n- 降低筛选阈值\n- 扩大时间范围\n- 尝试其他关键词" )] # 生成 Markdown 报告 report = f"# 🔥 YouTube Shorts 热门视频报告\n\n" report += f"**搜索参数**\n" report += f"- 关键词: {keyword or '全局趋势'}\n" report += f"- 时间范围: 最近 {hours_ago} 小时\n" if min_views > 0: report += f"- 播放量阈值: {min_views:,}\n" report += f"- 找到视频: {len(videos)} 个\n\n" report += "---\n\n" report += VideoData.markdown_header() + "\n" for video in videos: report += video.to_markdown_row() + "\n" # 添加统计摘要 report += f"\n---\n\n## 📊 数据摘要\n\n" avg_engagement = sum(v.engagement_rate for v in videos) / len(videos) avg_views = sum(v.views for v in videos) / len(videos) top_video = videos[0] report += f"- **平均播放量**: {avg_views:,.0f}\n" report += f"- **平均互动率**: {avg_engagement:.2f}%\n" report += f"- **最高播放量**: {top_video.views:,} ([{top_video.title[:30]}...]({top_video.url}))\n" # 返回 TextContent 对象 return [TextContent(type="text", text=report)] except Exception as e: return [TextContent( type="text", text=f"❌ 获取趋势失败: {str(e)}" )]
• src/server.py:144-180 (schema)

  The Tool schema definition that declares the tool name, description, and inputSchema with JSON Schema validation for parameters: keyword (string), hours_ago (integer, 1-720), max_results (integer, 1-50), and min_views (integer, minimum 0).
  Tool( name="get_youtube_shorts_trends", description=( "发现 YouTube Shorts 上的潜在热门视频。" "通过分析播放量、互动率和发布时间,识别高质量的短视频。" ), inputSchema={ "type": "object", "properties": { "keyword": { "type": "string", "description": "搜索关键词(留空表示全局趋势)", "default": "" }, "hours_ago": { "type": "integer", "description": "时间范围(小时),默认 24 小时,最大 30 天", "default": 24, "minimum": 1, "maximum": 720 }, "max_results": { "type": "integer", "description": "返回结果数量,默认 10", "default": 10, "minimum": 1, "maximum": 50 }, "min_views": { "type": "integer", "description": "最低播放量阈值,默认 100,000", "default": 100000, "minimum": 0 } } } ),
• src/server.py:302-368 (registration)

  The call_tool registration handler decorated with @app.call_tool() that routes tool invocation requests. Lines 325-331 show the routing logic for get_youtube_shorts_trends, extracting arguments and calling the handler function.
  @app.call_tool() async def call_tool(name: str, arguments: dict) -> list[TextContent]: """ 当 Claude 调用工具时,这个函数会被执行 参数: name: 工具名称(来自 list_tools 中定义的 name) arguments: 工具参数(JSON 对象,符合 inputSchema) 返回值: TextContent 列表,包含要返回给 Claude 的文本内容 工作流程: 1. Claude 解析用户的自然语言请求 2. 选择合适的工具和参数 3. 调用这个函数 4. 我们执行业务逻辑(调用 YouTube API、分析数据) 5. 返回格式化的结果 6. Claude 将结果展示给用户 """ try: # 根据工具名称路由到对应的处理函数 if name == "get_youtube_shorts_trends": return await get_youtube_shorts_trends( keyword=arguments.get("keyword", ""), hours_ago=arguments.get("hours_ago", 24), max_results=arguments.get("max_results", 10), min_views=arguments.get("min_views", 100000) ) elif name == "analyze_video_potential": return await analyze_video_potential( video_url=arguments["video_url"] ) elif name == "get_trending_topics": return await get_trending_topics( category=arguments.get("category", "all"), hours_ago=arguments.get("hours_ago", 24) ) elif name == "summarize_video_story": return await summarize_video_story( video_url=arguments["video_url"] ) elif name == "discover_niche_trends": return await discover_niche_trends( main_topic=arguments["main_topic"], hours_ago=arguments.get("hours_ago", 24), min_videos=arguments.get("min_videos", 3), top_niches=arguments.get("top_niches", 10) ) else: return [TextContent( type="text", text=f"未知工具: {name}" )] except Exception as e: # 错误处理:返回友好的错误信息 return [TextContent( type="text", text=f"❌ 执行失败: {str(e)}" )]
• src/youtube/client.py:182-220 (helper)

  Helper method YouTubeClient.get_trending_shorts() that orchestrates the complete YouTube API workflow: searching for shorts by keyword, fetching video details, and optionally retrieving channel information. This is called by the main handler.
  def get_trending_shorts( self, keyword: str = "", hours_ago: int = 24, max_results: int = 10, include_channel_info: bool = True ) -> List[VideoData]: """ 获取热门 Shorts(完整流程) Args: keyword: 搜索关键词 hours_ago: 时间范围 max_results: 最大结果数 include_channel_info: 是否包含频道信息 Returns: VideoData 列表 """ # 1. 搜索视频 video_ids = self.search_shorts(keyword, hours_ago, max_results) if not video_ids: return [] # 2. 获取视频详情 videos = self.get_video_details(video_ids) # 3. 获取频道信息(可选) if include_channel_info: channel_ids = [v.channel_id for v in videos] channel_map = self.get_channel_info(channel_ids) # 填充频道订阅数 for video in videos: if video.channel_id in channel_map: video.channel_subscribers = channel_map[video.channel_id].subscribers return videos
• src/youtube/analyzer.py:58-102 (helper)

  Helper methods in ViralAnalyzer used by the handler: analyze_videos() for calculating engagement rates, filter_by_views() for applying minimum view thresholds, and rank_videos() for sorting by view count (descending).
  @classmethod def analyze_videos(cls, videos: List[VideoData]) -> List[VideoData]: """ 批量分析视频 Args: videos: 视频列表 Returns: 分析后的视频列表 """ return [cls.analyze_video(video) for video in videos] @staticmethod def rank_videos(videos: List[VideoData]) -> List[VideoData]: """ 按播放量排序视频 Args: videos: 视频列表 Returns: 排序后的视频列表(降序) """ return sorted(videos, key=lambda v: v.views, reverse=True) @staticmethod def filter_by_views( videos: List[VideoData], min_views: int = 0 ) -> List[VideoData]: """ 按播放量阈值筛选视频 Args: videos: 视频列表 min_views: 最低播放量阈值 (默认 0,不筛选) Returns: 筛选后的视频列表 """ if min_views == 0: return videos return [v for v in videos if v.views >= min_views]