get_trending_topics

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"
              "|------|------|--------|--------|--------|--------|----------|"
          )