analyze_sentiment

Instructions

用户："分析特斯拉最近7天的新闻情感"
推荐调用流程：
1. resolve_date_range("最近7天") → {"date_range": {"start": "2025-11-20", "end": "2025-11-26"}}
2. analyze_sentiment(topic="特斯拉", date_range={"start": "2025-11-20", "end": "2025-11-26"})

Implementation Reference

• mcp_server/tools/analytics.py:631-816 (handler)

  The actual implementation of the 'analyze_sentiment' tool, which gathers news data, performs deduplication, sorts by weight, and generates an AI prompt for sentiment analysis.

  def analyze_sentiment(
      self,
      topic: Optional[str] = None,
      platforms: Optional[List[str]] = None,
      date_range: Optional[Dict[str, str]] = None,
      limit: int = 50,
      sort_by_weight: bool = True,
      include_url: bool = False
  ) -> Dict:
      """
      情感倾向分析 - 生成用于 AI 情感分析的结构化提示词
  
      本工具收集新闻数据并生成优化的 AI 提示词，你可以将其发送给 AI 进行深度情感分析。
  
      Args:
          topic: 话题关键词（可选），只分析包含该关键词的新闻
          platforms: 平台过滤列表（可选），如 ['zhihu', 'weibo']
          date_range: 日期范围（可选），格式: {"start": "YYYY-MM-DD", "end": "YYYY-MM-DD"}
                     不指定则默认查询今天的数据
          limit: 返回新闻数量限制，默认50，最大100
          sort_by_weight: 是否按权重排序，默认True（推荐）
          include_url: 是否包含URL链接，默认False（节省token）
  
      Returns:
          包含 AI 提示词和新闻数据的结构化结果
  
      Examples:
          用户询问示例：
          - "分析一下今天新闻的情感倾向"
          - "看看'特斯拉'相关新闻是正面还是负面的"
          - "分析各平台对'人工智能'的情感态度"
          - "看看'特斯拉'相关新闻是正面还是负面的，请选择一周内的前10条新闻来分析"
  
          代码调用示例：
          >>> tools = AnalyticsTools()
          >>> # 分析今天的特斯拉新闻，返回前10条
          >>> result = tools.analyze_sentiment(
          ...     topic="特斯拉",
          ...     limit=10
          ... )
          >>> # 分析一周内的特斯拉新闻（假设今天是 2025-11-17）
          >>> result = tools.analyze_sentiment(
          ...     topic="特斯拉",
          ...     date_range={"start": "2025-11-11", "end": "2025-11-17"},
          ...     limit=10
          ... )
          >>> print(result['ai_prompt'])  # 获取生成的提示词
      """
      try:
          # 参数验证
          if topic:
              topic = validate_keyword(topic)
          platforms = validate_platforms(platforms)
          limit = validate_limit(limit, default=50)
  
          # 处理日期范围
          if date_range:
              date_range_tuple = validate_date_range(date_range)
              start_date, end_date = date_range_tuple
          else:
              # 默认今天
              start_date = end_date = datetime.now()
  
          # 收集新闻数据（支持多天）
          all_news_items = []
          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,
                      platform_ids=platforms
                  )
  
                  # 收集该日期的新闻
                  for platform_id, titles in all_titles.items():
                      platform_name = id_to_name.get(platform_id, platform_id)
                      for title, info in titles.items():
                          # 如果指定了话题，只收集包含话题的标题
                          if topic and topic.lower() not in title.lower():
                              continue
  
                          news_item = {
                              "platform": platform_name,
                              "title": title,
                              "ranks": info.get("ranks", []),
                              "count": len(info.get("ranks", [])),
                              "date": current_date.strftime("%Y-%m-%d")
                          }
  
                          # 条件性添加 URL 字段
                          if include_url:
                              news_item["url"] = info.get("url", "")
                              news_item["mobileUrl"] = info.get("mobileUrl", "")
  
                          all_news_items.append(news_item)
  
              except DataNotFoundError:
                  # 该日期没有数据，继续下一天
                  pass
  
              # 下一天
              current_date += timedelta(days=1)
  
          if not all_news_items:
              time_desc = "今天" if start_date == end_date else f"{start_date.strftime('%Y-%m-%d')} 至 {end_date.strftime('%Y-%m-%d')}"
              raise DataNotFoundError(
                  f"未找到相关新闻（{time_desc}）",
                  suggestion="请尝试其他话题、日期范围或平台"
              )
  
          # 去重（同一标题只保留一次）
          unique_news = {}
          for item in all_news_items:
              key = f"{item['platform']}::{item['title']}"
              if key not in unique_news:
                  unique_news[key] = item
              else:
                  # 合并 ranks（如果同一新闻在多天出现）
                  existing = unique_news[key]
                  existing["ranks"].extend(item["ranks"])
                  existing["count"] = len(existing["ranks"])
  
          deduplicated_news = list(unique_news.values())
  
          # 按权重排序（如果启用）
          if sort_by_weight:
              deduplicated_news.sort(
                  key=lambda x: calculate_news_weight(x),
                  reverse=True
              )
  
          # 限制返回数量
          selected_news = deduplicated_news[:limit]
  
          # 生成 AI 提示词
          ai_prompt = self._create_sentiment_analysis_prompt(
              news_data=selected_news,
              topic=topic
          )
  
          # 构建时间范围描述
          if start_date == end_date:
              time_range_desc = start_date.strftime("%Y-%m-%d")
          else:
              time_range_desc = f"{start_date.strftime('%Y-%m-%d')} 至 {end_date.strftime('%Y-%m-%d')}"
  
          result = {
              "success": True,
              "method": "ai_prompt_generation",
              "summary": {
                  "total_found": len(deduplicated_news),
                  "returned_count": len(selected_news),
                  "requested_limit": limit,
                  "duplicates_removed": len(all_news_items) - len(deduplicated_news),
                  "topic": topic,
                  "time_range": time_range_desc,
                  "platforms": list(set(item["platform"] for item in selected_news)),
                  "sorted_by_weight": sort_by_weight
              },
              "ai_prompt": ai_prompt,
              "news_sample": selected_news,
              "usage_note": "请将 ai_prompt 字段的内容发送给 AI 进行情感分析"
          }
  
          # 如果返回数量少于请求数量，增加提示
          if len(selected_news) < limit and len(deduplicated_news) >= limit:
              result["note"] = "返回数量少于请求数量是因为去重逻辑（同一标题在不同平台只保留一次）"
          elif len(deduplicated_news) < limit:
              result["note"] = f"在指定时间范围内仅找到 {len(deduplicated_news)} 条匹配的新闻"
  
          return result
  
      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)
              }
          }