get_news_by_date

Instructions

Implementation Reference

• mcp_server/tools/data_query.py:211-283 (handler)

  Tool handler for get_news_by_date in DataQueryTools class.

  def get_news_by_date(
      self,
      date_query: Optional[str] = None,
      platforms: Optional[List[str]] = None,
      limit: Optional[int] = None,
      include_url: bool = False
  ) -> Dict:
      """
      按日期查询新闻，支持自然语言日期
  
      Args:
          date_query: 日期查询字符串（可选，默认"今天"），支持：
              - 相对日期：今天、昨天、前天、3天前、yesterday、3 days ago
              - 星期：上周一、本周三、last monday、this friday
              - 绝对日期：2025-10-10、10月10日、2025年10月10日
          platforms: 平台ID列表，如 ['zhihu', 'weibo']
          limit: 返回条数限制，默认50
          include_url: 是否包含URL链接，默认False（节省token）
  
      Returns:
          新闻列表字典
  
      Example:
          >>> tools = DataQueryTools()
          >>> # 不指定日期，默认查询今天
          >>> result = tools.get_news_by_date(platforms=['zhihu'], limit=20)
          >>> # 指定日期
          >>> result = tools.get_news_by_date(
          ...     date_query="昨天",
          ...     platforms=['zhihu'],
          ...     limit=20
          ... )
          >>> print(result['total'])
          20
      """
      try:
          # 参数验证 - 默认今天
          if date_query is None:
              date_query = "今天"
          target_date = validate_date_query(date_query)
          platforms = validate_platforms(platforms)
          limit = validate_limit(limit, default=50)
  
          # 获取数据
          news_list = self.data_service.get_news_by_date(
              target_date=target_date,
              platforms=platforms,
              limit=limit,
              include_url=include_url
          )
  
          return {
              "news": news_list,
              "total": len(news_list),
              "date": target_date.strftime("%Y-%m-%d"),
              "date_query": date_query,
              "platforms": platforms,
              "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)
              }
          }

• mcp_server/services/data_service.py:104-182 (handler)

  Implementation of get_news_by_date in DataService class which performs the actual data retrieval.

  def get_news_by_date(
      self,
      target_date: datetime,
      platforms: Optional[List[str]] = None,
      limit: int = 50,
      include_url: bool = False
  ) -> List[Dict]:
      """
      按指定日期获取新闻
  
      Args:
          target_date: 目标日期
          platforms: 平台ID列表,None表示所有平台
          limit: 返回条数限制
          include_url: 是否包含URL链接,默认False(节省token)
  
      Returns:
          新闻列表
  
      Raises:
          DataNotFoundError: 数据不存在
  
      Examples:
          >>> service = DataService()
          >>> news = service.get_news_by_date(
          ...     target_date=datetime(2025, 10, 10),
          ...     platforms=['zhihu'],
          ...     limit=20
          ... )
      """
      # 尝试从缓存获取
      date_str = target_date.strftime("%Y-%m-%d")
      cache_key = f"news_by_date:{date_str}:{','.join(platforms or [])}:{limit}:{include_url}"
      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(
          date=target_date,
          platform_ids=platforms
      )
  
      # 转换为新闻列表
      news_list = []
      for platform_id, titles in all_titles.items():
          platform_name = id_to_name.get(platform_id, platform_id)
  
          for title, info in titles.items():
              # 计算平均排名
              avg_rank = sum(info["ranks"]) / len(info["ranks"]) if info["ranks"] else 0
  
              news_item = {
                  "title": title,
                  "platform": platform_id,
                  "platform_name": platform_name,
                  "rank": info["ranks"][0] if info["ranks"] else 0,
                  "avg_rank": round(avg_rank, 2),
                  "count": len(info["ranks"]),
                  "date": date_str
              }
  
              # 条件性添加 URL 字段
              if include_url:
                  news_item["url"] = info.get("url", "")
                  news_item["mobileUrl"] = info.get("mobileUrl", "")
  
              news_list.append(news_item)
  
      # 按排名排序
      news_list.sort(key=lambda x: x["rank"])
  
      # 限制返回数量
      result = news_list[:limit]
  
      # 缓存结果(历史数据缓存更久)
      self.cache.set(cache_key, result)
  
      return result