generate_summary_report

Instructions

Implementation Reference

• mcp_server/tools/analytics.py:1158-1336 (handler)

  The actual implementation of the tool 'generate_summary_report' is within the 'AnalyticsTools' class in 'mcp_server/tools/analytics.py'.

      def generate_summary_report(
          self,
          report_type: str = "daily",
          date_range: Optional[Dict[str, str]] = None
      ) -> Dict:
          """
          每日/每周摘要生成器 - 自动生成热点摘要报告
  
          Args:
              report_type: 报告类型（daily/weekly）
              date_range: 自定义日期范围（可选）
  
          Returns:
              Markdown格式的摘要报告
  
          Examples:
              用户询问示例：
              - "生成今天的新闻摘要报告"
              - "给我一份本周的热点总结"
              - "生成过去7天的新闻分析报告"
  
              代码调用示例：
              >>> tools = AnalyticsTools()
              >>> result = tools.generate_summary_report(
              ...     report_type="daily"
              ... )
              >>> print(result['markdown_report'])
          """
          try:
              # 参数验证
              if report_type not in ["daily", "weekly"]:
                  raise InvalidParameterError(
                      f"无效的报告类型: {report_type}",
                      suggestion="支持的类型: daily, weekly"
                  )
  
              # 确定日期范围
              if date_range:
                  date_range_tuple = validate_date_range(date_range)
                  start_date, end_date = date_range_tuple
              else:
                  if report_type == "daily":
                      start_date = end_date = datetime.now()
                  else:  # weekly
                      end_date = datetime.now()
                      start_date = end_date - timedelta(days=6)
  
              # 收集数据
              all_keywords = Counter()
              all_platforms_news = defaultdict(int)
              all_titles_list = []
  
              current_date = start_date
              while current_date <= end_date:
                  try:
                      all_titles, id_to_name, _ = self.data_service.parser.read_all_titles_for_date(
                          date=current_date
                      )
  
                      for platform_id, titles in all_titles.items():
                          platform_name = id_to_name.get(platform_id, platform_id)
                          all_platforms_news[platform_name] += len(titles)
  
                          for title in titles.keys():
                              all_titles_list.append({
                                  "title": title,
                                  "platform": platform_name,
                                  "date": current_date.strftime("%Y-%m-%d")
                              })
  
                              # 提取关键词
                              keywords = self._extract_keywords(title)
                              all_keywords.update(keywords)
  
                  except DataNotFoundError:
                      pass
  
                  current_date += timedelta(days=1)
  
              # 生成报告
              report_title = f"{'每日' if report_type == 'daily' else '每周'}新闻热点摘要"
              date_str = f"{start_date.strftime('%Y-%m-%d')}" if report_type == "daily" else f"{start_date.strftime('%Y-%m-%d')} 至 {end_date.strftime('%Y-%m-%d')}"
  
              # 构建Markdown报告
              markdown = f"""# {report_title}
  
  **报告日期**: {date_str}
  **生成时间**: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
  
  ---
  
  ## 📊 数据概览
  
  - **总新闻数**: {len(all_titles_list)}
  - **覆盖平台**: {len(all_platforms_news)}
  - **热门关键词数**: {len(all_keywords)}
  
  ## 🔥 TOP 10 热门话题
  
  """
  
              # 添加TOP 10关键词
              for i, (keyword, count) in enumerate(all_keywords.most_common(10), 1):
                  markdown += f"{i}. **{keyword}** - 出现 {count} 次\n"
  
              # 平台分析
              markdown += "\n## 📱 平台活跃度\n\n"
              sorted_platforms = sorted(all_platforms_news.items(), key=lambda x: x[1], reverse=True)
  
              for platform, count in sorted_platforms:
                  markdown += f"- **{platform}**: {count} 条新闻\n"
  
              # 趋势变化（如果是周报）
              if report_type == "weekly":
                  markdown += "\n## 📈 趋势分析\n\n"
                  markdown += "本周热度持续的话题（样本数据）：\n\n"
  
                  # 简单的趋势分析
                  top_keywords = [kw for kw, _ in all_keywords.most_common(5)]
                  for keyword in top_keywords:
                      markdown += f"- **{keyword}**: 持续热门\n"
  
              # 添加样本新闻（按权重选择，确保确定性）
              markdown += "\n## 📰 精选新闻样本\n\n"
  
              # 确定性选取：按标题的权重排序，取前5条
              # 这样相同输入总是返回相同结果
              if all_titles_list:
                  # 计算每条新闻的权重分数（基于关键词出现次数）
                  news_with_scores = []
                  for news in all_titles_list:
                      # 简单权重：统计包含TOP关键词的次数
                      score = 0
                      title_lower = news['title'].lower()
                      for keyword, count in all_keywords.most_common(10):
                          if keyword.lower() in title_lower:
                              score += count
                      news_with_scores.append((news, score))
  
                  # 按权重降序排序，权重相同则按标题字母顺序（确保确定性）
                  news_with_scores.sort(key=lambda x: (-x[1], x[0]['title']))
  
                  # 取前5条
                  sample_news = [item[0] for item in news_with_scores[:5]]
  
                  for news in sample_news:
                      markdown += f"- [{news['platform']}] {news['title']}\n"
  
              markdown += "\n---\n\n*本报告由 TrendRadar MCP 自动生成*\n"
  
              return {
                  "success": True,
                  "report_type": report_type,
                  "date_range": {
                      "start": start_date.strftime("%Y-%m-%d"),
                      "end": end_date.strftime("%Y-%m-%d")
                  },
                  "markdown_report": markdown,
                  "statistics": {
                      "total_news": len(all_titles_list),
                      "platforms_count": len(all_platforms_news),
                      "keywords_count": len(all_keywords),
                      "top_keyword": all_keywords.most_common(1)[0] if all_keywords else None
                  }
              }
  
          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)
                  }
              }