sentry-issues-query

name	sentry-issues-query
description	查询Sentry Issues并支持关键词过滤和忽略词排除。当用户需要从Sentry获取error issues列表、按关键词过滤、排除特定错误、或获取issues的统计数据时使用。

Sentry Issues 查询技能

从 Sentry 获取 issues 列表，支持关键词过滤、忽略词排除，并补充统计数据。

输入参数

参数	类型	必填	默认值	说明
sentry_token	string	是	-	Sentry API Token
sentry_org	string	否	"test"	Sentry 组织名称
project_ids	list[int]	是	-	Sentry 项目ID列表
keywords	list[str]	否	None	关键词列表（不传则全量捞取）
ignore_words	list[str]	否	None	忽略词列表（不传则不过滤）
stats_period	string	否	"24h"	时间周期，支持 24h, 30d, 90d 等

输出格式

返回标准化的 issues 列表，每个 issue 包含：

{
  "id": "issue_id",
  "title": "issue标题",
  "culprit": "culprit信息",
  "status": "unresolved/resolved/ignored",
  "platform": "平台",
  "permalink": "Sentry链接",
  "count_24h": 10,
  "count_30d": 100,
  "first_seen": "2024-01-01 10:00:00",
  "last_seen": "2024-01-01 12:00:00",
  "hit_keywords": "命中的关键词"  // 仅当keywords参数传入时返回
}

执行流程

1. 构建 Sentry API 请求

使用 Sentry Issues API：

GET https://sentry.io/api/0/organizations/{org}/issues/

查询参数：

• project: 项目ID
• statsPeriod: 时间周期（默认24h）
• query: 构建查询字符串

2. 构建查询字符串

• 基础查询：is:unresolved event.type:error
• 如果传入 keywords：message:"keyword1" OR message:"keyword2"
• 拼接完整查询条件

3. 获取 Issues 列表

调用 Sentry API 获取 issues，使用分页处理确保获取完整数据。

4. 过滤和补充数据

1. 忽略词过滤：如果传入 ignore_words，对 title/culprit 进行大小写不敏感匹配
2. 获取统计数据：调用 issues-stats API 获取每个 issue 的：
   • 24h count
   • 30d count
   • firstSeen
   • lastSeen
3. 关键词命中标记：如果传入 keywords，检查并记录每个 issue 命中的关键词

5. 返回结果

返回过滤后的 issues 列表，包含原始数据和补充的统计数据。

代码实现

import requests
from datetime import datetime
from typing import List, Optional, Dict, Any

def query_sentry_issues(
    sentry_token: str,
    sentry_org: str = "zhihu",
    project_ids: List[int] = None,
    keywords: List[str] = None,
    ignore_words: List[str] = None,
    stats_period: str = "24h"
) -> List[Dict[str, Any]]:
    """
    查询 Sentry Issues 并返回标准化数据
    
    Args:
        sentry_token: Sentry API Token
        sentry_org: Sentry 组织名称，默认 "zhihu"
        project_ids: Sentry 项目ID列表
        keywords: 关键词列表，用于过滤和标记命中
        ignore_words: 忽略词列表，命中则过滤掉
        stats_period: 时间周期，默认 "24h"
    
    Returns:
        issues 列表，每个 issue 包含基本信息和统计数据
    """
    if not project_ids:
        raise ValueError("project_ids is required")
    
    headers = {
        "Authorization": f"Bearer {sentry_token}",
        "Content-Type": "application/json"
    }
    
    base_url = f"https://sentry.io/api/0/organizations/{sentry_org}"
    
    # 构建查询字符串
    query_parts = ["is:unresolved", "event.type:error"]
    
    if keywords:
        keyword_queries = [f'message:"{kw}"' for kw in keywords]
        query_parts.append(f"({' OR '.join(keyword_queries)})")
    
    query = " ".join(query_parts)
    
    all_issues = []
    
    # 遍历每个项目获取 issues
    for project_id in project_ids:
        params = {
            "project": project_id,
            "statsPeriod": stats_period,
            "query": query,
            "limit": 100,
            "shortIdLookup": 0
        }
        
        # 处理分页
        cursor = None
        while True:
            if cursor:
                params["cursor"] = cursor
            
            response = requests.get(
                f"{base_url}/issues/",
                headers=headers,
                params=params,
                timeout=30
            )
            
            if response.status_code != 200:
                print(f"API请求失败: {response.status_code} - {response.text}")
                break
            
            issues = response.json()
            if not issues:
                break
            
            all_issues.extend(issues)
            
            # 检查是否有下一页
            # Sentry API 使用 cursor 进行分页
            if isinstance(issues, list) and len(issues) < 100:
                break
            cursor = response.headers.get("X-Sentry-Driven-Cursor")
            if not cursor:
                break
    
    # 过滤忽略词
    if ignore_words:
        filtered_issues = []
        for issue in all_issues:
            title = issue.get("title", "").lower()
            culprit = issue.get("culprit", "").lower()
            should_ignore = False
            for word in ignore_words:
                if word.lower() in title or word.lower() in culprit:
                    should_ignore = True
                    break
            if not should_ignore:
                filtered_issues.append(issue)
        all_issues = filtered_issues
    
    # 去重
    seen_ids = set()
    unique_issues = []
    for issue in all_issues:
        if issue["id"] not in seen_ids:
            seen_ids.add(issue["id"])
            unique_issues.append(issue)
    all_issues = unique_issues
    
    # 获取统计数据
    issue_ids = [issue["id"] for issue in all_issues]
    stats = _get_issues_stats(issue_ids, project_ids[0], headers, sentry_org)
    
    # 构建返回结果
    result = []
    for issue in all_issues:
        issue_id = issue["id"]
        stat = stats.get(issue_id, {})
        
        # 检查关键词命中
        hit_keywords = ""
        if keywords:
            title = issue.get("title", "")
            culprit = issue.get("culprit", "")
            hits = []
            for kw in keywords:
                if kw.lower() in title.lower() or kw.lower() in culprit.lower():
                    hits.append(kw)
            hit_keywords = ", ".join(hits)
        
        item = {
            "id": issue_id,
            "title": issue.get("title", ""),
            "culprit": issue.get("culprit", ""),
            "status": issue.get("status", "unknown"),
            "platform": issue.get("platform", ""),
            "permalink": issue.get("permalink", ""),
            "count_24h": stat.get("count_24h", 0),
            "count_30d": stat.get("count_30d", 0),
            "first_seen": stat.get("first_seen", ""),
            "last_seen": stat.get("last_seen", ""),
        }
        
        if keywords:
            item["hit_keywords"] = hit_keywords
        
        result.append(item)
    
    return result


def _get_issues_stats(
    issue_ids: List[str], 
    project_id: int, 
    headers: Dict[str, str],
    sentry_org: str
) -> Dict[str, Dict[str, Any]]:
    """获取 issues 统计数据"""
    if not issue_ids:
        return {}
    
    stats = {}
    chunk_size = 25
    
    for i in range(0, len(issue_ids), chunk_size):
        chunk = issue_ids[i:i + chunk_size]
        params = {
            "groups": chunk,
            "project": project_id
        }
        
        response = requests.get(
            f"https://sentry.io/api/0/organizations/{sentry_org}/issues-stats/",
            headers=headers,
            params=params,
            timeout=30
        )
        
        if response.status_code != 200:
            continue
        
        data = response.json()
        if isinstance(data, dict) and "detail" in data:
            continue
        
        for item in data:
            uid = str(item.get("id"))
            lifetime = item.get("lifetime", {})
            # 24h 统计从 countHistoricalShorthand 字段获取
            # 30d 统计使用 lifetime 数据
            stats[uid] = {
                "count_24h": item.get("count", 0),
                "count_30d": lifetime.get("count", 0),
                "first_seen": _format_datetime(lifetime.get("firstSeen")),
                "last_seen": _format_datetime(lifetime.get("lastSeen"))
            }
    
    return stats


def _format_datetime(dt_str: Optional[str]) -> str:
    """格式化 datetime 字符串"""
    if not dt_str:
        return ""
    try:
        # Sentry 返回 ISO 格式: 2024-01-01T10:00:00Z
        dt_str = dt_str.replace("Z", "+00:00")
        dt = datetime.fromisoformat(dt_str)
        return dt.strftime("%Y-%m-%d %H:%M:%S")
    except Exception:
        return dt_str

使用示例

示例1：查询单个项目的全量 issues

issues = query_sentry_issues(
    sentry_token="your-sentry-token",
    project_ids=[123]
)

示例2：按关键词过滤并标记命中

issues = query_sentry_issues(
    sentry_token="your-sentry-token",
    project_ids=[123],
    keywords=["NullPointerException", "Timeout"]
)

示例3：排除特定错误并限定时间周期

issues = query_sentry_issues(
    sentry_token="your-sentry-token",
    project_ids=[123, 456],
    ignore_words=["test error", "debug log"],
    stats_period="30d"
)

注意事项

1. Token 权限：确保 Sentry Token 具有 project:read 和 event:read 权限
2. API 限制：Sentry API 有速率限制，大批量查询时注意添加延迟
3. 分页处理：自动处理 Sentry 的 cursor 分页，确保获取完整数据
4. 大小写不敏感：关键词和忽略词匹配均采用大小写不敏感方式