Viral Shorts

Instructions

Implementation Reference

• src/server.py:504-572 (handler)

  The main handler function that implements the get_trending_topics tool. It searches for videos by category keywords, analyzes them with ViralAnalyzer, removes duplicates, ranks them, and generates a markdown report with the top 10 trending videos.
  async def get_trending_topics( category: str = "all", hours_ago: int = 24 ) -> list[TextContent]: """工具 3 的实现:获取热门话题""" # 定义分类关键词 category_keywords = { "tech": ["AI", "coding", "tech", "programming", "software"], "entertainment": ["funny", "comedy", "entertainment", "viral"], "education": ["tutorial", "learn", "education", "how to"], "gaming": ["gaming", "gameplay", "game", "esports"], "all": [""] # 空关键词表示全局搜索 } keywords = category_keywords.get(category, [""]) try: client = YouTubeClient() analyzer = ViralAnalyzer() all_videos = [] # 对每个关键词搜索 for keyword in keywords: videos = client.get_trending_shorts( keyword=keyword, hours_ago=hours_ago, max_results=5 # 每个关键词取 5 个 ) all_videos.extend(videos) if not all_videos: return [TextContent( type="text", text=f"未找到 {category} 分类的热门视频" )] # 分析并排序 all_videos = analyzer.analyze_videos(all_videos) all_videos = analyzer.rank_videos(all_videos) # 去重(按视频 ID) seen = set() unique_videos = [] for v in all_videos: if v.video_id not in seen: seen.add(v.video_id) unique_videos.append(v) # 取前 10 top_videos = unique_videos[:10] # 生成报告 report = f"# 🎯 {category.upper()} 分类热门话题\n\n" report += f"**时间范围**: 最近 {hours_ago} 小时\n\n" report += "---\n\n" report += VideoData.markdown_header() + "\n" for video in top_videos: report += video.to_markdown_row() + "\n" return [TextContent(type="text", text=report)] except Exception as e: return [TextContent( type="text", text=f"❌ 获取热门话题失败: {str(e)}" )]
• src/server.py:206-230 (registration)

  Tool registration in the list_tools() function that defines the schema for get_trending_topics, including its name, description, and input parameters (category with enum values, and hours_ago with constraints).
  Tool( name="get_trending_topics", description=( "发现当前 YouTube Shorts 上的热门话题和趋势。" "分析多个分类的热门内容,帮助选题。" ), inputSchema={ "type": "object", "properties": { "category": { "type": "string", "description": "内容分类", "enum": ["tech", "entertainment", "education", "gaming", "all"], "default": "all" }, "hours_ago": { "type": "integer", "description": "时间范围(小时),最大 30 天", "default": 24, "minimum": 1, "maximum": 720 } } } ),
• src/server.py:338-342 (registration)

  The routing logic in the call_tool() handler that maps the tool name 'get_trending_topics' to its handler function, extracting and passing the arguments.
  elif name == "get_trending_topics": return await get_trending_topics( category=arguments.get("category", "all"), hours_ago=arguments.get("hours_ago", 24) )
• src/models/video.py:7-55 (helper)

  The VideoData model used by get_trending_topics handler for formatting output. Provides to_markdown_row() and markdown_header() methods to generate the report table.
  class VideoData(BaseModel): """YouTube Shorts 视频数据模型""" video_id: str = Field(..., description="视频 ID") title: str = Field(..., description="视频标题") channel_name: str = Field(..., description="频道名称") channel_id: str = Field(..., description="频道 ID") channel_subscribers: Optional[int] = Field(None, description="频道订阅数") views: int = Field(..., description="播放量") likes: int = Field(default=0, description="点赞数") comments: int = Field(default=0, description="评论数") published_at: datetime = Field(..., description="发布时间") duration: str = Field(..., description="视频时长") url: str = Field(..., description="视频链接") thumbnail_url: Optional[str] = Field(None, description="缩略图链接") description: Optional[str] = Field(None, description="视频描述") engagement_rate: float = Field(default=0.0, description="互动率 (%)") class Config: """Pydantic 配置""" json_encoders = { datetime: lambda v: v.isoformat() } def to_markdown_row(self) -> str: """转换为 Markdown 表格行""" published_time = self.published_at.strftime('%Y-%m-%d %H:%M') return ( f"| [{self.title[:40]}...]({self.url}) " f"| {self.channel_name[:20]} " f"| {self.views:,} " f"| {self.likes:,} " f"| {self.comments:,} " f"| {self.engagement_rate:.2f}% " f"| {published_time} |" ) @staticmethod def markdown_header() -> str: """Markdown 表格头""" return ( "| 标题 | 频道 | 播放量 | 点赞数 | 评论数 | 互动率 | 发布时间 |\n" "|------|------|--------|--------|--------|--------|----------|" )