get_trending_topics

Instructions

Implementation Reference

• mcp_server/services/data_service.py:285-401 (handler)

  The core implementation of get_trending_topics logic in DataService.

  def get_trending_topics(
      self,
      top_n: int = 10,
      mode: str = "current"
  ) -> Dict:
      """
      获取个人关注词的新闻出现频率统计
  
      注意:本工具基于 config/frequency_words.txt 中的个人关注词列表进行统计,
      而不是自动从新闻中提取热点话题。用户可以自定义这个关注词列表。
  
      Args:
          top_n: 返回TOP N关注词
          mode: 模式 - daily(当日累计), current(最新一批)
  
      Returns:
          关注词频率统计字典
  
      Raises:
          DataNotFoundError: 数据不存在
      """
      # 尝试从缓存获取
      cache_key = f"trending_topics:{top_n}:{mode}"
      cached = self.cache.get(cache_key, ttl=1800)  # 30分钟缓存
      if cached:
          return cached
  
      # 读取今天的数据
      all_titles, id_to_name, timestamps = self.parser.read_all_titles_for_date()
  
      if not all_titles:
          raise DataNotFoundError(
              "未找到今天的新闻数据",
              suggestion="请确保爬虫已经运行并生成了数据"
          )
  
      # 加载关键词配置
      word_groups = self.parser.parse_frequency_words()
  
      # 根据mode选择要处理的标题数据
      titles_to_process = {}
  
      if mode == "daily":
          # daily模式:处理当天所有累计数据
          titles_to_process = all_titles
  
      elif mode == "current":
          # current模式:只处理最新一批数据(最新时间戳的文件)
          if timestamps:
              # 找出最新的时间戳
              latest_timestamp = max(timestamps.values())
  
              # 重新读取,只获取最新时间的数据
              # 这里我们通过timestamps字典反查找最新文件对应的平台
              latest_titles, _, _ = self.parser.read_all_titles_for_date()
  
              # 由于read_all_titles_for_date返回所有文件的合并数据,
              # 我们需要通过timestamps来过滤出最新批次
              # 简化实现:使用当前所有数据作为最新批次
              # (更精确的实现需要解析服务支持按时间过滤)
              titles_to_process = latest_titles
          else:
              titles_to_process = all_titles
  
      else:
          raise ValueError(
              f"不支持的模式: {mode}。支持的模式: daily, current"
          )
  
      # 统计词频
      word_frequency = Counter()
      keyword_to_news = {}
  
      # 遍历要处理的标题
      for platform_id, titles in titles_to_process.items():
          for title in titles.keys():
              # 对每个关键词组进行匹配
              for group in word_groups:
                  all_words = group.get("required", []) + group.get("normal", [])
  
                  for word in all_words:
                      if word and word in title:
                          word_frequency[word] += 1
  
                          if word not in keyword_to_news:
                              keyword_to_news[word] = []
                          keyword_to_news[word].append(title)
  
      # 获取TOP N关键词
      top_keywords = word_frequency.most_common(top_n)
  
      # 构建话题列表
      topics = []
      for keyword, frequency in top_keywords:
          matched_news = keyword_to_news.get(keyword, [])
  
          topics.append({
              "keyword": keyword,
              "frequency": frequency,
              "matched_news": len(set(matched_news)),  # 去重后的新闻数量
              "trend": "stable",  # TODO: 需要历史数据来计算趋势
              "weight_score": 0.0  # TODO: 需要实现权重计算
          })
  
      # 构建结果
      result = {
          "topics": topics,
          "generated_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
          "mode": mode,
          "total_keywords": len(word_frequency),
          "description": self._get_mode_description(mode)
      }
  
      # 缓存结果
      self.cache.set(cache_key, result)
  
      return result

• mcp_server/tools/data_query.py:154-209 (handler)

  The tool handler in DataQueryTools that validates parameters and calls the underlying service.

  def get_trending_topics(
      self,
      top_n: Optional[int] = None,
      mode: Optional[str] = None
  ) -> Dict:
      """
      获取个人关注词的新闻出现频率统计
  
      注意：本工具基于 config/frequency_words.txt 中的个人关注词列表进行统计，
      而不是自动从新闻中提取热点话题。这是一个个人可定制的关注词列表，
      用户可以根据自己的兴趣添加或删除关注词。
  
      Args:
          top_n: 返回TOP N关注词，默认10
          mode: 模式 - daily(当日累计), current(最新一批), incremental(增量)
  
      Returns:
          关注词频率统计字典，包含每个关注词在新闻中出现的次数
  
      Example:
          >>> tools = DataQueryTools()
          >>> result = tools.get_trending_topics(top_n=5, mode="current")
          >>> print(len(result['topics']))
          5
          >>> # 返回的是你在 frequency_words.txt 中设置的关注词的频率统计
      """
      try:
          # 参数验证
          top_n = validate_top_n(top_n, default=10)
          valid_modes = ["daily", "current", "incremental"]
          mode = validate_mode(mode, valid_modes, default="current")
  
          # 获取趋势话题
          trending_result = self.data_service.get_trending_topics(
              top_n=top_n,
              mode=mode
          )
  
          return {
              **trending_result,
              "success": True
          }
  
      except MCPError as e:
          return {
              "success": False,
              "error": e.to_dict()
          }
      except Exception as e:
          return {
              "success": False,
              "error": {
                  "code": "INTERNAL_ERROR",
                  "message": str(e)
              }
          }