129 lines
4.5 KiB
Python
129 lines
4.5 KiB
Python
"""任务三TAPD API - 整合Story和Bug查询"""
|
||
import os
|
||
import requests
|
||
from typing import List, Dict
|
||
from requests.auth import HTTPBasicAuth
|
||
from datetime import datetime
|
||
|
||
|
||
class TAPDUnifiedApi:
|
||
"""统一的TAPD API，整合Story和Bug查询"""
|
||
|
||
BASE_URL = "https://tapd-api.bilibili.co/tapd"
|
||
|
||
# 需求终态
|
||
STORY_TERMINAL_STATUSES = ['status_8', 'status_12'] # 已完成、取消
|
||
# 缺陷终态
|
||
BUG_TERMINAL_STATUSES = ['rejected', 'closed'] # 取消、验证通过
|
||
|
||
def __init__(self, workspace_id: str, logger):
|
||
self.workspace_id = workspace_id
|
||
self.logger = logger
|
||
|
||
self.api_user = os.environ.get('TAPD_API_USER')
|
||
self.api_password = os.environ.get('TAPD_API_PASSWORD')
|
||
|
||
if not self.api_user or not self.api_password:
|
||
raise ValueError("TAPD认证信息未设置")
|
||
|
||
self.auth = HTTPBasicAuth(self.api_user, self.api_password)
|
||
self.session = requests.Session()
|
||
|
||
def _make_request(self, endpoint: str, params: Dict) -> Dict:
|
||
"""发起TAPD API请求"""
|
||
url = f"{self.BASE_URL}/{endpoint}"
|
||
|
||
try:
|
||
response = self.session.get(url, params=params, auth=self.auth, timeout=30)
|
||
response.raise_for_status()
|
||
result = response.json()
|
||
|
||
if result.get('status') != 1:
|
||
raise RuntimeError(f"TAPD API返回错误: {result}")
|
||
|
||
self.logger.log_api_call("tapd", endpoint, params, result, True)
|
||
return result
|
||
except Exception as e:
|
||
self.logger.log_api_call("tapd", endpoint, params, {}, False, str(e))
|
||
raise
|
||
|
||
def get_overdue_stories(self, owner_list: List[str]) -> List[dict]:
|
||
"""获取过期需求"""
|
||
today = datetime.now().strftime('%Y-%m-%d')
|
||
all_stories = []
|
||
|
||
for owner in owner_list:
|
||
params = {
|
||
'workspace_id': self.workspace_id,
|
||
'owner': owner,
|
||
'fields': 'id,name,due,status'
|
||
}
|
||
|
||
try:
|
||
result = self._make_request("stories", params)
|
||
data = result.get('data', [])
|
||
|
||
for item in data:
|
||
story = item.get('Story', {})
|
||
due_date = story.get('due', '')
|
||
status = story.get('status', '')
|
||
|
||
# 过滤：有截止日期、已过期、非终态
|
||
if due_date and due_date < today and status not in self.STORY_TERMINAL_STATUSES:
|
||
all_stories.append({
|
||
'id': story.get('id'),
|
||
'name': story.get('name'),
|
||
'owner': owner,
|
||
'due': due_date,
|
||
'status': status,
|
||
'type': 'story'
|
||
})
|
||
except Exception as e:
|
||
pass
|
||
|
||
return all_stories
|
||
|
||
def get_overdue_bugs(self, owner_list: List[str]) -> List[dict]:
|
||
"""获取过期缺陷"""
|
||
today = datetime.now().strftime('%Y-%m-%d')
|
||
all_bugs = []
|
||
|
||
for owner in owner_list:
|
||
params = {
|
||
'workspace_id': self.workspace_id,
|
||
'current_owner': owner,
|
||
'fields': 'id,title,deadline,status'
|
||
}
|
||
|
||
try:
|
||
result = self._make_request("bugs", params)
|
||
data = result.get('data', [])
|
||
|
||
for item in data:
|
||
bug = item.get('Bug', {})
|
||
deadline = bug.get('deadline', '')
|
||
status = bug.get('status', '')
|
||
|
||
# 过滤：有截止日期、已过期、非终态
|
||
if deadline and deadline < today and status not in self.BUG_TERMINAL_STATUSES:
|
||
all_bugs.append({
|
||
'id': bug.get('id'),
|
||
'title': bug.get('title'),
|
||
'owner': owner,
|
||
'due': deadline,
|
||
'status': status,
|
||
'type': 'bug'
|
||
})
|
||
except Exception as e:
|
||
pass
|
||
|
||
return all_bugs
|
||
|
||
def get_story_url(self, story_id: str) -> str:
|
||
"""生成需求URL"""
|
||
return f"https://www.tapd.cn/{self.workspace_id}/prong/stories/view/{story_id}"
|
||
|
||
def get_bug_url(self, bug_id: str) -> str:
|
||
"""生成缺陷URL"""
|
||
return f"https://www.tapd.cn/{self.workspace_id}/bugtrace/bugs/view?bug_id={bug_id}"
|