96 lines
2.6 KiB
Python
96 lines
2.6 KiB
Python
|
"""
|
||
|
|
Jira API 客户端
|
||
|
|
|
||
|
|
用于查询 Jira issue 数量。
|
||
|
|
"""
|
||
|
|
|
||
|
|
import requests
|
||
|
|
import json
|
||
|
|
from typing import Optional, Dict, List, Any
|
||
|
|
import os
|
||
|
|
|
||
|
|
|
||
|
|
class JiraClient:
|
||
|
|
"""Jira API 客户端"""
|
||
|
|
|
||
|
|
def __init__(self, base_url: str, username: str, token: str):
|
||
|
|
"""
|
||
|
|
初始化 Jira 客户端
|
||
|
|
|
||
|
|
Args:
|
||
|
|
base_url: Jira 基础 URL (如 https://jira.example.com)
|
||
|
|
username: Jira 用户名
|
||
|
|
token: Jira API Token
|
||
|
|
"""
|
||
|
|
self.base_url = base_url.rstrip("/")
|
||
|
|
self.username = username
|
||
|
|
self.token = token
|
||
|
|
self.auth = (username, token)
|
||
|
|
|
||
|
|
def search_issues(self, jql: str, max_results: int = 100) -> List[Dict[str, Any]]:
|
||
|
|
"""
|
||
|
|
搜索 Jira issues
|
||
|
|
|
||
|
|
Args:
|
||
|
|
jql: JQL 查询语句
|
||
|
|
max_results: 最大返回数量
|
||
|
|
|
||
|
|
Returns:
|
||
|
|
Issue 列表
|
||
|
|
"""
|
||
|
|
url = f"{self.base_url}/rest/api/2/search"
|
||
|
|
headers = {"Content-Type": "application/json", "Accept": "application/json"}
|
||
|
|
params = {"jql": jql, "maxResults": max_results, "fields": "key"}
|
||
|
|
|
||
|
|
try:
|
||
|
|
response = requests.get(
|
||
|
|
url, headers=headers, auth=self.auth, params=params, timeout=30
|
||
|
|
)
|
||
|
|
response.raise_for_status()
|
||
|
|
data = response.json()
|
||
|
|
return data.get("issues", [])
|
||
|
|
except requests.exceptions.RequestException as e:
|
||
|
|
print(f"Jira 查询失败: {e}")
|
||
|
|
return []
|
||
|
|
|
||
|
|
def count_issues(self, jql: str) -> int:
|
||
|
|
"""
|
||
|
|
统计满足条件的 issue 数量
|
||
|
|
|
||
|
|
Args:
|
||
|
|
jql: JQL 查询语句
|
||
|
|
|
||
|
|
Returns:
|
||
|
|
Issue 数量
|
||
|
|
"""
|
||
|
|
url = f"{self.base_url}/rest/api/2/search"
|
||
|
|
headers = {"Content-Type": "application/json", "Accept": "application/json"}
|
||
|
|
params = {
|
||
|
|
"jql": jql,
|
||
|
|
"maxResults": 0, # 不返回实际数据，只返回总数
|
||
|
|
}
|
||
|
|
|
||
|
|
try:
|
||
|
|
response = requests.get(
|
||
|
|
url, headers=headers, auth=self.auth, params=params, timeout=30
|
||
|
|
)
|
||
|
|
response.raise_for_status()
|
||
|
|
data = response.json()
|
||
|
|
return data.get("total", 0)
|
||
|
|
except requests.exceptions.RequestException as e:
|
||
|
|
print(f"Jira 查询失败: {e}")
|
||
|
|
return 0
|
||
|
|
|
||
|
|
def get_issue_keys(self, jql: str) -> List[str]:
|
||
|
|
"""
|
||
|
|
获取满足条件的 issue keys
|
||
|
|
|
||
|
|
Args:
|
||
|
|
jql: JQL 查询语句
|
||
|
|
|
||
|
|
Returns:
|
||
|
|
Issue key 列表 (如 ["FZ-2042", "FZ-2043"])
|
||
|
|
"""
|
||
|
|
issues = self.search_issues(jql)
|
||
|
|
return [issue.get("key") for issue in issues if issue.get("key")]
|